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%2FAsyncReadEntry.java;h=0dd5730a5f3a1f5c6c232063741bf4343303feb4;hp=e13ecab72cd0819f71c3c337af62ad242db709c2;hb=90a52d58489b6c518b13dde238923125c8f63eec;hpb=bb61be97905f72c01fd99e21c263546c88edc5f7

diff --git a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/AsyncReadEntry.java b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/AsyncReadEntry.java
index e13ecab72..0dd5730a5 100644
--- a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/AsyncReadEntry.java
+++ b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/AsyncReadEntry.java
@@ -11,13 +11,13 @@
  *******************************************************************************/
 package org.simantics.db.impl.query;
 
+import java.util.Collection;
+
 import org.simantics.databoard.Bindings;
 import org.simantics.db.AsyncReadGraph;
 import org.simantics.db.DevelopmentKeys;
 import org.simantics.db.exception.DatabaseException;
 import org.simantics.db.impl.BlockingAsyncProcedure;
-import org.simantics.db.impl.graph.AsyncBarrierImpl;
-import org.simantics.db.impl.graph.BarrierTracing;
 import org.simantics.db.impl.graph.ReadGraphImpl;
 import org.simantics.db.impl.query.QueryProcessor.SessionTask;
 import org.simantics.db.procedure.AsyncProcedure;
@@ -26,11 +26,12 @@ import org.simantics.utils.Development;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-final public class AsyncReadEntry<T> extends CacheEntryBase<AsyncProcedure<T>> implements AsyncProcedure<T> {
+final public class AsyncReadEntry<T> extends CacheEntryBase<AsyncProcedure<T>> implements AsyncProcedure<T>, IPending {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(AsyncReadEntry.class);
 
     protected AsyncRead<T> id;
+    protected PendingTaskSupport pendingTaskSupport;
 
     AsyncReadEntry(AsyncRead<T> request) {
         this.id = request;
@@ -77,7 +78,7 @@ final public class AsyncReadEntry<T> extends CacheEntryBase<AsyncProcedure<T>> i
 
                 try {
 
-                    BlockingAsyncProcedure<T> proc = new BlockingAsyncProcedure<>(graph.asyncBarrier, graph, new AsyncProcedure<T>() {
+                    BlockingAsyncProcedure<T> proc = new BlockingAsyncProcedure(graph, AsyncReadEntry.this, new AsyncProcedure<T>() {
 
                         @Override
                         public void execute(AsyncReadGraph graph, T result) {
@@ -90,10 +91,11 @@ final public class AsyncReadEntry<T> extends CacheEntryBase<AsyncProcedure<T>> i
                             except(t);
                         }
 
-                    }, id);
-
-                    id.perform(graph, proc);
+                    }, id, true);
 
+                    id.perform(proc.queryGraph, proc);
+                    
+                    proc.dec();
                     proc.get();
 
                 } catch (Throwable t) {
@@ -152,103 +154,24 @@ final public class AsyncReadEntry<T> extends CacheEntryBase<AsyncProcedure<T>> i
 
     }
 
-    public static <T> T computeForEach(ReadGraphImpl graph, AsyncRead<T> request, AsyncReadEntry<T> entry,
+    public static <T> T computeForEach(ReadGraphImpl callerGraph, AsyncRead<T> request, AsyncReadEntry<T> entry,
             AsyncProcedure<T> procedure_, boolean needsToBlock) throws DatabaseException {
 
-        AsyncProcedure<T> procedure = entry != null ? entry : procedure_;
-
-        ReadGraphImpl queryGraph = graph.withParent(entry);
-        queryGraph.asyncBarrier.inc();
-        
-        BlockingAsyncProcedure<T> proc = new BlockingAsyncProcedure<>(queryGraph.asyncBarrier, graph, null, request);
-        
-        class AsyncTask extends SessionTask {
-
-            int counter = 0;
-            T result;
-            DatabaseException exception;
-            
-            public AsyncTask(ReadGraphImpl graph) {
-                this(graph, 1);
-            }
-
-            public AsyncTask(ReadGraphImpl graph, int pos) {
-                super(graph);
-                this.position = pos;
-                if(this.position < 1024)
-                    this.position *= 2;
-            }
-
-            @Override
-            public void run0(int thread) {
-                if(needsToBlock) proc.waitBarrier();
-                if(proc.isDone()) {
-                    ReadGraphImpl executeGraph = graph.withParent(graph.parent);
-                    executeGraph.asyncBarrier.inc();
-                    try {
-                        result = (T)proc.get();
-                        if(procedure != null) {
-                            procedure.execute(executeGraph, result);
-                        }
-                    } catch (DatabaseException e) {
-                        if(procedure != null) procedure.exception(executeGraph, e);
-                        exception = e;
-                    } catch (Throwable t) {
-                        DatabaseException dbe = new DatabaseException(t);
-                        if(procedure != null) procedure.exception(executeGraph, dbe);
-                        exception = dbe;
-                    } finally {
-                        if (entry != null) {
-                            // This does not throw
-                            entry.performFromCache(executeGraph, procedure_);
-                        }
-                        executeGraph.asyncBarrier.dec();
-                        executeGraph.asyncBarrier.waitBarrier(procedure, executeGraph);
-                    }
-                } else {
-                    if(counter++ > 10000) {
-                        if(BarrierTracing.BOOKKEEPING) {
-                            AsyncBarrierImpl.printReverse(queryGraph.asyncBarrier, 2);
-                            AsyncBarrierImpl caller = queryGraph.asyncBarrier.caller;
-                            while(caller != null) {
-                                System.err.println("called by " + AsyncBarrierImpl.report(caller));
-                                caller = caller.caller;
-                            }
-                            for(AsyncBarrierImpl ab : BarrierTracing.debuggerMap.keySet()) {
-                                AsyncBarrierImpl.printReverse(ab, 2);
-                            }
-                        }
-                        throw new IllegalStateException("Eternal loop in queries.");
-                    }
-                    graph.processor.scheduleLater(new AsyncTask(graph, position));
-                }
-            }
-            
-            @Override
-            public boolean maybeReady() {
-            	return proc.isDone();
-            }
-            
-        }
+        BlockingAsyncProcedure<T> proc = new BlockingAsyncProcedure(callerGraph, entry, procedure_, request, needsToBlock);
 
         try {
-            request.perform(queryGraph, proc);
+            request.perform(proc.queryGraph, proc);
         } finally {
-            queryGraph.asyncBarrier.dec();
+            proc.queryGraph.asyncBarrier.dec();
         }
 
-        AsyncTask task = new AsyncTask(graph);
-
-        if(needsToBlock) task.run(0);
-        else if (proc.isDone()) task.run(0);
-        else  {
-            graph.processor.scheduleLater(task);
+        if(needsToBlock) {
+            proc.waitBarrier();
+            return proc.get();
+        } else {
             return null;
         }
 
-        if(task.exception != null) throw task.exception;
-        else return task.result;
-
     }
 
     @Override
@@ -263,13 +186,41 @@ final public class AsyncReadEntry<T> extends CacheEntryBase<AsyncProcedure<T>> i
 
     @Override
     public void execute(AsyncReadGraph graph, T result) {
-        setResult(result);
-        setReady();
+        Collection<SessionTask> tasks = null;
+        synchronized(this) {
+            setResult(result);
+            setReady();
+            if(pendingTaskSupport != null)
+                tasks = pendingTaskSupport.executePending();
+        }
+        if(tasks != null)
+            for(SessionTask task : tasks)
+                ((ReadGraphImpl)graph).processor.scheduleNow(task);
     }
 
     @Override
-    public void exception(AsyncReadGraph graph, Throwable throwable) {
-        except(throwable);
+    public synchronized void exception(AsyncReadGraph graph, Throwable throwable) {
+        Collection<SessionTask> tasks = null;
+        synchronized(this) {
+            except(throwable);
+            if(pendingTaskSupport != null)
+                tasks = pendingTaskSupport.executePending();
+        }
+        if(tasks != null)
+            for(SessionTask task : tasks)
+                ((ReadGraphImpl)graph).processor.scheduleNow(task);
+    }
+
+    public void executeWhenResultIsAvailable(QueryProcessor processor, SessionTask task) {
+        boolean ready = false;
+        synchronized(this) {
+            if(pendingTaskSupport == null)
+                pendingTaskSupport = new PendingTaskSupport(this);
+            ready = pendingTaskSupport.executeWhenResultIsAvailable(task);
+        }
+        if(ready) {
+            processor.scheduleNow(task);
+        }
     }
 
 }