X-Git-Url: https://gerrit.simantics.org/r/gitweb?a=blobdiff_plain;f=bundles%2Forg.simantics.db.impl%2Fsrc%2Forg%2Fsimantics%2Fdb%2Fimpl%2Fquery%2FQueryProcessor.java;h=5a37257e2531c051c39de0bea04af271e1adccce;hb=d85a4d990c5cac9d0c70781a265f02888b3aaa43;hp=65b277e6c8d2985410b82e12ce4edca3f1fea09d;hpb=277b1c7fb3fdd8effb4bf2b447358d0e4ef0e302;p=simantics%2Fplatform.git

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..5a37257e2 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,7 +172,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
 	public void close() {
 	}
 
-	SessionTask getOwnTask(ReadGraphImpl impl) {
+	public SessionTask getOwnTask(ReadGraphImpl impl) {
 		Set<ReadGraphImpl> ancestors = impl.ancestorSet();
 		synchronized(querySupportLock) {
 			int index = 0;
@@ -186,6 +187,21 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
 		return null;
 	}
 
+    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) {
@@ -201,7 +217,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
 //	}
 
 	final public void schedule(SessionTask request) {
-
+	    
 		//int performer = request.thread;
 
 //		if(DebugPolicy.SCHEDULE)
@@ -218,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;
@@ -265,9 +277,12 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
 
 		public final ReadGraphImpl graph;
 		private Set<ReadGraphImpl> 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<ReadGraphImpl> otherAncestors) {
@@ -276,7 +291,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() {
@@ -2266,67 +2296,16 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
 	@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);
+	    }
 
 	}