import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.Semaphore;
-import java.util.concurrent.atomic.AtomicBoolean;
+import org.simantics.databoard.Bindings;
import org.simantics.db.AsyncReadGraph;
+import org.simantics.db.DevelopmentKeys;
import org.simantics.db.ObjectResourceIdMap;
import org.simantics.db.ReadGraph;
import org.simantics.db.RelationInfo;
import org.simantics.db.common.utils.Logger;
import org.simantics.db.exception.DatabaseException;
-import org.simantics.db.impl.DebugPolicy;
import org.simantics.db.impl.graph.ReadGraphImpl;
import org.simantics.db.impl.procedure.InternalProcedure;
-import org.simantics.db.impl.query.QueryProcessor.SessionTask;
import org.simantics.db.procedure.AsyncMultiProcedure;
import org.simantics.db.procedure.AsyncProcedure;
-import org.simantics.db.procedure.Listener;
import org.simantics.db.procedure.ListenerBase;
import org.simantics.db.procedure.Procedure;
import org.simantics.db.procedure.SyncMultiProcedure;
import org.simantics.db.request.ExternalRead;
import org.simantics.db.request.MultiRead;
import org.simantics.db.request.Read;
+import org.simantics.utils.Development;
import gnu.trove.map.hash.THashMap;
import gnu.trove.map.hash.TObjectIntHashMap;
//
// }
- public <T> Object performQuery(ReadGraphImpl parentGraph, final ExternalRead<T> query, final CacheEntryBase entry_, AsyncProcedure procedure_) throws DatabaseException {
-
- ExternalReadEntry entry = (ExternalReadEntry)entry_;
- AsyncProcedure<T> procedure = (AsyncProcedure<T>)procedure_;
-
- try {
-
- query.register(parentGraph, new Listener<T>() {
-
- AtomicBoolean used = new AtomicBoolean(false);
-
- @Override
- public void execute(T result) {
-
- // Just for safety
- if(entry.isDiscarded()) return;
-
- if(used.compareAndSet(false, true)) {
- //entry.setPending();
- entry.addOrSet(parentGraph.processor, result);
- procedure.execute(parentGraph, result);
- } else {
- entry.queue(result);
- parentGraph.processor.updatePrimitive(query);
- }
-
- }
-
- @Override
- public void exception(Throwable t) {
-
- entry.except(t);
-
- if(used.compareAndSet(false, true)) {
- procedure.exception(parentGraph, t);
- } else {
-// entry.queue(result);
- parentGraph.processor.updatePrimitive(query);
- }
-
- }
-
- @Override
- public String toString() {
- return procedure.toString();
- }
-
- @Override
- public boolean isDisposed() {
- return entry.isDiscarded() || !parentGraph.processor.isBound(entry);
- }
-
- });
-
- return entry.getResult();
-
- } catch (Throwable t) {
-
- entry.except(t);
- procedure.exception(parentGraph, t);
- return entry.getResult();
-
- }
-
- }
public <T> Object performQuery(ReadGraphImpl parentGraph, final AsyncMultiRead<T> query, final CacheEntryBase entry_, Object procedure_) throws DatabaseException {
} catch (DatabaseException e) {
Logger.defaultLogError(e);
}
- if(DebugPolicy.DEPENDENCIES) System.out.println(child + " -> " + parent);
+ if (Development.DEVELOPMENT) {
+ if(Development.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_DEPENDENCIES, Bindings.BOOLEAN)) {
+ System.err.println(child + " -> " + parent);
+ }
+ }
}
if (listener != null) {
list.add(result);
}
- if(DebugPolicy.LISTENER) {
- new Exception().printStackTrace();
- System.out.println("addListener -> " + list.size() + " " + entry + " " + base + " " + procedure);
+ if (Development.DEVELOPMENT) {
+ if(Development.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_LISTENERS, Bindings.BOOLEAN)) {
+ new Exception().printStackTrace();
+ System.err.println("addListener -> " + list.size() + " " + entry + " " + base + " " + procedure);
+ }
}
return result;
}
}
- public static void waitPending(QueryProcessor processor, CacheEntry entry) throws DatabaseException {
+ public static void waitPending(ReadGraphImpl graph, CacheEntry entry) throws DatabaseException {
int counter = 0;
while(entry.isPending()) {
try {
- SessionTask task = null;//processor.getOwnTask(processor.thread.get());
- if(task != null) {
- task.run(processor.thread.get());
- } else {
+ boolean performed = graph.performPending();
+ if(!performed) {
Thread.sleep(1);
counter++;
- if(counter > 5000) {
+ if(counter > 30000) {
CacheEntryBase base = ((CacheEntryBase)entry);
// if(base.created != null) {
// System.err.println("created:");
// System.err.println("ready:");
// base.ready.printStackTrace();
// }
- new Exception("Timeout waiting for request to complete: " + entry.getOriginalRequest().toString()).printStackTrace();
- throw new DatabaseException("Timeout waiting for request to complete.");
+ new Exception("Timeout waiting for request to complete: " + entry.getOriginalRequest()).printStackTrace();
+ throw new DatabaseException("Timeout waiting for request to complete." + entry.getOriginalRequest());
//System.err.println("asd");
//base.getQuery().recompute(null, null, entry);
}
}
public static <T> T resultReadEntry(ReadGraphImpl graph, Read r, CacheEntry parent, ListenerBase listener, AsyncProcedure<T> procedure) throws DatabaseException {
- AsyncProcedureWrapper<T> wrap = new AsyncProcedureWrapper<>(procedure);
- QueryCache.runnerReadEntry(graph, r, parent, listener, wrap, true);
- return wrap.get();
+ return (T)QueryCache.runnerReadEntry(graph, r, parent, listener, procedure, true);
}
public static <T> T resultAsyncReadEntry(ReadGraphImpl graph, AsyncRead r, CacheEntry parent, ListenerBase listener, AsyncProcedure<T> procedure) throws DatabaseException {
- AsyncProcedureWrapper<T> wrap = new AsyncProcedureWrapper<>(procedure);
- QueryCache.runnerAsyncReadEntry(graph, r, parent, listener, wrap, true);
- return wrap.get();
+ return (T)QueryCache.runnerAsyncReadEntry(graph, r, parent, listener, procedure, true);
}
public static byte[] resultValueQuery(ReadGraphImpl graph, int r, CacheEntry parent, ListenerBase listener) throws DatabaseException {
Code Review / simantics / platform.git / blobdiff