You are viewing a plain text version of this content. The canonical link for it is here.
Posted to oak-commits@jackrabbit.apache.org by mr...@apache.org on 2022/07/01 07:32:12 UTC
[jackrabbit-oak] 01/02: OAK-9780: PoC for DocumentNodeStore cache warming
This is an automated email from the ASF dual-hosted git repository.
mreutegg pushed a commit to branch OAK-9780
in repository https://gitbox.apache.org/repos/asf/jackrabbit-oak.git
commit ddb1ef6b57cc519dcf87efecfdf9cda0b1414b9f
Author: Marcel Reutegger <ma...@gmail.com>
AuthorDate: Thu Jun 30 14:55:21 2022 +0200
OAK-9780: PoC for DocumentNodeStore cache warming
Cache absence of document for a given id
Ensure cache consistency with locks and use of CacheChangesTracker
More tests
Basic DocumentStore metrics for prefetch operations
---
.../oak/plugins/document/DocumentStoreStats.java | 31 ++++
.../document/DocumentStoreStatsCollector.java | 10 ++
.../plugins/document/DocumentStoreStatsMBean.java | 4 +
.../plugins/document/mongo/MongoDocumentStore.java | 160 +++++++++++----------
.../document/prefetch/CacheWarmingTest.java | 49 ++++++-
5 files changed, 179 insertions(+), 75 deletions(-)
diff --git a/oak-store-document/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentStoreStats.java b/oak-store-document/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentStoreStats.java
index d2e2d21463..95ed595533 100644
--- a/oak-store-document/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentStoreStats.java
+++ b/oak-store-document/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentStoreStats.java
@@ -73,6 +73,10 @@ public class DocumentStoreStats implements DocumentStoreStatsCollector, Document
static final String NODES_REMOVE = "DOCUMENT_NODES_REMOVE";
static final String NODES_REMOVE_TIMER = "DOCUMENT_NODES_REMOVE_TIMER";
+ static final String NODES_PREFETCH = "DOCUMENT_NODES_PREFETCH";
+
+ static final String NODES_PREFETCH_TIMER = "DOCUMENT_NODES_PREFETCH_TIMER";
+
static final String JOURNAL_QUERY = "DOCUMENT_JOURNAL_QUERY";
static final String JOURNAL_CREATE = "DOCUMENT_JOURNAL_CREATE";
static final String JOURNAL_QUERY_TIMER = "DOCUMENT_JOURNAL_QUERY_TIMER";
@@ -109,6 +113,8 @@ public class DocumentStoreStats implements DocumentStoreStatsCollector, Document
private final MeterStats updateNodeRetryCountMeter;
private final MeterStats removeNodes;
private final TimerStats removeNodesTimer;
+ private final MeterStats prefetchNodes;
+ private final TimerStats prefetchNodesTimer;
public DocumentStoreStats(StatisticsProvider provider) {
statisticsProvider = checkNotNull(provider);
@@ -148,6 +154,9 @@ public class DocumentStoreStats implements DocumentStoreStatsCollector, Document
removeNodes = provider.getMeter(NODES_REMOVE, StatsOptions.DEFAULT);
removeNodesTimer = provider.getTimer(NODES_REMOVE_TIMER, StatsOptions.METRICS_ONLY);
+
+ prefetchNodes = provider.getMeter(NODES_PREFETCH, StatsOptions.DEFAULT);
+ prefetchNodesTimer = provider.getTimer(NODES_PREFETCH_TIMER, StatsOptions.METRICS_ONLY);
}
//~------------------------------------------< DocumentStoreStatsCollector >
@@ -292,6 +301,17 @@ public class DocumentStoreStats implements DocumentStoreStatsCollector, Document
perfLog(timeTakenNanos, "remove [{}]", removeCount);
}
+ @Override
+ public void donePrefetch(long timeTakenNanos,
+ Collection<? extends Document> collection,
+ List<String> ids) {
+ if (collection == Collection.NODES) {
+ prefetchNodes.mark(ids.size());
+ prefetchNodesTimer.update(timeTakenNanos, TimeUnit.NANOSECONDS);
+ }
+ perfLog(timeTakenNanos, "prefetch {}", ids);
+ }
+
private void perfLog(long timeTakenNanos, String logMessagePrefix, Object... arguments){
if (!perfLog.isDebugEnabled()){
return;
@@ -356,6 +376,11 @@ public class DocumentStoreStats implements DocumentStoreStatsCollector, Document
return queryJournal.getCount();
}
+ @Override
+ public long getNodesPrefetchCount() {
+ return prefetchNodes.getCount();
+ }
+
@Override
public CompositeData getFindCachedNodesHistory() {
return getTimeSeriesData(NODES_FIND_CACHED,
@@ -446,6 +471,12 @@ public class DocumentStoreStats implements DocumentStoreStatsCollector, Document
"Number of removed node documents.");
}
+ @Override
+ public CompositeData getPrefetchNodesHistory() {
+ return getTimeSeriesData(NODES_PREFETCH,
+ "Number of prefetched node documents.");
+ }
+
private CompositeData getTimeSeriesData(String name, String desc){
return TimeSeriesStatsUtil.asCompositeData(getTimeSeries(name), desc);
}
diff --git a/oak-store-document/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentStoreStatsCollector.java b/oak-store-document/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentStoreStatsCollector.java
index 5e8e1f6cde..031b58420d 100644
--- a/oak-store-document/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentStoreStatsCollector.java
+++ b/oak-store-document/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentStoreStatsCollector.java
@@ -94,4 +94,14 @@ public interface DocumentStoreStatsCollector {
void doneRemove(long timeTakenNanos,
Collection<? extends Document> collection,
int removeCount);
+
+ /**
+ * Called when a prefetch operation for documents was completed.
+ * @param timeTakenNanos time taken
+ * @param collection the collection
+ * @param ids list of IDs to prefetch / read
+ */
+ void donePrefetch(long timeTakenNanos,
+ Collection<? extends Document> collection,
+ List<String> ids);
}
diff --git a/oak-store-document/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentStoreStatsMBean.java b/oak-store-document/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentStoreStatsMBean.java
index 3118233043..38346e27ba 100644
--- a/oak-store-document/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentStoreStatsMBean.java
+++ b/oak-store-document/src/main/java/org/apache/jackrabbit/oak/plugins/document/DocumentStoreStatsMBean.java
@@ -46,6 +46,8 @@ public interface DocumentStoreStatsMBean {
long getJournalReadCount();
+ long getNodesPrefetchCount();
+
CompositeData getFindCachedNodesHistory();
CompositeData getFindSplitNodesHistory();
@@ -75,4 +77,6 @@ public interface DocumentStoreStatsMBean {
CompositeData getUpdateNodesFailureHistory();
CompositeData getRemoveNodesHistory();
+
+ CompositeData getPrefetchNodesHistory();
}
diff --git a/oak-store-document/src/main/java/org/apache/jackrabbit/oak/plugins/document/mongo/MongoDocumentStore.java b/oak-store-document/src/main/java/org/apache/jackrabbit/oak/plugins/document/mongo/MongoDocumentStore.java
index bfe99b829b..abb4fe5c61 100644
--- a/oak-store-document/src/main/java/org/apache/jackrabbit/oak/plugins/document/mongo/MongoDocumentStore.java
+++ b/oak-store-document/src/main/java/org/apache/jackrabbit/oak/plugins/document/mongo/MongoDocumentStore.java
@@ -111,6 +111,7 @@ import static com.google.common.collect.Sets.difference;
import static org.apache.jackrabbit.oak.plugins.document.DocumentStoreException.asDocumentStoreException;
import static org.apache.jackrabbit.oak.plugins.document.NodeDocument.DELETED_ONCE;
import static org.apache.jackrabbit.oak.plugins.document.NodeDocument.MODIFIED_IN_SECS;
+import static org.apache.jackrabbit.oak.plugins.document.NodeDocument.NULL;
import static org.apache.jackrabbit.oak.plugins.document.NodeDocument.SD_MAX_REV_TIME_IN_SECS;
import static org.apache.jackrabbit.oak.plugins.document.NodeDocument.SD_TYPE;
import static org.apache.jackrabbit.oak.plugins.document.UpdateOp.Condition.newEqualsCondition;
@@ -1390,6 +1391,92 @@ public class MongoDocumentStore implements DocumentStore {
}
}
+ @SuppressWarnings("unchecked")
+ @Override
+ public <T extends Document> void prefetch(Collection<T> collection,
+ Iterable<String> keysToPrefetch) {
+ log("prefetch", keysToPrefetch);
+
+ Set<String> keys = new HashSet<>();
+ for (String k : keysToPrefetch) {
+ if (nodesCache.getIfPresent(k) == null) {
+ keys.add(k);
+ }
+ }
+ if (keys.size() < minPrefetch) {
+ return;
+ }
+
+ final Stopwatch watch = startWatch();
+ List<String> resultKeys = new ArrayList<>(keys.size());
+ CacheChangesTracker tracker = null;
+ if (collection == Collection.NODES) {
+ tracker = nodesCache.registerTracker(keys);
+ }
+ Throwable t;
+ try {
+ ReadPreference readPreference = getMongoReadPreference(collection, null, getDefaultReadPreference(collection));
+ MongoCollection<BasicDBObject> dbCollection = getDBCollection(collection, readPreference);
+
+ if (readPreference.isSlaveOk()) {
+ LOG.trace("Routing call to secondary for prefetching [{}]", keys);
+ }
+
+ List<BasicDBObject> result = new ArrayList<>(keys.size());
+ execute(session -> {
+ final Bson query = Filters.in(Document.ID, keys);
+ if (session != null) {
+ dbCollection.find(session, query).into(result);
+ } else {
+ dbCollection.find(query).into(result);
+ }
+ return null;
+ }, collection);
+
+ List<T> docs = new ArrayList<>(keys.size());
+ for (BasicDBObject dbObject : result) {
+ final T d = convertFromDBObject(collection, dbObject);
+ if (d == null) {
+ continue;
+ }
+ d.seal();
+ String key = String.valueOf(d.get(Document.ID));
+ resultKeys.add(key);
+ keys.remove(key);
+ docs.add(d);
+ }
+
+ if (tracker != null) {
+ nodesCache.putNonConflictingDocs(tracker, (List<NodeDocument>) docs);
+
+ // documents for remaining ids in keys do not exist
+ for (String id : keys) {
+ Lock lock = nodeLocks.acquire(id);
+ try {
+ // load NULL document into cache unless it may have
+ // been affected by another concurrent operation
+ if (!tracker.mightBeenAffected(id)) {
+ nodesCache.get(id, () -> NULL);
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+ }
+ return;
+ } catch (UncheckedExecutionException | ExecutionException e) {
+ t = e.getCause();
+ } catch (RuntimeException e) {
+ t = e;
+ } finally {
+ if (tracker != null) {
+ tracker.close();
+ }
+ stats.donePrefetch(watch.elapsed(TimeUnit.NANOSECONDS), collection, resultKeys);
+ }
+ throw handleException(t, collection, keysToPrefetch);
+ }
+
/**
* Returns the {@link Document#MOD_COUNT} and
* {@link NodeDocument#MODIFIED_IN_SECS} values of the documents with the
@@ -1934,7 +2021,6 @@ public class MongoDocumentStore implements DocumentStore {
}
return dbConnection.createClientSession();
}
-
interface DocumentStoreCallable<T> {
T call(@Nullable ClientSession session) throws DocumentStoreException;
@@ -1975,76 +2061,4 @@ public class MongoDocumentStore implements DocumentStore {
return toString();
}
}
-
- @Override
- public <T extends Document> void prefetch(Collection<T> collection,
- Iterable<String> keysToPrefetch) {
- ArrayList<String> keys = new ArrayList<>();
- for (String k : keysToPrefetch) {
- if (nodesCache.getIfPresent(k) == null) {
- keys.add(k);
- }
- }
- if (keys.size() < minPrefetch) {
- return;
- }
-
- final DocumentReadPreference docReadPref = DocumentReadPreference.PRIMARY;
- log("prefetch", keys, docReadPref);
- final Stopwatch watch = startWatch();
- boolean isSlaveOk = false;
- boolean docFound = true;
-
- try {
- ReadPreference readPreference = getMongoReadPreference(collection, null, docReadPref);
- MongoCollection<BasicDBObject> dbCollection = getDBCollection(collection, readPreference);
-
- if(readPreference.isSlaveOk()){
- LOG.trace("Routing call to secondary for fetching [{}]", keys);
- isSlaveOk = true;
- }
-
- List<BasicDBObject> result = new ArrayList<>(1);
- execute(session -> {
- final Bson query = Filters.in(Document.ID, keys);;
- if (session != null) {
- dbCollection.find(session, query).into(result);
- } else {
- dbCollection.find(query).into(result);
- }
- return null;
- }, collection);
-
- if(result.isEmpty()) {
- docFound = false;
- return;
- }
- for (BasicDBObject basicDBObject : result) {
-// final NodeDocument d = (NodeDocument) findUncachedWithRetry(
-// collection, key,
-// getReadPreference(maxCacheAge));
- final NodeDocument d = (NodeDocument) convertFromDBObject(collection, basicDBObject);
- if (d == null) {
- continue;
- }
- String key = String.valueOf(d.get(Document.ID));
- d.seal();
- try {
- // load into cache
- invalidateCache(collection, key);
- nodesCache.get(key, new Callable<NodeDocument>() {
- @Override
- public NodeDocument call() throws Exception {
- return d == null ? NodeDocument.NULL : d;
- }
- });
- } catch (ExecutionException e) {
- LOG.error("prefetch: error prefetching id {}", key, e);
- }
- }
-// return doc;
- } finally {
-// stats.doneFindUncached(watch.elapsed(TimeUnit.NANOSECONDS), collection, key, docFound, isSlaveOk);
- }
- }
}
diff --git a/oak-store-document/src/test/java/org/apache/jackrabbit/oak/plugins/document/prefetch/CacheWarmingTest.java b/oak-store-document/src/test/java/org/apache/jackrabbit/oak/plugins/document/prefetch/CacheWarmingTest.java
index e3e851b886..c17f7bdc79 100644
--- a/oak-store-document/src/test/java/org/apache/jackrabbit/oak/plugins/document/prefetch/CacheWarmingTest.java
+++ b/oak-store-document/src/test/java/org/apache/jackrabbit/oak/plugins/document/prefetch/CacheWarmingTest.java
@@ -16,10 +16,14 @@
*/
package org.apache.jackrabbit.oak.plugins.document.prefetch;
+import static org.apache.jackrabbit.oak.plugins.document.util.Utils.getIdFromPath;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -40,6 +44,7 @@ import org.apache.jackrabbit.oak.plugins.document.Path;
import org.apache.jackrabbit.oak.plugins.document.memory.MemoryDocumentStore;
import org.apache.jackrabbit.oak.plugins.document.mongo.MongoDocumentStore;
import org.apache.jackrabbit.oak.plugins.document.util.MongoConnection;
+import org.apache.jackrabbit.oak.plugins.document.util.Utils;
import org.apache.jackrabbit.oak.spi.commit.CommitInfo;
import org.apache.jackrabbit.oak.spi.commit.EmptyHook;
import org.apache.jackrabbit.oak.spi.state.NodeBuilder;
@@ -114,6 +119,40 @@ public class CacheWarmingTest {
doSimple(false, true);
}
+ @Test
+ public void prefetch() throws Exception {
+ DocumentStore ds = newMongoDocumentStore();
+ CacheWarming cw = new CacheWarming(ds);
+ DocumentNodeStore store = builderProvider.newBuilder().setAsyncDelay(0)
+ .setDocumentStore(ds).getNodeStore();
+ SortedSet<String> children = new TreeSet<String>();
+ NodeBuilder builder = store.getRoot().builder();
+ for (int i = 0; i < 10; i++) {
+ String name = "c" + i;
+ children.add("/" + name + "/" + name);
+ builder.child(name).child(name);
+ }
+ store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+ store.runBackgroundOperations();
+
+ store.getMBean().cleanAllCaches();
+
+ cw.prefetch(children, store.getRoot());
+
+ for (String p : children) {
+ assertNotNull(ds.getIfCached(Collection.NODES, getIdFromPath(p)));
+ }
+
+ List<String> paths = new ArrayList<>();
+ for (int i = 0; i < 10; i++) {
+ paths.add("/does/not/exist-" + i);
+ }
+ cw.prefetch(paths, store.getRoot());
+ int numRawFindCalls = getRawFindCalls();
+ assertNull(ds.find(Collection.NODES, getIdFromPath(paths.get(0))));
+ assertEquals(0, getRawFindCalls() - numRawFindCalls);
+ }
+
private void doSimple(boolean cleanCaches, boolean prefetch)
throws InterruptedException, CommitFailedException {
System.out.println("=== doSimple( cleanCaches = " + cleanCaches + ", prefetch = " + prefetch + " )");
@@ -121,7 +160,8 @@ public class CacheWarmingTest {
DocumentStore ds = newMongoDocumentStore();
final CountingDocumentStore cds = new CountingDocumentStore(ds);
logAndReset("after init ", cds, sw);
- DocumentNodeStore store = builderProvider.newBuilder().setDocumentStore(cds).getNodeStore();
+ DocumentNodeStore store = builderProvider.newBuilder().setAsyncDelay(0)
+ .setDocumentStore(cds).getNodeStore();
NodeBuilder builder = store.getRoot().builder();
SortedSet<String> children = new TreeSet<String>();
// create a bunch of nodes
@@ -133,6 +173,7 @@ public class CacheWarmingTest {
}
logAndReset("before merge ", cds, sw);
store.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY);
+ store.runBackgroundOperations();
logAndReset("after merge ", cds, sw);
if (cleanCaches) {
// clear the cache via mbean
@@ -176,10 +217,14 @@ public class CacheWarmingTest {
+ ", findCalls = " + cds.getNumFindCalls(Collection.NODES)
+ ", queryCalls = " + cds.getNumQueryCalls(Collection.NODES)
+ ", removalCalls = " + cds.getNumRemoveCalls(Collection.NODES)
- + ", rawFindCalls = " + db.getCachedCountingCollection("nodes").getFindCounter()
+ + ", rawFindCalls = " + getRawFindCalls()
+ ", " + sw);
db.getCachedCountingCollection("nodes").resetFindCounter();
cds.resetCounters();
sw.reset().start();
}
+
+ private int getRawFindCalls() {
+ return db.getCachedCountingCollection("nodes").getFindCounter();
+ }
}