X-Git-Url: https://gerrit.simantics.org/r/gitweb?p=simantics%2Fplatform.git;a=blobdiff_plain;f=bundles%2Forg.simantics.db.impl%2Fsrc%2Forg%2Fsimantics%2Fdb%2Fimpl%2Fquery%2FAsyncReadEntry.java;h=2fdb43d113332e4f33b6b131685fd8c80c112b22;hp=972bd381f222207b2ff91d02d33b13269321913c;hb=HEAD;hpb=ff1c29bf73b3e5ea939dc1987540aa0240dc4ae6 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 972bd381f..2fdb43d11 100644 --- 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 @@ -11,39 +11,45 @@ *******************************************************************************/ package org.simantics.db.impl.query; +import java.util.Collection; + +import org.simantics.databoard.Bindings; import org.simantics.db.AsyncReadGraph; +import org.simantics.db.DevelopmentKeys; import org.simantics.db.exception.DatabaseException; import org.simantics.db.impl.BlockingAsyncProcedure; -import org.simantics.db.impl.DebugPolicy; -import org.simantics.db.impl.graph.AsyncBarrierImpl; -import org.simantics.db.impl.graph.BarrierTracing; import org.simantics.db.impl.graph.ReadGraphImpl; import org.simantics.db.impl.query.QueryProcessor.SessionTask; import org.simantics.db.procedure.AsyncProcedure; import org.simantics.db.request.AsyncRead; +import org.simantics.utils.Development; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -final public class AsyncReadEntry extends CacheEntryBase> implements AsyncProcedure { +final public class AsyncReadEntry extends CacheEntryBase> implements AsyncProcedure, IPending { private static final Logger LOGGER = LoggerFactory.getLogger(AsyncReadEntry.class); - protected AsyncRead request; + protected AsyncRead id; + protected PendingTaskSupport pendingTaskSupport; AsyncReadEntry(AsyncRead request) { - this.request = request; - if (DebugPolicy.QUERY_STATE) - System.out.println("[QUERY STATE]: created " + this); + this.id = request; + if (Development.DEVELOPMENT) { + if(Development.getProperty(DevelopmentKeys.CACHE_ENTRY_STATE, Bindings.BOOLEAN)) { + System.err.println("[QUERY STATE]: created " + this); + } + } } @Override int makeHash() { - return request.hashCode(); + return id.hashCode(); } @Override public Object getOriginalRequest() { - return request; + return id; } @Override @@ -72,7 +78,7 @@ final public class AsyncReadEntry extends CacheEntryBase> i try { - BlockingAsyncProcedure proc = new BlockingAsyncProcedure<>(graph.asyncBarrier, graph, new AsyncProcedure() { + BlockingAsyncProcedure proc = new BlockingAsyncProcedure(graph, AsyncReadEntry.this, new AsyncProcedure() { @Override public void execute(AsyncReadGraph graph, T result) { @@ -85,11 +91,9 @@ final public class AsyncReadEntry extends CacheEntryBase> i except(t); } - }, request); - - request.perform(graph, proc); - - proc.get(); + }, id, true); + + proc.performSync(id); } catch (Throwable t) { except(t); @@ -104,17 +108,17 @@ final public class AsyncReadEntry extends CacheEntryBase> i @Override public int type() { - return request.getFlags(); + return id.getFlags(); } @Override public String toString() { - if (request == null) + if (id == null) return "DISCARDED"; else if (isExcepted()) - return request.toString() + " " + getResult(); + return id.toString() + " " + getResult(); else - return request.toString() + " " + statusOrException; + return id.toString() + " " + statusOrException; } }; @@ -147,112 +151,71 @@ final public class AsyncReadEntry extends CacheEntryBase> i } - public static T computeForEach(ReadGraphImpl graph, AsyncRead request, AsyncReadEntry entry, + public static T computeForEach(ReadGraphImpl callerGraph, AsyncRead request, AsyncReadEntry entry, AsyncProcedure procedure_, boolean needsToBlock) throws DatabaseException { - AsyncProcedure procedure = entry != null ? entry : procedure_; - - ReadGraphImpl queryGraph = graph.withParent(entry); - queryGraph.asyncBarrier.inc(); - - BlockingAsyncProcedure proc = new BlockingAsyncProcedure<>(queryGraph.asyncBarrier, graph, null, request); - - class AsyncTask extends SessionTask { - - int counter = 0; - T result; - DatabaseException exception; - - public AsyncTask(ReadGraphImpl graph) { - super(graph); - } - - @Override - public void run0(int thread) { - if(needsToBlock) proc.waitBarrier(); - if(proc.isDone()) { - ReadGraphImpl executeGraph = graph.withParent(graph.parent); - executeGraph.asyncBarrier.inc(); - try { - result = (T)proc.get(); - if(procedure != null) { - procedure.execute(executeGraph, result); - } - } catch (DatabaseException e) { - if(procedure != null) procedure.exception(executeGraph, e); - exception = e; - } catch (Throwable t) { - DatabaseException dbe = new DatabaseException(t); - if(procedure != null) procedure.exception(executeGraph, dbe); - exception = dbe; - } finally { - if (entry != null) { - // This does not throw - entry.performFromCache(executeGraph, procedure_); - } - executeGraph.asyncBarrier.dec(); - executeGraph.asyncBarrier.waitBarrier(procedure, executeGraph); - } - } else { - if(counter++ > 10000) { - if(BarrierTracing.BOOKKEEPING) { - AsyncBarrierImpl.printReverse(queryGraph.asyncBarrier, 2); - AsyncBarrierImpl caller = queryGraph.asyncBarrier.caller; - while(caller != null) { - System.err.println("called by " + AsyncBarrierImpl.report(caller)); - caller = caller.caller; - } - for(AsyncBarrierImpl ab : BarrierTracing.debuggerMap.keySet()) { - AsyncBarrierImpl.printReverse(ab, 2); - } - } - throw new IllegalStateException("Eternal loop in queries."); - } - graph.processor.schedule(new AsyncTask(graph)); - } - } - - } - - try { - request.perform(queryGraph, proc); - } finally { - queryGraph.asyncBarrier.dec(); - } - - AsyncTask task = new AsyncTask(graph); - - if(needsToBlock) task.run(0); - else if (proc.isDone()) task.run(0); - else { - graph.processor.schedule(task); + BlockingAsyncProcedure proc = new BlockingAsyncProcedure(callerGraph, entry, procedure_, request, needsToBlock); + if(needsToBlock) { + return proc.performSync(request); + } else { + proc.performAsync(request); return null; } - if(task.exception != null) throw task.exception; - else return task.result; - } @Override public String toString() { if (isDiscarded()) - return "DISCARDED " + request.toString(); + return "DISCARDED " + id.toString(); else if (isExcepted()) - return request.toString() + " " + getResult(); + return id.toString() + " " + getResult(); else - return request.toString() + " " + statusOrException; + return id.toString() + " " + statusOrException; } @Override public void execute(AsyncReadGraph graph, T result) { - setResult(result); - setReady(); + Collection tasks = null; + synchronized(this) { + setResult(result); + setReady(); + if(pendingTaskSupport != null) + tasks = pendingTaskSupport.executePending(); + } + if(tasks != null) + for(SessionTask task : tasks) + ((ReadGraphImpl)graph).processor.scheduleNow(task); + } + + @Override + public synchronized void exception(AsyncReadGraph graph, Throwable throwable) { + Collection tasks = null; + synchronized(this) { + except(throwable); + if(pendingTaskSupport != null) + tasks = pendingTaskSupport.executePending(); + } + if(tasks != null) + for(SessionTask task : tasks) + ((ReadGraphImpl)graph).processor.scheduleNow(task); + } + + public void executeWhenResultIsAvailable(QueryProcessor processor, SessionTask task) { + boolean ready = false; + synchronized(this) { + if(pendingTaskSupport == null) + pendingTaskSupport = new PendingTaskSupport(this); + ready = pendingTaskSupport.executeWhenResultIsAvailable(task); + } + if(ready) { + processor.scheduleNow(task); + } } @Override - public void exception(AsyncReadGraph graph, Throwable throwable) { - except(throwable); + public String classId() { + return null; } }