Do not wait forever for querythreads to dispose

[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..79124893137d67dc058892edce823eb993ed939c 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
@@ -2,6 +2,7 @@ package org.simantics.db.impl.query;
 
 import java.util.ArrayList;
 import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.simantics.db.Session;
@@ -33,6 +34,8 @@ class QueryThread extends Thread implements SessionThread {
        final private int THREADS;
        final private AtomicInteger sleepers;
        final private ThreadState[] threadStates;
+
+       private SessionTask currentTask;
 //     final private ArrayList<SessionTask>[] delayQueues;
 //     final private QueryThread[] executors;
 //     final private ReentrantLock[] threadLocks;
@@ -71,7 +74,12 @@ class QueryThread extends Thread implements SessionThread {
 //             lock.unlock();
                
                try {
-                       exited.acquire();
+                       // we are not willing to wait forever here..
+                       boolean acquired = exited.tryAcquire(60, TimeUnit.SECONDS);
+                       if (!acquired) {
+                               LOGGER.error("Could not safely dispose query thread {} - we were processing task {} and still have tasks to do {}", index, currentTask.graph != null ? String.valueOf(currentTask) : "SessionTask with null graph", tasks.size());
+                               interrupt();
+                       }
                } catch (InterruptedException e) {
                        LOGGER.error("dispose was interrupted", e);
                }
@@ -215,14 +223,15 @@ class QueryThread extends Thread implements SessionThread {
 
                while(!tasks.isEmpty()) {
 
-                       SessionTask task = tasks.remove(tasks.size() - 1);
+                       SessionTask t = currentTask = tasks.remove(tasks.size() - 1);
 
 //                     if(task.syncCaller == index) {
 //                             ownSyncTasks[index].add(task);
 //                     } else {
-                               task.run(index);
+                               t.run(index);
 //                             System.err.println("QT(s) " + index + " runs " + task);
                                didExecute = true;
+                               currentTask = null;
 //                     }
 
                }
@@ -249,10 +258,10 @@ class QueryThread extends Thread implements SessionThread {
                                
                                while(!tasks.isEmpty()) {
 
-                                       SessionTask task = tasks.remove(tasks.size()-1);
+                                       SessionTask t = currentTask = tasks.remove(tasks.size()-1);
 //                                     System.err.println("QT " + index + " runs " + task);
-                                       task.run(index);
-
+                                       t.run(index);
+                                       currentTask = null;
                                }
 
 //                             for(int performer=0;performer<THREADS;performer++) {