X-Git-Url: https://gerrit.simantics.org/r/gitweb?p=simantics%2Fplatform.git;a=blobdiff_plain;f=bundles%2Forg.simantics.db.procore%2Fsrc%2Ffi%2Fvtt%2Fsimantics%2Fprocore%2Finternal%2FSessionImplSocket.java;h=30e0dde5244de43dbd11ff9582015dad0d831f76;hp=9a86dc1fa2ae813b0b3a137b51a9063546af27af;hb=4aba159170fc72d39c2f930ea224aa71f4cdc2e7;hpb=56a799c9b7d395cefb59e101cd0f7ce8d68f88e6 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 9a86dc1fa..30e0dde52 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 @@ -1,5 +1,5 @@ /******************************************************************************* - * Copyright (c) 2007, 2010 Association for Decentralized Information Management + * Copyright (c) 2007, 2018 Association for Decentralized Information Management * in Industry THTH ry. * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v1.0 @@ -34,10 +34,12 @@ import org.simantics.databoard.Bindings; import org.simantics.db.AsyncReadGraph; import org.simantics.db.ChangeSet; import org.simantics.db.DevelopmentKeys; +import org.simantics.db.Disposable; import org.simantics.db.ExternalValueSupport; import org.simantics.db.Metadata; import org.simantics.db.MonitorContext; import org.simantics.db.MonitorHandler; +import org.simantics.db.ReadGraph; import org.simantics.db.Resource; import org.simantics.db.ResourceSerializer; import org.simantics.db.Session; @@ -51,10 +53,13 @@ import org.simantics.db.common.Indexing; import org.simantics.db.common.TransactionPolicyRelease; import org.simantics.db.common.procedure.adapter.AsyncMultiProcedureAdapter; import org.simantics.db.common.procedure.adapter.ProcedureAdapter; +import org.simantics.db.common.procedure.adapter.SyncMultiProcedureAdapter; import org.simantics.db.common.procedure.wrapper.NoneToAsyncListener; import org.simantics.db.common.procedure.wrapper.NoneToAsyncMultiListener; import org.simantics.db.common.procedure.wrapper.NoneToAsyncMultiProcedure; import org.simantics.db.common.procedure.wrapper.NoneToAsyncProcedure; +import org.simantics.db.common.procedure.wrapper.NoneToSyncMultiListener; +import org.simantics.db.common.procedure.wrapper.NoneToSyncMultiProcedure; import org.simantics.db.common.procedure.wrapper.SyncToAsyncListener; import org.simantics.db.common.procedure.wrapper.SyncToAsyncMultiListener; import org.simantics.db.common.procedure.wrapper.SyncToAsyncMultiProcedure; @@ -74,6 +79,7 @@ import org.simantics.db.exception.ResourceNotFoundException; import org.simantics.db.exception.RuntimeDatabaseException; import org.simantics.db.exception.ServiceException; import org.simantics.db.exception.ServiceNotFoundException; +import org.simantics.db.impl.BlockingAsyncProcedure; import org.simantics.db.impl.ClusterBase; import org.simantics.db.impl.ClusterI; import org.simantics.db.impl.ClusterTraitsBase; @@ -87,7 +93,9 @@ import org.simantics.db.impl.graph.WriteGraphImpl; 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.procedure.ResultCallWrappedSyncQueryProcedure; +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; @@ -408,7 +416,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) { @@ -429,6 +437,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule try { flushCounter = 0; + Disposable.safeDispose(clientChanges); clientChanges = new ClientChangesImpl(SessionImplSocket.this); VirtualGraph vg = getProvider(request.getProvider()); @@ -464,7 +473,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule } - assert(!queryProvider2.dirty); + assert(!queryProvider2.cache.dirty); } catch (Throwable e) { @@ -497,7 +506,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule // state.cancelCommit(context, clusterStream); // if (!empty) { // if (!context.isOk()) // this is a blocking operation -// throw new InternalException("Cancel failed. This should never happen. Contact application support."); +// throw new InternalException("Cancel failed. This should never happen."); // getQueryProvider2().performDirtyUpdates(writeState.getGraph()); // } // state.cancelCommit2(context, clusterStream); @@ -545,7 +554,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) { @@ -555,6 +564,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule fireSessionVariableChange(SessionVariables.QUEUED_WRITES); flushCounter = 0; + Disposable.safeDispose(clientChanges); clientChanges = new ClientChangesImpl(SessionImplSocket.this); VirtualGraph vg = getProvider(request.getProvider()); @@ -630,7 +640,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) { @@ -677,6 +687,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule fireSessionVariableChange(SessionVariables.QUEUED_WRITES); flushCounter = 0; + Disposable.safeDispose(clientChanges); clientChanges = new ClientChangesImpl(SessionImplSocket.this); acquireWriteOnly(); @@ -1018,15 +1029,13 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule try { int sResourceKey = ClusterTraits.getResourceIndexFromResourceKey(s); - int pResourceKey = ClusterTraits.getResourceIndexFromResourceKey(p); - int oResourceKey = ClusterTraits.getResourceIndexFromResourceKey(o); ClusterI pc = clusterTable.getClusterProxyByResourceKey(p); ClusterI oc = clusterTable.getClusterProxyByResourceKey(o); clusterTranslator.addStatementIndex(cluster, sResourceKey, cluster.getClusterUID(), ClusterChange.REMOVE_OPERATION); - clusterTranslator.addStatementIndex(cluster, pResourceKey, pc.getClusterUID(), ClusterStream.NULL_OPERATION); - clusterTranslator.addStatementIndex(cluster, oResourceKey, oc.getClusterUID(), ClusterStream.NULL_OPERATION); + clusterTranslator.addStatementIndex(cluster, p, pc.getClusterUID(), ClusterStream.NULL_OPERATION); + clusterTranslator.addStatementIndex(cluster, o, oc.getClusterUID(), ClusterStream.NULL_OPERATION); clusterTranslator.removeStatement(cluster); queryProvider2.invalidateResource(s); @@ -1296,6 +1305,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule fireSessionVariableChange(SessionVariables.QUEUED_WRITES); flushCounter = 0; + Disposable.safeDispose(clientChanges); clientChanges = new ClientChangesImpl(SessionImplSocket.this); acquireWriteOnly(); @@ -1369,7 +1379,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) { @@ -1381,6 +1391,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule fireSessionVariableChange(SessionVariables.QUEUED_WRITES); flushCounter = 0; + Disposable.safeDispose(clientChanges); clientChanges = new ClientChangesImpl(SessionImplSocket.this); acquireWriteOnly(); @@ -1468,7 +1479,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) { @@ -1490,9 +1501,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) { @@ -1508,7 +1519,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) { @@ -1526,7 +1538,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); @@ -1606,13 +1621,77 @@ 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) { + + fireSessionVariableChange(SessionVariables.QUEUED_READS); + + final ReadGraphImpl newGraph = ReadGraphImpl.create(getQueryProvider2()); + + try { + + if (listener != null) { + + 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"); + + BlockingAsyncProcedure wrap = new BlockingAsyncProcedure(newGraph, procedure, request); + + try { + + request.perform(newGraph, wrap); + wrap.get(); + + } catch (DatabaseException e) { + + Logger.defaultLogError(e); + + } + + } + + } finally { + + fireSessionVariableChange(SessionVariables.QUEUED_READS); + + } + + } + + }); + + } + + public void scheduleRequest(final MultiRead request, final SyncMultiProcedure procedure, final Semaphore notify) { + + assert (request != null); + assert (procedure != null); + + int thread = request.hashCode() & queryProvider2.THREAD_MASK; + + int sync = notify != null ? thread : -1; + + requestManager.scheduleRead(new SessionRead(null, notify) { @Override public void run(int thread) { fireSessionVariableChange(SessionVariables.QUEUED_READS); + ListenerBase listener = getListenerBase(procedure); + final ReadGraphImpl newGraph = ReadGraphImpl.create(getQueryProvider2()); try { @@ -1625,22 +1704,15 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule } else { - final ResultCallWrappedSingleQueryProcedure4 wrapper = new ResultCallWrappedSingleQueryProcedure4( - procedure, "request"); + final ResultCallWrappedSyncQueryProcedure wrapper = new ResultCallWrappedSyncQueryProcedure(procedure); try { -// newGraph.state.barrier.inc(); - request.perform(newGraph, wrapper); -// newGraph.waitAsync(request); - } catch (Throwable t) { - wrapper.exception(newGraph, t); -// newGraph.waitAsync(request); - + t.printStackTrace(); } @@ -1657,7 +1729,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule }); } - + public void scheduleRequest(final AsyncMultiRead request, final AsyncMultiProcedure procedure, final Semaphore notify) { assert (request != null); @@ -1667,7 +1739,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) { @@ -1721,7 +1793,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) { @@ -1736,25 +1808,11 @@ 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); - -// newGraph.waitAsync(request); + try { + QueryCacheBase.resultExternalReadEntry(newGraph, request, null, listener, procedure); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } else { @@ -3044,21 +3102,21 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule final ArrayList result = new ArrayList(); final DataContainer exception = new DataContainer(); - syncRequest(request, new AsyncMultiProcedure() { + syncRequest(request, new SyncMultiProcedure() { @Override - public void execute(AsyncReadGraph graph, T t) { + public void execute(ReadGraph graph, T t) { synchronized(result) { result.add(t); } } @Override - public void finished(AsyncReadGraph graph) { + public void finished(ReadGraph graph) { } @Override - public void exception(AsyncReadGraph graph, Throwable t) { + public void exception(ReadGraph graph, Throwable t) { exception.set(t); } @@ -3076,33 +3134,27 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule } @Override - public Collection syncRequest(MultiRead request, AsyncMultiListener procedure) throws DatabaseException { + public Collection syncRequest(MultiRead arg0, SyncMultiProcedure arg1) throws DatabaseException { assertNotSession(); - return syncRequest(request, (AsyncMultiProcedure)procedure); + throw new Error("Not implemented!"); } @Override public Collection syncRequest(MultiRead request, SyncMultiListener procedure) throws DatabaseException { assertNotSession(); - return syncRequest(request, new SyncToAsyncMultiListener(procedure)); + return syncRequest(request, (SyncMultiProcedure)procedure); } @Override public Collection syncRequest(MultiRead request, MultiListener procedure) throws DatabaseException { assertNotSession(); - return syncRequest(request, new NoneToAsyncMultiListener(procedure)); - } - - @Override - public Collection syncRequest(MultiRead request, SyncMultiProcedure procedure) throws DatabaseException { - assertNotSession(); - return syncRequest(request, new SyncToAsyncMultiProcedure(procedure)); + return syncRequest(request, new NoneToSyncMultiListener(procedure)); } @Override public Collection syncRequest(MultiRead request, MultiProcedure procedure) throws DatabaseException { assertNotSession(); - return syncRequest(request, new NoneToAsyncMultiProcedure(procedure)); + return syncRequest(request, new NoneToSyncMultiProcedure(procedure)); } @Override @@ -3258,38 +3310,33 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule assert(request != null); - asyncRequest(request, new AsyncMultiProcedureAdapter() { + asyncRequest(request, new SyncMultiProcedureAdapter() { @Override - public void exception(AsyncReadGraph graph, Throwable t) { + public void exception(ReadGraph graph, Throwable t) { t.printStackTrace(); } }); } - @Override - public void asyncRequest(MultiRead request, AsyncMultiListener procedure) { - asyncRequest(request, (AsyncMultiProcedure)procedure); - } - @Override public void asyncRequest(MultiRead request, SyncMultiListener procedure) { - asyncRequest(request, new SyncToAsyncMultiListener(procedure)); + asyncRequest(request, (SyncMultiProcedure)procedure); } @Override public void asyncRequest(MultiRead request, MultiListener procedure) { - asyncRequest(request, new NoneToAsyncMultiListener(procedure)); + asyncRequest(request, new NoneToSyncMultiListener(procedure)); } @Override public void asyncRequest(MultiRead request, SyncMultiProcedure procedure) { - asyncRequest(request, new SyncToAsyncMultiProcedure(procedure)); + scheduleRequest(request, procedure, null); } @Override public void asyncRequest(MultiRead request, MultiProcedure procedure) { - asyncRequest(request, new NoneToAsyncMultiProcedure(procedure)); + asyncRequest(request, new NoneToSyncMultiProcedure(procedure)); } @Override @@ -3331,17 +3378,6 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule asyncRequest(request, new NoneToAsyncMultiProcedure(procedure)); } - @Override - public Collection syncRequest(MultiRead arg0, AsyncMultiProcedure arg1) throws DatabaseException { - assertNotSession(); - throw new Error("Not implemented!"); - } - - @Override - public void asyncRequest(MultiRead arg0, AsyncMultiProcedure arg1) { - throw new Error("Not implemented!"); - } - @Override final public void asyncRequest(final ExternalRead request) {