X-Git-Url: https://gerrit.simantics.org/r/gitweb?a=blobdiff_plain;f=bundles%2Forg.simantics.db.impl%2Fsrc%2Forg%2Fsimantics%2Fdb%2Fimpl%2Fquery%2FQueryProcessor.java;h=5945589a331c2460f83e0479f4f56f2b7425ebf2;hb=1f8b50d81a1aa1bbd67a77f7cbc1060f2eb805d4;hp=f1e9233eb685b7fe87cb47ca4c00704503edec9a;hpb=8e67ae458c3ee4711eff252f971d9a4219d51210;p=simantics%2Fplatform.git

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 f1e9233eb..5945589a3 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
@@ -151,7 +151,9 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
 
 	QueryThread[]                                   executors;
 
-	public ArrayList<SessionTask>[]                           queues;
+//	public ArrayList<SessionTask>[]                           queues;
+	
+	public LinkedList<SessionTask>                           freeScheduling = new LinkedList<SessionTask>();
 
 	enum ThreadState {
 
@@ -160,14 +162,14 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
 	}
 
 	public ThreadState[]									threadStates;
-	public ReentrantLock[]									threadLocks;
-	public Condition[]									    threadConditions;
+//	public ReentrantLock[]									threadLocks;
+//	public Condition[]									    threadConditions;
 
-	public ArrayList<SessionTask>[]                           ownTasks;
+	//public ArrayList<SessionTask>[]                           ownTasks;
 
-	public ArrayList<SessionTask>[]                           ownSyncTasks;
+	//public ArrayList<SessionTask>[]                           ownSyncTasks;
 
-	ArrayList<SessionTask>[]                           delayQueues;
+	//ArrayList<SessionTask>[]                           delayQueues;
 	
 	final Object querySupportLock;
 	
@@ -176,45 +178,85 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
 	public void close() {
 	}
 
-	final public void scheduleOwn(int caller, SessionTask request) {
-		ownTasks[caller].add(request);
+	SessionTask getOwnTask(int thread) {
+		synchronized(querySupportLock) {
+			int index = 0;
+			while(index < freeScheduling.size()) {
+				SessionTask task = freeScheduling.get(index);
+				if(task.thread == thread && !task.systemCall)
+					return freeScheduling.remove(index);
+				index++;
+			}
+		}
+		return null;
 	}
-
-	final public void scheduleAlways(int caller, SessionTask request) {
-
-		int performer = request.thread;
-		if(caller == performer) {
-			ownTasks[caller].add(request);
+	
+	public boolean performPending(int thread) {
+		SessionTask task = getOwnTask(thread);
+		if(task != null) {
+			task.run(thread);
+			return true;
 		} else {
-			schedule(caller, request);
+			return false;
 		}
-
 	}
 
-	final public void schedule(int caller, SessionTask request) {
+//	final public void scheduleOwn(int caller, SessionTask request) {
+//		ownTasks[caller].add(request);
+//	}
+
+	final public void schedule(SessionTask request) {
 
 		int performer = request.thread;
 
 		if(DebugPolicy.SCHEDULE)
-			System.out.println("schedule " + request + " " + caller + " -> " + performer);
+			System.out.println("schedule " + request + " " + " -> " + performer);
 
-		assert(performer >= 0);
+		//assert(performer >= 0);
 
 		assert(request != null);
 
-		if(caller == performer) {
-			request.run(caller);
-		} else {
-			ReentrantLock queueLock = threadLocks[performer];
-			queueLock.lock();
-			queues[performer].add(request);
-			// This thread could have been sleeping
-			if(queues[performer].size() == 1) {
-				if(ThreadState.SLEEP == threadStates[performer]) sleepers.decrementAndGet();
-				threadConditions[performer].signalAll();
-			}
-			queueLock.unlock();
-		}
+//		if(caller == performer) {
+//			request.run(caller);
+//		} else {
+			
+//			if(performer == THREADS) {
+				
+				synchronized(querySupportLock) {
+					
+					//new Exception().printStackTrace();
+					
+					freeScheduling.add(request);
+					
+					querySupportLock.notifyAll();
+
+					//System.err.println("schedule free task " + request + " => " + freeScheduling.size());
+
+//					for(int i=0;i<THREADS;i++) {
+//						ReentrantLock queueLock = threadLocks[i];
+//						queueLock.lock();
+//						//queues[performer].add(request);
+//						//if(ThreadState.SLEEP == threadStates[i]) sleepers.decrementAndGet();
+//						threadConditions[i].signalAll();
+//						queueLock.unlock();
+//					}
+
+				}
+
+				return;
+				
+//			}
+//			
+//			ReentrantLock queueLock = threadLocks[performer];
+//			queueLock.lock();
+//			queues[performer].add(request);
+//			// This thread could have been sleeping
+//			if(queues[performer].size() == 1) {
+//				//if(ThreadState.SLEEP == threadStates[performer]) sleepers.decrementAndGet();
+//				threadConditions[performer].signalAll();
+//			}
+//			queueLock.unlock();
+//		}
 
 	}
 
@@ -227,26 +269,28 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
 	public static abstract class SessionTask {
 
 		final public int thread;
-		final public int syncCaller;
-		final public Object object;
-
-		public SessionTask(WriteTraits object, int thread) {
-			this.thread = thread;
-			this.syncCaller = -1;
-			this.object = object;
+		final public boolean systemCall;
+//		final public int syncCaller;
+		//final public Object object;
+
+		public SessionTask(boolean systemCall) {
+			this.thread = QueryProcessor.thread.get();
+			this.systemCall = systemCall;
+//			this.syncCaller = -1;
+			//this.object = object;
 		}
 
-		public SessionTask(Object object, int thread, int syncCaller) {
-			this.thread = thread;
-			this.syncCaller = syncCaller;
-			this.object = object;
-		}
+//		public SessionTask(Object object, int syncCaller) {
+//			this.thread = QueryProcessor.thread.get();
+//			this.syncCaller = syncCaller;
+//			this.object = object;
+//		}
 
 		public abstract void run(int thread);
 
 		@Override
 		public String toString() {
-			return "SessionTask[" + object + "]";
+			return "SessionTask[" + super.toString() + "]";
 		}
 
 	}
@@ -256,14 +300,8 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
 		final public Semaphore notify;
 		final public DataContainer<Throwable> throwable; 
 
-		public SessionRead(Object object, DataContainer<Throwable> throwable, Semaphore notify, int thread) {
-			super(object, thread, thread);
-			this.throwable = throwable;
-			this.notify = notify;
-		}
-
-		public SessionRead(Object object, DataContainer<Throwable> throwable, Semaphore notify, int thread, int syncThread) {
-			super(object, thread, syncThread);
+		public SessionRead(DataContainer<Throwable> throwable, Semaphore notify) {
+			super(true);
 			this.throwable = throwable;
 			this.notify = notify;
 		}
@@ -311,28 +349,28 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
 		querySupportLock = core.getLock();
 
 		executors = new QueryThread[THREADS];
-		queues = new ArrayList[THREADS];
-		threadLocks = new ReentrantLock[THREADS];
-		threadConditions = new Condition[THREADS];
+//		queues = new ArrayList[THREADS];
+//		threadLocks = new ReentrantLock[THREADS];
+//		threadConditions = new Condition[THREADS];
 		threadStates = new ThreadState[THREADS];
-		ownTasks = new ArrayList[THREADS];
-		ownSyncTasks = new ArrayList[THREADS];
-		delayQueues = new ArrayList[THREADS * THREADS];
+//		ownTasks = new ArrayList[THREADS];
+//		ownSyncTasks = new ArrayList[THREADS];
+//		delayQueues = new ArrayList[THREADS * THREADS];
 
 		//        freeSchedule = new AtomicInteger(0);
 
-		for (int i = 0; i < THREADS * THREADS; i++) {
-			delayQueues[i] = new ArrayList<SessionTask>();
-		}
+//		for (int i = 0; i < THREADS * THREADS; i++) {
+//			delayQueues[i] = new ArrayList<SessionTask>();
+//		}
 
 		for (int i = 0; i < THREADS; i++) {
 
 			//            tasks[i] = new ArrayList<Runnable>();
-			ownTasks[i] = new ArrayList<SessionTask>();
-			ownSyncTasks[i] = new ArrayList<SessionTask>();
-			queues[i] = new ArrayList<SessionTask>();
-			threadLocks[i] = new ReentrantLock();
-			threadConditions[i] = threadLocks[i].newCondition();
+//			ownTasks[i] = new ArrayList<SessionTask>();
+//			ownSyncTasks[i] = new ArrayList<SessionTask>();
+//			queues[i] = new ArrayList<SessionTask>();
+//			threadLocks[i] = new ReentrantLock();
+//			threadConditions[i] = threadLocks[i].newCondition();
 			//            limits[i] = false;
 			threadStates[i] = ThreadState.INIT;
 
@@ -4431,5 +4469,11 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
 		}
 		return L0;
 	}
+
+    public static ThreadLocal<Integer> thread = new ThreadLocal<Integer>() {
+        protected Integer initialValue() {
+            return -1;
+        }
+    };
 	
 }