X-Git-Url: https://gerrit.simantics.org/r/gitweb?a=blobdiff_plain;ds=sidebyside;f=bundles%2Forg.simantics.db.impl%2Fsrc%2Forg%2Fsimantics%2Fdb%2Fimpl%2Fquery%2FAsyncReadEntry.java;h=0dd5730a5f3a1f5c6c232063741bf4343303feb4;hb=a88c02c2d0f4250caf887a130b9f4314c6564722;hp=00abbebf65859e76d0e8dad481d01097cab04621;hpb=9f0fd59be54719b1fe9322d8fd37e4950857308c;p=simantics%2Fplatform.git

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 00abbebf6..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,37 +11,45 @@
  *******************************************************************************/
 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.exception.RuntimeDatabaseException;
 import org.simantics.db.impl.BlockingAsyncProcedure;
-import org.simantics.db.impl.DebugPolicy;
 import org.simantics.db.impl.graph.ReadGraphImpl;
+import org.simantics.db.impl.query.QueryProcessor.SessionTask;
 import org.simantics.db.procedure.AsyncProcedure;
 import org.simantics.db.request.AsyncRead;
+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> request;
+    protected AsyncRead<T> id;
+    protected PendingTaskSupport pendingTaskSupport;
 
     AsyncReadEntry(AsyncRead<T> request) {
-        this.request = request;
-        if (DebugPolicy.QUERY_STATE)
-            System.out.println("[QUERY STATE]: created " + this);
+        this.id = request;
+		if (Development.DEVELOPMENT) {
+			if(Development.<Boolean>getProperty(DevelopmentKeys.CACHE_ENTRY_STATE, Bindings.BOOLEAN)) {
+		        System.err.println("[QUERY STATE]: created " + this);
+			}
+		}
     }
 
     @Override
     int makeHash() {
-        return request.hashCode();
+        return id.hashCode();
     }
 
     @Override
     public Object getOriginalRequest() {
-        return request;
+        return id;
     }
 
     @Override
@@ -70,7 +78,7 @@ final public class AsyncReadEntry<T> extends CacheEntryBase<AsyncProcedure<T>> i
 
                 try {
 
-                    BlockingAsyncProcedure<T> proc = new BlockingAsyncProcedure<>(graph, new AsyncProcedure<T>() {
+                    BlockingAsyncProcedure<T> proc = new BlockingAsyncProcedure(graph, AsyncReadEntry.this, new AsyncProcedure<T>() {
 
                         @Override
                         public void execute(AsyncReadGraph graph, T result) {
@@ -83,10 +91,11 @@ final public class AsyncReadEntry<T> extends CacheEntryBase<AsyncProcedure<T>> i
                             except(t);
                         }
 
-                    }, request);
-
-                    request.perform(graph, proc);
+                    }, id, true);
 
+                    id.perform(proc.queryGraph, proc);
+                    
+                    proc.dec();
                     proc.get();
 
                 } catch (Throwable t) {
@@ -102,17 +111,17 @@ final public class AsyncReadEntry<T> extends CacheEntryBase<AsyncProcedure<T>> i
 
             @Override
             public int type() {
-                return request.getFlags();
+                return id.getFlags();
             }
 
             @Override
             public String toString() {
-                if (request == null)
+                if (id == null)
                     return "DISCARDED";
                 else if (isExcepted())
-                    return request.toString() + " " + getResult();
+                    return id.toString() + " " + getResult();
                 else
-                    return request.toString() + " " + statusOrException;
+                    return id.toString() + " " + statusOrException;
             }
 
         };
@@ -133,7 +142,8 @@ final public class AsyncReadEntry<T> extends CacheEntryBase<AsyncProcedure<T>> i
         } else {
 
             try {
-                proc.execute(graph, (T) getResult());
+                T result = (T) getResult();
+                proc.execute(graph, result);
             } catch (Throwable t) {
                 LOGGER.error("performFromCache proc.execute failed", t);
             }
@@ -144,68 +154,73 @@ final public class AsyncReadEntry<T> extends CacheEntryBase<AsyncProcedure<T>> i
 
     }
 
-    public static <T> void computeForEach(ReadGraphImpl parentGraph, AsyncRead<T> request, AsyncReadEntry<T> entry,
-            AsyncProcedure<T> procedure_) throws DatabaseException {
-
-        AsyncProcedure<T> procedure = entry != null ? entry : procedure_;
-
-        ReadGraphImpl queryGraph = parentGraph.withParent(entry);
-        
-        BlockingAsyncProcedure<T> proc = new BlockingAsyncProcedure<>(queryGraph, new AsyncProcedure<T>() {
-
-            @Override
-            public void execute(AsyncReadGraph returnGraph, T result) {
-                try {
-                    procedure.execute(parentGraph, result);
-                } catch (Throwable t) {
-                    LOGGER.error("computeForEach procedure.execute failed", t);
-                }
-            }
-
-            @Override
-            public void exception(AsyncReadGraph returnGraph, Throwable t) {
-                try {
-                    procedure.exception(parentGraph, t);
-                } catch (Throwable t2) {
-                    LOGGER.error("computeForEach procedure.exception failed", t2);
-                }
-            }
-
-            @Override
-            public String toString() {
-                return procedure.toString();
-            }
+    public static <T> T computeForEach(ReadGraphImpl callerGraph, AsyncRead<T> request, AsyncReadEntry<T> entry,
+            AsyncProcedure<T> procedure_, boolean needsToBlock) throws DatabaseException {
 
-        }, request);
+        BlockingAsyncProcedure<T> proc = new BlockingAsyncProcedure(callerGraph, entry, procedure_, request, needsToBlock);
 
-        request.perform(queryGraph, proc);
-        
-        proc.get();
+        try {
+            request.perform(proc.queryGraph, proc);
+        } finally {
+            proc.queryGraph.asyncBarrier.dec();
+        }
 
-        if (entry != null)
-            entry.performFromCache(parentGraph, procedure_);
+        if(needsToBlock) {
+            proc.waitBarrier();
+            return proc.get();
+        } else {
+            return null;
+        }
 
     }
 
     @Override
     public String toString() {
         if (isDiscarded())
-            return "DISCARDED " + request.toString();
+            return "DISCARDED " + id.toString();
         else if (isExcepted())
-            return request.toString() + " " + getResult();
+            return id.toString() + " " + getResult();
         else
-            return request.toString() + " " + statusOrException;
+            return id.toString() + " " + statusOrException;
     }
 
     @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);
+        }
     }
 
 }