]> gerrit.simantics Code Review - simantics/platform.git/blobdiff - bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/Predicates.java
Multiple reader thread support for db client
[simantics/platform.git] / bundles / org.simantics.db.impl / src / org / simantics / db / impl / query / Predicates.java
index 960cdb6c02aba26fe709af0fac79e1c4c28412ae..9445db03b4e048ca1488cb22f45eeaa56da9bdd1 100644 (file)
  *******************************************************************************/
 package org.simantics.db.impl.query;
 
-import gnu.trove.procedure.TIntProcedure;
-
-import java.util.concurrent.Semaphore;
-
 import org.simantics.db.common.exception.DebugException;
+import org.simantics.db.exception.DatabaseException;
 import org.simantics.db.impl.graph.ReadGraphImpl;
-import org.simantics.db.procedure.ListenerBase;
+import org.simantics.db.impl.procedure.InternalProcedure;
 import org.simantics.db.request.RequestFlags;
 
-final public class Predicates extends UnaryQuery<IntProcedure> {
-       
-//     public ArrayList<IntProcedure> procs;
-
-       public Predicates(final int r) {
-        super(r);
-    }
-
-    public static Predicates newInstance(final int r) {
-        return new Predicates(r);
-    }
-    
-    final static Predicates entry(final QueryProcessor provider, final int r) {
-        
-        return (Predicates)provider.predicatesMap.get(r);
-
-    }
-    
-    final static void runner(ReadGraphImpl graph, final int r, final QueryProcessor provider, Predicates cached, final CacheEntry parent, final ListenerBase listener, final IntProcedure procedure) {
-
-       Predicates entry = cached != null ? cached : (Predicates)provider.predicatesMap.get(r); 
-        if(entry == null) {
-               
-               entry = new Predicates(r);
-               entry.setPending();
-               entry.clearResult(provider.querySupport);
-               entry.putEntry(provider);
-               
-            provider.performForEach(graph, entry, parent, listener, procedure);
-            
-        } else {
-               
-            if(entry.isPending()) {
-               synchronized(entry) {
-                    if(entry.isPending()) {
-                        throw new IllegalStateException();
-                    }
-               }
-            }
-            provider.performForEach(graph, entry, parent, listener, procedure);
-        }
-
-    }
-
-    final static IntSet runner2(ReadGraphImpl graph, final int r, final QueryProcessor provider, final CacheEntry parent) throws Throwable {
-
-       Predicates entry = (Predicates)provider.predicatesMap.get(r); 
-        if(entry == null) {
-               
-               entry = new Predicates(r);
-               entry.setPending();
-               entry.clearResult(provider.querySupport);
-               entry.putEntry(provider);
-               
-            return (IntSet)provider.performForEach2(graph, entry, parent, null, null);
-            
-        } else {
-               
-            if(entry.isPending()) {
-               synchronized(entry) {
-                    if(entry.isPending()) {
-                        throw new IllegalStateException();
-                    }
-               }
-            }
-            return (IntSet)provider.performForEach(graph, entry, parent, null, null);
-            
-        }
-
-    }
-    
-    final public static void queryEach(ReadGraphImpl graph, final int r, final QueryProcessor provider, final CacheEntry parent, final ListenerBase listener, final IntProcedure procedure) {
-        
-       assert(r != 0);
-       
-       final Predicates entry = (Predicates)provider.predicatesMap.get(r);
-       
-       if(parent == null && listener == null) {
-               if(entry != null && entry.isReady()) { 
-                       entry.performFromCache(graph, provider, procedure);
-                       return;
-               }
-        }
-
-        runner(graph, r, provider, entry, parent, listener, procedure);
-         
-    }
+import gnu.trove.procedure.TIntProcedure;
 
-    final public static IntSet queryEach2(ReadGraphImpl graph, final int r, final QueryProcessor provider, final CacheEntry parent) throws Throwable {
-        
-       if(parent == null) {
-               final Predicates entry = (Predicates)provider.predicatesMap.get(r);
-               if(entry != null && entry.isReady()) {
-                       return (IntSet)entry.get(graph, provider, null);
-               }
-        }
+final public class Predicates extends UnaryQueryP<IntSet> {
 
-        return runner2(graph, r, provider, parent);
-         
-    }
-    
-    @Override
-    public UnaryQuery<IntProcedure> getEntry(QueryProcessor provider) {
-        return provider.predicatesMap.get(id);
+       Predicates(final int r) {
+        super(r);
     }
-       
-       @Override
-       public void putEntry(QueryProcessor provider) {
-           provider.predicatesMap.put(id, this);
-       }
 
        @Override
        final public void removeEntry(QueryProcessor provider) {
-        provider.predicatesMap.remove(id);
+        provider.cache.remove(this);
        }
     
-    final private void forAssertions(ReadGraphImpl graph, final QueryProcessor queryProvider, final IntProcedure procedure, final boolean store) {
+    final static private void forAssertions(ReadGraphImpl graph, int r, Predicates parent, final IntSet set) throws DatabaseException {
 
-        PrincipalTypes.queryEach(graph, id, queryProvider, store ? Predicates.this : null, null, new SyncIntProcedure() {
+       QueryCache.runnerPrincipalTypes(graph, r, parent, null, new SyncIntProcedure() {
             
             @Override
-            public void run(ReadGraphImpl graph) {
-                
-                finish(graph, queryProvider);
-                procedure.finished(graph);
-                
+            public void run(ReadGraphImpl graph) throws DatabaseException {
             }
             
             IntProcedure proc = new IntProcedure() {
 
                 @Override
-                public void execute(ReadGraphImpl graph, int i) {
-                    if(addOrSet(queryProvider, i))
-                       procedure.execute(graph, i);
+                public void execute(ReadGraphImpl graph, int i) throws DatabaseException {
+                       set.add(i);
                 }
 
                 @Override
-                public void finished(ReadGraphImpl graph) {
+                public void finished(ReadGraphImpl graph) throws DatabaseException {
                     dec(graph);
                 }
                                
                                @Override
-                               public void exception(ReadGraphImpl graph, Throwable t) {
+                               public void exception(ReadGraphImpl graph, Throwable t) throws DatabaseException {
                     if(DebugException.DEBUG) new DebugException(t).printStackTrace();
-                                       procedure.exception(graph, t);
                    }
 
             }; 
 
             @Override
-            public void execute(ReadGraphImpl graph, int type) {
+            public void execute(ReadGraphImpl graph, int type) throws DatabaseException {
 
                inc();
-                
-                AssertedPredicates.queryEach(graph, type, queryProvider, store ? Predicates.this : null, null, proc);
+                QueryCache.runnerAssertedPredicates(graph, type, parent, null, proc);
                 
             }
             
             @Override
-            public void finished(ReadGraphImpl graph) {
+            public void finished(ReadGraphImpl graph) throws DatabaseException {
                 dec(graph);       
             }
             
@@ -188,153 +75,50 @@ final public class Predicates extends UnaryQuery<IntProcedure> {
 
     }
 
-    @Override
-    public Object computeForEach(ReadGraphImpl graph, final QueryProcessor provider, final IntProcedure procedure, final boolean store) {
-
-               DirectPredicates.queryEach(graph, id, provider, store ? Predicates.this : null, null, new IntProcedure() {
-
-                       @Override
-                       public void execute(ReadGraphImpl graph, final int pred) {
-
-                               if(addOrSet(provider, pred))
-                                       procedure.execute(graph, pred);
-
-                       }
-
-                       @Override
-                       public void finished(ReadGraphImpl graph) {
-
-                               forAssertions(graph, provider, procedure, store);
-
-                       }
-
-                       @Override
-                       public void exception(ReadGraphImpl graph, Throwable t) {
-                               procedure.exception(graph, t);
-                       }
+       @Override
+       public void compute(ReadGraphImpl graph, final InternalProcedure<IntSet> procedure) throws DatabaseException {
+               computeForEach(graph, id, this, procedure);
+       }
 
-               });
-               
-               return getResult();
-        
-    }
-    
-    @Override
-    public String toString() {
-       return "Predicates2[" + id + "]";
-    }
-
-    final public void finish(final ReadGraphImpl graph, QueryProcessor provider) {
-        
-//        ArrayList<IntProcedure> p = null;
-
-        synchronized(this) {
-
-               setReady();
-//            p = procs;
-//            procs = null; 
-        
-        }
-        
-//        if(p != null) {
-//        
-//             final ArrayList<IntProcedure> finalP = p;
-//
-//             IntSet v = (IntSet)getResult();
-//             v.forEach(new TIntProcedure() {
-//
-//                     @Override
-//                     public boolean execute(int arg0) {
-//                             for(IntProcedure proc : finalP) proc.execute(graph, arg0);
-//                             return true;
-//                     }
-//
-//             });
-//
-//             for(IntProcedure proc : p) proc.finished(graph);
-//        
-//        }
-
-    }
-
-    synchronized private boolean addOrSet(QueryProcessor processor, int add) {
-
-       if(!isPending()) {
-               setResult(new IntSet(null));
-       }
-       
-       IntSet value = (IntSet)getResult();
-        return value.add(add);
-        
-    }
-
-    @Override
-    public void clearResult(QuerySupport support) {
-       setResult(new IntSet(support));
-    }
-    
-    @Override
-    public Object performFromCache(final ReadGraphImpl graph, QueryProcessor provider, final IntProcedure procedure) {
+       public static void computeForEach(ReadGraphImpl graph, final int r, final Predicates entry, final InternalProcedure<IntSet> procedure_) throws DatabaseException {
+           
+           InternalProcedure<IntSet> procedure = entry != null ? entry : procedure_;
        
-       assert(isReady());
-
-       if(handleException(graph, procedure)) return EXCEPTED;
+       IntSet direct = QueryCache.resultDirectPredicates(graph, r, entry, null);
+               
+               IntSet result = new IntSet(graph.processor.querySupport);
+       forAssertions(graph, r, entry, result);
        
-        IntSet v = getResult();
-        if(procedure != null) {
-               v.forEach(new TIntProcedure() {
-
+       if(result.isEmpty()) {
+               procedure.execute(graph, direct);
+       } else {
+               direct.forEach(new TIntProcedure() {
                        @Override
-                       public boolean execute(int arg0) {
-                               procedure.execute(graph, arg0);                                         
+                       public boolean execute(int value) {
+                               result.add(value);
                                return true;
                        }
-               });
-            procedure.finished(graph);
-        }
-        
-        return v;
+               });
+               procedure.execute(graph, result);
+       }
+       
+       if(entry != null) entry.performFromCache(graph, procedure_);
         
     }
     
     @Override
-    public void recompute(ReadGraphImpl graph, QueryProcessor provider) {
-        
-        final Semaphore s = new Semaphore(0);
-        
-        computeForEach(graph, provider, new IntProcedure() {
-
-            @Override
-            public void finished(ReadGraphImpl graph) {
-                s.release();
-            }
-                       
-                       @Override
-                       public void exception(ReadGraphImpl graph, Throwable t) {
-                               throw new Error("Error in recompute.", t);
-            }
-
-                       @Override
-                       public void execute(ReadGraphImpl graph, int i) {
-                       }
+    public String toString() {
+       return "Predicates[" + id + "]";
+    }
 
-        }, true);
-        
-        while(!s.tryAcquire()) {
-               provider.resume(graph);
-        }
-        
+    @Override
+    public void clearResult(QuerySupport support) {
+       setResult(new IntSet(support));
     }
     
     @Override
     public int type() {
         return RequestFlags.IMMEDIATE_UPDATE;
     }
-
-
-    @Override
-    boolean isImmutable(ReadGraphImpl graph) {
-       return graph.processor.isImmutable(id);
-    }
     
 }