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=55855ee3e823bed936c00c558848ceaf05a04915;hp=65b277e6c8d2985410b82e12ce4edca3f1fea09d;hb=cb12888f3415720276da57f2d162e13ea5b3f46c;hpb=277b1c7fb3fdd8effb4bf2b447358d0e4ef0e302 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 65b277e6c..55855ee3e 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 @@ -49,8 +49,8 @@ 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; @@ -171,7 +171,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap public void close() { } - SessionTask getOwnTask(ReadGraphImpl impl) { + public SessionTask getOwnTask(ReadGraphImpl impl) { Set ancestors = impl.ancestorSet(); synchronized(querySupportLock) { int index = 0; @@ -186,6 +186,21 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap return null; } + public SessionTask getSubTask(ReadGraphImpl impl) { + Set onlyThis = Collections.singleton(impl); + synchronized(querySupportLock) { + int index = 0; + while(index < freeScheduling.size()) { + SessionTask task = freeScheduling.get(index); + if(task.hasCommonParent(onlyThis)) { + return freeScheduling.remove(index); + } + index++; + } + } + return null; + } + public boolean performPending(ReadGraphImpl graph) { SessionTask task = getOwnTask(graph); if(task != null) { @@ -201,7 +216,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap // } final public void schedule(SessionTask request) { - + //int performer = request.thread; // if(DebugPolicy.SCHEDULE) @@ -218,24 +233,20 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap // if(performer == THREADS) { synchronized(querySupportLock) { - - //new Exception().printStackTrace(); - + + if(BarrierTracing.BOOKKEEPING) { + Exception current = new Exception(); + Exception previous = BarrierTracing.tasks.put(request, current); + if(previous != null) { + previous.printStackTrace(); + current.printStackTrace(); + } + } + freeScheduling.add(request); querySupportLock.notifyAll(); - //System.err.println("schedule free task " + request + " => " + freeScheduling.size()); - -// for(int i=0;i ancestors; + private int counter = 0; + private Exception trace; public SessionTask(ReadGraphImpl graph) { this.graph = graph; + if(graph != null) graph.asyncBarrier.inc(); } public boolean hasCommonParent(Set otherAncestors) { @@ -276,7 +290,22 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap return !Collections.disjoint(ancestors, otherAncestors); } - public abstract void run(int thread); + public abstract void run0(int thread); + + public final void run(int thread) { + if(counter++ > 0) { + if(BarrierTracing.BOOKKEEPING) { + trace.printStackTrace(); + new Exception().printStackTrace(); + } + throw new IllegalStateException("Multiple invocations of SessionTask!"); + } + if(BarrierTracing.BOOKKEEPING) { + trace = new Exception(); + } + run0(thread); + if(graph != null) graph.asyncBarrier.dec(); + } @Override public String toString() { @@ -683,7 +712,11 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap } catch (DatabaseException e) { Logger.defaultLogError(e); } - if(DebugPolicy.DEPENDENCIES) System.out.println(child + " -> " + parent); + if (Development.DEVELOPMENT) { + if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_DEPENDENCIES, Bindings.BOOLEAN)) { + System.out.println(child + " -> " + parent); + } + } } if (listener != null) { @@ -933,9 +966,11 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap list.add(result); } - if(DebugPolicy.LISTENER) { - new Exception().printStackTrace(); - System.out.println("addListener -> " + list.size() + " " + entry + " " + base + " " + procedure); + if (Development.DEVELOPMENT) { + if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_LISTENERS, Bindings.BOOLEAN)) { + new Exception().printStackTrace(); + System.err.println("addListener -> " + list.size() + " " + entry + " " + base + " " + procedure); + } } return result; @@ -944,7 +979,11 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap private void scheduleListener(ListenerEntry entry) { assert (entry != null); - if(DebugPolicy.LISTENER) System.out.println("Scheduled " + entry.procedure); + if (Development.DEVELOPMENT) { + if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_LISTENERS, Bindings.BOOLEAN)) { + System.err.println("Scheduled " + entry.procedure); + } + } scheduledListeners.add(entry); } @@ -1396,8 +1435,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. @@ -1405,32 +1442,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"); } } } @@ -1438,7 +1475,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"); } } } @@ -1447,10 +1484,10 @@ 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()); } } @@ -1462,9 +1499,9 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap if (Development.DEVELOPMENT) { if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_UPDATE, Bindings.BOOLEAN)) { if(hasListener(entry)) { - System.out.println(" (L)"); + System.err.println(" (L)"); } else { - System.out.println(""); + System.err.println(""); } } } @@ -1586,7 +1623,11 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap 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); @@ -1599,10 +1640,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; } @@ -1618,10 +1661,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,7 +1711,12 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap for (ListenerEntry listenerEntry : entries) { if (pruneListener(listenerEntry)) { - if(DebugPolicy.LISTENER) System.out.println("Pruned " + listenerEntry.procedure); + if (Development.DEVELOPMENT) { + if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_LISTENERS, Bindings.BOOLEAN)) { + new Exception().printStackTrace(); + System.err.println("Pruned " + listenerEntry.procedure); + } + } continue; } @@ -1676,8 +1726,12 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap 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); + if (Development.DEVELOPMENT) { + if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_LISTENERS, Bindings.BOOLEAN)) { + new Exception().printStackTrace(); + System.err.println("Add to schedule " + listenerEntry.procedure + " with " + newValue); + } + } schedule.add(listenerEntry); listenerEntry.setLastKnown(entry.getResult()); } @@ -1686,11 +1740,17 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap for(ListenerEntry listenerEntry : schedule) { final CacheEntry entry = listenerEntry.entry; - if(DebugPolicy.LISTENER) - System.out.println("Firing " + listenerEntry.procedure); + if (Development.DEVELOPMENT) { + if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_LISTENERS, Bindings.BOOLEAN)) { + System.err.println("Firing " + listenerEntry.procedure); + } + } try { - if(DebugPolicy.LISTENER) - System.out.println("Firing " + listenerEntry.procedure + " for " + listenerEntry.entry); + if (Development.DEVELOPMENT) { + if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_LISTENERS, Bindings.BOOLEAN)) { + System.err.println("Firing " + listenerEntry.procedure + " for " + listenerEntry.entry); + } + } entry.performFromCache(graph, listenerEntry.procedure); } catch (Throwable t) { t.printStackTrace(); @@ -1791,6 +1851,22 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap private Object primitiveUpdateLock = new Object(); private THashSet scheduledPrimitiveUpdates = new THashSet(); + 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 performDirtyUpdates(final ReadGraphImpl graph) { cache.dirty = false; @@ -1810,28 +1886,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; } @@ -1842,9 +1927,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; } @@ -1858,30 +1952,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; } @@ -1893,15 +1969,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); @@ -1920,14 +1996,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); @@ -1944,12 +2020,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; @@ -1963,7 +2039,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; @@ -1971,21 +2047,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 =="); + } + } } @@ -2266,67 +2356,16 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap @Override final public void forEachPredicate(final ReadGraphImpl impl, final Resource subject, final AsyncMultiProcedure procedure) { - throw new UnsupportedOperationException(); + try { -// assert(subject != null); -// assert(procedure != null); -// -// final ListenerBase listener = getListenerBase(procedure); -// -// IntProcedure ip = new IntProcedure() { -// -// AtomicBoolean first = new AtomicBoolean(true); -// -// @Override -// public void execute(ReadGraphImpl graph, int i) { -// try { -// if(first.get()) { -// procedure.execute(graph, querySupport.getResource(i)); -// } else { -// procedure.execute(impl.newRestart(graph), querySupport.getResource(i)); -// } -// } catch (Throwable t2) { -// Logger.defaultLogError(t2); -// } -// } -// -// @Override -// public void finished(ReadGraphImpl graph) { -// try { -// if(first.compareAndSet(true, false)) { -// procedure.finished(graph); -//// impl.state.barrier.dec(this); -// } else { -// procedure.finished(impl.newRestart(graph)); -// } -// -// } catch (Throwable t2) { -// Logger.defaultLogError(t2); -// } -// } -// -// @Override -// public void exception(ReadGraphImpl graph, Throwable t) { -// try { -// if(first.compareAndSet(true, false)) { -// procedure.exception(graph, t); -// } else { -// procedure.exception(impl.newRestart(graph), t); -// } -// } catch (Throwable t2) { -// Logger.defaultLogError(t2); -// } -// } -// -// }; -// -// int sId = querySupport.getId(subject); -// -// try { -// QueryCache.runnerPredicates(impl, sId, impl.parent, listener, ip); -// } catch (DatabaseException e) { -// Logger.defaultLogError(e); -// } + for(Resource predicate : getPredicates(impl, subject)) + procedure.execute(impl, predicate); + + procedure.finished(impl); + + } catch (Throwable e) { + procedure.exception(impl, e); + } }