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%2FQueryProcessor.java;h=7bfa0f4c6195659c21efdb0e6bcf99fed0dfb7bb;hp=5a37257e2531c051c39de0bea04af271e1adccce;hb=0db1af75cd14f99f16c42c920562708abf37be93;hpb=d85a4d990c5cac9d0c70781a265f02888b3aaa43 diff --git a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryProcessor.java b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryProcessor.java index 5a37257e2..7bfa0f4c6 100644 --- a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryProcessor.java +++ b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryProcessor.java @@ -44,17 +44,14 @@ import org.simantics.db.Statement; import org.simantics.db.VirtualGraph; import org.simantics.db.common.procedure.adapter.AsyncMultiProcedureAdapter; import org.simantics.db.common.utils.Logger; -import org.simantics.db.debug.ListenerReport; import org.simantics.db.exception.DatabaseException; import org.simantics.db.exception.ManyObjectsForFunctionalRelationException; import org.simantics.db.exception.NoInverseException; import org.simantics.db.exception.ResourceNotFoundException; -import org.simantics.db.impl.DebugPolicy; import org.simantics.db.impl.ResourceImpl; import org.simantics.db.impl.graph.BarrierTracing; import org.simantics.db.impl.graph.ReadGraphImpl; import org.simantics.db.impl.graph.ReadGraphSupport; -import org.simantics.db.impl.graph.WriteGraphImpl; import org.simantics.db.impl.procedure.IntProcedureAdapter; import org.simantics.db.impl.procedure.InternalProcedure; import org.simantics.db.impl.procedure.TripleIntProcedureAdapter; @@ -131,17 +128,15 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap AtomicInteger sleepers = new AtomicInteger(0); - private boolean updating = false; + boolean updating = false; - private boolean firingListeners = false; - final public QueryCache cache; final public QuerySupport querySupport; final public Session session; final public ResourceSupport resourceSupport; - - private THashSet scheduledListeners = new THashSet(); + + final public QueryListening listening = new QueryListening(this); QueryThread[] executors; @@ -505,7 +500,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap } final public void releaseWrite(ReadGraphImpl graph) { - performDirtyUpdates(graph); + propagateChangesInQueryCache(graph); modificationCounter++; } @@ -700,31 +695,10 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap boolean isBound(ExternalReadEntry entry) { if(entry.hasParents()) return true; - else if(hasListener(entry)) return true; + else if(listening.hasListener(entry)) return true; else return false; } - synchronized public ListenerEntry registerDependencies(ReadGraphImpl graph, CacheEntry child, CacheEntry parent, ListenerBase listener, Object procedure, boolean inferred) { - - if (parent != null && !inferred) { - try { - if(!child.isImmutable(graph)) - child.addParent(parent); - } catch (DatabaseException e) { - Logger.defaultLogError(e); - } - if(DebugPolicy.DEPENDENCIES) System.out.println(child + " -> " + parent); - } - - if (listener != null) { - return registerListener(child, listener, procedure); - } else { - return null; - } - - } - - static class Dummy implements InternalProcedure, IntProcedure { @Override @@ -926,175 +900,6 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap } - public ListenerEntry registerListener(final CacheEntry entry, final ListenerBase base, final Object procedure) { - - assert (entry != null); - - if (base.isDisposed()) - return null; - - return addListener(entry, base, procedure); - - } - - private void primeListenerEntry(final ListenerEntry entry, final Object result) { - entry.setLastKnown(result); - } - - private ListenerEntry addListener(CacheEntry entry, ListenerBase base, Object procedure) { - - assert (entry != null); - assert (procedure != null); - - ArrayList list = cache.listeners.get(entry); - if (list == null) { - list = new ArrayList(1); - cache.listeners.put(entry, list); - } - - ListenerEntry result = new ListenerEntry(entry, base, procedure); - int currentIndex = list.indexOf(result); - // There was already a listener - if(currentIndex > -1) { - ListenerEntry current = list.get(currentIndex); - if(!current.base.isDisposed()) return null; - list.set(currentIndex, result); - } else { - list.add(result); - } - - if(DebugPolicy.LISTENER) { - new Exception().printStackTrace(); - System.out.println("addListener -> " + list.size() + " " + entry + " " + base + " " + procedure); - } - - return result; - - } - - private void scheduleListener(ListenerEntry entry) { - assert (entry != null); - if(DebugPolicy.LISTENER) System.out.println("Scheduled " + entry.procedure); - scheduledListeners.add(entry); - } - - private void removeListener(ListenerEntry entry) { - assert (entry != null); - ArrayList list = cache.listeners.get(entry.entry); - if(list == null) return; - boolean success = list.remove(entry); - assert (success); - if (list.isEmpty()) - cache.listeners.remove(entry.entry); - } - - private boolean hasListener(CacheEntry entry) { - if(cache.listeners.get(entry) != null) return true; - return false; - } - - boolean hasListenerAfterDisposing(CacheEntry entry) { - if(cache.listeners.get(entry) != null) { - ArrayList entries = cache.listeners.get(entry); - ArrayList list = null; - for (ListenerEntry e : entries) { - if (e.base.isDisposed()) { - if(list == null) list = new ArrayList(); - list.add(e); - } - } - if(list != null) { - for (ListenerEntry e : list) { - entries.remove(e); - } - } - if (entries.isEmpty()) { - cache.listeners.remove(entry); - return false; - } - return true; - } - return false; - } - - List getListenerEntries(CacheEntry entry) { - hasListenerAfterDisposing(entry); - if(cache.listeners.get(entry) != null) - return cache.listeners.get(entry); - else - return Collections.emptyList(); - } - - void processListenerReport(CacheEntry entry, Map> workarea) { - - if(!workarea.containsKey(entry)) { - - HashSet ls = new HashSet(); - for(ListenerEntry e : getListenerEntries(entry)) - ls.add(e.base); - - workarea.put(entry, ls); - - for(CacheEntry parent : entry.getParents(this)) { - processListenerReport(parent, workarea); - ls.addAll(workarea.get(parent)); - } - - } - - } - - public synchronized ListenerReport getListenerReport() throws IOException { - - class ListenerReportImpl implements ListenerReport { - - Map> workarea = new HashMap>(); - - @Override - public void print(PrintStream b) { - Map hist = new HashMap(); - for(Map.Entry> e : workarea.entrySet()) { - for(ListenerBase l : e.getValue()) { - Integer i = hist.get(l); - hist.put(l, i != null ? i-1 : -1); - } - } - - for(Pair p : CollectionUtils.valueSortedEntries(hist)) { - b.print("" + -p.second + " " + p.first + "\n"); - } - - b.flush(); - } - - } - - ListenerReportImpl result = new ListenerReportImpl(); - - Collection all = allCaches(new CacheCollectionResult()).toCollection(); - for(CacheEntryBase entry : all) { - hasListenerAfterDisposing(entry); - } - for(CacheEntryBase entry : all) { - processListenerReport(entry, result.workarea); - } - - return result; - - } - - public synchronized String reportListeners(File file) throws IOException { - - if (!isAlive()) - return "Disposed!"; - - PrintStream b = new PrintStream(new BufferedOutputStream(new FileOutputStream(file))); - ListenerReport report = getListenerReport(); - report.print(b); - - return "Done reporting listeners."; - - } void processParentReport(CacheEntry entry, Map> workarea) { @@ -1167,7 +972,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap int listeners = 0; for(CacheEntry entry : workarea.keySet()) { - boolean listener = hasListenerAfterDisposing(entry); + boolean listener = listening.hasListenerAfterDisposing(entry); boolean hasParents = entry.getParents(this).iterator().hasNext(); if(listener) { // Bound @@ -1257,11 +1062,11 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap for(CacheEntry entry : workarea.keySet()) { Class clazz = entry.getClass(); - if(entry instanceof ReadEntry) clazz = ((ReadEntry)entry).request.getClass(); - else if(entry instanceof MultiReadEntry) clazz = ((MultiReadEntry)entry).request.getClass(); - else if(entry instanceof AsyncReadEntry) clazz = ((AsyncReadEntry)entry).request.getClass(); - else if(entry instanceof AsyncMultiReadEntry) clazz = ((AsyncMultiReadEntry)entry).request.getClass(); - else if(entry instanceof ExternalReadEntry) clazz = ((ExternalReadEntry)entry).request.getClass(); + if(entry instanceof ReadEntry) clazz = ((ReadEntry)entry).id.getClass(); + else if(entry instanceof MultiReadEntry) clazz = ((MultiReadEntry)entry).id.getClass(); + else if(entry instanceof AsyncReadEntry) clazz = ((AsyncReadEntry)entry).id.getClass(); + else if(entry instanceof AsyncMultiReadEntry) clazz = ((AsyncMultiReadEntry)entry).id.getClass(); + else if(entry instanceof ExternalReadEntry) clazz = ((ExternalReadEntry)entry).id.getClass(); Integer c = counts.get(clazz); if(c == null) counts.put(clazz, -1); else counts.put(clazz, c-1); @@ -1331,7 +1136,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap b.print("// Entry parent listing\n"); for(CacheEntry entry : workarea.keySet()) { int status = flagMap.get(entry); - boolean hasListener = hasListenerAfterDisposing(entry); + boolean hasListener = listening.hasListenerAfterDisposing(entry); b.print("Q " + entry.toString()); if(hasListener) { b.print(" (L" + status + ")"); @@ -1377,22 +1182,6 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap } - class UpdateEntry { - - public CacheEntry caller; - - public CacheEntry entry; - - public int indent; - - public UpdateEntry(CacheEntry caller, CacheEntry entry, int indent) { - this.caller = caller; - this.entry = entry; - this.indent = indent; - } - - }; - boolean removeQuery(CacheEntry entry) { // This entry has been removed before. No need to do anything here. @@ -1426,8 +1215,6 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap CacheEntry entry = e.entry; - //System.err.println("updateQuery " + entry); - /* * If the dependency graph forms a DAG, some entries are inserted in the * todo list many times. They only need to be processed once though. @@ -1435,32 +1222,32 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap if (entry.isDiscarded()) { if (Development.DEVELOPMENT) { if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_UPDATE, Bindings.BOOLEAN)) { - System.out.print("D"); + System.err.print("D"); for (int i = 0; i < e.indent; i++) - System.out.print(" "); - System.out.println(entry.getQuery()); + System.err.print(" "); + System.err.println(entry.getQuery()); } } // System.err.println(" => DISCARDED"); return false; } - if (entry.isRefuted()) { - if (Development.DEVELOPMENT) { - if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_UPDATE, Bindings.BOOLEAN)) { - System.out.print("R"); - for (int i = 0; i < e.indent; i++) - System.out.print(" "); - System.out.println(entry.getQuery()); - } - } - return false; - } +// if (entry.isRefuted()) { +// if (Development.DEVELOPMENT) { +// if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_UPDATE, Bindings.BOOLEAN)) { +// System.err.print("R"); +// for (int i = 0; i < e.indent; i++) +// System.err.print(" "); +// System.err.println(entry.getQuery()); +// } +// } +// return false; +// } if (entry.isExcepted()) { if (Development.DEVELOPMENT) { if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_UPDATE, Bindings.BOOLEAN)) { - System.out.print("E"); + System.err.print("E"); } } } @@ -1468,7 +1255,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap if (entry.isPending()) { if (Development.DEVELOPMENT) { if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_UPDATE, Bindings.BOOLEAN)) { - System.out.print("P"); + System.err.print("P"); } } } @@ -1477,24 +1264,24 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap if (Development.DEVELOPMENT) { if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_UPDATE, Bindings.BOOLEAN)) { - System.out.print("U "); + System.err.print("U "); for (int i = 0; i < e.indent; i++) - System.out.print(" "); - System.out.print(entry.getQuery()); + System.err.print(" "); + System.err.print(entry.getQuery()); } } Query query = entry.getQuery(); int type = query.type(); - boolean hasListener = hasListener(entry); + boolean hasListener = listening.hasListener(entry); if (Development.DEVELOPMENT) { if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_UPDATE, Bindings.BOOLEAN)) { - if(hasListener(entry)) { - System.out.println(" (L)"); + if(listening.hasListener(entry)) { + System.err.println(" (L)"); } else { - System.out.println(""); + System.err.println(""); } } } @@ -1538,43 +1325,23 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap // System.err.println(" => FOO " + type); if (hasListener) { - ArrayList entries = cache.listeners.get(entry); + ArrayList entries = listening.listeners.get(entry); if(entries != null) { for (ListenerEntry le : entries) { - scheduleListener(le); + listening.scheduleListener(le); } } } // If invalid, update parents if (type == RequestFlags.INVALIDATE) { - updateParents(e.indent, entry, todo); + listening.updateParents(e.indent, entry, todo); } return hasListener; } - private void updateParents(int indent, CacheEntry entry, LinkedList todo) { - - Iterable oldParents = entry.getParents(this); - for (CacheEntry parent : oldParents) { -// System.err.println("updateParents " + entry + " => " + parent); - if(!parent.isDiscarded()) - todo.push(new UpdateEntry(entry, parent, indent + 2)); - } - - } - - private boolean pruneListener(ListenerEntry entry) { - if (entry.base.isDisposed()) { - removeListener(entry); - return true; - } else { - return false; - } - } - /** * @param av1 an array (guaranteed) * @param av2 any object @@ -1610,13 +1377,17 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap - final private Object compareTo(ReadGraphImpl graph, final CacheEntry entry, final Object oldValue) { + final Object compareTo(ReadGraphImpl graph, final CacheEntry entry, final Object oldValue) { try { Query query = entry.getQuery(); - if(DebugPolicy.RECOMPUTE) System.out.println("R " + query); + if (Development.DEVELOPMENT) { + if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_RECOMPUTE, Bindings.BOOLEAN)) { + System.err.println("R " + query); + } + } entry.prepareRecompute(querySupport); @@ -1629,10 +1400,12 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap Object newValue = entry.getResult(); if (ListenerEntry.NO_VALUE == oldValue) { - if(DebugPolicy.CHANGES) { - System.out.println("C " + query); - System.out.println("- " + oldValue); - System.out.println("- " + newValue); + if (Development.DEVELOPMENT) { + if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_CHANGES, Bindings.BOOLEAN)) { + System.out.println("C " + query); + System.out.println("- " + oldValue); + System.out.println("- " + newValue); + } } return newValue; } @@ -1648,10 +1421,12 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap } else changed = (oldValue != null); - if(DebugPolicy.CHANGES && changed) { - System.out.println("C " + query); - System.out.println("- " + oldValue); - System.out.println("- " + newValue); + if (Development.DEVELOPMENT) { + if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_CHANGES, Bindings.BOOLEAN)) { + System.err.println("C " + query); + System.err.println("- " + oldValue); + System.err.println("- " + newValue); + } } return changed ? newValue : ListenerEntry.NOT_CHANGED; @@ -1666,78 +1441,6 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap } - public boolean hasScheduledUpdates() { - return !scheduledListeners.isEmpty(); - } - - public void performScheduledUpdates(WriteGraphImpl graph) { - - assert (!updating); - assert (!cache.collecting); - assert (!firingListeners); - - firingListeners = true; - - try { - - // Performing may cause further events to be scheduled. - while (!scheduledListeners.isEmpty()) { - -// graph.restart(); -// graph.state.barrier.inc(); - - // Clone current events to make new entries possible during - // firing. - THashSet entries = scheduledListeners; - scheduledListeners = new THashSet(); - - ArrayList schedule = new ArrayList(); - - for (ListenerEntry listenerEntry : entries) { - - if (pruneListener(listenerEntry)) { - if(DebugPolicy.LISTENER) System.out.println("Pruned " + listenerEntry.procedure); - continue; - } - - final CacheEntry entry = listenerEntry.entry; - assert (entry != null); - - Object newValue = compareTo(graph, entry, listenerEntry.getLastKnown()); - - if (newValue != ListenerEntry.NOT_CHANGED) { - if(DebugPolicy.LISTENER) - System.out.println("Add to schedule " + listenerEntry.procedure + " with " + newValue); - schedule.add(listenerEntry); - listenerEntry.setLastKnown(entry.getResult()); - } - - } - - for(ListenerEntry listenerEntry : schedule) { - final CacheEntry entry = listenerEntry.entry; - if(DebugPolicy.LISTENER) - System.out.println("Firing " + listenerEntry.procedure); - try { - if(DebugPolicy.LISTENER) - System.out.println("Firing " + listenerEntry.procedure + " for " + listenerEntry.entry); - entry.performFromCache(graph, listenerEntry.procedure); - } catch (Throwable t) { - t.printStackTrace(); - } - } - -// graph.state.barrier.dec(); -// graph.waitAsync(null); -// graph.state.barrier.assertReady(); - - } - - } finally { - firingListeners = false; - } - - } /** * @@ -1780,7 +1483,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap Object newValue = compareTo(graph, immediate, ListenerEntry.NO_VALUE); if (newValue != ListenerEntry.NOT_CHANGED) - updateParents(0, immediate, todo); + listening.updateParents(0, immediate, todo); } else { @@ -1788,7 +1491,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap Object newValue = compareTo(graph, immediate, oldValue); if (newValue != ListenerEntry.NOT_CHANGED) { - updateParents(0, immediate, todo); + listening.updateParents(0, immediate, todo); } else { // If not changed, keep the old value immediate.setResult(oldValue); @@ -1821,7 +1524,26 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap private Object primitiveUpdateLock = new Object(); private THashSet scheduledPrimitiveUpdates = new THashSet(); - public void performDirtyUpdates(final ReadGraphImpl graph) { + private ArrayList refutations = new ArrayList<>(); + + private void markForUpdate(ReadGraphImpl graph, CacheEntry e) { + e.refute(); + refutations.add(e); + } + + private void updateRefutations(ReadGraphImpl graph) { + + for(CacheEntry e : refutations) + update(graph, e); + + refutations.clear(); + + } + + public void propagateChangesInQueryCache(final ReadGraphImpl graph) { + + // Make sure that listening has performed its work + listening.sync(); cache.dirty = false; lastInvalidate = 0; @@ -1840,28 +1562,37 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap final int subject = (int)(arg0 >>> 32); final int predicate = (int)(arg0 & 0xffffffff); - for(Objects o : QueryCache.entriesObjects(QueryProcessor.this, subject)) update(graph, o); - for(DirectObjects o : QueryCache.entriesDirectObjects(QueryProcessor.this, subject)) update(graph, o); - for(Statements o : QueryCache.entriesStatements(QueryProcessor.this, subject)) update(graph, o); + for(Objects o : QueryCache.entriesObjects(QueryProcessor.this, subject)) markForUpdate(graph, o); + for(DirectObjects o : QueryCache.entriesDirectObjects(QueryProcessor.this, subject)) markForUpdate(graph, o); + for(Statements o : QueryCache.entriesStatements(QueryProcessor.this, subject)) markForUpdate(graph, o); if(predicate == instanceOf || predicate == inherits || predicate == subrelationOf) { PrincipalTypes principalTypes = QueryCache.entryPrincipalTypes(QueryProcessor.this, subject); - if(principalTypes != null) update(graph, principalTypes); + if(principalTypes != null) markForUpdate(graph, principalTypes); Types types = QueryCache.entryTypes(QueryProcessor.this, subject); - if(types != null) update(graph, types); + if(types != null) markForUpdate(graph, types); } if(predicate == subrelationOf) { SuperRelations superRelations = SuperRelations.entry(QueryProcessor.this, subject); - if(superRelations != null) update(graph, superRelations); + if(superRelations != null) markForUpdate(graph, superRelations); } DirectPredicates dp = QueryCache.entryDirectPredicates(QueryProcessor.this, subject); - if(dp != null) update(graph, dp); + if(dp != null) markForUpdate(graph, dp); OrderedSet os = QueryCache.entryOrderedSet(QueryProcessor.this, predicate); - if(os != null) update(graph, os); + if(os != null) markForUpdate(graph, os); + updateRefutations(graph); + scheduledObjectUpdates.clear(); + + if (Development.DEVELOPMENT) { + if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_UPDATE, Bindings.BOOLEAN)) { + System.err.println("== Query update ends =="); + } + } + return; } @@ -1872,9 +1603,18 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap int arg0 = scheduledValueUpdates.getFirst(); ValueQuery valueQuery = QueryCache.entryValueQuery(QueryProcessor.this, arg0); - if(valueQuery != null) update(graph, valueQuery); + if(valueQuery != null) markForUpdate(graph, valueQuery); + + updateRefutations(graph); scheduledValueUpdates.clear(); + + if (Development.DEVELOPMENT) { + if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_UPDATE, Bindings.BOOLEAN)) { + System.err.println("== Query update ends =="); + } + } + return; } @@ -1888,30 +1628,12 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap scheduledPrimitiveUpdates = new THashSet(); } - primitiveUpdates.forEach(new TObjectProcedure() { - - @Override - public boolean execute(Object arg0) { - - ExternalReadEntry query = (ExternalReadEntry)cache.externalReadEntryMap.get(arg0); - if (query != null) { - boolean listening = update(graph, query); - if (!listening && !query.hasParents()) { - cache.externalReadEntryMap.remove(arg0); - query.discard(); - } - } - return true; - } - - }); - scheduledValueUpdates.forEach(new TIntProcedure() { @Override public boolean execute(int arg0) { ValueQuery valueQuery = QueryCache.entryValueQuery(QueryProcessor.this, arg0); - if(valueQuery != null) update(graph, valueQuery); + if(valueQuery != null) markForUpdate(graph, valueQuery); return true; } @@ -1923,15 +1645,15 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap public boolean execute(int resource) { ValueQuery valueQuery = QueryCache.entryValueQuery(QueryProcessor.this, resource); - if(valueQuery != null) update(graph, valueQuery); + if(valueQuery != null) markForUpdate(graph, valueQuery); PrincipalTypes principalTypes = QueryCache.entryPrincipalTypes(QueryProcessor.this, resource); - if(principalTypes != null) update(graph, principalTypes); + if(principalTypes != null) markForUpdate(graph, principalTypes); Types types = QueryCache.entryTypes(QueryProcessor.this, resource); - if(types != null) update(graph, types); + if(types != null) markForUpdate(graph, types); SuperRelations superRelations = SuperRelations.entry(QueryProcessor.this, resource); - if(superRelations != null) update(graph, superRelations); + if(superRelations != null) markForUpdate(graph, superRelations); predicates.add(resource); @@ -1950,14 +1672,14 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap if(predicate == instanceOf || predicate == inherits || predicate == subrelationOf) { PrincipalTypes principalTypes = QueryCache.entryPrincipalTypes(QueryProcessor.this, subject); - if(principalTypes != null) update(graph, principalTypes); + if(principalTypes != null) markForUpdate(graph, principalTypes); Types types = QueryCache.entryTypes(QueryProcessor.this, subject); - if(types != null) update(graph, types); + if(types != null) markForUpdate(graph, types); } if(predicate == subrelationOf) { SuperRelations superRelations = SuperRelations.entry(QueryProcessor.this, subject); - if(superRelations != null) update(graph, superRelations); + if(superRelations != null) markForUpdate(graph, superRelations); } predicates.add(subject); @@ -1974,12 +1696,12 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap @Override public boolean execute(final int subject) { - for(Objects o : QueryCache.entriesObjects(QueryProcessor.this, subject)) update(graph, o); - for(DirectObjects o : QueryCache.entriesDirectObjects(QueryProcessor.this, subject)) update(graph, o); - for(Statements o : QueryCache.entriesStatements(QueryProcessor.this, subject)) update(graph, o); + for(Objects o : QueryCache.entriesObjects(QueryProcessor.this, subject)) markForUpdate(graph, o); + for(DirectObjects o : QueryCache.entriesDirectObjects(QueryProcessor.this, subject)) markForUpdate(graph, o); + for(Statements o : QueryCache.entriesStatements(QueryProcessor.this, subject)) markForUpdate(graph, o); DirectPredicates entry = QueryCache.entryDirectPredicates(QueryProcessor.this, subject); - if(entry != null) update(graph, entry); + if(entry != null) markForUpdate(graph, entry); return true; @@ -1993,7 +1715,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap public boolean execute(int orderedSet) { OrderedSet entry = QueryCache.entryOrderedSet(QueryProcessor.this, orderedSet); - if(entry != null) update(graph, entry); + if(entry != null) markForUpdate(graph, entry); return true; @@ -2001,21 +1723,35 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap }); - // for (Integer subject : predicates) { - // DirectPredicates entry = DirectPredicates.entry(QueryProcessor.this, subject); - // if(entry != null) update(graph, entry); - // } + updateRefutations(graph); + primitiveUpdates.forEach(new TObjectProcedure() { - if (Development.DEVELOPMENT) { - if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_UPDATE, Bindings.BOOLEAN)) { - System.err.println("== Query update ends =="); + @Override + public boolean execute(Object arg0) { + + ExternalReadEntry query = (ExternalReadEntry)cache.externalReadEntryMap.get(arg0); + if (query != null) { + boolean listening = update(graph, query); + if (!listening && !query.hasParents()) { + cache.externalReadEntryMap.remove(arg0); + query.discard(); + } + } + return true; } - } + }); + scheduledValueUpdates.clear(); scheduledObjectUpdates.clear(); scheduledInvalidates.clear(); + + if (Development.DEVELOPMENT) { + if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_UPDATE, Bindings.BOOLEAN)) { + System.err.println("== Query update ends =="); + } + } }