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=2fca451d66c14f97e6c70c31fdbf43d1969e0ce1;hb=90a52d58489b6c518b13dde238923125c8f63eec;hpb=45548cb3447414845d4672b70e22181eeb3a3487 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 2fca451d6..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 @@ -20,6 +20,7 @@ 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); } @@ -30,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; } @@ -66,12 +68,14 @@ public class QueryCache extends QueryCacheBase { } 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()); } } @@ -87,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; } @@ -123,12 +128,14 @@ public class QueryCache extends QueryCacheBase { } 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()); } } @@ -144,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; } @@ -180,12 +188,14 @@ public class QueryCache extends QueryCacheBase { } 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()); } } @@ -201,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; } @@ -237,12 +248,14 @@ public class QueryCache extends QueryCacheBase { } 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()); } } @@ -258,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; } @@ -294,12 +308,14 @@ public class QueryCache extends QueryCacheBase { } 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()); } } @@ -315,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; } @@ -351,12 +368,14 @@ public class QueryCache extends QueryCacheBase { } 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()); } } @@ -372,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; } @@ -408,12 +428,14 @@ public class QueryCache extends QueryCacheBase { } 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()); } } @@ -429,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; } @@ -465,12 +488,14 @@ public class QueryCache extends QueryCacheBase { } 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()); } } @@ -486,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; } @@ -522,12 +548,14 @@ public class QueryCache extends QueryCacheBase { } 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()); } } @@ -543,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; } @@ -579,12 +608,14 @@ public class QueryCache extends QueryCacheBase { } 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()); } } @@ -594,38 +625,39 @@ public class QueryCache extends QueryCacheBase { } } - ReadEntry getOrCreateReadEntry(ReadGraphImpl graph, Read r, boolean needsToBlock) throws DatabaseException { + 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)) { if (SINGLE) { @@ -634,33 +666,39 @@ public class QueryCache extends QueryCacheBase { return e.performFromCache(graph, procedure); } } - return ReadEntry.computeForEach(graph, r, null, 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 run0(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; } } - private ReadEntry peekReadEntry(Read r) { synchronized(readEntryMap) { @@ -668,38 +706,49 @@ public class QueryCache extends QueryCacheBase { } } - AsyncReadEntry getOrCreateAsyncReadEntry(ReadGraphImpl graph, AsyncRead r, boolean needsToBlock) throws DatabaseException { + 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) { @@ -710,31 +759,26 @@ public class QueryCache extends QueryCacheBase { } 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 run0(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) { @@ -748,18 +792,19 @@ public class QueryCache extends QueryCacheBase { 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; } @@ -784,12 +829,14 @@ public class QueryCache extends QueryCacheBase { } 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()); } } @@ -805,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; } @@ -841,12 +889,14 @@ public class QueryCache extends QueryCacheBase { } 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()); } } @@ -862,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; } @@ -898,12 +949,14 @@ public class QueryCache extends QueryCacheBase { } 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()); } } @@ -919,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; } @@ -955,12 +1009,14 @@ public class QueryCache extends QueryCacheBase { } 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()); } } @@ -976,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; } @@ -1012,12 +1069,14 @@ public class QueryCache extends QueryCacheBase { } 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()); } } @@ -1033,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; } @@ -1058,12 +1118,14 @@ 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()); } } @@ -1079,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; } @@ -1104,12 +1167,14 @@ 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()); } } @@ -1125,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; } @@ -1150,12 +1216,14 @@ 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()); } } @@ -1171,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); } } @@ -1196,12 +1265,14 @@ 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()); } } @@ -1217,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); } } @@ -1242,12 +1314,14 @@ 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()); } } @@ -1263,25 +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); } } @@ -1289,12 +1363,14 @@ 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()); } }