Working towards multiple readers.

[simantics/platform.git] / bundles / org.simantics.db.procore / src / fi / vtt / simantics / procore / internal / SessionImplSocket.java

diff --git a/bundles/org.simantics.db.procore/src/fi/vtt/simantics/procore/internal/SessionImplSocket.java b/bundles/org.simantics.db.procore/src/fi/vtt/simantics/procore/internal/SessionImplSocket.java
index f48ccaf6b799a8348a2b8e0341b26a79bb243cee..6d34d9341c425bbdd465616af54a12d6e3e62cf4 100644 (file)
--- a/bundles/org.simantics.db.procore/src/fi/vtt/simantics/procore/internal/SessionImplSocket.java
+++ b/bundles/org.simantics.db.procore/src/fi/vtt/simantics/procore/internal/SessionImplSocket.java
@@ -39,6 +39,7 @@ import org.simantics.db.ExternalValueSupport;
 import org.simantics.db.Metadata;
 import org.simantics.db.MonitorContext;
 import org.simantics.db.MonitorHandler;
+import org.simantics.db.ReadGraph;
 import org.simantics.db.Resource;
 import org.simantics.db.ResourceSerializer;
 import org.simantics.db.Session;
@@ -50,13 +51,15 @@ import org.simantics.db.authentication.UserAuthenticationAgent;
 import org.simantics.db.authentication.UserAuthenticator;
 import org.simantics.db.common.Indexing;
 import org.simantics.db.common.TransactionPolicyRelease;
-import org.simantics.db.common.procedure.BlockingAsyncProcedure;
 import org.simantics.db.common.procedure.adapter.AsyncMultiProcedureAdapter;
 import org.simantics.db.common.procedure.adapter.ProcedureAdapter;
+import org.simantics.db.common.procedure.adapter.SyncMultiProcedureAdapter;
 import org.simantics.db.common.procedure.wrapper.NoneToAsyncListener;
 import org.simantics.db.common.procedure.wrapper.NoneToAsyncMultiListener;
 import org.simantics.db.common.procedure.wrapper.NoneToAsyncMultiProcedure;
 import org.simantics.db.common.procedure.wrapper.NoneToAsyncProcedure;
+import org.simantics.db.common.procedure.wrapper.NoneToSyncMultiListener;
+import org.simantics.db.common.procedure.wrapper.NoneToSyncMultiProcedure;
 import org.simantics.db.common.procedure.wrapper.SyncToAsyncListener;
 import org.simantics.db.common.procedure.wrapper.SyncToAsyncMultiListener;
 import org.simantics.db.common.procedure.wrapper.SyncToAsyncMultiProcedure;
@@ -76,6 +79,7 @@ import org.simantics.db.exception.ResourceNotFoundException;
 import org.simantics.db.exception.RuntimeDatabaseException;
 import org.simantics.db.exception.ServiceException;
 import org.simantics.db.exception.ServiceNotFoundException;
+import org.simantics.db.impl.BlockingAsyncProcedure;
 import org.simantics.db.impl.ClusterBase;
 import org.simantics.db.impl.ClusterI;
 import org.simantics.db.impl.ClusterTraitsBase;
@@ -89,7 +93,7 @@ import org.simantics.db.impl.graph.WriteGraphImpl;
 import org.simantics.db.impl.graph.WriteSupport;
 import org.simantics.db.impl.internal.RandomAccessValueSupport;
 import org.simantics.db.impl.procedure.ResultCallWrappedQueryProcedure4;
-import org.simantics.db.impl.procedure.ResultCallWrappedSingleQueryProcedure4;
+import org.simantics.db.impl.procedure.ResultCallWrappedSyncQueryProcedure;
 import org.simantics.db.impl.query.QueryCache;
 import org.simantics.db.impl.query.QueryCacheBase;
 import org.simantics.db.impl.query.QueryProcessor;
@@ -1674,6 +1678,62 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
 
     }
 
+    public <T> void scheduleRequest(final MultiRead<T> request, final SyncMultiProcedure<T> procedure, final Semaphore notify) {
+
+        assert (request != null);
+        assert (procedure != null);
+
+        int thread = request.hashCode() & queryProvider2.THREAD_MASK;
+
+        int sync = notify != null ? thread : -1;
+
+        requestManager.scheduleRead(new SessionRead(null, notify) {
+
+            @Override
+            public void run(int thread) {
+
+                fireSessionVariableChange(SessionVariables.QUEUED_READS);
+
+                ListenerBase listener = getListenerBase(procedure);
+
+                final ReadGraphImpl newGraph = ReadGraphImpl.create(getQueryProvider2());
+
+                try {
+
+                    if (listener != null) {
+
+                        newGraph.processor.query(newGraph, request, null, procedure, listener);
+
+//                        newGraph.waitAsync(request);
+
+                    } else {
+
+                        final ResultCallWrappedSyncQueryProcedure<T> wrapper = new ResultCallWrappedSyncQueryProcedure<T>(procedure);
+
+                        try {
+
+                            request.perform(newGraph, wrapper);
+
+                        } catch (Throwable t) {
+
+                            t.printStackTrace();
+
+                        }
+
+                    }
+
+                } finally {
+
+                    fireSessionVariableChange(SessionVariables.QUEUED_READS);
+
+                }
+
+            }
+
+        });
+
+    }
+    
     public <T> void scheduleRequest(final AsyncMultiRead<T> request, final AsyncMultiProcedure<T> procedure, final Semaphore notify) {
 
         assert (request != null);
@@ -3067,21 +3127,21 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
         final ArrayList<T> result = new ArrayList<T>();
         final DataContainer<Throwable> exception = new DataContainer<Throwable>();
 
-        syncRequest(request, new AsyncMultiProcedure<T>() {
+        syncRequest(request, new SyncMultiProcedure<T>() {
 
             @Override
-            public void execute(AsyncReadGraph graph, T t) {
+            public void execute(ReadGraph graph, T t) {
                 synchronized(result) {
                     result.add(t);
                 }
             }
 
             @Override
-            public void finished(AsyncReadGraph graph) {
+            public void finished(ReadGraph graph) {
             }
 
             @Override
-            public void exception(AsyncReadGraph graph, Throwable t) {
+            public void exception(ReadGraph graph, Throwable t) {
                 exception.set(t);
             }
 
@@ -3099,33 +3159,27 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
     }
 
     @Override
-    public <T> Collection<T> syncRequest(MultiRead<T> request, AsyncMultiListener<T> procedure) throws DatabaseException {
+    public <T> Collection<T> syncRequest(MultiRead<T> arg0, SyncMultiProcedure<T> arg1) throws DatabaseException {
         assertNotSession();
-        return syncRequest(request, (AsyncMultiProcedure<T>)procedure);
+        throw new Error("Not implemented!");
     }
 
     @Override
     public <T> Collection<T> syncRequest(MultiRead<T> request, SyncMultiListener<T> procedure) throws DatabaseException {
         assertNotSession();
-        return syncRequest(request, new SyncToAsyncMultiListener<T>(procedure));
+        return syncRequest(request, (SyncMultiProcedure<T>)procedure);
     }
 
     @Override
     public <T> Collection<T> syncRequest(MultiRead<T> request, MultiListener<T> procedure) throws DatabaseException {
         assertNotSession();
-        return syncRequest(request, new NoneToAsyncMultiListener<T>(procedure));
-    }
-
-    @Override
-    public <T> Collection<T> syncRequest(MultiRead<T> request, SyncMultiProcedure<T> procedure) throws DatabaseException {
-        assertNotSession();
-        return syncRequest(request, new SyncToAsyncMultiProcedure<T>(procedure));
+        return syncRequest(request, new NoneToSyncMultiListener<T>(procedure));
     }
 
     @Override
     public <T> Collection<T> syncRequest(MultiRead<T> request, MultiProcedure<T> procedure) throws DatabaseException {
         assertNotSession();
-        return syncRequest(request, new NoneToAsyncMultiProcedure<T>(procedure));
+        return syncRequest(request, new NoneToSyncMultiProcedure<T>(procedure));
     }
 
     @Override
@@ -3281,38 +3335,33 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
 
         assert(request != null);
 
-        asyncRequest(request, new AsyncMultiProcedureAdapter<T>() {
+        asyncRequest(request, new SyncMultiProcedureAdapter<T>() {
             @Override
-            public void exception(AsyncReadGraph graph, Throwable t) {
+            public void exception(ReadGraph graph, Throwable t) {
                 t.printStackTrace();
             }
         });
 
     }
 
-    @Override
-    public <T> void asyncRequest(MultiRead<T> request, AsyncMultiListener<T> procedure) {
-        asyncRequest(request, (AsyncMultiProcedure<T>)procedure);
-    }
-
     @Override
     public <T> void asyncRequest(MultiRead<T> request, SyncMultiListener<T> procedure) {
-        asyncRequest(request, new SyncToAsyncMultiListener<T>(procedure));
+        asyncRequest(request, (SyncMultiProcedure<T>)procedure);
     }
 
     @Override
     public <T> void asyncRequest(MultiRead<T> request, MultiListener<T> procedure) {
-        asyncRequest(request, new NoneToAsyncMultiListener<T>(procedure));
+        asyncRequest(request, new NoneToSyncMultiListener<T>(procedure));
     }
 
     @Override
     public <T> void asyncRequest(MultiRead<T> request, SyncMultiProcedure<T> procedure) {
-        asyncRequest(request, new SyncToAsyncMultiProcedure<T>(procedure));
+       scheduleRequest(request, procedure, null);
     }
 
     @Override
     public <T> void asyncRequest(MultiRead<T> request, MultiProcedure<T> procedure) {
-        asyncRequest(request, new NoneToAsyncMultiProcedure<T>(procedure));
+        asyncRequest(request, new NoneToSyncMultiProcedure<T>(procedure));
     }
 
     @Override
@@ -3354,17 +3403,6 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
         asyncRequest(request, new NoneToAsyncMultiProcedure<T>(procedure));
     }
 
-    @Override
-    public <T> Collection<T> syncRequest(MultiRead<T> arg0, AsyncMultiProcedure<T> arg1) throws DatabaseException {
-        assertNotSession();
-        throw new Error("Not implemented!");
-    }
-
-    @Override
-    public <T> void asyncRequest(MultiRead<T> arg0, AsyncMultiProcedure<T> arg1) {
-        throw new Error("Not implemented!");
-    }
-
     @Override
     final public <T> void asyncRequest(final ExternalRead<T> request) {
 
@@ -3524,7 +3562,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
 
     public int getAmountOfQueryThreads() {
         // This must be a power of two
-        return 16;
+        return 1;
 //        return Integer.highestOneBit(Runtime.getRuntime().availableProcessors());
     }