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=b2951cb27e3c122322aeca97aabe124cac8b9702;hp=9c60691fad7a0a850228997e85ffb47168ca1aae;hb=ff1c29bf73b3e5ea939dc1987540aa0240dc4ae6;hpb=0d9b90834ce56b292c00b1a39850ed842c3e4d42

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 9c60691fa..b2951cb27 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
@@ -51,6 +51,7 @@ 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,23 +172,40 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
 	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;
@@ -199,11 +217,11 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
 //	}
 
 	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);
 
@@ -216,24 +234,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<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;
@@ -256,34 +270,47 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
 
 	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 + "]";
 		}
 
 	}
@@ -294,7 +321,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
 		final public DataContainer<Throwable> throwable; 
 
 		public SessionRead(DataContainer<Throwable> throwable, Semaphore notify) {
-			super(true);
+			super(null);
 			this.throwable = throwable;
 			this.notify = notify;
 		}
@@ -1593,7 +1620,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
 
 			entry.prepareRecompute(querySupport);
 			
-			ReadGraphImpl parentGraph = graph.withParent(entry);
+			ReadGraphImpl parentGraph = graph.forRecompute(entry);
 
 			query.recompute(parentGraph);
 
@@ -1765,6 +1792,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
 						} else {
 							// If not changed, keep the old value
 							immediate.setResult(oldValue);
+							immediate.setReady();
 							listenersUnknown = true;
 						}