Multiple readers in db client

[simantics/platform.git] / bundles / org.simantics.db.impl / src / org / simantics / db / impl / query / QueryThread.java

diff --git a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryThread.java b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryThread.java
index ab10efa0baab252a1ff5ef8b8485c313b2b32db1..cfa088a2dbfd7a966e0af3db7f4d9a5aa5d1ad4c 100644 (file)
--- a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryThread.java
+++ b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryThread.java
@@ -22,6 +22,7 @@ class QueryThread extends Thread implements SessionThread {
        private Session session;
        private QuerySupport querySupport;
        private final QueryProcessor processor;
+       private final Semaphore requests;
 
        final private ArrayList<SessionTask> tasks = new ArrayList<SessionTask>();
 //     final private ArrayList<SessionTask> own;
@@ -54,6 +55,7 @@ class QueryThread extends Thread implements SessionThread {
                sleepers = processor.sleepers;
                querySupport = processor.querySupport;
                threadStates = processor.threadStates;
+               requests = processor.requests;
 //             delayQueues = processor.delayQueues;
 //             executors = processor.executors;
 //             threadLocks = processor.threadLocks;
@@ -96,10 +98,16 @@ class QueryThread extends Thread implements SessionThread {
        }
 
        private boolean pumpTask() {
-               if(!processor.freeScheduling.isEmpty()) {
-                       tasks.add(processor.freeScheduling.removeFirst());
-                       return true;
-               }
+           // First finish existing executions
+        if(!processor.freeScheduling.isEmpty()) {
+            tasks.add(processor.freeScheduling.removeFirst());
+            return true;
+        }
+        // Check for new tasks
+        if(!processor.topLevelTasks.isEmpty()) {
+            tasks.add(processor.topLevelTasks.removeFirst());
+            return true;
+        }
                return false;
        }
        
@@ -155,18 +163,7 @@ class QueryThread extends Thread implements SessionThread {
                                
                                threadStates[index] = ThreadState.SLEEP;
                                
-                               synchronized (querySupportLock) {
-                                       querySupportLock.wait(100);
-                                       
-                               }
-                               
-//                             boolean woken = condition.await(10, TimeUnit.MILLISECONDS);
-//                             if(!woken) {
-//                                     synchronized (querySupportLock) {
-//                                             if(!processor.freeScheduling.isEmpty())
-//                                                     System.err.println("some tasks are available!");
-//                                     }
-//                             }
+                               requests.acquire();
 
                                sleepers.decrementAndGet();
 
@@ -234,8 +231,6 @@ class QueryThread extends Thread implements SessionThread {
        @Override
        public void run() {
 
-               processor.thread.set(index);
-               
                QuerySupport support = this.querySupport;
 
                try {
@@ -250,23 +245,10 @@ class QueryThread extends Thread implements SessionThread {
                                while(!tasks.isEmpty()) {
 
                                        SessionTask task = tasks.remove(tasks.size()-1);
-//                                     System.err.println("QT " + index + " runs " + task);
-                                       task.run(index);
+                                       task.run(0);
 
                                }
 
-//                             for(int performer=0;performer<THREADS;performer++) {
-//                                     if(!delayQueues[index * THREADS + performer].isEmpty()) {
-//                                             synchronized(executors[performer]) {
-//                                                     threadLocks[performer].lock();
-//                                                     queues[performer].addAll(delayQueues[index * THREADS + performer]);
-//                                                     delayQueues[index * THREADS + performer].clear();
-//                                                     executors[performer].notify();
-//                                                     threadLocks[performer].unlock();
-//                                             }
-//                                     }
-//                             }
-
                        }
 
                } catch (Throwable t) {