24662130060

Committed 20 Apr 2026 10:42AM UTC coverage: 60.236% (+0.04%) from 60.194%

Build # 24662130060

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #74 from knowledgepixels/fix/metrics-off-event-loop-and-http-pool

perf: changes 5+6 — metrics off event loop, raise HTTP pool, consolidate clients

Coverage Stats

293 of 540 branches covered (54.26%)

Branch coverage included in aggregate %.

828 of 1321 relevant lines covered (62.68%)

9.14 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

61.54

src/main/java/com/knowledgepixels/query/TripleStore.java

package com.knowledgepixels.query;

import org.apache.commons.exec.environment.EnvironmentUtils;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpUriRequest;
import org.apache.http.client.methods.RequestBuilder;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.BasicResponseHandler;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.eclipse.rdf4j.common.transaction.IsolationLevels;
import org.eclipse.rdf4j.model.ValueFactory;
import org.eclipse.rdf4j.model.impl.SimpleValueFactory;
import org.eclipse.rdf4j.repository.Repository;
import org.eclipse.rdf4j.repository.RepositoryConnection;
import org.eclipse.rdf4j.repository.base.RepositoryConnectionWrapper;
import org.eclipse.rdf4j.repository.http.HTTPRepository;
import org.nanopub.NanopubUtils;
import org.nanopub.vocabulary.NPA;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * Class to access the database in the form of triple stores.
 */
public class TripleStore {

    /**
     * Name of the admin graph.
     */
    public static final String ADMIN_REPO = "admin";

    private static ValueFactory vf = SimpleValueFactory.getInstance();

    private static final Logger log = LoggerFactory.getLogger(TripleStore.class);

    private final Map<String, Repository> repositories = new LinkedHashMap<>();

    /**
     * Per-repo open-connection counter, read by the eviction loop to skip repos that
     * still have live connections. Incremented in {@link #getRepoConnection(String)}
     * just before the connection is handed out and decremented via a
     * {@link RepositoryConnectionWrapper} that intercepts {@code close()} exactly once.
     */
    private final ConcurrentHashMap<String, AtomicInteger> openConnections = new ConcurrentHashMap<>();

    private String endpointBase = null;
    private String endpointType = null;

    private static TripleStore tripleStoreInstance;

    /**
     * Returns singleton triple store instance.
     *
     * @return Triple store instance
     */
    public static TripleStore get() {
        if (tripleStoreInstance == null) {
            try {
                tripleStoreInstance = new TripleStore();
            } catch (IOException ex) {
                log.info("Could not init TripleStore. ", ex);
            }
        }
        return tripleStoreInstance;
    }

    private TripleStore() throws IOException {
        Map<String, String> env = EnvironmentUtils.getProcEnvironment();
        endpointBase = env.get("ENDPOINT_BASE");
        log.info("Endpoint base: {}", endpointBase);
        endpointType = env.get("ENDPOINT_TYPE");

        getRepository("empty");  // Make sure empty repo exists
    }

    /**
     * Shared HTTP client for all RDF4J traffic. Apache HttpClient treats all requests
     * to a given host + port + protocol as one "route", so every `HTTPRepository` in
     * this process funnels through this single client's connection pool — plus the
     * two ad-hoc users in {@link #createRepo} and {@link #getRepositoryNames}, which
     * have been consolidated onto this client too.
     *
     * <p>The Apache defaults (maxPerRoute=2 / maxTotal=20) throttle four concurrent
     * loader-pool threads down to two-way parallelism at the HTTP layer — invisible
     * client-side serialisation the code is actively fighting. Raised to 10 / 40
     * here: comfortable headroom for the 4-thread loader pool plus admin-repo
     * transactions and the metrics tick, small enough to be a conservative first
     * step with room to grow later.
     */
    private final CloseableHttpClient httpclient = HttpClients.custom()
            .setMaxConnPerRoute(10)
            .setMaxConnTotal(40)
            .build();

    @GeneratedFlagForDependentElements
    Repository getRepository(String name) {
        synchronized (this) {
            if (repositories.size() > 100) {
                evictIdleRepos();
            }
            if (repositories.containsKey(name)) {
                // Move to the end of the list:
                Repository repo = repositories.remove(name);
                repositories.put(name, repo);
            } else {
                Repository repository = null;
                if (endpointType == null || endpointType.equals("rdf4j")) {
                    HTTPRepository hr = new HTTPRepository(endpointBase + "repositories/" + name);
                    hr.setHttpClient(httpclient);
                    repository = hr;
//                        } else if (endpointType.equals("virtuoso")) {
//                                repository = new VirtuosoRepository(endpointBase + name, username, password);
                } else {
                    throw new RuntimeException("Unknown repository type: " + endpointType);
                }
                repositories.put(name, repository);
                createRepo(name);
                getRepoConnection(name).close();
            }
            return repositories.get(name);
        }
    }

    /**
     * Return the repository connection for the given repository name.
     *
     * @param name repository name
     * @return repository connection
     */
    @GeneratedFlagForDependentElements
    public RepositoryConnection getRepoConnection(String name) {
        // The increment has to happen under the same monitor that guards eviction,
        // otherwise another thread could evict this repo in the window between
        // getRepository() returning and the counter going above zero. getConnection()
        // on HTTPRepository is local (it doesn't do HTTP), so holding the lock here
        // is cheap.
        synchronized (this) {
            Repository repo = getRepository(name);
            if (repo == null) {
                return null;
            }
            AtomicInteger counter = openConnections.computeIfAbsent(name, k -> new AtomicInteger());
            counter.incrementAndGet();
            return new CountingRepositoryConnection(repo, repo.getConnection(), counter);
        }
    }

    /**
     * Evicts the eldest cache entries until either the size is back within the
     * 100-entry cap or every remaining entry has at least one open connection.
     * The cap is load-bearing — each cached entry keeps an LMDB environment alive
     * on the RDF4J server (in-memory cache, mmap pages, native memory), so
     * exceeding it by much risks server-side OOM. Actively-used repos are skipped
     * rather than shut down, because {@link org.eclipse.rdf4j.repository.http.HTTPRepository#shutDown()}
     * closes the session manager and kills any live transaction on that repo with
     * a connection-close error (the {@code MMapIndexInput – Already closed}
     * failure mode observed on query-3 in the April test). The cache self-converges
     * as active repos become idle.
     */
    @GeneratedFlagForDependentElements
    private void evictIdleRepos() {
        List<String> skipped = new ArrayList<>();
        Iterator<Entry<String, Repository>> iter = repositories.entrySet().iterator();
        while (iter.hasNext() && repositories.size() > 100) {
            Entry<String, Repository> e = iter.next();
            AtomicInteger active = openConnections.get(e.getKey());
            if (active != null && active.get() > 0) {
                skipped.add(e.getKey());
                continue;
            }
            iter.remove();
            log.info("Shutting down repo: {}", e.getKey());
            e.getValue().shutDown();
            log.info("Shutdown complete");
        }
        if (!skipped.isEmpty()) {
            log.warn("Skipped eviction for {} active repo(s); cache size is now {} (cap 100). Active names: {}",
                    skipped.size(), repositories.size(), skipped);
        }
    }

    /**
     * Minimal wrapper around a delegate {@link RepositoryConnection} whose only job
     * is to decrement the per-repo open-connection counter exactly once when the
     * caller closes it. Uses {@link AtomicBoolean} so that repeated/idempotent
     * {@code close()} calls (common with try-with-resources plus explicit close)
     * don't decrement more than once.
     */
    private static final class CountingRepositoryConnection extends RepositoryConnectionWrapper {

        private final AtomicInteger counter;
        private final AtomicBoolean closed = new AtomicBoolean();

        CountingRepositoryConnection(Repository repo, RepositoryConnection delegate, AtomicInteger counter) {
            super(repo, delegate);
            this.counter = counter;
        }

        @Override
        public void close() {
            try {
                super.close();
            } finally {
                if (closed.compareAndSet(false, true)) {
                    counter.decrementAndGet();
                }
            }
        }

    }

    @GeneratedFlagForDependentElements
    private void createRepo(String repoName) {
        if (!repoName.equals(ADMIN_REPO)) {
            getRepository(ADMIN_REPO);  // make sure admin repo is loaded first
        }
        // Uses the shared this.httpclient rather than a per-call client so it inherits
        // the configured pool sizes (and, once change 1 of the fix plan lands, the
        // socket/connection-request timeouts).
        try {
            //log.info("Trying to creating repo " + name);

            // TODO new syntax somehow doesn't work for the Lucene case:

//                        String createRegularRepoQueryString = "@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#>.\n"
//                                        + "@prefix config: <tag:rdf4j.org,2023:config/>.\n"
//                                        + "[] a config:Repository ;\n"
//                                        + "    config:rep.id \"" + name + "\" ;\n"
//                                        + "    rdfs:label \"" + name + " native store\" ;\n"
//                                        + "    config:rep.impl [\n"
//                                        + "        config:rep.type \"openrdf:SailRepository\" ;\n"
//                                        + "        config:sail.impl [\n"
//                                        + "            config:sail.type \"openrdf:NativeStore\" ;\n"
//                                        + "            config:sail.iterationCacheSyncThreshold \"10000\";\n"
//                                        + "            config:native.tripleIndexes \"spoc,posc,ospc,opsc,psoc,sopc,spoc,cpos,cosp,cops,cpso,csop\" ;\n"
//                                        + "            config:sail.defaultQueryEvaluationMode \"STANDARD\"\n"
//                                        + "        ]\n"
//                                        + "    ].";
//                        String createTextRepoQueryString = "@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#>.\n"
//                                        + "@prefix config: <tag:rdf4j.org,2023:config/>.\n"
//                                        + "[] a config:Repository ;\n"
//                                        + "    config:rep.id \"" + name + "\" ;\n"
//                                        + "    rdfs:label \"" + name + " native store\" ;\n"
//                                        + "    config:rep.impl [\n"
//                                        + "        config:rep.type \"openrdf:SailRepository\" ;\n"
//                                        + "        config:sail.impl [\n"
//                                        + "            config:sail.type \"openrdf:LuceneSail\" ;\n"
//                                        + "            config:sail.lucene.indexDir \"index/\" ;\n"
//                                        + "            config:delegate [\n"
//                                        + "                config:rep.type \"openrdf:SailRepository\" ;\n"
//                                        + "                config:sail.impl [\n"
//                                        + "                    config:sail.type \"openrdf:NativeStore\" ;\n"
//                                        + "                    config:sail.iterationCacheSyncThreshold \"10000\";\n"
//                                        + "                    config:native.tripleIndexes \"spoc,posc,ospc,opsc,psoc,sopc,spoc,cpos,cosp,cops,cpso,csop\" ;\n"
//                                        + "                    config:sail.defaultQueryEvaluationMode \"STANDARD\"\n"
//                                        + "                ]\n"
//                                        + "            ]\n"
//                                        + "        ]\n"
//                                        + "    ].";

            String indexTypes = "spoc,posc,ospc,cspo,cpos,cosp";
            if (repoName.startsWith("meta") || repoName.startsWith("text")) {
                indexTypes = "spoc,posc,ospc";
            }

            String createRegularRepoQueryString =
                    "@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#>.\n" +
                    "@prefix rep: <http://www.openrdf.org/config/repository#>.\n" +
                    "@prefix sr: <http://www.openrdf.org/config/repository/sail#>.\n" +
                    "@prefix sail: <http://www.openrdf.org/config/sail#>.\n" +
                    "@prefix sail-luc: <http://www.openrdf.org/config/sail/lucene#>.\n" +
                    "@prefix lmdb: <http://rdf4j.org/config/sail/lmdb#>.\n" +
                    "@prefix sb: <http://www.openrdf.org/config/sail/base#>.\n" +
                    "\n" +
                    "[] a rep:Repository ;\n" +
                    "    rep:repositoryID \"" + repoName + "\" ;\n" +
                    "    rdfs:label \"" + repoName + " LMDB store\" ;\n" +
                    "    rep:repositoryImpl [\n" +
                    "        rep:repositoryType \"openrdf:SailRepository\" ;\n" +
                    "        sr:sailImpl [\n" +
                    "            sail:sailType \"rdf4j:LmdbStore\" ;\n" +
                    "            sail:iterationCacheSyncThreshold \"10000\";\n" +
                    "            lmdb:tripleIndexes \"" + indexTypes + "\" ;\n" +
                    "            sb:defaultQueryEvaluationMode \"STANDARD\"\n" +
                    "        ]\n"
                    + "    ].\n";

            // TODO Index npa:hasFilterLiteral predicate too (see https://groups.google.com/g/rdf4j-users/c/epF4Af1jXGU):
            String createTextRepoQueryString =
                    "@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#>.\n" +
                    "@prefix rep: <http://www.openrdf.org/config/repository#>.\n" +
                    "@prefix sr: <http://www.openrdf.org/config/repository/sail#>.\n" +
                    "@prefix sail: <http://www.openrdf.org/config/sail#>.\n" +
                    "@prefix sail-luc: <http://www.openrdf.org/config/sail/lucene#>.\n" +
                    "@prefix lmdb: <http://rdf4j.org/config/sail/lmdb#>.\n" +
                    "@prefix sb: <http://www.openrdf.org/config/sail/base#>.\n" +
                    "\n"
                    + "[] a rep:Repository ;\n" +
                    "    rep:repositoryID \"" + repoName + "\" ;\n" +
                    "    rdfs:label \"" + repoName + " store\" ;\n" +
                    "    rep:repositoryImpl [\n" +
                    "        rep:repositoryType \"openrdf:SailRepository\" ;\n" +
                    "        sr:sailImpl [\n" +
                    "            sail:sailType \"openrdf:LuceneSail\" ;\n" +
                    "            sail-luc:indexDir \"index/\" ;\n" +
                    "            sail-luc:transactional false ;\n" +
                    "            sail:delegate [\n" +
                    "              sail:sailType \"rdf4j:LmdbStore\" ;\n" +
                    "              sail:iterationCacheSyncThreshold \"10000\";\n" +
                    "              lmdb:tripleIndexes \"" + indexTypes + "\" ;\n" +
                    "              sb:defaultQueryEvaluationMode \"STANDARD\"\n" +
                    "            ]\n" +
                    "        ]\n" +
                    "    ].";

            String createRepoQueryString = createRegularRepoQueryString;
            if (repoName.startsWith("text")) {
                createRepoQueryString = createTextRepoQueryString;
            }

            HttpUriRequest createRepoRequest = RequestBuilder.put().setUri(endpointBase + "repositories/" + repoName).addHeader("Content-Type", "text/turtle").setEntity(new StringEntity(createRepoQueryString)).build();

            // Response is try-with-resources'd so the connection is released back to
            // the shared pool rather than leaked.
            try (CloseableHttpResponse response = httpclient.execute(createRepoRequest)) {
                int statusCode = response.getStatusLine().getStatusCode();
                if (statusCode == 409) {
                    //log.info("Already exists.");
                    getRepository(repoName).init();
                } else if (statusCode >= 200 && statusCode < 300) {
                    //log.info("Successfully created.");
                    initNewRepo(repoName);
                } else {
                    log.info("Status code: {}", response.getStatusLine().getStatusCode());
                    log.info(response.getStatusLine().getReasonPhrase());
                    String handledResponse = new BasicResponseHandler().handleResponse(response);
                    log.info("Response: {}", handledResponse);
                }
            }
        } catch (IOException ex) {
            log.info("Could not create repo.", ex);
        }
    }

    /**
     * Sends shutdown signal to all repositories.
     */
    @GeneratedFlagForDependentElements
    public void shutdownRepositories() {
        for (Repository repo : repositories.values()) {
            if (repo != null && repo.isInitialized()) {
                repo.shutDown();
            }
        }
    }

    /**
     * Returns admin repo connection.
     *
     * @return repository connection to admin repository
     */
    @GeneratedFlagForDependentElements
    public RepositoryConnection getAdminRepoConnection() {
        return get().getRepoConnection(ADMIN_REPO);
    }

    private Set<String> cachedRepositoryNames = Set.of();
    private boolean repoNamesCacheValid = false;
    private final ReadWriteLock repoNamesCacheLock = new ReentrantReadWriteLock();

    /**
     * Returns set of all repository names.
     *
     * @return Repository name set
     */
    public Set<String> getRepositoryNames() {
        // See if the repository names are cached:
        final var readLock = repoNamesCacheLock.readLock();
        try {
            readLock.lock();
            if (repoNamesCacheValid) {
                return cachedRepositoryNames;
            }
        } finally {
            readLock.unlock();
        }

        // Not cached, get from server:
        final var writeLock = repoNamesCacheLock.writeLock();
        try {
            writeLock.lock();
            // Check again if another thread has already updated the cache:
            if (repoNamesCacheValid) {
                return cachedRepositoryNames;
            }
            Map<String, Boolean> repositoryNames = null;
            // Uses the shared this.httpclient; response try-with-resources releases
            // the pooled connection when done.
            try (CloseableHttpResponse resp = httpclient.execute(RequestBuilder.get()
                    .setUri(endpointBase + "/repositories")
                    .addHeader("Content-Type", "text/csv")
                    .build())) {
                BufferedReader reader = new BufferedReader(new InputStreamReader(resp.getEntity().getContent()));
                int code = resp.getStatusLine().getStatusCode();
                if (code < 200 || code >= 300) return null;
                repositoryNames = new HashMap<>();
                int lineCount = 0;
                while (true) {
                    String line = reader.readLine();
                    if (line == null) break;
                    if (lineCount > 0) {
                        String repoName = line.split(",")[1];
                        repositoryNames.put(repoName, true);
                    }
                    lineCount = lineCount + 1;
                }
            } catch (IOException ex) {
                log.info("Could not get repository names.", ex);
                return null;
            }
            cachedRepositoryNames = repositoryNames.keySet();
            repoNamesCacheValid = true;
            return cachedRepositoryNames;
        } finally {
            writeLock.unlock();
        }
    }

    /**
     * Invalidates the repository names cache. Call this method when a repository is created or deleted.
     */
    private void invalidateRepositoryNamesCache() {
        final var writeLock = repoNamesCacheLock.writeLock();
        try {
            writeLock.lock();
            repoNamesCacheValid = false;
        } finally {
            writeLock.unlock();
        }
    }

    @GeneratedFlagForDependentElements
    private void initNewRepo(String repoName) {
        String repoInitId = new Random().nextLong() + "";
        getRepository(repoName).init();
        if (!repoName.equals("empty")) {
            RepositoryConnection conn = getRepoConnection(repoName);
            try (conn) {
                // Full isolation, just in case.
                conn.begin(IsolationLevels.SERIALIZABLE);
                conn.add(NPA.THIS_REPO, NPA.HAS_REPO_INIT_ID, vf.createLiteral(repoInitId), NPA.GRAPH);
                if (tracksNanopubCountAndChecksum(repoName)) {
                    conn.add(NPA.THIS_REPO, NPA.HAS_NANOPUB_COUNT, vf.createLiteral(0L), NPA.GRAPH);
                    conn.add(NPA.THIS_REPO, NPA.HAS_NANOPUB_CHECKSUM, vf.createLiteral(NanopubUtils.INIT_CHECKSUM), NPA.GRAPH);
                }
                if (repoName.startsWith("pubkey_") || repoName.startsWith("type_")) {
                    String h = repoName.replaceFirst("^[^_]+_", "");
                    conn.add(NPA.THIS_REPO, NPA.HAS_COVERAGE_ITEM, Utils.getObjectForHash(h), NPA.GRAPH);
                    conn.add(NPA.THIS_REPO, NPA.HAS_COVERAGE_HASH, vf.createLiteral(h), NPA.GRAPH);
                    conn.add(NPA.THIS_REPO, NPA.HAS_COVERAGE_FILTER, vf.createLiteral("_" + repoName), NPA.GRAPH);
                }
                conn.commit();
            }
            // Refresh repository names cache
            invalidateRepositoryNamesCache();
        }
    }

    /**
     * Whether the given repo participates in the cumulative nanopub-count / XOR-checksum
     * tracking. Repos that don't hold raw nanopubs skip the
     * {@code npa:hasNanopubCount} and {@code npa:hasNanopubChecksum} initial triples —
     * leaving them at {@code 0} and the empty-XOR placeholder forever would just be
     * misleading. Currently excluded:
     * <ul>
     *   <li>{@code admin} — holds metadata only.</li>
     *   <li>{@code last30d} — content expires on a periodic cleanup.</li>
     *   <li>{@code trust} and {@code spaces} — hold derived state, not raw nanopubs.</li>
     * </ul>
     */
    private static boolean tracksNanopubCountAndChecksum(String repoName) {
        return !repoName.equals(ADMIN_REPO)
                && !repoName.equals("last30d")
                && !repoName.equals("trust")
                && !repoName.equals("spaces");
    }

}

1	package com.knowledgepixels.query;
2
3	import org.apache.commons.exec.environment.EnvironmentUtils;
4	import org.apache.http.client.methods.CloseableHttpResponse;
5	import org.apache.http.client.methods.HttpUriRequest;
6	import org.apache.http.client.methods.RequestBuilder;
7	import org.apache.http.entity.StringEntity;
8	import org.apache.http.impl.client.BasicResponseHandler;
9	import org.apache.http.impl.client.CloseableHttpClient;
10	import org.apache.http.impl.client.HttpClients;
11	import org.eclipse.rdf4j.common.transaction.IsolationLevels;
12	import org.eclipse.rdf4j.model.ValueFactory;
13	import org.eclipse.rdf4j.model.impl.SimpleValueFactory;
14	import org.eclipse.rdf4j.repository.Repository;
15	import org.eclipse.rdf4j.repository.RepositoryConnection;
16	import org.eclipse.rdf4j.repository.base.RepositoryConnectionWrapper;
17	import org.eclipse.rdf4j.repository.http.HTTPRepository;
18	import org.nanopub.NanopubUtils;
19	import org.nanopub.vocabulary.NPA;
20	import org.slf4j.Logger;
21	import org.slf4j.LoggerFactory;
22
23	import java.io.BufferedReader;
24	import java.io.IOException;
25	import java.io.InputStreamReader;
26	import java.util.*;
27	import java.util.Map.Entry;
28	import java.util.concurrent.ConcurrentHashMap;
29	import java.util.concurrent.atomic.AtomicBoolean;
30	import java.util.concurrent.atomic.AtomicInteger;
31	import java.util.concurrent.locks.ReadWriteLock;
32	import java.util.concurrent.locks.ReentrantReadWriteLock;
33
34	/**
35	* Class to access the database in the form of triple stores.
36	*/
37	public class TripleStore {
38
39	/**
40	* Name of the admin graph.
41	*/
42	public static final String ADMIN_REPO = "admin";
43
44	private static ValueFactory vf = SimpleValueFactory.getInstance();	6✔
45
46	private static final Logger log = LoggerFactory.getLogger(TripleStore.class);	12✔
47
48	private final Map<String, Repository> repositories = new LinkedHashMap<>();	15✔
49
50	/**
51	* Per-repo open-connection counter, read by the eviction loop to skip repos that
52	* still have live connections. Incremented in {@link #getRepoConnection(String)}
53	* just before the connection is handed out and decremented via a
54	* {@link RepositoryConnectionWrapper} that intercepts {@code close()} exactly once.
55	*/
56	private final ConcurrentHashMap<String, AtomicInteger> openConnections = new ConcurrentHashMap<>();	15✔
57
58	private String endpointBase = null;	9✔
59	private String endpointType = null;	9✔
60
61	private static TripleStore tripleStoreInstance;
62
63	/**
64	* Returns singleton triple store instance.
65	*
66	* @return Triple store instance
67	*/
68	public static TripleStore get() {
69	if (tripleStoreInstance == null) {	6!
70	try {
71	tripleStoreInstance = new TripleStore();	×
72	} catch (IOException ex) {	×
73	log.info("Could not init TripleStore. ", ex);	×
74	}	×
75	}
76	return tripleStoreInstance;	×
77	}
78
79	private TripleStore() throws IOException {	6✔
80	Map<String, String> env = EnvironmentUtils.getProcEnvironment();	6✔
81	endpointBase = env.get("ENDPOINT_BASE");	18✔
82	log.info("Endpoint base: {}", endpointBase);	15✔
83	endpointType = env.get("ENDPOINT_TYPE");	18✔
84
85	getRepository("empty"); // Make sure empty repo exists	×
86	}	×
87
88	/**
89	* Shared HTTP client for all RDF4J traffic. Apache HttpClient treats all requests
90	* to a given host + port + protocol as one "route", so every `HTTPRepository` in
91	* this process funnels through this single client's connection pool — plus the
92	* two ad-hoc users in {@link #createRepo} and {@link #getRepositoryNames}, which
93	* have been consolidated onto this client too.
94	*
95	* <p>The Apache defaults (maxPerRoute=2 / maxTotal=20) throttle four concurrent
96	* loader-pool threads down to two-way parallelism at the HTTP layer — invisible
97	* client-side serialisation the code is actively fighting. Raised to 10 / 40
98	* here: comfortable headroom for the 4-thread loader pool plus admin-repo
99	* transactions and the metrics tick, small enough to be a conservative first
100	* step with room to grow later.
101	*/
102	private final CloseableHttpClient httpclient = HttpClients.custom()	9✔
103	.setMaxConnPerRoute(10)	6✔
104	.setMaxConnTotal(40)	3✔
105	.build();	6✔
106
107	@GeneratedFlagForDependentElements
108	Repository getRepository(String name) {
109	synchronized (this) {
110	if (repositories.size() > 100) {
111	evictIdleRepos();
112	}
113	if (repositories.containsKey(name)) {
114	// Move to the end of the list:
115	Repository repo = repositories.remove(name);
116	repositories.put(name, repo);
117	} else {
118	Repository repository = null;
119	if (endpointType == null \|\| endpointType.equals("rdf4j")) {
120	HTTPRepository hr = new HTTPRepository(endpointBase + "repositories/" + name);
121	hr.setHttpClient(httpclient);
122	repository = hr;
123	// } else if (endpointType.equals("virtuoso")) {
124	// repository = new VirtuosoRepository(endpointBase + name, username, password);
125	} else {
126	throw new RuntimeException("Unknown repository type: " + endpointType);
127	}
128	repositories.put(name, repository);
129	createRepo(name);
130	getRepoConnection(name).close();
131	}
132	return repositories.get(name);
133	}
134	}
135
136	/**
137	* Return the repository connection for the given repository name.
138	*
139	* @param name repository name
140	* @return repository connection
141	*/
142	@GeneratedFlagForDependentElements
143	public RepositoryConnection getRepoConnection(String name) {
144	// The increment has to happen under the same monitor that guards eviction,
145	// otherwise another thread could evict this repo in the window between
146	// getRepository() returning and the counter going above zero. getConnection()
147	// on HTTPRepository is local (it doesn't do HTTP), so holding the lock here
148	// is cheap.
149	synchronized (this) {
150	Repository repo = getRepository(name);
151	if (repo == null) {
152	return null;
153	}
154	AtomicInteger counter = openConnections.computeIfAbsent(name, k -> new AtomicInteger());	×
155	counter.incrementAndGet();
156	return new CountingRepositoryConnection(repo, repo.getConnection(), counter);
157	}
158	}
159
160	/**
161	* Evicts the eldest cache entries until either the size is back within the
162	* 100-entry cap or every remaining entry has at least one open connection.
163	* The cap is load-bearing — each cached entry keeps an LMDB environment alive
164	* on the RDF4J server (in-memory cache, mmap pages, native memory), so
165	* exceeding it by much risks server-side OOM. Actively-used repos are skipped
166	* rather than shut down, because {@link org.eclipse.rdf4j.repository.http.HTTPRepository#shutDown()}
167	* closes the session manager and kills any live transaction on that repo with
168	* a connection-close error (the {@code MMapIndexInput – Already closed}
169	* failure mode observed on query-3 in the April test). The cache self-converges
170	* as active repos become idle.
171	*/
172	@GeneratedFlagForDependentElements
173	private void evictIdleRepos() {
174	List<String> skipped = new ArrayList<>();
175	Iterator<Entry<String, Repository>> iter = repositories.entrySet().iterator();
176	while (iter.hasNext() && repositories.size() > 100) {
177	Entry<String, Repository> e = iter.next();
178	AtomicInteger active = openConnections.get(e.getKey());
179	if (active != null && active.get() > 0) {
180	skipped.add(e.getKey());
181	continue;
182	}
183	iter.remove();
184	log.info("Shutting down repo: {}", e.getKey());
185	e.getValue().shutDown();
186	log.info("Shutdown complete");
187	}
188	if (!skipped.isEmpty()) {
189	log.warn("Skipped eviction for {} active repo(s); cache size is now {} (cap 100). Active names: {}",
190	skipped.size(), repositories.size(), skipped);
191	}
192	}
193
194	/**
195	* Minimal wrapper around a delegate {@link RepositoryConnection} whose only job
196	* is to decrement the per-repo open-connection counter exactly once when the
197	* caller closes it. Uses {@link AtomicBoolean} so that repeated/idempotent
198	* {@code close()} calls (common with try-with-resources plus explicit close)
199	* don't decrement more than once.
200	*/
201	private static final class CountingRepositoryConnection extends RepositoryConnectionWrapper {
202
203	private final AtomicInteger counter;
204	private final AtomicBoolean closed = new AtomicBoolean();	×
205
206	CountingRepositoryConnection(Repository repo, RepositoryConnection delegate, AtomicInteger counter) {
207	super(repo, delegate);	×
208	this.counter = counter;	×
209	}	×
210
211	@Override
212	public void close() {
213	try {
214	super.close();	×
215	} finally {
216	if (closed.compareAndSet(false, true)) {	×
217	counter.decrementAndGet();	×
218	}
219	}
220	}	×
221
222	}
223
224	@GeneratedFlagForDependentElements
225	private void createRepo(String repoName) {
226	if (!repoName.equals(ADMIN_REPO)) {
227	getRepository(ADMIN_REPO); // make sure admin repo is loaded first
228	}
229	// Uses the shared this.httpclient rather than a per-call client so it inherits
230	// the configured pool sizes (and, once change 1 of the fix plan lands, the
231	// socket/connection-request timeouts).
232	try {
233	//log.info("Trying to creating repo " + name);
234
235	// TODO new syntax somehow doesn't work for the Lucene case:
236
237	// String createRegularRepoQueryString = "@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#>.\n"
238	// + "@prefix config: <tag:rdf4j.org,2023:config/>.\n"
239	// + "[] a config:Repository ;\n"
240	// + " config:rep.id \"" + name + "\" ;\n"
241	// + " rdfs:label \"" + name + " native store\" ;\n"
242	// + " config:rep.impl [\n"
243	// + " config:rep.type \"openrdf:SailRepository\" ;\n"
244	// + " config:sail.impl [\n"
245	// + " config:sail.type \"openrdf:NativeStore\" ;\n"
246	// + " config:sail.iterationCacheSyncThreshold \"10000\";\n"
247	// + " config:native.tripleIndexes \"spoc,posc,ospc,opsc,psoc,sopc,spoc,cpos,cosp,cops,cpso,csop\" ;\n"
248	// + " config:sail.defaultQueryEvaluationMode \"STANDARD\"\n"
249	// + " ]\n"
250	// + " ].";
251	// String createTextRepoQueryString = "@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#>.\n"
252	// + "@prefix config: <tag:rdf4j.org,2023:config/>.\n"
253	// + "[] a config:Repository ;\n"
254	// + " config:rep.id \"" + name + "\" ;\n"
255	// + " rdfs:label \"" + name + " native store\" ;\n"
256	// + " config:rep.impl [\n"
257	// + " config:rep.type \"openrdf:SailRepository\" ;\n"
258	// + " config:sail.impl [\n"
259	// + " config:sail.type \"openrdf:LuceneSail\" ;\n"
260	// + " config:sail.lucene.indexDir \"index/\" ;\n"
261	// + " config:delegate [\n"
262	// + " config:rep.type \"openrdf:SailRepository\" ;\n"
263	// + " config:sail.impl [\n"
264	// + " config:sail.type \"openrdf:NativeStore\" ;\n"
265	// + " config:sail.iterationCacheSyncThreshold \"10000\";\n"
266	// + " config:native.tripleIndexes \"spoc,posc,ospc,opsc,psoc,sopc,spoc,cpos,cosp,cops,cpso,csop\" ;\n"
267	// + " config:sail.defaultQueryEvaluationMode \"STANDARD\"\n"
268	// + " ]\n"
269	// + " ]\n"
270	// + " ]\n"
271	// + " ].";
272
273	String indexTypes = "spoc,posc,ospc,cspo,cpos,cosp";
274	if (repoName.startsWith("meta") \|\| repoName.startsWith("text")) {
275	indexTypes = "spoc,posc,ospc";
276	}
277
278	String createRegularRepoQueryString =
279	"@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#>.\n" +
280	"@prefix rep: <http://www.openrdf.org/config/repository#>.\n" +
281	"@prefix sr: <http://www.openrdf.org/config/repository/sail#>.\n" +
282	"@prefix sail: <http://www.openrdf.org/config/sail#>.\n" +
283	"@prefix sail-luc: <http://www.openrdf.org/config/sail/lucene#>.\n" +
284	"@prefix lmdb: <http://rdf4j.org/config/sail/lmdb#>.\n" +
285	"@prefix sb: <http://www.openrdf.org/config/sail/base#>.\n" +
286	"\n" +
287	"[] a rep:Repository ;\n" +
288	" rep:repositoryID \"" + repoName + "\" ;\n" +
289	" rdfs:label \"" + repoName + " LMDB store\" ;\n" +
290	" rep:repositoryImpl [\n" +
291	" rep:repositoryType \"openrdf:SailRepository\" ;\n" +
292	" sr:sailImpl [\n" +
293	" sail:sailType \"rdf4j:LmdbStore\" ;\n" +
294	" sail:iterationCacheSyncThreshold \"10000\";\n" +
295	" lmdb:tripleIndexes \"" + indexTypes + "\" ;\n" +
296	" sb:defaultQueryEvaluationMode \"STANDARD\"\n" +
297	" ]\n"
298	+ " ].\n";
299
300	// TODO Index npa:hasFilterLiteral predicate too (see https://groups.google.com/g/rdf4j-users/c/epF4Af1jXGU):
301	String createTextRepoQueryString =
302	"@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#>.\n" +
303	"@prefix rep: <http://www.openrdf.org/config/repository#>.\n" +
304	"@prefix sr: <http://www.openrdf.org/config/repository/sail#>.\n" +
305	"@prefix sail: <http://www.openrdf.org/config/sail#>.\n" +
306	"@prefix sail-luc: <http://www.openrdf.org/config/sail/lucene#>.\n" +
307	"@prefix lmdb: <http://rdf4j.org/config/sail/lmdb#>.\n" +
308	"@prefix sb: <http://www.openrdf.org/config/sail/base#>.\n" +
309	"\n"
310	+ "[] a rep:Repository ;\n" +
311	" rep:repositoryID \"" + repoName + "\" ;\n" +
312	" rdfs:label \"" + repoName + " store\" ;\n" +
313	" rep:repositoryImpl [\n" +
314	" rep:repositoryType \"openrdf:SailRepository\" ;\n" +
315	" sr:sailImpl [\n" +
316	" sail:sailType \"openrdf:LuceneSail\" ;\n" +
317	" sail-luc:indexDir \"index/\" ;\n" +
318	" sail-luc:transactional false ;\n" +
319	" sail:delegate [\n" +
320	" sail:sailType \"rdf4j:LmdbStore\" ;\n" +
321	" sail:iterationCacheSyncThreshold \"10000\";\n" +
322	" lmdb:tripleIndexes \"" + indexTypes + "\" ;\n" +
323	" sb:defaultQueryEvaluationMode \"STANDARD\"\n" +
324	" ]\n" +
325	" ]\n" +
326	" ].";
327
328	String createRepoQueryString = createRegularRepoQueryString;
329	if (repoName.startsWith("text")) {
330	createRepoQueryString = createTextRepoQueryString;
331	}
332
333	HttpUriRequest createRepoRequest = RequestBuilder.put().setUri(endpointBase + "repositories/" + repoName).addHeader("Content-Type", "text/turtle").setEntity(new StringEntity(createRepoQueryString)).build();
334
335	// Response is try-with-resources'd so the connection is released back to
336	// the shared pool rather than leaked.
337	try (CloseableHttpResponse response = httpclient.execute(createRepoRequest)) {
338	int statusCode = response.getStatusLine().getStatusCode();
339	if (statusCode == 409) {
340	//log.info("Already exists.");
341	getRepository(repoName).init();
342	} else if (statusCode >= 200 && statusCode < 300) {
343	//log.info("Successfully created.");
344	initNewRepo(repoName);
345	} else {
346	log.info("Status code: {}", response.getStatusLine().getStatusCode());
347	log.info(response.getStatusLine().getReasonPhrase());
348	String handledResponse = new BasicResponseHandler().handleResponse(response);
349	log.info("Response: {}", handledResponse);
350	}
351	}
352	} catch (IOException ex) {
353	log.info("Could not create repo.", ex);
354	}
355	}
356
357	/**
358	* Sends shutdown signal to all repositories.
359	*/
360	@GeneratedFlagForDependentElements
361	public void shutdownRepositories() {
362	for (Repository repo : repositories.values()) {
363	if (repo != null && repo.isInitialized()) {
364	repo.shutDown();
365	}
366	}
367	}
368
369	/**
370	* Returns admin repo connection.
371	*
372	* @return repository connection to admin repository
373	*/
374	@GeneratedFlagForDependentElements
375	public RepositoryConnection getAdminRepoConnection() {
376	return get().getRepoConnection(ADMIN_REPO);
377	}
378
379	private Set<String> cachedRepositoryNames = Set.of();	9✔
380	private boolean repoNamesCacheValid = false;	9✔
381	private final ReadWriteLock repoNamesCacheLock = new ReentrantReadWriteLock();	15✔
382
383	/**
384	* Returns set of all repository names.
385	*
386	* @return Repository name set
387	*/
388	public Set<String> getRepositoryNames() {
389	// See if the repository names are cached:
390	final var readLock = repoNamesCacheLock.readLock();	12✔
391	try {
392	readLock.lock();	6✔
393	if (repoNamesCacheValid) {	9✔
394	return cachedRepositoryNames;	15✔
395	}
396	} finally {
397	readLock.unlock();	6✔
398	}
399
400	// Not cached, get from server:
401	final var writeLock = repoNamesCacheLock.writeLock();	12✔
402	try {
403	writeLock.lock();	6✔
404	// Check again if another thread has already updated the cache:
405	if (repoNamesCacheValid) {	9!
406	return cachedRepositoryNames;	×
407	}
408	Map<String, Boolean> repositoryNames = null;	6✔
409	// Uses the shared this.httpclient; response try-with-resources releases
410	// the pooled connection when done.
411	try (CloseableHttpResponse resp = httpclient.execute(RequestBuilder.get()	24✔
412	.setUri(endpointBase + "/repositories")	9✔
413	.addHeader("Content-Type", "text/csv")	3✔
414	.build())) {	3✔
415	BufferedReader reader = new BufferedReader(new InputStreamReader(resp.getEntity().getContent()));	30✔
416	int code = resp.getStatusLine().getStatusCode();	12✔
417	if (code < 200 \|\| code >= 300) return null;	30!
418	repositoryNames = new HashMap<>();	12✔
419	int lineCount = 0;	6✔
420	while (true) {
421	String line = reader.readLine();	9✔
422	if (line == null) break;	9✔
423	if (lineCount > 0) {	6✔
424	String repoName = line.split(",")[1];	18✔
425	repositoryNames.put(repoName, true);	18✔
426	}
427	lineCount = lineCount + 1;	12✔
428	}	3✔
429	} catch (IOException ex) {	15!
430	log.info("Could not get repository names.", ex);	12✔
431	return null;	12✔
432	}	3✔
433	cachedRepositoryNames = repositoryNames.keySet();	12✔
434	repoNamesCacheValid = true;	9✔
435	return cachedRepositoryNames;	15✔
436	} finally {
437	writeLock.unlock();	6✔
438	}
439	}
440
441	/**
442	* Invalidates the repository names cache. Call this method when a repository is created or deleted.
443	*/
444	private void invalidateRepositoryNamesCache() {
445	final var writeLock = repoNamesCacheLock.writeLock();	×
446	try {
447	writeLock.lock();	×
448	repoNamesCacheValid = false;	×
449	} finally {
450	writeLock.unlock();	×
451	}
452	}	×
453
454	@GeneratedFlagForDependentElements
455	private void initNewRepo(String repoName) {
456	String repoInitId = new Random().nextLong() + "";
457	getRepository(repoName).init();
458	if (!repoName.equals("empty")) {
459	RepositoryConnection conn = getRepoConnection(repoName);
460	try (conn) {
461	// Full isolation, just in case.
462	conn.begin(IsolationLevels.SERIALIZABLE);
463	conn.add(NPA.THIS_REPO, NPA.HAS_REPO_INIT_ID, vf.createLiteral(repoInitId), NPA.GRAPH);
464	if (tracksNanopubCountAndChecksum(repoName)) {
465	conn.add(NPA.THIS_REPO, NPA.HAS_NANOPUB_COUNT, vf.createLiteral(0L), NPA.GRAPH);
466	conn.add(NPA.THIS_REPO, NPA.HAS_NANOPUB_CHECKSUM, vf.createLiteral(NanopubUtils.INIT_CHECKSUM), NPA.GRAPH);
467	}
468	if (repoName.startsWith("pubkey_") \|\| repoName.startsWith("type_")) {
469	String h = repoName.replaceFirst("^[^_]+_", "");
470	conn.add(NPA.THIS_REPO, NPA.HAS_COVERAGE_ITEM, Utils.getObjectForHash(h), NPA.GRAPH);
471	conn.add(NPA.THIS_REPO, NPA.HAS_COVERAGE_HASH, vf.createLiteral(h), NPA.GRAPH);
472	conn.add(NPA.THIS_REPO, NPA.HAS_COVERAGE_FILTER, vf.createLiteral("_" + repoName), NPA.GRAPH);
473	}
474	conn.commit();
475	}
476	// Refresh repository names cache
477	invalidateRepositoryNamesCache();
478	}
479	}
480
481	/**
482	* Whether the given repo participates in the cumulative nanopub-count / XOR-checksum
483	* tracking. Repos that don't hold raw nanopubs skip the
484	* {@code npa:hasNanopubCount} and {@code npa:hasNanopubChecksum} initial triples —
485	* leaving them at {@code 0} and the empty-XOR placeholder forever would just be
486	* misleading. Currently excluded:
487	* <ul>
488	* <li>{@code admin} — holds metadata only.</li>
489	* <li>{@code last30d} — content expires on a periodic cleanup.</li>
490	* <li>{@code trust} and {@code spaces} — hold derived state, not raw nanopubs.</li>
491	* </ul>
492	*/
493	private static boolean tracksNanopubCountAndChecksum(String repoName) {
494	return !repoName.equals(ADMIN_REPO)	×
495	&& !repoName.equals("last30d")	×
496	&& !repoName.equals("trust")	×
497	&& !repoName.equals("spaces");	×
498	}
499
500	}

knowledgepixels / nanopub-query / 24662130060

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous