X-Git-Url: https://gerrit.simantics.org/r/gitweb?p=simantics%2Fplatform.git;a=blobdiff_plain;f=bundles%2Forg.simantics.db.impl%2Fsrc%2Forg%2Fsimantics%2Fdb%2Fimpl%2Fquery%2FQueryCache.java;h=35ebdbc65ba4ce82e5e7e7dced301310c30f8b17;hp=57c1c68c76df0ca930a4655d40e83e18b1ddc9db;hb=90a52d58489b6c518b13dde238923125c8f63eec;hpb=782843a248c4f70d28e8146098f9a77977cc6b3c diff --git a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryCache.java b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryCache.java index 57c1c68c7..35ebdbc65 100644 --- a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryCache.java +++ b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryCache.java @@ -19,6 +19,8 @@ import org.simantics.db.request.Read; public class QueryCache extends QueryCacheBase { + private static final boolean SINGLE = true; + public QueryCache(QuerySupport querySupport, int threads) { super(querySupport, threads); } @@ -29,18 +31,19 @@ public class QueryCache extends QueryCacheBase { existing = (Objects)objectsMap.get(r1,r2); if(existing == null) { existing = new Objects(r1,r2); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); objectsMap.put(keyR2(r1,r2), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } - if(existing.isPending()) waitPending(graph, existing); + if(existing.isPending()) { + waitPending(graph, existing); + } return existing; } @@ -53,17 +56,32 @@ public class QueryCache extends QueryCacheBase { public static void runnerObjects(ReadGraphImpl graph, int r1, int r2, CacheEntry parent, ListenerBase listener, final IntProcedure procedure) throws DatabaseException { QueryCache cache = graph.processor.cache; if(parent == null && listener == null && !cache.shouldCache(graph.processor, r1,r2)) { + if (SINGLE) { + Objects e = cache.peekObjects(r1,r2); + if (e != null && e.isReady()) { + e.performFromCache(graph, procedure); + return; + } + } Objects.computeForEach(graph, r1,r2, null, procedure); return; } Objects entry = (Objects)cache.getOrCreateObjects(graph, r1,r2); IntProcedure procedure_ = procedure != null ? procedure : emptyProcedureObjects; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + entry.performFromCache(graph, procedure_); + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); Objects.computeForEach(graph, r1,r2, entry, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + } + } + + private Objects peekObjects(int r1, int r2) { + synchronized(objectsMap) { + return (Objects) objectsMap.get(r1,r2); } } @@ -73,18 +91,19 @@ public class QueryCache extends QueryCacheBase { existing = (Statements)statementsMap.get(r1,r2); if(existing == null) { existing = new Statements(r1,r2); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); statementsMap.put(keyR2(r1,r2), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } - if(existing.isPending()) waitPending(graph, existing); + if(existing.isPending()) { + waitPending(graph, existing); + } return existing; } @@ -97,17 +116,32 @@ public class QueryCache extends QueryCacheBase { public static void runnerStatements(ReadGraphImpl graph, int r1, int r2, CacheEntry parent, ListenerBase listener, final TripleIntProcedure procedure) throws DatabaseException { QueryCache cache = graph.processor.cache; if(parent == null && listener == null && !cache.shouldCache(graph.processor, r1,r2)) { + if (SINGLE) { + Statements e = cache.peekStatements(r1,r2); + if (e != null && e.isReady()) { + e.performFromCache(graph, procedure); + return; + } + } Statements.computeForEach(graph, r1,r2, null, procedure); return; } Statements entry = (Statements)cache.getOrCreateStatements(graph, r1,r2); TripleIntProcedure procedure_ = procedure != null ? procedure : emptyProcedureStatements; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + entry.performFromCache(graph, procedure_); + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); Statements.computeForEach(graph, r1,r2, entry, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + } + } + + private Statements peekStatements(int r1, int r2) { + synchronized(statementsMap) { + return (Statements) statementsMap.get(r1,r2); } } @@ -117,18 +151,19 @@ public class QueryCache extends QueryCacheBase { existing = (DirectObjects)directObjectsMap.get(r1,r2); if(existing == null) { existing = new DirectObjects(r1,r2); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); directObjectsMap.put(keyR2(r1,r2), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } - if(existing.isPending()) waitPending(graph, existing); + if(existing.isPending()) { + waitPending(graph, existing); + } return existing; } @@ -141,17 +176,32 @@ public class QueryCache extends QueryCacheBase { public static void runnerDirectObjects(ReadGraphImpl graph, int r1, int r2, CacheEntry parent, ListenerBase listener, final IntProcedure procedure) throws DatabaseException { QueryCache cache = graph.processor.cache; if(parent == null && listener == null && !cache.shouldCache(graph.processor, r1,r2)) { + if (SINGLE) { + DirectObjects e = cache.peekDirectObjects(r1,r2); + if (e != null && e.isReady()) { + e.performFromCache(graph, procedure); + return; + } + } DirectObjects.computeForEach(graph, r1,r2, null, procedure); return; } DirectObjects entry = (DirectObjects)cache.getOrCreateDirectObjects(graph, r1,r2); IntProcedure procedure_ = procedure != null ? procedure : emptyProcedureDirectObjects; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + entry.performFromCache(graph, procedure_); + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); DirectObjects.computeForEach(graph, r1,r2, entry, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + } + } + + private DirectObjects peekDirectObjects(int r1, int r2) { + synchronized(directObjectsMap) { + return (DirectObjects) directObjectsMap.get(r1,r2); } } @@ -161,18 +211,19 @@ public class QueryCache extends QueryCacheBase { existing = (RelationInfoQuery)relationInfoQueryMap.get(r); if(existing == null) { existing = new RelationInfoQuery(r); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); relationInfoQueryMap.put(keyR(r), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } - if(existing.isPending()) waitPending(graph, existing); + if(existing.isPending()) { + waitPending(graph, existing); + } return existing; } @@ -185,17 +236,32 @@ public class QueryCache extends QueryCacheBase { public static void runnerRelationInfoQuery(ReadGraphImpl graph, int r, CacheEntry parent, ListenerBase listener, final InternalProcedure procedure) throws DatabaseException { QueryCache cache = graph.processor.cache; if(parent == null && listener == null && !cache.shouldCache(graph.processor, r)) { + if (SINGLE) { + RelationInfoQuery e = cache.peekRelationInfoQuery(r); + if (e != null && e.isReady()) { + e.performFromCache(graph, procedure); + return; + } + } RelationInfoQuery.computeForEach(graph, r, null, procedure); return; } RelationInfoQuery entry = (RelationInfoQuery)cache.getOrCreateRelationInfoQuery(graph, r); InternalProcedure procedure_ = procedure != null ? procedure : emptyProcedureRelationInfoQuery; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + entry.performFromCache(graph, procedure_); + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); RelationInfoQuery.computeForEach(graph, r, entry, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + } + } + + private RelationInfoQuery peekRelationInfoQuery(int r) { + synchronized(relationInfoQueryMap) { + return (RelationInfoQuery) relationInfoQueryMap.get(r); } } @@ -205,18 +271,19 @@ public class QueryCache extends QueryCacheBase { existing = (URIToResource)uRIToResourceMap.get(id); if(existing == null) { existing = new URIToResource(id); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); uRIToResourceMap.put(keyID(id), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } - if(existing.isPending()) waitPending(graph, existing); + if(existing.isPending()) { + waitPending(graph, existing); + } return existing; } @@ -229,17 +296,32 @@ public class QueryCache extends QueryCacheBase { public static void runnerURIToResource(ReadGraphImpl graph, String id, CacheEntry parent, ListenerBase listener, final InternalProcedure procedure) throws DatabaseException { QueryCache cache = graph.processor.cache; if(parent == null && listener == null && !cache.shouldCache(graph.processor, id)) { + if (SINGLE) { + URIToResource e = cache.peekURIToResource(id); + if (e != null && e.isReady()) { + e.performFromCache(graph, procedure); + return; + } + } URIToResource.computeForEach(graph, id, null, procedure); return; } URIToResource entry = (URIToResource)cache.getOrCreateURIToResource(graph, id); InternalProcedure procedure_ = procedure != null ? procedure : emptyProcedureURIToResource; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + entry.performFromCache(graph, procedure_); + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); URIToResource.computeForEach(graph, id, entry, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + } + } + + private URIToResource peekURIToResource(String id) { + synchronized(uRIToResourceMap) { + return (URIToResource) uRIToResourceMap.get(id); } } @@ -249,18 +331,19 @@ public class QueryCache extends QueryCacheBase { existing = (ValueQuery)valueQueryMap.get(r); if(existing == null) { existing = new ValueQuery(r); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); valueQueryMap.put(keyR(r), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } - if(existing.isPending()) waitPending(graph, existing); + if(existing.isPending()) { + waitPending(graph, existing); + } return existing; } @@ -273,17 +356,32 @@ public class QueryCache extends QueryCacheBase { public static void runnerValueQuery(ReadGraphImpl graph, int r, CacheEntry parent, ListenerBase listener, final InternalProcedure procedure) throws DatabaseException { QueryCache cache = graph.processor.cache; if(parent == null && listener == null && !cache.shouldCache(graph.processor, r)) { + if (SINGLE) { + ValueQuery e = cache.peekValueQuery(r); + if (e != null && e.isReady()) { + e.performFromCache(graph, procedure); + return; + } + } ValueQuery.computeForEach(graph, r, null, procedure); return; } ValueQuery entry = (ValueQuery)cache.getOrCreateValueQuery(graph, r); InternalProcedure procedure_ = procedure != null ? procedure : emptyProcedureValueQuery; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + entry.performFromCache(graph, procedure_); + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); ValueQuery.computeForEach(graph, r, entry, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + } + } + + private ValueQuery peekValueQuery(int r) { + synchronized(valueQueryMap) { + return (ValueQuery) valueQueryMap.get(r); } } @@ -293,18 +391,19 @@ public class QueryCache extends QueryCacheBase { existing = (OrderedSet)orderedSetMap.get(r); if(existing == null) { existing = new OrderedSet(r); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); orderedSetMap.put(keyR(r), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } - if(existing.isPending()) waitPending(graph, existing); + if(existing.isPending()) { + waitPending(graph, existing); + } return existing; } @@ -317,17 +416,32 @@ public class QueryCache extends QueryCacheBase { public static void runnerOrderedSet(ReadGraphImpl graph, int r, CacheEntry parent, ListenerBase listener, final IntProcedure procedure) throws DatabaseException { QueryCache cache = graph.processor.cache; if(parent == null && listener == null && !cache.shouldCache(graph.processor, r)) { + if (SINGLE) { + OrderedSet e = cache.peekOrderedSet(r); + if (e != null && e.isReady()) { + e.performFromCache(graph, procedure); + return; + } + } OrderedSet.computeForEach(graph, r, null, procedure); return; } OrderedSet entry = (OrderedSet)cache.getOrCreateOrderedSet(graph, r); IntProcedure procedure_ = procedure != null ? procedure : emptyProcedureOrderedSet; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + entry.performFromCache(graph, procedure_); + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); OrderedSet.computeForEach(graph, r, entry, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + } + } + + private OrderedSet peekOrderedSet(int r) { + synchronized(orderedSetMap) { + return (OrderedSet) orderedSetMap.get(r); } } @@ -337,18 +451,19 @@ public class QueryCache extends QueryCacheBase { existing = (PrincipalTypes)principalTypesMap.get(r); if(existing == null) { existing = new PrincipalTypes(r); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); principalTypesMap.put(keyR(r), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } - if(existing.isPending()) waitPending(graph, existing); + if(existing.isPending()) { + waitPending(graph, existing); + } return existing; } @@ -361,17 +476,32 @@ public class QueryCache extends QueryCacheBase { public static void runnerPrincipalTypes(ReadGraphImpl graph, int r, CacheEntry parent, ListenerBase listener, final IntProcedure procedure) throws DatabaseException { QueryCache cache = graph.processor.cache; if(parent == null && listener == null && !cache.shouldCache(graph.processor, r)) { + if (SINGLE) { + PrincipalTypes e = cache.peekPrincipalTypes(r); + if (e != null && e.isReady()) { + e.performFromCache(graph, procedure); + return; + } + } PrincipalTypes.computeForEach(graph, r, null, procedure); return; } PrincipalTypes entry = (PrincipalTypes)cache.getOrCreatePrincipalTypes(graph, r); IntProcedure procedure_ = procedure != null ? procedure : emptyProcedurePrincipalTypes; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + entry.performFromCache(graph, procedure_); + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); PrincipalTypes.computeForEach(graph, r, entry, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + } + } + + private PrincipalTypes peekPrincipalTypes(int r) { + synchronized(principalTypesMap) { + return (PrincipalTypes) principalTypesMap.get(r); } } @@ -381,18 +511,19 @@ public class QueryCache extends QueryCacheBase { existing = (DirectPredicates)directPredicatesMap.get(r); if(existing == null) { existing = new DirectPredicates(r); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); directPredicatesMap.put(keyR(r), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } - if(existing.isPending()) waitPending(graph, existing); + if(existing.isPending()) { + waitPending(graph, existing); + } return existing; } @@ -405,17 +536,32 @@ public class QueryCache extends QueryCacheBase { public static void runnerDirectPredicates(ReadGraphImpl graph, int r, CacheEntry parent, ListenerBase listener, final InternalProcedure procedure) throws DatabaseException { QueryCache cache = graph.processor.cache; if(parent == null && listener == null && !cache.shouldCache(graph.processor, r)) { + if (SINGLE) { + DirectPredicates e = cache.peekDirectPredicates(r); + if (e != null && e.isReady()) { + e.performFromCache(graph, procedure); + return; + } + } DirectPredicates.computeForEach(graph, r, null, procedure); return; } DirectPredicates entry = (DirectPredicates)cache.getOrCreateDirectPredicates(graph, r); InternalProcedure procedure_ = procedure != null ? procedure : emptyProcedureDirectPredicates; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + entry.performFromCache(graph, procedure_); + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); DirectPredicates.computeForEach(graph, r, entry, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + } + } + + private DirectPredicates peekDirectPredicates(int r) { + synchronized(directPredicatesMap) { + return (DirectPredicates) directPredicatesMap.get(r); } } @@ -425,18 +571,19 @@ public class QueryCache extends QueryCacheBase { existing = (Predicates)predicatesMap.get(r); if(existing == null) { existing = new Predicates(r); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); predicatesMap.put(keyR(r), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } - if(existing.isPending()) waitPending(graph, existing); + if(existing.isPending()) { + waitPending(graph, existing); + } return existing; } @@ -449,158 +596,215 @@ public class QueryCache extends QueryCacheBase { public static void runnerPredicates(ReadGraphImpl graph, int r, CacheEntry parent, ListenerBase listener, final InternalProcedure procedure) throws DatabaseException { QueryCache cache = graph.processor.cache; if(parent == null && listener == null && !cache.shouldCache(graph.processor, r)) { + if (SINGLE) { + Predicates e = cache.peekPredicates(r); + if (e != null && e.isReady()) { + e.performFromCache(graph, procedure); + return; + } + } Predicates.computeForEach(graph, r, null, procedure); return; } Predicates entry = (Predicates)cache.getOrCreatePredicates(graph, r); InternalProcedure procedure_ = procedure != null ? procedure : emptyProcedurePredicates; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + entry.performFromCache(graph, procedure_); + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); Predicates.computeForEach(graph, r, entry, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); } } - ReadEntry getOrCreateReadEntry(ReadGraphImpl graph, Read r, boolean needsToBlock) throws DatabaseException { + private Predicates peekPredicates(int r) { + synchronized(predicatesMap) { + return (Predicates) predicatesMap.get(r); + } + } + + private final ReadEntry getOrCreateReadEntry(ReadGraphImpl graph, Read r, boolean needsToBlock) throws DatabaseException { ReadEntry existing = null; synchronized(readEntryMap) { existing = (ReadEntry)readEntryMap.get(r); if(existing == null) { existing = new ReadEntry(r); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); readEntryMap.put(id(r), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } if(existing.isPending()) { - if(needsToBlock) waitPending(graph, existing); - else return null; + if(needsToBlock) + waitPending(graph, existing); + else { + return null; + } } return existing; } void remove(ReadEntry entry) { synchronized(readEntryMap) { - readEntryMap.remove(entry.request); + readEntryMap.remove(entry.id); } } - public static Object runnerReadEntry(ReadGraphImpl graph, Read r, CacheEntry parent, ListenerBase listener, final AsyncProcedure procedure, final boolean needsToBlock) throws DatabaseException { + public static Object runnerReadEntry(ReadGraphImpl graph, Read r, CacheEntry parent, ListenerBase listener, final AsyncProcedure procedure, boolean needsToBlock) throws DatabaseException { QueryCache cache = graph.processor.cache; if(parent == null && listener == null && !cache.shouldCache(graph.processor, r)) { - return ReadEntry.computeForEach(graph, r, null, procedure); + if (SINGLE) { + ReadEntry e = cache.peekReadEntry(r); + if (e != null && e.isReady()) { + return e.performFromCache(graph, procedure); + } + } + return ReadEntry.computeForEach(graph, r, null, procedure, needsToBlock); } ReadEntry entry = (ReadEntry)cache.getOrCreateReadEntry(graph, r, needsToBlock); if(entry == null) { - graph.processor.schedule(new SessionTask(graph) { - @Override - public void run(int thread) { - try { - runnerReadEntry(graph, r, parent, listener, procedure, needsToBlock); - } catch (DatabaseException e) { - Logger.defaultLogError(e); - } - } - }); - return null; + graph.asyncBarrier.inc(); + graph.processor.scheduleNow(new SessionTask() { + @Override + public void run0(int thread) { + try { + runnerReadEntry(graph, r, parent, listener, procedure, needsToBlock); + graph.asyncBarrier.dec(); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } + } + }); + return null; } AsyncProcedure procedure_ = procedure != null ? procedure : emptyProcedureReadEntry; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) return entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + Object result = entry.performFromCache(graph, procedure_); + graph.processor.listening.registerFirstKnown(listener, result); + return result; + } else { assert(entry.isPending()); - Object result = ReadEntry.computeForEach(graph, r, entry, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + Object result = ReadEntry.computeForEach(graph, r, entry, procedure_, needsToBlock); + graph.processor.listening.registerFirstKnown(listener, result); return result; } } - AsyncReadEntry getOrCreateAsyncReadEntry(ReadGraphImpl graph, AsyncRead r, boolean needsToBlock) throws DatabaseException { + private ReadEntry peekReadEntry(Read r) { + synchronized(readEntryMap) { + return (ReadEntry) readEntryMap.get(r); + } + } + + AsyncReadEntry getOrCreateAsyncReadEntry(ReadGraphImpl graph, AsyncRead r, CacheEntry parent, ListenerBase listener, final AsyncProcedure procedure, boolean needsToBlock) throws DatabaseException { AsyncReadEntry existing = null; synchronized(asyncReadEntryMap) { existing = (AsyncReadEntry)asyncReadEntryMap.get(r); if(existing == null) { existing = new AsyncReadEntry(r); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); asyncReadEntryMap.put(id(r), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } if(existing.isPending()) { - if(needsToBlock) waitPending(graph, existing); - else return null; + if(needsToBlock) + waitPending(graph, existing); + else { + existing.executeWhenResultIsAvailable(graph.processor, new SessionTask(graph) { + @Override + public void run0(int thread) { + try { + runnerAsyncReadEntry(graph, r, parent, listener, procedure, needsToBlock); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } + } + }); + return null; + } } return existing; } void remove(AsyncReadEntry entry) { synchronized(asyncReadEntryMap) { - asyncReadEntryMap.remove(entry.request); + asyncReadEntryMap.remove(entry.id); } } - public static Object runnerAsyncReadEntry(ReadGraphImpl graph, AsyncRead r, CacheEntry parent, ListenerBase listener, final AsyncProcedure procedure, final boolean needsToBlock) throws DatabaseException { + public static Object runnerAsyncReadEntry(ReadGraphImpl graph, AsyncRead r, CacheEntry parent, ListenerBase listener, final AsyncProcedure procedure, boolean needsToBlock) throws DatabaseException { QueryCache cache = graph.processor.cache; if(parent == null && listener == null && !cache.shouldCache(graph.processor, r)) { + if (SINGLE) { + AsyncReadEntry e = cache.peekAsyncReadEntry(r); + if (e != null && e.isReady()) { + return e.performFromCache(graph, procedure); + } + } return AsyncReadEntry.computeForEach(graph, r, null, procedure, needsToBlock); } - AsyncReadEntry entry = (AsyncReadEntry)cache.getOrCreateAsyncReadEntry(graph, r, needsToBlock); + AsyncReadEntry entry = (AsyncReadEntry)cache.getOrCreateAsyncReadEntry(graph, r, parent, listener, procedure, needsToBlock); if(entry == null) { - graph.processor.schedule(new SessionTask(graph) { - @Override - public void run(int thread) { - try { - runnerAsyncReadEntry(graph, r, parent, listener, procedure, needsToBlock); - } catch (DatabaseException e) { - Logger.defaultLogError(e); - } - } - }); + // Entry was pending and this request has been queued return null; } AsyncProcedure procedure_ = procedure != null ? procedure : emptyProcedureAsyncReadEntry; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) return entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + Object result = entry.performFromCache(graph, procedure_); + graph.processor.listening.registerFirstKnown(listener, result); + return result; + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); Object result = AsyncReadEntry.computeForEach(graph, r, entry, procedure_, needsToBlock); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + graph.processor.listening.registerFirstKnown(listener, result); return result; } } + private AsyncReadEntry peekAsyncReadEntry(AsyncRead r) { + synchronized(asyncReadEntryMap) { + return (AsyncReadEntry) asyncReadEntryMap.get(r); + } + } + Types getOrCreateTypes(ReadGraphImpl graph, int r) throws DatabaseException { Types existing = null; synchronized(typesMap) { existing = (Types)typesMap.get(r); if(existing == null) { existing = new Types(r); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); typesMap.put(keyR(r), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } - if(existing.isPending()) waitPending(graph, existing); + if(existing.isPending()) { + waitPending(graph, existing); + } return existing; } @@ -613,17 +817,32 @@ public class QueryCache extends QueryCacheBase { public static void runnerTypes(ReadGraphImpl graph, int r, CacheEntry parent, ListenerBase listener, final InternalProcedure procedure) throws DatabaseException { QueryCache cache = graph.processor.cache; if(parent == null && listener == null && !cache.shouldCache(graph.processor, r)) { + if (SINGLE) { + Types e = cache.peekTypes(r); + if (e != null && e.isReady()) { + e.performFromCache(graph, procedure); + return; + } + } Types.computeForEach(graph, r, null, procedure); return; } Types entry = (Types)cache.getOrCreateTypes(graph, r); InternalProcedure procedure_ = procedure != null ? procedure : emptyProcedureTypes; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + entry.performFromCache(graph, procedure_); + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); Types.computeForEach(graph, r, entry, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + } + } + + private Types peekTypes(int r) { + synchronized(typesMap) { + return (Types) typesMap.get(r); } } @@ -633,18 +852,19 @@ public class QueryCache extends QueryCacheBase { existing = (ChildMap)childMapMap.get(r); if(existing == null) { existing = new ChildMap(r); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); childMapMap.put(keyR(r), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } - if(existing.isPending()) waitPending(graph, existing); + if(existing.isPending()) { + waitPending(graph, existing); + } return existing; } @@ -657,17 +877,32 @@ public class QueryCache extends QueryCacheBase { public static void runnerChildMap(ReadGraphImpl graph, int r, CacheEntry parent, ListenerBase listener, final InternalProcedure> procedure) throws DatabaseException { QueryCache cache = graph.processor.cache; if(parent == null && listener == null && !cache.shouldCache(graph.processor, r)) { + if (SINGLE) { + ChildMap e = cache.peekChildMap(r); + if (e != null && e.isReady()) { + e.performFromCache(graph, procedure); + return; + } + } ChildMap.computeForEach(graph, r, null, procedure); return; } ChildMap entry = (ChildMap)cache.getOrCreateChildMap(graph, r); InternalProcedure> procedure_ = procedure != null ? procedure : emptyProcedureChildMap; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + entry.performFromCache(graph, procedure_); + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); ChildMap.computeForEach(graph, r, entry, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + } + } + + private ChildMap peekChildMap(int r) { + synchronized(childMapMap) { + return (ChildMap) childMapMap.get(r); } } @@ -677,18 +912,19 @@ public class QueryCache extends QueryCacheBase { existing = (TypeHierarchy)typeHierarchyMap.get(r); if(existing == null) { existing = new TypeHierarchy(r); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); typeHierarchyMap.put(keyR(r), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } - if(existing.isPending()) waitPending(graph, existing); + if(existing.isPending()) { + waitPending(graph, existing); + } return existing; } @@ -701,17 +937,32 @@ public class QueryCache extends QueryCacheBase { public static void runnerTypeHierarchy(ReadGraphImpl graph, int r, CacheEntry parent, ListenerBase listener, final InternalProcedure procedure) throws DatabaseException { QueryCache cache = graph.processor.cache; if(parent == null && listener == null && !cache.shouldCache(graph.processor, r)) { + if (SINGLE) { + TypeHierarchy e = cache.peekTypeHierarchy(r); + if (e != null && e.isReady()) { + e.performFromCache(graph, procedure); + return; + } + } TypeHierarchy.computeForEach(graph, r, null, procedure); return; } TypeHierarchy entry = (TypeHierarchy)cache.getOrCreateTypeHierarchy(graph, r); InternalProcedure procedure_ = procedure != null ? procedure : emptyProcedureTypeHierarchy; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + entry.performFromCache(graph, procedure_); + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); TypeHierarchy.computeForEach(graph, r, entry, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + } + } + + private TypeHierarchy peekTypeHierarchy(int r) { + synchronized(typeHierarchyMap) { + return (TypeHierarchy) typeHierarchyMap.get(r); } } @@ -721,18 +972,19 @@ public class QueryCache extends QueryCacheBase { existing = (SuperTypes)superTypesMap.get(r); if(existing == null) { existing = new SuperTypes(r); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); superTypesMap.put(keyR(r), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } - if(existing.isPending()) waitPending(graph, existing); + if(existing.isPending()) { + waitPending(graph, existing); + } return existing; } @@ -745,17 +997,32 @@ public class QueryCache extends QueryCacheBase { public static void runnerSuperTypes(ReadGraphImpl graph, int r, CacheEntry parent, ListenerBase listener, final InternalProcedure procedure) throws DatabaseException { QueryCache cache = graph.processor.cache; if(parent == null && listener == null && !cache.shouldCache(graph.processor, r)) { + if (SINGLE) { + SuperTypes e = cache.peekSuperTypes(r); + if (e != null && e.isReady()) { + e.performFromCache(graph, procedure); + return; + } + } SuperTypes.computeForEach(graph, r, null, procedure); return; } SuperTypes entry = (SuperTypes)cache.getOrCreateSuperTypes(graph, r); InternalProcedure procedure_ = procedure != null ? procedure : emptyProcedureSuperTypes; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + entry.performFromCache(graph, procedure_); + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); SuperTypes.computeForEach(graph, r, entry, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + } + } + + private SuperTypes peekSuperTypes(int r) { + synchronized(superTypesMap) { + return (SuperTypes) superTypesMap.get(r); } } @@ -765,18 +1032,19 @@ public class QueryCache extends QueryCacheBase { existing = (SuperRelations)superRelationsMap.get(r); if(existing == null) { existing = new SuperRelations(r); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); superRelationsMap.put(keyR(r), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } - if(existing.isPending()) waitPending(graph, existing); + if(existing.isPending()) { + waitPending(graph, existing); + } return existing; } @@ -789,17 +1057,32 @@ public class QueryCache extends QueryCacheBase { public static void runnerSuperRelations(ReadGraphImpl graph, int r, CacheEntry parent, ListenerBase listener, final InternalProcedure procedure) throws DatabaseException { QueryCache cache = graph.processor.cache; if(parent == null && listener == null && !cache.shouldCache(graph.processor, r)) { + if (SINGLE) { + SuperRelations e = cache.peekSuperRelations(r); + if (e != null && e.isReady()) { + e.performFromCache(graph, procedure); + return; + } + } SuperRelations.computeForEach(graph, r, null, procedure); return; } SuperRelations entry = (SuperRelations)cache.getOrCreateSuperRelations(graph, r); InternalProcedure procedure_ = procedure != null ? procedure : emptyProcedureSuperRelations; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + entry.performFromCache(graph, procedure_); + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); SuperRelations.computeForEach(graph, r, entry, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + } + } + + private SuperRelations peekSuperRelations(int r) { + synchronized(superRelationsMap) { + return (SuperRelations) superRelationsMap.get(r); } } @@ -809,18 +1092,19 @@ public class QueryCache extends QueryCacheBase { existing = (AssertedPredicates)assertedPredicatesMap.get(r); if(existing == null) { existing = new AssertedPredicates(r); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); assertedPredicatesMap.put(keyR(r), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } - if(existing.isPending()) waitPending(graph, existing); + if(existing.isPending()) { + waitPending(graph, existing); + } return existing; } @@ -834,12 +1118,20 @@ public class QueryCache extends QueryCacheBase { QueryCache cache = graph.processor.cache; AssertedPredicates entry = (AssertedPredicates)cache.getOrCreateAssertedPredicates(graph, r); IntProcedure procedure_ = procedure != null ? procedure : emptyProcedureAssertedPredicates; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + entry.performFromCache(graph, procedure_); + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); entry.compute(graph, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + } + } + + private AssertedPredicates peekAssertedPredicates(int r) { + synchronized(assertedPredicatesMap) { + return (AssertedPredicates) assertedPredicatesMap.get(r); } } @@ -849,18 +1141,19 @@ public class QueryCache extends QueryCacheBase { existing = (AssertedStatements)assertedStatementsMap.get(r1,r2); if(existing == null) { existing = new AssertedStatements(r1,r2); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); assertedStatementsMap.put(keyR2(r1,r2), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } - if(existing.isPending()) waitPending(graph, existing); + if(existing.isPending()) { + waitPending(graph, existing); + } return existing; } @@ -874,12 +1167,20 @@ public class QueryCache extends QueryCacheBase { QueryCache cache = graph.processor.cache; AssertedStatements entry = (AssertedStatements)cache.getOrCreateAssertedStatements(graph, r1,r2); TripleIntProcedure procedure_ = procedure != null ? procedure : emptyProcedureAssertedStatements; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + entry.performFromCache(graph, procedure_); + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); entry.compute(graph, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + } + } + + private AssertedStatements peekAssertedStatements(int r1, int r2) { + synchronized(assertedStatementsMap) { + return (AssertedStatements) assertedStatementsMap.get(r1,r2); } } @@ -889,18 +1190,19 @@ public class QueryCache extends QueryCacheBase { existing = (DirectSuperRelations)directSuperRelationsMap.get(r); if(existing == null) { existing = new DirectSuperRelations(r); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); directSuperRelationsMap.put(keyR(r), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } - if(existing.isPending()) waitPending(graph, existing); + if(existing.isPending()) { + waitPending(graph, existing); + } return existing; } @@ -914,12 +1216,20 @@ public class QueryCache extends QueryCacheBase { QueryCache cache = graph.processor.cache; DirectSuperRelations entry = (DirectSuperRelations)cache.getOrCreateDirectSuperRelations(graph, r); IntProcedure procedure_ = procedure != null ? procedure : emptyProcedureDirectSuperRelations; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + entry.performFromCache(graph, procedure_); + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); entry.compute(graph, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + } + } + + private DirectSuperRelations peekDirectSuperRelations(int r) { + synchronized(directSuperRelationsMap) { + return (DirectSuperRelations) directSuperRelationsMap.get(r); } } @@ -929,24 +1239,25 @@ public class QueryCache extends QueryCacheBase { existing = (MultiReadEntry)multiReadEntryMap.get(r); if(existing == null) { existing = new MultiReadEntry(r); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); multiReadEntryMap.put(id(r), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } - if(existing.isPending()) waitPending(graph, existing); + if(existing.isPending()) { + waitPending(graph, existing); + } return existing; } void remove(MultiReadEntry entry) { synchronized(multiReadEntryMap) { - multiReadEntryMap.remove(entry.request); + multiReadEntryMap.remove(entry.id); } } @@ -954,12 +1265,20 @@ public class QueryCache extends QueryCacheBase { QueryCache cache = graph.processor.cache; MultiReadEntry entry = (MultiReadEntry)cache.getOrCreateMultiReadEntry(graph, r); SyncMultiProcedure procedure_ = procedure != null ? procedure : emptyProcedureMultiReadEntry; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + entry.performFromCache(graph, procedure_); + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); entry.compute(graph, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + } + } + + private MultiReadEntry peekMultiReadEntry(MultiRead r) { + synchronized(multiReadEntryMap) { + return (MultiReadEntry) multiReadEntryMap.get(r); } } @@ -969,24 +1288,25 @@ public class QueryCache extends QueryCacheBase { existing = (AsyncMultiReadEntry)asyncMultiReadEntryMap.get(r); if(existing == null) { existing = new AsyncMultiReadEntry(r); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); asyncMultiReadEntryMap.put(id(r), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } - if(existing.isPending()) waitPending(graph, existing); + if(existing.isPending()) { + waitPending(graph, existing); + } return existing; } void remove(AsyncMultiReadEntry entry) { synchronized(asyncMultiReadEntryMap) { - asyncMultiReadEntryMap.remove(entry.request); + asyncMultiReadEntryMap.remove(entry.id); } } @@ -994,12 +1314,20 @@ public class QueryCache extends QueryCacheBase { QueryCache cache = graph.processor.cache; AsyncMultiReadEntry entry = (AsyncMultiReadEntry)cache.getOrCreateAsyncMultiReadEntry(graph, r); AsyncMultiProcedure procedure_ = procedure != null ? procedure : emptyProcedureAsyncMultiReadEntry; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + entry.performFromCache(graph, procedure_); + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); entry.compute(graph, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + } + } + + private AsyncMultiReadEntry peekAsyncMultiReadEntry(AsyncMultiRead r) { + synchronized(asyncMultiReadEntryMap) { + return (AsyncMultiReadEntry) asyncMultiReadEntryMap.get(r); } } @@ -1009,24 +1337,25 @@ public class QueryCache extends QueryCacheBase { existing = (ExternalReadEntry)externalReadEntryMap.get(r); if(existing == null) { existing = new ExternalReadEntry(r, graph); - existing.clearResult(querySupport); - existing.setPending(); + existing.setPending(querySupport); externalReadEntryMap.put(id(r), existing); size++; return existing; } if(existing.requiresComputation()) { - existing.setPending(); + existing.setPending(querySupport); return existing; } } - if(existing.isPending()) waitPending(graph, existing); + if(existing.isPending()) { + waitPending(graph, existing); + } return existing; } void remove(ExternalReadEntry entry) { synchronized(externalReadEntryMap) { - externalReadEntryMap.remove(entry.request); + externalReadEntryMap.remove(entry.id); } } @@ -1034,12 +1363,20 @@ public class QueryCache extends QueryCacheBase { QueryCache cache = graph.processor.cache; ExternalReadEntry entry = (ExternalReadEntry)cache.getOrCreateExternalReadEntry(graph, r); AsyncProcedure procedure_ = procedure != null ? procedure : emptyProcedureExternalReadEntry; - ListenerEntry listenerEntry = cache.registerDependencies(graph, entry, parent, listener, procedure_, false); - if(entry.isReady()) entry.performFromCache(graph, procedure_); + if(entry.isReady()) { + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); + entry.performFromCache(graph, procedure_); + } else { assert(entry.isPending()); + graph.processor.listening.registerDependencies(graph, entry, parent, listener, procedure_, false); entry.compute(graph, procedure_); - if(listenerEntry != null) cache.primeListenerEntry(listenerEntry, entry.getResult()); + } + } + + private ExternalReadEntry peekExternalReadEntry(ExternalRead r) { + synchronized(externalReadEntryMap) { + return (ExternalReadEntry) externalReadEntryMap.get(r); } }