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;
public void close() {
}
- SessionTask getOwnTask(int thread) {
+ public SessionTask getOwnTask(ReadGraphImpl impl) {
+ Set<ReadGraphImpl> ancestors = impl.ancestorSet();
synchronized(querySupportLock) {
int index = 0;
while(index < freeScheduling.size()) {
SessionTask task = freeScheduling.get(index);
- if(task.thread == thread && !task.systemCall)
+ if(task.hasCommonParent(ancestors)) {
return freeScheduling.remove(index);
+ }
index++;
}
}
return null;
}
-
- public boolean performPending(int thread) {
- SessionTask task = getOwnTask(thread);
+
+ public SessionTask getSubTask(ReadGraphImpl impl) {
+ Set<ReadGraphImpl> 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) {
- task.run(thread);
+ task.run(QueryProcessor.thread.get());
return true;
} else {
return false;
// }
final public void schedule(SessionTask request) {
+
+ //int performer = request.thread;
- int performer = request.thread;
-
- if(DebugPolicy.SCHEDULE)
- System.out.println("schedule " + request + " " + " -> " + performer);
+// if(DebugPolicy.SCHEDULE)
+// System.out.println("schedule " + request + " " + " -> " + performer);
//assert(performer >= 0);
// 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<THREADS;i++) {
-// ReentrantLock queueLock = threadLocks[i];
-// queueLock.lock();
-// //queues[performer].add(request);
-// //if(ThreadState.SLEEP == threadStates[i]) sleepers.decrementAndGet();
-// threadConditions[i].signalAll();
-// queueLock.unlock();
-// }
-
}
return;
final int THREADS;
final public int THREAD_MASK;
-
- final public static ThreadGroup QueryThreadGroup = new ThreadGroup("Query Thread Group");
+
+ final public static ThreadGroup QueryThreadGroup = new ThreadGroup("Query Thread Group");
public static abstract class SessionTask {
- final public int thread;
- final public boolean systemCall;
-// final public int syncCaller;
- //final public Object object;
+ public final ReadGraphImpl graph;
+ private Set<ReadGraphImpl> ancestors;
+ private int counter = 0;
+ private Exception trace;
- public SessionTask(boolean systemCall) {
- this.thread = QueryProcessor.thread.get();
- this.systemCall = systemCall;
-// this.syncCaller = -1;
- //this.object = object;
+ public SessionTask(ReadGraphImpl graph) {
+ this.graph = graph;
+ if(graph != null) graph.asyncBarrier.inc();
}
-// public SessionTask(Object object, int syncCaller) {
-// this.thread = QueryProcessor.thread.get();
-// this.syncCaller = syncCaller;
-// this.object = object;
-// }
+ public boolean hasCommonParent(Set<ReadGraphImpl> otherAncestors) {
+ if(graph == null) return false;
+ if(ancestors == null) ancestors = graph.ancestorSet();
+ 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() {
- return "SessionTask[" + super.toString() + "]";
+ return "SessionTask[" + graph.parent + "]";
}
}
final public DataContainer<Throwable> throwable;
public SessionRead(DataContainer<Throwable> throwable, Semaphore notify) {
- super(true);
+ super(null);
this.throwable = throwable;
this.notify = notify;
}
} catch (DatabaseException e) {
Logger.defaultLogError(e);
}
- if(DebugPolicy.DEPENDENCIES) System.out.println(child + " -> " + parent);
+ if (Development.DEVELOPMENT) {
+ if(Development.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_DEPENDENCIES, Bindings.BOOLEAN)) {
+ System.out.println(child + " -> " + parent);
+ }
+ }
}
if (listener != null) {
list.add(result);
}
- if(DebugPolicy.LISTENER) {
- new Exception().printStackTrace();
- System.out.println("addListener -> " + list.size() + " " + entry + " " + base + " " + procedure);
+ if (Development.DEVELOPMENT) {
+ if(Development.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_LISTENERS, Bindings.BOOLEAN)) {
+ new Exception().printStackTrace();
+ System.err.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);
+ if (Development.DEVELOPMENT) {
+ if(Development.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_LISTENERS, Bindings.BOOLEAN)) {
+ System.err.println("Scheduled " + entry.procedure);
+ }
+ }
scheduledListeners.add(entry);
}
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.
if (entry.isDiscarded()) {
if (Development.DEVELOPMENT) {
if(Development.<Boolean>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.<Boolean>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.<Boolean>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.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_UPDATE, Bindings.BOOLEAN)) {
- System.out.print("E");
+ System.err.print("E");
}
}
}
if (entry.isPending()) {
if (Development.DEVELOPMENT) {
if(Development.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_UPDATE, Bindings.BOOLEAN)) {
- System.out.print("P");
+ System.err.print("P");
}
}
}
if (Development.DEVELOPMENT) {
if(Development.<Boolean>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());
}
}
if (Development.DEVELOPMENT) {
if(Development.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_UPDATE, Bindings.BOOLEAN)) {
if(hasListener(entry)) {
- System.out.println(" (L)");
+ System.err.println(" (L)");
} else {
- System.out.println("");
+ System.err.println("");
}
}
}
Query query = entry.getQuery();
- if(DebugPolicy.RECOMPUTE) System.out.println("R " + query);
+ if (Development.DEVELOPMENT) {
+ if(Development.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_RECOMPUTE, Bindings.BOOLEAN)) {
+ System.err.println("R " + query);
+ }
+ }
entry.prepareRecompute(querySupport);
- ReadGraphImpl parentGraph = graph.withParent(entry);
+ ReadGraphImpl parentGraph = graph.forRecompute(entry);
query.recompute(parentGraph);
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.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_CHANGES, Bindings.BOOLEAN)) {
+ System.out.println("C " + query);
+ System.out.println("- " + oldValue);
+ System.out.println("- " + newValue);
+ }
}
return newValue;
}
} 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.<Boolean>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;
for (ListenerEntry listenerEntry : entries) {
if (pruneListener(listenerEntry)) {
- if(DebugPolicy.LISTENER) System.out.println("Pruned " + listenerEntry.procedure);
+ if (Development.DEVELOPMENT) {
+ if(Development.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_LISTENERS, Bindings.BOOLEAN)) {
+ new Exception().printStackTrace();
+ System.err.println("Pruned " + listenerEntry.procedure);
+ }
+ }
continue;
}
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.<Boolean>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());
}
for(ListenerEntry listenerEntry : schedule) {
final CacheEntry entry = listenerEntry.entry;
- if(DebugPolicy.LISTENER)
- System.out.println("Firing " + listenerEntry.procedure);
+ if (Development.DEVELOPMENT) {
+ if(Development.<Boolean>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.<Boolean>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();
} else {
// If not changed, keep the old value
immediate.setResult(oldValue);
+ immediate.setReady();
listenersUnknown = true;
}
private Object primitiveUpdateLock = new Object();
private THashSet scheduledPrimitiveUpdates = new THashSet();
+ private ArrayList<CacheEntry> 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;
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.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_UPDATE, Bindings.BOOLEAN)) {
+ System.err.println("== Query update ends ==");
+ }
+ }
+
return;
}
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.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_UPDATE, Bindings.BOOLEAN)) {
+ System.err.println("== Query update ends ==");
+ }
+ }
+
return;
}
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;
}
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);
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);
@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;
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;
});
- // 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.<Boolean>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.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_UPDATE, Bindings.BOOLEAN)) {
+ System.err.println("== Query update ends ==");
+ }
+ }
}
@Override
final public void forEachPredicate(final ReadGraphImpl impl, final Resource subject, final AsyncMultiProcedure<Resource> 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);
+ }
}
Code Review / simantics / platform.git / blobdiff