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=c6942b5ed7b7972eb321c28f11fa5c4ec0be91b3;hb=1fb7e00de0895266e08097371c6f84e4322d6530;hp=6c3497f1569b4040ae10200e8a78d1a913bc7cdc;hpb=0ae2b770234dfc3cbb18bd38f324125cf0faca07;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 6c3497f15..c6942b5ed 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 @@ -27,12 +27,14 @@ import java.util.TreeMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.Semaphore; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; import org.eclipse.core.runtime.Platform; 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; @@ -87,6 +89,7 @@ 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.QueryProcessor; import org.simantics.db.impl.query.QueryProcessor.SessionRead; import org.simantics.db.impl.query.QueryProcessor.SessionTask; @@ -118,7 +121,6 @@ import org.simantics.db.request.ExternalRead; import org.simantics.db.request.MultiRead; import org.simantics.db.request.Read; import org.simantics.db.request.ReadInterface; -import org.simantics.db.request.UndoTraits; import org.simantics.db.request.Write; import org.simantics.db.request.WriteInterface; import org.simantics.db.request.WriteOnly; @@ -156,7 +158,6 @@ import org.simantics.db.service.XSupport; import org.simantics.layer0.Layer0; import org.simantics.utils.DataContainer; import org.simantics.utils.Development; -import org.simantics.utils.datastructures.Callback; import org.simantics.utils.threads.logger.ITask; import org.simantics.utils.threads.logger.ThreadLogger; @@ -325,40 +326,29 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule private Object result; final Semaphore sema = new Semaphore(0); private Throwable throwable = null; - final Callback callback = new Callback() { - @Override - public void run(DatabaseException e) { - synchronized (TaskHelper.this) { - throwable = e; - } + final Consumer callback = e -> { + synchronized (TaskHelper.this) { + throwable = e; } }; final Procedure proc = new Procedure() { @Override public void execute(Object result) { - callback.run(null); + callback.accept(null); } @Override public void exception(Throwable t) { if (t instanceof DatabaseException) - callback.run((DatabaseException)t); + callback.accept((DatabaseException)t); else - callback.run(new DatabaseException("" + name + "operation failed.", t)); - } - }; - final WriteTraits writeTraits = new WriteTraits() { - @Override - public UndoTraits getUndoTraits() { - return null; - } - @Override - public VirtualGraph getProvider() { - return null; + callback.accept(new DatabaseException("" + name + "operation failed.", t)); } }; + final WriteTraits writeTraits = new WriteTraits() {}; TaskHelper(String name) { this.name = name; } + @SuppressWarnings("unchecked") T getResult() { return (T)result; } @@ -397,7 +387,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule return null; } - public void scheduleRequest(final Write request, final Callback callback, final Semaphore notify) { + public void scheduleRequest(final Write request, final Consumer callback, final Semaphore notify) { scheduleRequest(request, callback, notify, null); } @@ -405,7 +395,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule * @see fi.vtt.simantics.procore.internal.WriteRequestScheduler#scheduleRequest(org.simantics.db.request.Write, org.simantics.utils.datastructures.Callback, java.util.concurrent.Semaphore, java.lang.Boolean) */ @Override - public void scheduleRequest(final Write request, final Callback callback, final Semaphore notify, Boolean combine) { + public void scheduleRequest(final Write request, final Consumer callback, final Semaphore notify, Boolean combine) { assert (request != null); @@ -441,6 +431,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule try { flushCounter = 0; + Disposable.safeDispose(clientChanges); clientChanges = new ClientChangesImpl(SessionImplSocket.this); VirtualGraph vg = getProvider(request.getProvider()); @@ -450,12 +441,12 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule @Override public void execute(Object result) { - if(callback != null) callback.run(null); + if(callback != null) callback.accept(null); } @Override public void exception(Throwable t) { - if(callback != null) callback.run((DatabaseException)t); + if(callback != null) callback.accept((DatabaseException)t); } }); @@ -476,7 +467,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule } - assert(!queryProvider2.dirty); + assert(!queryProvider2.cache.dirty); } catch (Throwable e) { @@ -567,6 +558,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()); @@ -626,7 +618,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule } - public void scheduleRequest(final DelayedWrite request, final Callback callback, final Semaphore notify) { + public void scheduleRequest(final DelayedWrite request, final Consumer callback, final Semaphore notify) { scheduleRequest(request, callback, notify, null); } @@ -634,7 +626,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule * @see fi.vtt.simantics.procore.internal.WriteRequestScheduler#scheduleRequest(org.simantics.db.request.DelayedWrite, org.simantics.utils.datastructures.Callback, java.util.concurrent.Semaphore, java.lang.Boolean) */ @Override - public void scheduleRequest(final DelayedWrite request, final Callback callback, final Semaphore notify, Boolean combine) { + public void scheduleRequest(final DelayedWrite request, final Consumer callback, final Semaphore notify, Boolean combine) { final ITask total = ThreadLogger.getInstance().begin("ScheduleDelayedWrite"); @@ -652,13 +644,13 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule @Override public void execute(Object result) { if (callback != null) - callback.run(null); + callback.accept(null); } @Override public void exception(Throwable t) { if (callback != null) { - if (t instanceof DatabaseException) callback.run((DatabaseException) t); - else callback.run(new DatabaseException(t)); + if (t instanceof DatabaseException) callback.accept((DatabaseException) t); + else callback.accept(new DatabaseException(t)); } else Logger.defaultLogError("Unhandled exception", t); } @@ -675,6 +667,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule } catch (Throwable e) { delayedWriteState.except(e); total.finish(); + dwg.close(); return; } finally { // newGraph.state.barrier.dec(); @@ -688,6 +681,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule fireSessionVariableChange(SessionVariables.QUEUED_WRITES); flushCounter = 0; + Disposable.safeDispose(clientChanges); clientChanges = new ClientChangesImpl(SessionImplSocket.this); acquireWriteOnly(); @@ -858,14 +852,17 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule private void maintainCluster(ClusterImpl before, ClusterI after_) { if(after_ != null && after_ != before) { ClusterImpl after = (ClusterImpl)after_; - if(currentCluster == before) currentCluster = after; + if(currentCluster == before) { + currentCluster = after; + } clusterTable.replaceCluster(after); } } public int createResourceKey(int foreignCounter) throws DatabaseException { - if(currentCluster == null) + if(currentCluster == null) { currentCluster = getNewResourceCluster(); + } if(currentCluster.getNumberOfResources(clusterTranslator) == ClusterTable.CLUSTER_FILL_SIZE) { ClusterWriteOnly newCluster = (ClusterWriteOnly)getNewResourceCluster(); newCluster.foreignLookup = new byte[foreignCounter]; @@ -1299,11 +1296,12 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule try { - int thread = request.hashCode() & queryProvider2.THREAD_MASK; + //int thread = request.hashCode() & queryProvider2.THREAD_MASK; fireSessionVariableChange(SessionVariables.QUEUED_WRITES); flushCounter = 0; + Disposable.safeDispose(clientChanges); clientChanges = new ClientChangesImpl(SessionImplSocket.this); acquireWriteOnly(); @@ -1364,12 +1362,12 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule } - public void scheduleRequest(final WriteOnly request, final Callback callback, final Semaphore notify) { + public void scheduleRequest(final WriteOnly request, final Consumer callback, final Semaphore notify) { scheduleRequest(request, callback, notify, null); } @Override - public void scheduleRequest(final WriteOnly request, final Callback callback, final Semaphore notify, Boolean combine) { + public void scheduleRequest(final WriteOnly request, final Consumer callback, final Semaphore notify, Boolean combine) { assertAlive(); @@ -1389,6 +1387,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule fireSessionVariableChange(SessionVariables.QUEUED_WRITES); flushCounter = 0; + Disposable.safeDispose(clientChanges); clientChanges = new ClientChangesImpl(SessionImplSocket.this); acquireWriteOnly(); @@ -1402,12 +1401,12 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule @Override public void execute(Object result) { - if(callback != null) callback.run(null); + if(callback != null) callback.accept(null); } @Override public void exception(Throwable t) { - if(callback != null) callback.run((DatabaseException)t); + if(callback != null) callback.accept((DatabaseException)t); } }); @@ -1430,7 +1429,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule if(!(e instanceof CancelTransactionException)) { if (callback != null) - callback.run(new DatabaseException(e)); + callback.accept(new DatabaseException(e)); } writeState.except(e); @@ -1516,7 +1515,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) { @@ -1534,7 +1534,10 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule procedure.execute(graph, t); } - }, listener); + }; + + QueryCache.runnerReadEntry(newGraph, request, null, listener, ap); + } catch (Throwable t) { // This is handled by the AsyncProcedure //Logger.defaultLogError("Internal error", t); @@ -1627,9 +1630,12 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule if (listener != null) { - newGraph.processor.query(newGraph, request, null, procedure, listener); - -// newGraph.waitAsync(request); + try { + QueryCache.runnerAsyncReadEntry(newGraph, request, null, listener, procedure); + //newGraph.processor.query(newGraph, request, null, procedure, listener); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } else { @@ -1969,14 +1975,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule assertAlive(); Semaphore notify = new Semaphore(0); final DataContainer exception = new DataContainer(); - scheduleRequest(request, new Callback() { - - @Override - public void run(DatabaseException e) { - exception.set(e); - } - - }, notify); + scheduleRequest(request, e -> exception.set(e), notify); acquire(notify, request); if(exception.get() != null) throw exception.get(); } @@ -2070,12 +2069,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule assertNotSession(); Semaphore notify = new Semaphore(0); final DataContainer exception = new DataContainer(); - scheduleRequest(request, new Callback() { - @Override - public void run(DatabaseException e) { - exception.set(e); - } - }, notify); + scheduleRequest(request, e -> exception.set(e), notify); acquire(notify, request); if(exception.get() != null) throw exception.get(); } @@ -2086,12 +2080,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule assertAlive(); Semaphore notify = new Semaphore(0); final DataContainer exception = new DataContainer(); - scheduleRequest(request, new Callback() { - @Override - public void run(DatabaseException e) { - exception.set(e); - } - }, notify); + scheduleRequest(request, e -> exception.set(e), notify); acquire(notify, request); if(exception.get() != null) throw exception.get(); } @@ -2123,7 +2112,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule } @Override - public void asyncRequest(final Write request, final Callback callback) { + public void asyncRequest(final Write request, final Consumer callback) { scheduleRequest(request, callback, null); @@ -2151,7 +2140,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule } @Override - public void asyncRequest(final DelayedWrite request, final Callback callback) { + public void asyncRequest(final DelayedWrite request, final Consumer callback) { scheduleRequest(request, callback, null); @@ -2168,7 +2157,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule } @Override - public void asyncRequest(final WriteOnly request, final Callback callback) { + public void asyncRequest(final WriteOnly request, final Consumer callback) { scheduleRequest(request, callback, null); @@ -2292,16 +2281,16 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule return true; } - static void loadVirtualStatements(final VirtualGraphServerSupportImpl support, ReadGraphImpl graph, int subject, final Callback runnable) { + static void loadVirtualStatements(final VirtualGraphServerSupportImpl support, ReadGraphImpl graph, int subject, final Consumer runnable) { - Callback composite = new Callback() { + Consumer composite = new Consumer() { AtomicInteger ready = new AtomicInteger(support.providers.size() + 1); @Override - public void run(ReadGraphImpl graph) { + public void accept(ReadGraphImpl graph) { if(ready.decrementAndGet() == 0) { - runnable.run(graph); + runnable.accept(graph); } } @@ -2315,24 +2304,24 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule e.printStackTrace(); } } else { - composite.run(graph); + composite.accept(graph); } } - composite.run(graph); + composite.accept(graph); } - static void loadVirtualStatements(final VirtualGraphServerSupportImpl support, ReadGraphImpl graph, int subject, int predicate, final Callback runnable) { + static void loadVirtualStatements(final VirtualGraphServerSupportImpl support, ReadGraphImpl graph, int subject, int predicate, final Consumer runnable) { - Callback composite = new Callback() { + Consumer composite = new Consumer() { AtomicInteger ready = new AtomicInteger(support.providers.size() + 1); @Override - public void run(ReadGraphImpl graph) { + public void accept(ReadGraphImpl graph) { if(ready.decrementAndGet() == 0) { - runnable.run(graph); + runnable.accept(graph); } } @@ -2346,11 +2335,11 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule e.printStackTrace(); } } else { - composite.run(graph); + composite.accept(graph); } } - composite.run(graph); + composite.accept(graph); } @@ -3526,7 +3515,7 @@ public abstract class SessionImplSocket implements Session, WriteRequestSchedule public int getAmountOfQueryThreads() { // This must be a power of two - return 1; + return 4; // return Integer.highestOneBit(Runtime.getRuntime().availableProcessors()); }