Merge branch 'private/db-threads' into private/threads-testing

[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 242fcfe3de8838867d4f7a8fe59ccdaa95eaa0ea..d5d9445ca4fb230520769af6f7cb1e4d0ea0e8fc 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
@@ -50,6 +50,7 @@ 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.wrapper.NoneToAsyncListener;
@@ -411,7 +412,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
 
         int thread = request.hashCode() & queryProvider2.THREAD_MASK;
 
-        requestManager.scheduleWrite(new SessionTask(request, thread, thread) {
+        requestManager.scheduleWrite(new SessionTask(true) {
 
             @Override
             public void run(int thread) {
@@ -549,7 +550,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
 
         int thread = request.hashCode() & queryProvider2.THREAD_MASK;
 
-        requestManager.scheduleWrite(new SessionTask(request, thread) {
+        requestManager.scheduleWrite(new SessionTask(true) {
 
             @Override
             public void run(int thread) {
@@ -635,7 +636,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
 
         int thread = request.hashCode() & queryProvider2.THREAD_MASK;
 
-        requestManager.scheduleWrite(new SessionTask(request, thread) {
+        requestManager.scheduleWrite(new SessionTask(true) {
 
             @Override
             public void run(int thread) {
@@ -1376,7 +1377,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
 
         int thread = request.hashCode() & queryProvider2.THREAD_MASK;
 
-        requestManager.scheduleWrite(new SessionTask(request, thread, thread) {
+        requestManager.scheduleWrite(new SessionTask(true) {
 
             @Override
             public void run(int thread) {
@@ -1476,7 +1477,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
 
         int thread = request.hashCode() & queryProvider2.THREAD_MASK;
 
-        requestManager.scheduleWrite(new SessionTask(request, thread) {
+        requestManager.scheduleWrite(new SessionTask(true) {
 
             @Override
             public void run(int thread) {
@@ -1498,9 +1499,9 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
         assert (request != null);
         assert (procedure != null);
 
-        int thread = request.hashCode() & queryProvider2.THREAD_MASK;
+        //int thread = request.hashCode() & queryProvider2.THREAD_MASK;
 
-        requestManager.scheduleRead(new SessionRead(request, throwable, notify, thread, thread) {
+        requestManager.scheduleRead(new SessionRead(throwable, notify) {
 
             @Override
             public void run(int thread) {
@@ -1537,7 +1538,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
 
                                };
                                
-                               QueryCache.runnerReadEntry(newGraph, request, null, listener, ap);
+                               QueryCache.runnerReadEntry(newGraph, request, null, listener, ap, true);
                                
                        } catch (Throwable t) {
                            // This is handled by the AsyncProcedure
@@ -1618,7 +1619,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
 
         int thread = request.hashCode() & queryProvider2.THREAD_MASK;
 
-        requestManager.scheduleRead(new SessionRead(request, null, notify, thread) {
+        requestManager.scheduleRead(new SessionRead(null, notify) {
 
             @Override
             public void run(int thread) {
@@ -1632,7 +1633,8 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
                     if (listener != null) {
 
                         try {
-                               QueryCache.runnerAsyncReadEntry(newGraph, request, null, listener, procedure);
+                               QueryCacheBase.resultAsyncReadEntry(newGraph, request, null, listener, procedure);
+                               //QueryCache.runnerAsyncReadEntry(newGraph, request, null, listener, procedure, true);
                                                        //newGraph.processor.query(newGraph, request, null, procedure, listener);
                                                } catch (DatabaseException e) {
                                                        Logger.defaultLogError(e);
@@ -1640,20 +1642,21 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
 
                     } else {
 
-                        final ResultCallWrappedSingleQueryProcedure4<T> wrapper = new ResultCallWrappedSingleQueryProcedure4<T>(
-                                procedure, "request");
-
-                        try {
+//                        final ResultCallWrappedSingleQueryProcedure4<T> wrapper = new ResultCallWrappedSingleQueryProcedure4<T>(
+//                                procedure, "request");
 
-//                            newGraph.state.barrier.inc();
+                       BlockingAsyncProcedure<T> wrap = new BlockingAsyncProcedure<T>(newGraph, procedure, request);
 
-                            request.perform(newGraph, wrapper);
+                       try {
 
-//                            newGraph.waitAsync(request);
+                            request.perform(newGraph, wrap);
+                            wrap.get();
 
                         } catch (Throwable t) {
 
-                            wrapper.exception(newGraph, t);
+                               wrap.exception(newGraph, t);
+                               
+//                            wrapper.exception(newGraph, t);
 //                            newGraph.waitAsync(request);
 
 
@@ -1682,7 +1685,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
 
         int sync = notify != null ? thread : -1;
 
-        requestManager.scheduleRead(new SessionRead(request, null, notify, thread, sync) {
+        requestManager.scheduleRead(new SessionRead(null, notify) {
 
             @Override
             public void run(int thread) {
@@ -1736,7 +1739,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
 
         int thread = request.hashCode() & queryProvider2.THREAD_MASK;
 
-        requestManager.scheduleRead(new SessionRead(request, throwable, notify, thread, thread) {
+        requestManager.scheduleRead(new SessionRead(throwable, notify) {
 
             @Override
             public void run(int thread) {
@@ -3523,7 +3526,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
 
     public int getAmountOfQueryThreads() {
         // This must be a power of two
-        return 4;
+        return 8;
 //        return Integer.highestOneBit(Runtime.getRuntime().availableProcessors());
     }