From: jsimomaa <jani.simomaa@gmail.com>
Date: Thu, 21 Nov 2019 14:11:48 +0000 (+0200)
Subject: Do not wait forever for querythreads to dispose
X-Git-Url: https://gerrit.simantics.org/r/gitweb?a=commitdiff_plain;h=refs%2Fchanges%2F99%2F3599%2F4;p=simantics%2Fplatform.git

Do not wait forever for querythreads to dispose

gitlab #421

Change-Id: I0b5a1bba5b08a9c8853dbdd974209a8095449945
---

diff --git a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryProcessor.java b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryProcessor.java
index 9b54d15f6..31a4a6e19 100644
--- a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryProcessor.java
+++ b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryProcessor.java
@@ -309,7 +309,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
 
 		@Override
 		public String toString() {
-			return "SessionTask[" + graph.parent + "]";
+			return "SessionTask[" + graph != null ? String.valueOf(graph.parent) : "null graph" + "]";
 		}
 
 	}
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 ab10efa0b..791248931 100644
--- 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++) {
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 d32c066bc..aff219622 100644
--- 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
@@ -1345,7 +1345,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
 
         } catch (Throwable e) {
 
-            e.printStackTrace();
+            LOGGER.error("Could not perform write only for request {}", request, e);
 
             releaseWriteOnly(writeState.getGraph());
 
@@ -2671,7 +2671,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
                 // Serialize as '<resource index>_<cluster id>'
                 return "" + r + "_" + getCluster(resourceImpl);
             } catch (Throwable e) {
-                e.printStackTrace();
+                LOGGER.error("Could not create random access id for resource {}", resource, e);
                 throw new InvalidResourceReferenceException(e);
             } finally {
             }
@@ -2719,7 +2719,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
             } catch (NumberFormatException e) {
                 throw new InvalidResourceReferenceException(e);
             } catch (Throwable e) {
-                e.printStackTrace();
+                LOGGER.error("Could not get resource for randomAccessId {}", randomAccessId, e);
                 throw new InvalidResourceReferenceException(e);
             } finally {
             }
@@ -3462,7 +3462,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule
 
             }
         } catch (InterruptedException e) {
-            e.printStackTrace();
+            LOGGER.error("Acquiring was interrupted", e);
             // FIXME: Should perhaps do something else in this case ??
         }
     }