X-Git-Url: https://gerrit.simantics.org/r/gitweb?a=blobdiff_plain;f=bundles%2Forg.simantics.db.procore%2Fsrc%2Ffi%2Fvtt%2Fsimantics%2Fprocore%2Finternal%2FSessionImplSocket.java;h=d5d9445ca4fb230520769af6f7cb1e4d0ea0e8fc;hb=d0e8f57df83c175bac2098412ee71646fb9ff0e9;hp=9290ac99fdd98a5792a542a0b51b7309941f1704;hpb=bdab0adcee131a18290393e1c2fca72de48c10d2;p=simantics%2Fplatform.git diff --git a/bundles/org.simantics.db.procore/src/fi/vtt/simantics/procore/internal/SessionImplSocket.java b/bundles/org.simantics.db.procore/src/fi/vtt/simantics/procore/internal/SessionImplSocket.java index 9290ac99f..d5d9445ca 100644 --- a/bundles/org.simantics.db.procore/src/fi/vtt/simantics/procore/internal/SessionImplSocket.java +++ b/bundles/org.simantics.db.procore/src/fi/vtt/simantics/procore/internal/SessionImplSocket.java @@ -50,6 +50,7 @@ import org.simantics.db.authentication.UserAuthenticationAgent; import org.simantics.db.authentication.UserAuthenticator; import org.simantics.db.common.Indexing; import org.simantics.db.common.TransactionPolicyRelease; +import org.simantics.db.common.procedure.BlockingAsyncProcedure; import org.simantics.db.common.procedure.adapter.AsyncMultiProcedureAdapter; import org.simantics.db.common.procedure.adapter.ProcedureAdapter; import org.simantics.db.common.procedure.wrapper.NoneToAsyncListener; @@ -89,6 +90,8 @@ import org.simantics.db.impl.graph.WriteSupport; import org.simantics.db.impl.internal.RandomAccessValueSupport; import org.simantics.db.impl.procedure.ResultCallWrappedQueryProcedure4; import org.simantics.db.impl.procedure.ResultCallWrappedSingleQueryProcedure4; +import org.simantics.db.impl.query.QueryCache; +import org.simantics.db.impl.query.QueryCacheBase; import org.simantics.db.impl.query.QueryProcessor; import org.simantics.db.impl.query.QueryProcessor.SessionRead; import org.simantics.db.impl.query.QueryProcessor.SessionTask; @@ -409,7 +412,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule int thread = request.hashCode() & queryProvider2.THREAD_MASK; - requestManager.scheduleWrite(new SessionTask(request, thread, thread) { + requestManager.scheduleWrite(new SessionTask(true) { @Override public void run(int thread) { @@ -466,7 +469,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule } - assert(!queryProvider2.dirty); + assert(!queryProvider2.cache.dirty); } catch (Throwable e) { @@ -547,7 +550,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule int thread = request.hashCode() & queryProvider2.THREAD_MASK; - requestManager.scheduleWrite(new SessionTask(request, thread) { + requestManager.scheduleWrite(new SessionTask(true) { @Override public void run(int thread) { @@ -633,7 +636,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule int thread = request.hashCode() & queryProvider2.THREAD_MASK; - requestManager.scheduleWrite(new SessionTask(request, thread) { + requestManager.scheduleWrite(new SessionTask(true) { @Override public void run(int thread) { @@ -1374,7 +1377,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule int thread = request.hashCode() & queryProvider2.THREAD_MASK; - requestManager.scheduleWrite(new SessionTask(request, thread, thread) { + requestManager.scheduleWrite(new SessionTask(true) { @Override public void run(int thread) { @@ -1474,7 +1477,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule int thread = request.hashCode() & queryProvider2.THREAD_MASK; - requestManager.scheduleWrite(new SessionTask(request, thread) { + requestManager.scheduleWrite(new SessionTask(true) { @Override public void run(int thread) { @@ -1496,9 +1499,9 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule assert (request != null); assert (procedure != null); - int thread = request.hashCode() & queryProvider2.THREAD_MASK; + //int thread = request.hashCode() & queryProvider2.THREAD_MASK; - requestManager.scheduleRead(new SessionRead(request, throwable, notify, thread, thread) { + requestManager.scheduleRead(new SessionRead(throwable, notify) { @Override public void run(int thread) { @@ -1514,7 +1517,8 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule if (listener != null) { try { - newGraph.processor.queryRead(newGraph, request, null, new AsyncProcedure() { + + AsyncProcedure ap = new AsyncProcedure() { @Override public void exception(AsyncReadGraph graph, Throwable t) { @@ -1532,7 +1536,10 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule procedure.execute(graph, t); } - }, listener); + }; + + QueryCache.runnerReadEntry(newGraph, request, null, listener, ap, true); + } catch (Throwable t) { // This is handled by the AsyncProcedure //Logger.defaultLogError("Internal error", t); @@ -1612,7 +1619,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule int thread = request.hashCode() & queryProvider2.THREAD_MASK; - requestManager.scheduleRead(new SessionRead(request, null, notify, thread) { + requestManager.scheduleRead(new SessionRead(null, notify) { @Override public void run(int thread) { @@ -1625,26 +1632,31 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule if (listener != null) { - newGraph.processor.query(newGraph, request, null, procedure, listener); - -// newGraph.waitAsync(request); + try { + QueryCacheBase.resultAsyncReadEntry(newGraph, request, null, listener, procedure); + //QueryCache.runnerAsyncReadEntry(newGraph, request, null, listener, procedure, true); + //newGraph.processor.query(newGraph, request, null, procedure, listener); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } else { - final ResultCallWrappedSingleQueryProcedure4 wrapper = new ResultCallWrappedSingleQueryProcedure4( - procedure, "request"); - - try { +// final ResultCallWrappedSingleQueryProcedure4 wrapper = new ResultCallWrappedSingleQueryProcedure4( +// procedure, "request"); -// newGraph.state.barrier.inc(); + BlockingAsyncProcedure wrap = new BlockingAsyncProcedure(newGraph, procedure, request); - request.perform(newGraph, wrapper); + try { -// newGraph.waitAsync(request); + request.perform(newGraph, wrap); + wrap.get(); } catch (Throwable t) { - wrapper.exception(newGraph, t); + wrap.exception(newGraph, t); + +// wrapper.exception(newGraph, t); // newGraph.waitAsync(request); @@ -1673,7 +1685,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule int sync = notify != null ? thread : -1; - requestManager.scheduleRead(new SessionRead(request, null, notify, thread, sync) { + requestManager.scheduleRead(new SessionRead(null, notify) { @Override public void run(int thread) { @@ -1727,7 +1739,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule int thread = request.hashCode() & queryProvider2.THREAD_MASK; - requestManager.scheduleRead(new SessionRead(request, throwable, notify, thread, thread) { + requestManager.scheduleRead(new SessionRead(throwable, notify) { @Override public void run(int thread) { @@ -1742,23 +1754,30 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule if (listener != null) { - newGraph.processor.query(newGraph, request, null, new Procedure() { - - @Override - public void exception(Throwable t) { - procedure.exception(t); - if(throwable != null) { - throwable.set(t); - } - } - - @Override - public void execute(T t) { - if(result != null) result.set(t); - procedure.execute(t); - } - - }, listener); + try { + QueryCacheBase.resultExternalReadEntry(newGraph, request, null, listener, procedure); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } + + +// newGraph.processor.query(newGraph, request, null, new Procedure() { +// +// @Override +// public void exception(Throwable t) { +// procedure.exception(t); +// if(throwable != null) { +// throwable.set(t); +// } +// } +// +// @Override +// public void execute(T t) { +// if(result != null) result.set(t); +// procedure.execute(t); +// } +// +// }, listener); // newGraph.waitAsync(request); @@ -3507,7 +3526,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule public int getAmountOfQueryThreads() { // This must be a power of two - return 1; + return 8; // return Integer.highestOneBit(Runtime.getRuntime().availableProcessors()); }