Author: reschke
Date: Fri Sep 2 14:16:20 2016
New Revision: 1758961
URL: http://svn.apache.org/viewvc?rev=1758961&view=rev
Log:
OAK-4725: Use the cache tracker in the RDB Document Store (ported to 1.4)
Modified:
jackrabbit/oak/branches/1.4/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/cache/NodeDocumentCache.java
jackrabbit/oak/branches/1.4/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/rdb/RDBDocumentStore.java
Modified:
jackrabbit/oak/branches/1.4/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/cache/NodeDocumentCache.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/branches/1.4/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/cache/NodeDocumentCache.java?rev=1758961&r1=1758960&r2=1758961&view=diff
==============================================================================
---
jackrabbit/oak/branches/1.4/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/cache/NodeDocumentCache.java
(original)
+++
jackrabbit/oak/branches/1.4/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/cache/NodeDocumentCache.java
Fri Sep 2 14:16:20 2016
@@ -96,9 +96,21 @@ public class NodeDocumentCache implement
nodeDocumentsCache.invalidate(new StringValue(key));
}
- for (CacheChangesTracker tracker : changeTrackers) {
- tracker.invalidateDocument(key);
- }
+ internalMarkChanged(key);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Mark that the document with the given key is being changed.
+ *
+ * @param key to mark
+ */
+ public void markChanged(@Nonnull String key) {
+ Lock lock = locks.acquire(key);
+ try {
+ internalMarkChanged(key);
} finally {
lock.unlock();
}
@@ -149,6 +161,10 @@ public class NodeDocumentCache implement
/**
* Return the document matching given key, optionally loading it from an
* external source.
+ * <p>
+ * This method can modify the cache, so it's synchronized. The {@link
#getIfPresent(String)}
+ * is not synchronized and will be faster if you need to get the cached
value
+ * outside the critical section.
*
* @see Cache#get(Object, Callable)
* @param key document key
@@ -443,6 +459,17 @@ public class NodeDocumentCache implement
//----------------------------< internal
>----------------------------------
/**
+ * Marks the document as potentially changed.
+ *
+ * @param key the document to be marked
+ */
+ private void internalMarkChanged(String key) {
+ for (CacheChangesTracker tracker : changeTrackers) {
+ tracker.invalidateDocument(key);
+ }
+ }
+
+ /**
* Puts a document into the cache without acquiring a lock.
*
* @param doc the document to put into the cache.
Modified:
jackrabbit/oak/branches/1.4/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/rdb/RDBDocumentStore.java
URL:
http://svn.apache.org/viewvc/jackrabbit/oak/branches/1.4/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/rdb/RDBDocumentStore.java?rev=1758961&r1=1758960&r2=1758961&view=diff
==============================================================================
---
jackrabbit/oak/branches/1.4/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/rdb/RDBDocumentStore.java
(original)
+++
jackrabbit/oak/branches/1.4/oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/document/rdb/RDBDocumentStore.java
Fri Sep 2 14:16:20 2016
@@ -77,6 +77,7 @@ import org.apache.jackrabbit.oak.plugins
import org.apache.jackrabbit.oak.plugins.document.UpdateOp.Key;
import org.apache.jackrabbit.oak.plugins.document.UpdateOp.Operation;
import org.apache.jackrabbit.oak.plugins.document.UpdateUtils;
+import org.apache.jackrabbit.oak.plugins.document.cache.CacheChangesTracker;
import org.apache.jackrabbit.oak.plugins.document.cache.CacheInvalidationStats;
import org.apache.jackrabbit.oak.plugins.document.cache.NodeDocumentCache;
import org.apache.jackrabbit.oak.plugins.document.locks.NodeDocumentLocks;
@@ -90,9 +91,6 @@ import com.google.common.collect.Immutab
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
-import com.google.common.hash.BloomFilter;
-import com.google.common.hash.Funnel;
-import com.google.common.hash.PrimitiveSink;
/**
* Implementation of {@link DocumentStore} for relational databases.
@@ -516,9 +514,6 @@ public class RDBDocumentStore implements
private <T extends Document> void invalidateCache(Collection<T>
collection, String id, boolean remove) {
if (collection == Collection.NODES) {
- for (QueryContext qc : qmap.values()) {
- qc.addKey(id);
- }
invalidateNodesCache(id, remove);
}
}
@@ -529,6 +524,7 @@ public class RDBDocumentStore implements
if (remove) {
nodesCache.invalidate(id);
} else {
+ nodesCache.markChanged(id);
NodeDocument entry = nodesCache.getIfPresent(id);
if (entry != null) {
entry.markUpToDate(0);
@@ -671,6 +667,15 @@ public class RDBDocumentStore implements
}
}
+ private <T extends Document> T getIfCached(Collection<T> collection,
String id, long modCount) {
+ T doc = getIfCached(collection, id);
+ if (doc != null && doc.getModCount() == modCount) {
+ return doc;
+ } else {
+ return null;
+ }
+ }
+
@Override
public Iterable<CacheStats> getCacheStats() {
return nodesCache.getCacheStats();
@@ -1334,18 +1339,9 @@ public class RDBDocumentStore implements
for (List<String> chunkedIds : Lists.partition(ids, CHUNKSIZE)) {
- Set<QueryContext> seenQueryContext = Collections.emptySet();
-
if (collection == Collection.NODES) {
- // keep concurrently running queries from updating
- // the cache entry for this key
- seenQueryContext = new HashSet<QueryContext>();
- for (QueryContext qc : qmap.values()) {
- qc.addKeys(chunkedIds);
- seenQueryContext.add(qc);
- }
- for (String id : chunkedIds) {
- nodesCache.invalidate(id);
+ for (String key : chunkedIds) {
+ nodesCache.invalidate(key);
}
}
@@ -1368,13 +1364,6 @@ public class RDBDocumentStore implements
}
if (success) {
if (collection == Collection.NODES) {
- // keep concurrently running queries from updating
- // the cache entry for this key
- for (QueryContext qc : qmap.values()) {
- if (!seenQueryContext.contains(qc)) {
- qc.addKeys(chunkedIds);
- }
- }
for (String id : chunkedIds) {
nodesCache.invalidate(id);
}
@@ -1396,72 +1385,6 @@ public class RDBDocumentStore implements
}
}
- /**
- * Class used to track which documents may have been updated since the
start
- * of the query and thus may not put into the cache.
- */
- private class QueryContext {
-
- private static final double FPP = 0.01d;
- private static final int ENTRIES_SCOPED = 1000;
- private static final int ENTRIES_OPEN = 10000;
-
- private final String fromKey, toKey;
- private volatile BloomFilter<String> filter = null;
-
- private BloomFilter<String> getFilter() {
- if (filter == null) {
- synchronized (this) {
- if (filter == null) {
- filter = BloomFilter.create(new Funnel<String>() {
- private static final long serialVersionUID =
-7114267990225941161L;
-
- @Override
- public void funnel(String from, PrimitiveSink
into) {
- into.putUnencodedChars(from);
- }
- }, toKey.equals(NodeDocument.MAX_ID_VALUE) ?
ENTRIES_OPEN : ENTRIES_SCOPED, FPP);
- }
- }
- }
- return filter;
- }
-
- public QueryContext(String fromKey, String toKey) {
- this.fromKey = fromKey;
- this.toKey = toKey;
- }
-
- synchronized public void addKey(String key) {
- if (fromKey.compareTo(key) < 0 && toKey.compareTo(key) > 0) {
- getFilter().put(key);
- }
- }
-
- synchronized public void addKeys(List<String> keys) {
- for (String key: keys) {
- addKey(key);
- }
- }
-
- synchronized public boolean mayUpdate(String key) {
- return filter == null ? true : !getFilter().mightContain(key);
- }
-
- synchronized public void dispose() {
- if (LOG.isDebugEnabled()) {
- if (filter != null) {
- LOG.debug("Disposing QueryContext for range " + fromKey +
"..." + toKey + " - filter fpp was: "
- + filter.expectedFpp());
- } else {
- LOG.debug("Disposing QueryContext for range " + fromKey +
"..." + toKey + " - no filter was needed");
- }
- }
- }
- }
-
- private Map<Thread, QueryContext> qmap = new ConcurrentHashMap<Thread,
QueryContext>();
-
private <T extends Document> List<T> internalQuery(Collection<T>
collection, String fromKey, String toKey,
List<String> excludeKeyPatterns, List<QueryCondition> conditions,
int limit) {
Connection connection = null;
@@ -1477,13 +1400,12 @@ public class RDBDocumentStore implements
final Stopwatch watch = startWatch();
int resultSize = 0;
+ CacheChangesTracker tracker = null;
try {
- long now = System.currentTimeMillis();
- QueryContext qp = null;
if (collection == Collection.NODES) {
- qp = new QueryContext(fromKey, toKey);
- qmap.put(Thread.currentThread(), qp);
+ tracker = nodesCache.registerTracker(fromKey, toKey);
}
+ long now = System.currentTimeMillis();
connection = this.ch.getROConnection();
String from = collection == Collection.NODES &&
NodeDocument.MIN_ID_VALUE.equals(fromKey) ? null : fromKey;
String to = collection == Collection.NODES &&
NodeDocument.MAX_ID_VALUE.equals(toKey) ? null : toKey;
@@ -1493,20 +1415,47 @@ public class RDBDocumentStore implements
int size = dbresult.size();
List<T> result = new ArrayList<T>(size);
for (int i = 0; i < size; i++) {
- RDBRow row = dbresult.set(i, null); // free RDBRow ASAP
- T doc = runThroughCache(collection, row, now, qp);
+ // free RDBRow as early as possible
+ RDBRow row = dbresult.set(i, null);
+ T doc = getIfCached(collection, row.getId(),
row.getModcount());
+ if (doc == null) {
+ // parse DB contents into document if and only if it's not
+ // already in the cache
+ doc = convertFromDBObject(collection, row);
+ } else {
+ // we got a document from the cache, thus collection is
NODES
+ // and a tracker is present
+ long lastmodified = modifiedOf(doc);
+ if (lastmodified == row.getModified() && lastmodified >=
1) {
+ Lock lock = locks.acquire(row.getId());
+ try {
+ if (!tracker.mightBeenAffected(row.getId())) {
+ // otherwise mark it as fresh
+ ((NodeDocument) doc).markUpToDate(now);
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+ else {
+ // we need a fresh document instance
+ doc = convertFromDBObject(collection, row);
+ }
+ }
result.add(doc);
}
- resultSize = result.size();
- if (qp != null) {
- qp.dispose();
+ if (collection == Collection.NODES) {
+ nodesCache.putNonConflictingDocs(tracker,
castAsNodeDocumentList(result));
}
+ resultSize = result.size();
return result;
} catch (Exception ex) {
LOG.error("SQL exception on query", ex);
throw new DocumentStoreException(ex);
} finally {
- qmap.remove(Thread.currentThread());
+ if (tracker != null) {
+ tracker.close();
+ }
this.ch.closeConnection(connection);
stats.doneQuery(watch.elapsed(TimeUnit.NANOSECONDS), collection,
fromKey, toKey,
!conditions.isEmpty(), resultSize, -1, false);
@@ -1844,6 +1793,11 @@ public class RDBDocumentStore implements
return (T) doc;
}
+ @SuppressWarnings("unchecked")
+ private static <T extends Document> List<NodeDocument>
castAsNodeDocumentList(List<T> list) {
+ return (List<NodeDocument>) list;
+ }
+
private NodeDocumentCache nodesCache;
private NodeDocumentLocks locks;
@@ -1883,54 +1837,6 @@ public class RDBDocumentStore implements
return ser.fromRow(collection, row);
}
- private <T extends Document> T runThroughCache(Collection<T> collection,
RDBRow row, long now, QueryContext qp) {
-
- if (collection != Collection.NODES) {
- // not in the cache anyway
- return convertFromDBObject(collection, row);
- }
-
- String id = row.getId();
- NodeDocument inCache = nodesCache.getIfPresent(id);
- Long modCount = row.getModcount();
-
- // do not overwrite document in cache if the
- // existing one in the cache is newer
- if (inCache != null && inCache != NodeDocument.NULL) {
- // check mod count
- Long cachedModCount = inCache.getModCount();
- if (cachedModCount == null) {
- throw new IllegalStateException("Missing " +
Document.MOD_COUNT);
- }
- long lastmodified = modifiedOf(inCache);
- if (lastmodified == row.getModified() && lastmodified >= 1) {
- if (modCount <= cachedModCount) {
- // we can use the cached document
- Lock lock = locks.acquire(row.getId());
- try {
- if (qp.mayUpdate(id)) {
- inCache.markUpToDate(now);
- }
- }
- finally {
- lock.unlock();
- }
- return castAsT(inCache);
- }
- }
- }
-
- NodeDocument fresh = (NodeDocument) convertFromDBObject(collection,
row);
- fresh.seal();
-
- if (!qp.mayUpdate(id)) {
- return castAsT(fresh);
- }
-
- nodesCache.putIfNewer(fresh);
- return castAsT(fresh);
- }
-
private static boolean hasChangesToCollisions(UpdateOp update) {
if (!USECMODCOUNT) {
return false;
