Fix livelock situation in QueryProcessor

gitlab #386

Change-Id: I91bb154e6410b125eb13171f9152996286a6d54d

diff --git a/bundles/org.simantics.browsing.ui.common/src/org/simantics/browsing/ui/common/internal/GECache.java b/bundles/org.simantics.browsing.ui.common/src/org/simantics/browsing/ui/common/internal/GECache.java
index b5545df0065264423b0bf9d240679a94063d05aa..014a5c0d54b37c0fbd06810137ca04b83a9f88f9 100644 (file)
--- a/bundles/org.simantics.browsing.ui.common/src/org/simantics/browsing/ui/common/internal/GECache.java
+++ b/bundles/org.simantics.browsing.ui.common/src/org/simantics/browsing/ui/common/internal/GECache.java
@@ -174,13 +174,13 @@ public class GECache implements IGECache {
     private TObjectIntHashMap<NodeContext> references = new TObjectIntHashMap<NodeContext>();
     
     @Override
     private TObjectIntHashMap<NodeContext> references = new TObjectIntHashMap<NodeContext>();
     
     @Override

-    public void incRef(NodeContext context) {
+    synchronized public void incRef(NodeContext context) {

        int exist = references.get(context);
        references.put(context, exist+1);
     }
     
     @Override
        int exist = references.get(context);
        references.put(context, exist+1);
     }
     
     @Override

-    public void decRef(NodeContext context) {
+    synchronized public void decRef(NodeContext context) {

        int exist = references.get(context);
        references.put(context, exist-1);
        if(exist == 1) {
        int exist = references.get(context);
        references.put(context, exist-1);
        if(exist == 1) {

diff --git a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/graph/ReadGraphImpl.java b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/graph/ReadGraphImpl.java
index cffbba51c75f852e44b215747e94efafa099e296..3e955febc54743e25134184a2d7c3d9d85b9135c 100644 (file)
--- a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/graph/ReadGraphImpl.java
+++ b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/graph/ReadGraphImpl.java
@@ -5152,7 +5152,7 @@ public class ReadGraphImpl implements AsyncReadGraph {
                assert (request != null);
                assert (procedure != null);
        
                assert (request != null);
                assert (procedure != null);
        

-               processor.schedule(new SessionTask(this) {
+               processor.scheduleNow(new SessionTask(this) {

 
                        @Override
                        public void run0(int thread) {
 
                        @Override
                        public void run0(int thread) {


@@ -5232,7 +5232,7 @@ public class ReadGraphImpl implements AsyncReadGraph {
                assert (request != null);
                assert (procedure != null);
 
                assert (request != null);
                assert (procedure != null);
 

-               processor.schedule(new SessionTask(this) {
+               processor.scheduleNow(new SessionTask(this) {

 
                        @Override
                        public void run0(int thread) {
 
                        @Override
                        public void run0(int thread) {


@@ -6312,7 +6312,7 @@ public class ReadGraphImpl implements AsyncReadGraph {
     public boolean performPending() {
         return processor.performPending(this);
     }
     public boolean performPending() {
         return processor.performPending(this);
     }

-
+    

     public Set<ReadGraphImpl> ancestorSet() {
         HashSet<ReadGraphImpl> result = new HashSet<>();
         ReadGraphImpl g = this;
     public Set<ReadGraphImpl> ancestorSet() {
         HashSet<ReadGraphImpl> result = new HashSet<>();
         ReadGraphImpl g = this;


@@ -6322,5 +6322,23 @@ public class ReadGraphImpl implements AsyncReadGraph {
         }
         return result;
     }
         }
         return result;
     }

+    
+    public int getLevel() {
+        return getLevelStatic(this);
+    }
+    
+    private static int getLevelStatic(ReadGraphImpl impl) {
+        if(impl == null) return 0;
+        else return 1 + getLevelStatic(impl.parentGraph);
+    }
+    
+    public ReadGraphImpl getTopLevelGraph() {
+        return getTopLevelGraphStatic(this);
+    }
+
+    private static ReadGraphImpl getTopLevelGraphStatic(ReadGraphImpl impl) {
+        if(impl.parentGraph == null) return impl;
+        else return getTopLevelGraphStatic(impl.parentGraph);
+    }

 
 }
 
 }

diff --git a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/AsyncReadEntry.java b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/AsyncReadEntry.java
index 72582ee605802bfe32fac38d10119f6200602586..e13ecab72cd0819f71c3c337af62ad242db709c2 100644 (file)
--- a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/AsyncReadEntry.java
+++ b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/AsyncReadEntry.java
@@ -169,7 +169,14 @@ final public class AsyncReadEntry<T> extends CacheEntryBase<AsyncProcedure<T>> i
             DatabaseException exception;
             
             public AsyncTask(ReadGraphImpl graph) {
             DatabaseException exception;
             
             public AsyncTask(ReadGraphImpl graph) {

+                this(graph, 1);
+            }
+
+            public AsyncTask(ReadGraphImpl graph, int pos) {

                 super(graph);
                 super(graph);

+                this.position = pos;
+                if(this.position < 1024)
+                    this.position *= 2;

             }
 
             @Override
             }
 
             @Override


@@ -213,10 +220,15 @@ final public class AsyncReadEntry<T> extends CacheEntryBase<AsyncProcedure<T>> i
                         }
                         throw new IllegalStateException("Eternal loop in queries.");
                     }
                         }
                         throw new IllegalStateException("Eternal loop in queries.");
                     }

-                    graph.processor.schedule(new AsyncTask(graph));
+                    graph.processor.scheduleLater(new AsyncTask(graph, position));

                 }
             }
             
                 }
             }
             

+            @Override
+            public boolean maybeReady() {
+               return proc.isDone();
+            }
+            

         }
 
         try {
         }
 
         try {


@@ -230,7 +242,7 @@ final public class AsyncReadEntry<T> extends CacheEntryBase<AsyncProcedure<T>> i
         if(needsToBlock) task.run(0);
         else if (proc.isDone()) task.run(0);
         else  {
         if(needsToBlock) task.run(0);
         else if (proc.isDone()) task.run(0);
         else  {

-            graph.processor.schedule(task);
+            graph.processor.scheduleLater(task);

             return null;
         }
 
             return null;
         }
 

diff --git a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryCache.java b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryCache.java
index 5c18d051089ef6026ce58048947bd82ae3e4f9b2..cc0ca919b584942a4d49a1a4c61b97b741e631f2 100644 (file)
--- a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryCache.java
+++ b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryCache.java
@@ -670,7 +670,7 @@ public class QueryCache extends QueryCacheBase {
         }
         ReadEntry entry = (ReadEntry)cache.getOrCreateReadEntry(graph, r, needsToBlock);
         if(entry == null) {
         }
         ReadEntry entry = (ReadEntry)cache.getOrCreateReadEntry(graph, r, needsToBlock);
         if(entry == null) {

-          graph.processor.schedule(new SessionTask(graph) {
+          graph.processor.scheduleNow(new SessionTask(graph) {

             @Override
             public void run0(int thread) {
               try {
             @Override
             public void run0(int thread) {
               try {


@@ -749,7 +749,7 @@ public class QueryCache extends QueryCacheBase {
         }
         AsyncReadEntry entry = (AsyncReadEntry)cache.getOrCreateAsyncReadEntry(graph, r, needsToBlock);
         if(entry == null) {
         }
         AsyncReadEntry entry = (AsyncReadEntry)cache.getOrCreateAsyncReadEntry(graph, r, needsToBlock);
         if(entry == null) {

-          graph.processor.schedule(new SessionTask(graph) {
+          graph.processor.scheduleNow(new SessionTask(graph) {

             @Override
             public void run0(int thread) {
               try {
             @Override
             public void run0(int thread) {
               try {

diff --git a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryListening.java b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryListening.java
index e524a080a99a2940bbc5da9a5564202d115bf931..9daab950e3964a8681a12fabc5ef0b2cfaf74ceb 100644 (file)
--- a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryListening.java
+++ b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryListening.java
@@ -14,8 +14,8 @@ import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;

+import java.util.concurrent.ArrayBlockingQueue;

 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.BlockingQueue;

-import java.util.concurrent.LinkedBlockingQueue;

 import java.util.concurrent.Semaphore;
 
 import org.simantics.databoard.Bindings;
 import java.util.concurrent.Semaphore;
 
 import org.simantics.databoard.Bindings;


@@ -41,7 +41,7 @@ public class QueryListening {
        private THashSet<ListenerEntry>                         scheduledListeners    = new THashSet<ListenerEntry>();
        private boolean                                         firingListeners       = false;
        final THashMap<CacheEntry, ArrayList<ListenerEntry>>    listeners = new THashMap<CacheEntry, ArrayList<ListenerEntry>>(10, 0.75f);
        private THashSet<ListenerEntry>                         scheduledListeners    = new THashSet<ListenerEntry>();
        private boolean                                         firingListeners       = false;
        final THashMap<CacheEntry, ArrayList<ListenerEntry>>    listeners = new THashMap<CacheEntry, ArrayList<ListenerEntry>>(10, 0.75f);

-       private BlockingQueue<Runnable>                         tasks = new LinkedBlockingQueue<Runnable>();
+       private BlockingQueue<Runnable>                         tasks = new ArrayBlockingQueue<Runnable>(2048);

        private Map<ListenerBase,ListenerEntry>                 addedEntries = new HashMap<>();
        
        QueryListening(QueryProcessor processor) {
        private Map<ListenerBase,ListenerEntry>                 addedEntries = new HashMap<>();
        
        QueryListening(QueryProcessor processor) {


@@ -51,10 +51,10 @@ public class QueryListening {
        
        public void sync() {
                Semaphore s = new Semaphore(0);
        
        public void sync() {
                Semaphore s = new Semaphore(0);

-               tasks.add(() -> {
-                       s.release();
-               });

                try {
                try {

+                       tasks.put(() -> {
+                               s.release();
+                       });

                        s.acquire();
                } catch (Throwable t) {
                        LOGGER.error("Error while waiting for query dependency management", t);
                        s.acquire();
                } catch (Throwable t) {
                        LOGGER.error("Error while waiting for query dependency management", t);


@@ -93,26 +93,30 @@ public class QueryListening {
 
        void registerDependencies(ReadGraphImpl graph, CacheEntry child, CacheEntry parent, ListenerBase listener, Object procedure, boolean inferred) {
                
 
        void registerDependencies(ReadGraphImpl graph, CacheEntry child, CacheEntry parent, ListenerBase listener, Object procedure, boolean inferred) {
                

-               tasks.offer(() -> {
+               try {
+                       tasks.put(() -> {

 
 

-                       if (parent != null && !inferred) {
-                               try {
-                                       if(!child.isImmutable(graph))
-                                               child.addParent(parent);
-                               } catch (DatabaseException e) {
-                                       LOGGER.error("Error while registering query dependencies", e);
-                               }
-                               if (Development.DEVELOPMENT) {
-                                       if(Development.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_DEPENDENCIES, Bindings.BOOLEAN)) {
-                                               System.out.println(child + " -> " + parent);
+                               if (parent != null && !inferred) {
+                                       try {
+                                               if(!child.isImmutable(graph))
+                                                       child.addParent(parent);
+                                       } catch (DatabaseException e) {
+                                               LOGGER.error("Error while registering query dependencies", e);
+                                       }
+                                       if (Development.DEVELOPMENT) {
+                                               if(Development.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_DEPENDENCIES, Bindings.BOOLEAN)) {
+                                                       System.out.println(child + " -> " + parent);
+                                               }

                                        }
                                }
                                        }
                                }

-                       }

 
 

-                       if (listener != null)
-                               registerListener(child, listener, procedure);
+                               if (listener != null)
+                                       registerListener(child, listener, procedure);

 
 

-               });
+                       });
+               } catch (InterruptedException e) {
+                       LOGGER.error("Error while registering dependencies", e);
+               }

                
        }
 
                
        }
 

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 7bfa0f4c6195659c21efdb0e6bcf99fed0dfb7bb..ee4b11175e4d2767cee0058d3938d8cd47125997 100644 (file)
--- 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
@@ -136,11 +136,11 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
        final public Session                                    session;
        final public ResourceSupport                            resourceSupport;
        
        final public Session                                    session;
        final public ResourceSupport                            resourceSupport;
        

+       final public Semaphore                                  requests = new Semaphore(1);
+       

        final public QueryListening                            listening = new QueryListening(this);
 
        QueryThread[]                                   executors;
        final public QueryListening                            listening = new QueryListening(this);
 
        QueryThread[]                                   executors;

-
-//     public ArrayList<SessionTask>[]                           queues;

        
        public LinkedList<SessionTask>                           freeScheduling = new LinkedList<SessionTask>();
 
        
        public LinkedList<SessionTask>                           freeScheduling = new LinkedList<SessionTask>();
 


@@ -151,14 +151,6 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
        }
 
        public ThreadState[]                                                                    threadStates;
        }
 
        public ThreadState[]                                                                    threadStates;

-//     public ReentrantLock[]                                                                  threadLocks;
-//     public Condition[]                                                                          threadConditions;
-
-       //public ArrayList<SessionTask>[]                           ownTasks;
-
-       //public ArrayList<SessionTask>[]                           ownSyncTasks;
-
-       //ArrayList<SessionTask>[]                           delayQueues;

        
        final Object querySupportLock;
        
        
        final Object querySupportLock;
        


@@ -167,28 +159,13 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
        public void close() {
        }
 
        public void close() {
        }
 

-       public SessionTask getOwnTask(ReadGraphImpl impl) {
-               Set<ReadGraphImpl> ancestors = impl.ancestorSet();
-               synchronized(querySupportLock) {
-                       int index = 0;
-                       while(index < freeScheduling.size()) {
-                               SessionTask task = freeScheduling.get(index);
-                               if(task.hasCommonParent(ancestors)) {
-                                       return freeScheduling.remove(index);
-                               }
-                               index++;
-                       }
-               }
-               return null;
-       }
-

     public SessionTask getSubTask(ReadGraphImpl impl) {
     public SessionTask getSubTask(ReadGraphImpl impl) {

-        Set<ReadGraphImpl> onlyThis = Collections.singleton(impl);

         synchronized(querySupportLock) {
             int index = 0;
             while(index < freeScheduling.size()) {
                 SessionTask task = freeScheduling.get(index);
         synchronized(querySupportLock) {
             int index = 0;
             while(index < freeScheduling.size()) {
                 SessionTask task = freeScheduling.get(index);

-                if(task.hasCommonParent(onlyThis)) {
+                if(task.hasCommonParent(task.graph, impl) && task.maybeReady()) {
+                    queueLength.decrementAndGet();

                     return freeScheduling.remove(index);
                 }
                 index++;
                     return freeScheduling.remove(index);
                 }
                 index++;


@@ -197,68 +174,84 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
         return null;
     }
 
         return null;
     }
 

-       public boolean performPending(ReadGraphImpl graph) {
-               SessionTask task = getOwnTask(graph);
+    /*
+     * We are running errands while waiting for requests to complete.
+     * We can only run work that is part of the current root request to avoid any deadlocks
+     */
+    public boolean performPending(ReadGraphImpl graph) {
+        SessionTask task = getSubTask(graph);

                if(task != null) {
                        task.run(QueryProcessor.thread.get());
                        return true;
                if(task != null) {
                        task.run(QueryProcessor.thread.get());
                        return true;

-               } else {
-                       return false;

                }
                }

+               return false;

        }
 
        }
 

-//     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 + " " + " -> " + performer);
+    final public void scheduleNow(SessionTask request) {
+        schedule(request, false);
+    }

 
 

-               //assert(performer >= 0);
+    final public void scheduleLater(SessionTask request) {
+        schedule(request, true);
+    }

 
 

+    AtomicInteger queueLength = new AtomicInteger(0);
+    
+    final public void schedule(SessionTask request, boolean late) {
+        
+        int queueLengthEstimate = queueLength.get();
+        if(!late && queueLengthEstimate > 80) {
+            request.run(thread.get());
+            return;
+        }
+        

                assert(request != null);
                assert(request != null);

+               
+               synchronized(querySupportLock) {

 
 

-//             if(caller == performer) {
-//                     request.run(caller);
-//             } else {
-                       
-//                     if(performer == THREADS) {
-                               
-                               synchronized(querySupportLock) {
-
-                                       if(BarrierTracing.BOOKKEEPING) {
-                                               Exception current = new Exception();
-                                               Exception previous = BarrierTracing.tasks.put(request, current);
-                                               if(previous != null) {
-                                                       previous.printStackTrace();
-                                                       current.printStackTrace();
-                                               }
-                                       }
-                                   
-                                       freeScheduling.add(request);
-                                       
-                                       querySupportLock.notifyAll();
+                   if(BarrierTracing.BOOKKEEPING) {
+                       Exception current = new Exception();
+                       Exception previous = BarrierTracing.tasks.put(request, current);
+                       if(previous != null) {
+                           previous.printStackTrace();
+                           current.printStackTrace();
+                       }
+                   }

 
 

-                               }
+                   if(late) {
+                       int pos = request.position - 1;
+                       if(pos < freeScheduling.size()) { 
+                           freeScheduling.add(pos, request);
+                           queueLength.incrementAndGet();
+                           requests.release();
+                       } else {
+                           freeScheduling.addLast(request);
+                    queueLength.incrementAndGet();
+                    requests.release();
+                       }
+                   }
+                   else {
+                       if(request.getLevel() < 4) {
+                           if(freeScheduling.size() < 100) {
+                               freeScheduling.addFirst(request);
+                           queueLength.incrementAndGet();
+                           requests.release();
+                           } else {
+                               request.run(thread.get());
+                           }
+                       } else {
+                           if(freeScheduling.size() < 20) {
+                               freeScheduling.addFirst(request);
+                           queueLength.incrementAndGet();
+                           requests.release();
+                           } else {
+                               request.run(thread.get());
+                           }
+                       }
+                   }

 
 

-                               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();
-//             }
+
+               }

 
        }
 
 
        }
 


@@ -271,8 +264,8 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
        public static abstract class SessionTask {
 
                public final ReadGraphImpl graph;
        public static abstract class SessionTask {
 
                public final ReadGraphImpl graph;

-               private Set<ReadGraphImpl> ancestors;

                private int counter = 0;
                private int counter = 0;

+               protected int position = 1;

                private Exception trace;
 
                public SessionTask(ReadGraphImpl graph) {
                private Exception trace;
 
                public SessionTask(ReadGraphImpl graph) {


@@ -280,13 +273,12 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
                        if(graph != null) graph.asyncBarrier.inc();
                }
 
                        if(graph != null) graph.asyncBarrier.inc();
                }
 

-               public boolean hasCommonParent(Set<ReadGraphImpl> otherAncestors) {
-                       if(graph == null) return false;
-                       if(ancestors == null) ancestors = graph.ancestorSet();
-                       return !Collections.disjoint(ancestors, otherAncestors);
-               }
+        public static boolean hasCommonParent(ReadGraphImpl r1, ReadGraphImpl r2) {
+            if(r1 == null || r2 == null) return false;
+            return r1.getTopLevelGraph() == r2.getTopLevelGraph();
+        } 

 
 

-               public abstract void run0(int thread);
+        public abstract void run0(int thread);

 
                public final void run(int thread) {
                    if(counter++ > 0) {
 
                public final void run(int thread) {
                    if(counter++ > 0) {


@@ -302,10 +294,22 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
                    run0(thread);
                    if(graph != null) graph.asyncBarrier.dec();
                }
                    run0(thread);
                    if(graph != null) graph.asyncBarrier.dec();
                }

+               
+               public boolean maybeReady() {
+                       return true;
+               }

 
                @Override
                public String toString() {
 
                @Override
                public String toString() {

-                       return "SessionTask[" + graph.parent + "]";
+                       if(graph == null)
+                               return "SessionTask[no graph]";
+                       else
+                               return "SessionTask[" + graph.parent + "]";
+               }
+               
+               public int getLevel() {
+                   if(graph == null) return 0;
+                   else return graph.getLevel();

                }
 
        }
                }
 
        }


@@ -1793,6 +1797,8 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap
        @Override
        protected void doDispose() {
 
        @Override
        protected void doDispose() {
 

+               requests.release(Integer.MAX_VALUE / 2);
+               

                for(int index = 0; index < THREADS; index++) { 
                        executors[index].dispose();
                }
                for(int index = 0; index < THREADS; index++) { 
                        executors[index].dispose();
                }

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..5510944dc319265e7e033e38a5462e62b280ee16 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 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;
 
        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;
                sleepers = processor.sleepers;
                querySupport = processor.querySupport;
                threadStates = processor.threadStates;

+               requests = processor.requests;

 //             delayQueues = processor.delayQueues;
 //             executors = processor.executors;
 //             threadLocks = processor.threadLocks;
 //             delayQueues = processor.delayQueues;
 //             executors = processor.executors;
 //             threadLocks = processor.threadLocks;


@@ -98,6 +100,7 @@ class QueryThread extends Thread implements SessionThread {
        private boolean pumpTask() {
                if(!processor.freeScheduling.isEmpty()) {
                        tasks.add(processor.freeScheduling.removeFirst());
        private boolean pumpTask() {
                if(!processor.freeScheduling.isEmpty()) {
                        tasks.add(processor.freeScheduling.removeFirst());

+                       processor.queueLength.decrementAndGet();

                        return true;
                }
                return false;
                        return true;
                }
                return false;


@@ -155,18 +158,7 @@ class QueryThread extends Thread implements SessionThread {
                                
                                threadStates[index] = ThreadState.SLEEP;
                                
                                
                                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();
 
 
                                sleepers.decrementAndGet();
 

diff --git a/bundles/org.simantics.db.layer0/src/org/simantics/db/layer0/adapter/impl/EntityInstances.java b/bundles/org.simantics.db.layer0/src/org/simantics/db/layer0/adapter/impl/EntityInstances.java
index 93903fb9b2d4371e01f521eca67c55cc925d33bf..ffbec2f5e01ca88222926bb1c03848ed940d692a 100644 (file)
--- a/bundles/org.simantics.db.layer0/src/org/simantics/db/layer0/adapter/impl/EntityInstances.java
+++ b/bundles/org.simantics.db.layer0/src/org/simantics/db/layer0/adapter/impl/EntityInstances.java
@@ -185,15 +185,17 @@ public class EntityInstances implements Instances {
        CollectionSupport coll = graph.getService(CollectionSupport.class);
        
        THashSet<Resource> visited = new THashSet<>();
        CollectionSupport coll = graph.getService(CollectionSupport.class);
        
        THashSet<Resource> visited = new THashSet<>();

-       List<Resource> rec = findRec(graph, index, filter, visited);
+       List<Resource> rec_ = findRec(graph, index, filter, visited);
+       // We must not modify rec_!
+       List<Resource> rec = rec_;

        for(Resource global : Layer0Utils.listGlobalOntologies(graph)) {
                if(!visited.add(global)) continue;
                List<Resource> rs = graph.syncRequest(new QueryIndex(global, type, filter), TransientCacheListener.<List<Resource>>instance());
        for(Resource global : Layer0Utils.listGlobalOntologies(graph)) {
                if(!visited.add(global)) continue;
                List<Resource> rs = graph.syncRequest(new QueryIndex(global, type, filter), TransientCacheListener.<List<Resource>>instance());

-               if(rec.isEmpty() && !rs.isEmpty()) {
-                       // TODO: rec could be an immutable empty list
-                       rec = new ArrayList<Resource>();
+               if(!rs.isEmpty()) {
+                   if(rec == rec_)
+                       rec = new ArrayList<>(rec);
+                rec.addAll(rs);

                }
                }

-               rec.addAll(rs);

        }
        Collection<Resource> result = coll.asSortedList(rec);
        return result; 
        }
        Collection<Resource> result = coll.asSortedList(rec);
        return result; 

diff --git a/bundles/org.simantics.db.procore/src/fi/vtt/simantics/procore/internal/SessionRequestManager.java b/bundles/org.simantics.db.procore/src/fi/vtt/simantics/procore/internal/SessionRequestManager.java
index a93476355abebb960eaa5b7f26582d9e89b066dc..fb421b6105ba1e526f6bfcd883b01b1fe41e4287 100644 (file)
--- a/bundles/org.simantics.db.procore/src/fi/vtt/simantics/procore/internal/SessionRequestManager.java
+++ b/bundles/org.simantics.db.procore/src/fi/vtt/simantics/procore/internal/SessionRequestManager.java
@@ -118,7 +118,7 @@ public class SessionRequestManager {
        
        public synchronized void startRead(int thread, final SessionRead task) {
                
        
        public synchronized void startRead(int thread, final SessionRead task) {
                

-               session.queryProvider2.schedule(new SessionTask(null) {
+               session.queryProvider2.scheduleNow(new SessionTask(null) {

 
                        @Override
             public void run0(int thread) {
 
                        @Override
             public void run0(int thread) {


@@ -142,7 +142,7 @@ public class SessionRequestManager {
        
        public synchronized void startReadUpdate(int thread) {
                
        
        public synchronized void startReadUpdate(int thread) {
                

-               session.queryProvider2.schedule(new SessionTask(null) {
+               session.queryProvider2.scheduleNow(new SessionTask(null) {

 
                        @Override
                        public void run0(int thread) {
 
                        @Override
                        public void run0(int thread) {


@@ -163,7 +163,7 @@ public class SessionRequestManager {
 
        public synchronized void startWrite(int thread, final SessionTask task) {
                
 
        public synchronized void startWrite(int thread, final SessionTask task) {
                

-               session.queryProvider2.schedule(new SessionTask(null) {
+               session.queryProvider2.scheduleNow(new SessionTask(null) {

 
                        @Override
                        public void run0(int thread) {
 
                        @Override
                        public void run0(int thread) {


@@ -184,7 +184,7 @@ public class SessionRequestManager {
 
        public synchronized void startWriteUpdate(int thread) {
                
 
        public synchronized void startWriteUpdate(int thread) {
                

-               session.queryProvider2.schedule(new SessionTask(null) {
+               session.queryProvider2.scheduleNow(new SessionTask(null) {

 
                        @Override
                        public void run0(int thread) {
 
                        @Override
                        public void run0(int thread) {


@@ -271,7 +271,7 @@ public class SessionRequestManager {
                        if (!reads.isEmpty()) {
 
                                final SessionRead read = reads.poll();
                        if (!reads.isEmpty()) {
 
                                final SessionRead read = reads.poll();

-                               session.queryProvider2.schedule(new SessionTask(null) {
+                               session.queryProvider2.scheduleNow(new SessionTask(null) {

 
                                        @Override
                                        public void run0(int thread) {
 
                                        @Override
                                        public void run0(int thread) {


@@ -312,7 +312,7 @@ public class SessionRequestManager {
                assert(State.INIT != state);
                
                if(State.READ == state) {
                assert(State.INIT != state);
                
                if(State.READ == state) {

-                       session.queryProvider2.schedule(new SessionTask(null) {
+                       session.queryProvider2.scheduleNow(new SessionTask(null) {

 
                                @Override
                                public void run0(int thread) {
 
                                @Override
                                public void run0(int thread) {