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%2Fgraph%2FReadGraphImpl.java;h=79fe974642d2a71e66054fe6de5d00a60424b7bf;hp=614a96df3c47d29ea45e249dfb10df9653aaedf4;hb=3bcd1023180e2cdc55bc001d0f60075dd7184cfa;hpb=bc5e6cb19f6af5f67bc5cfaad7d602841b8fdd0b diff --git a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/graph/ReadGraphImpl.java b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/graph/ReadGraphImpl.java index 614a96df3..79fe97464 100644 --- a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/graph/ReadGraphImpl.java +++ b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/graph/ReadGraphImpl.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 @@ -18,14 +18,15 @@ import java.io.IOException; import java.io.PrintStream; import java.lang.reflect.Array; import java.lang.reflect.InvocationTargetException; -import java.nio.BufferUnderflowException; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; +import java.util.HashSet; import java.util.IdentityHashMap; import java.util.Iterator; import java.util.List; import java.util.ListIterator; +import java.util.Map; import java.util.Set; import java.util.function.Consumer; @@ -44,6 +45,7 @@ import org.simantics.databoard.type.Datatype; import org.simantics.databoard.util.binary.BinaryFile; import org.simantics.databoard.util.binary.RandomAccessBinary; import org.simantics.db.AsyncReadGraph; +import org.simantics.db.ComputationalValue; import org.simantics.db.DevelopmentKeys; import org.simantics.db.ExternalValueSupport; import org.simantics.db.ReadGraph; @@ -88,7 +90,9 @@ import org.simantics.db.common.primitiverequest.Value; import org.simantics.db.common.primitiverequest.ValueImplied; import org.simantics.db.common.primitiverequest.VariantValueImplied; import org.simantics.db.common.procedure.adapter.AsyncMultiProcedureAdapter; +import org.simantics.db.common.procedure.adapter.AsyncProcedureAdapter; import org.simantics.db.common.procedure.adapter.ProcedureAdapter; +import org.simantics.db.common.procedure.adapter.SyncMultiProcedureAdapter; import org.simantics.db.common.procedure.adapter.TransientCacheAsyncListener; import org.simantics.db.common.procedure.adapter.TransientCacheListener; import org.simantics.db.common.procedure.single.SyncReadProcedure; @@ -104,6 +108,8 @@ 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.NoneToAsyncSetProcedure; +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; @@ -133,11 +139,12 @@ import org.simantics.db.impl.RelationContextImpl; import org.simantics.db.impl.ResourceImpl; import org.simantics.db.impl.internal.RandomAccessValueSupport; import org.simantics.db.impl.internal.ResourceData; -import org.simantics.db.impl.procedure.CallWrappedSingleQueryProcedure4; -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.CacheEntry; +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.SessionTask; import org.simantics.db.impl.query.QuerySupport; import org.simantics.db.impl.query.TripleIntProcedure; import org.simantics.db.impl.support.ResourceSupport; @@ -166,7 +173,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.RequestFlags; import org.simantics.db.request.Write; import org.simantics.db.request.WriteInterface; import org.simantics.db.request.WriteOnly; @@ -182,16 +188,22 @@ import org.simantics.utils.DataContainer; import org.simantics.utils.Development; import org.simantics.utils.datastructures.Pair; import org.simantics.utils.datastructures.collections.CollectionUtils; +import org.slf4j.LoggerFactory; import gnu.trove.map.hash.TObjectIntHashMap; -public class ReadGraphImpl implements ReadGraph { +public class ReadGraphImpl implements AsyncReadGraph { + + private static final org.slf4j.Logger LOGGER = LoggerFactory.getLogger(ReadGraphImpl.class); final static boolean EMPTY_RESOURCE_CHECK = false; final public CacheEntry parent; + public final ReadGraphImpl parentGraph; final public QueryProcessor processor; + public final AsyncBarrierImpl asyncBarrier; + final static Binding DATA_TYPE_BINDING_INTERNAL = Bindings.getBindingUnchecked(Datatype.class); final static Serializer DATA_TYPE_SERIALIZER = Bindings.getSerializerUnchecked(DATA_TYPE_BINDING_INTERNAL); @@ -278,8 +290,10 @@ public class ReadGraphImpl implements ReadGraph { try { - return syncRequest(new org.simantics.db.common.primitiverequest.Resource( - id)); + Integer rid = QueryCache.resultURIToResource(this, id, parent, null); + // FIXME: stupid to throw this here and catch and wrap it right away + if(rid == 0) throw new ResourceNotFoundException(id); + return processor.querySupport.getResource(rid); } catch (ResourceNotFoundException e) { @@ -309,8 +323,7 @@ public class ReadGraphImpl implements ReadGraph { try { - return syncRequest(new org.simantics.db.common.primitiverequest.Resource( - id)); + return getResource(id); } catch (ResourceNotFoundException e) { @@ -331,6 +344,32 @@ public class ReadGraphImpl implements ReadGraph { } } + + @Override + public Map getChildren(Resource resource) throws ValidationException, ServiceException { + + assert (resource != null); + + try { + + int rId = processor.querySupport.getId(resource); + return QueryCache.resultChildMap(this, rId, parent, null); + + } catch (ValidationException e) { + + throw new ValidationException(e); + + } catch (ServiceException e) { + + throw new ServiceException(e); + + } catch (DatabaseException e) { + + throw new ServiceException(INTERNAL_ERROR_STRING, e); + + } + + } final public Resource getRootLibrary() { return processor.getRootLibraryResource(); @@ -1067,9 +1106,10 @@ public class ReadGraphImpl implements ReadGraph { assert (subject != null); + byte[] bytes = null; try { - byte[] bytes = processor.getValue(this, subject); + bytes = processor.getValue(this, subject); if (bytes == null) throw new DoesNotContainValueException("No value for resource " + subject); Serializer serializer = getSerializer(binding); @@ -1079,20 +1119,9 @@ public class ReadGraphImpl implements ReadGraph { throw new DoesNotContainValueException(e); - } catch (IOException e) { - - throw new ServiceException(e); - - } catch (DatabaseException e) { - - throw new ServiceException(e); - - } catch (BufferUnderflowException e) { - // This is sometimes thrown when deserialize fails because wrong format. - // For callers of this method this is just an service exception. - throw new ServiceException(e); - } - + } catch (Throwable t) { + throw new ServiceException("Could not getValue for subject " + debugString(subject) + " and binding " + String.valueOf(binding) + " with bytes " + safeArrayToString(bytes), t); + } } @Override @@ -1893,39 +1922,8 @@ public class ReadGraphImpl implements ReadGraph { @Override public T syncRequest(final Read request) throws DatabaseException { - assert (request != null); - - if (parent != null) { - - try { - return processor.queryRead(this, request, parent, null, null); - } catch (Throwable e) { - if(e instanceof DatabaseException) throw (DatabaseException)e; - else throw new DatabaseException(e); - } - - } else { - - try { - - return processor.tryQuery(this, request); - - } catch (Throwable throwable) { - - //Logger.defaultLogError("Internal read request failure", throwable); - - if (throwable instanceof DatabaseException) - throw (DatabaseException) throwable; - else - throw new DatabaseException( - "Unexpected exception in ReadGraph.syncRequest(Read,Procedure)", - throwable); - - } - - } - + return (T)QueryCache.runnerReadEntry(this, request, parent, null, null, true); } @Override @@ -1947,42 +1945,7 @@ public class ReadGraphImpl implements ReadGraph { ListenerBase listener = procedure != null ? getListenerBase(procedure) : null; - if (parent != null || listener != null) { - - try { - return processor.queryRead(this, request, parent, procedure, listener); - } catch (Throwable e) { - if(e instanceof DatabaseException) throw (DatabaseException)e; - else throw new DatabaseException(e); - } - - } else { - - try { - - T t = processor.tryQuery(this, request); - if(procedure != null) - procedure.execute(this, t); - - return t; - - } catch (Throwable throwable) { - - Logger.defaultLogError("Internal read request failure", throwable); - - if(procedure != null) - procedure.exception(this, throwable); - - if (throwable instanceof DatabaseException) - throw (DatabaseException) throwable; - else - throw new DatabaseException( - "Unexpected exception in ReadGraph.syncRequest(Read,Procedure)", - throwable); - - } - - } + return QueryCache.resultReadEntry(this, request, parent, listener, procedure); } @@ -2022,7 +1985,7 @@ public class ReadGraphImpl implements ReadGraph { throw (DatabaseException) exception; else throw new DatabaseException( - "Unexpected exception in ReadGraph.syncRequest(AsyncMultiRead)", + "Unexpected exception in ReadGraph.syncRequest(AsyncRead)", exception); } } @@ -2038,12 +2001,7 @@ public class ReadGraphImpl implements ReadGraph { throws DatabaseException { assert (request != null); - AsyncReadProcedure procedure = new AsyncReadProcedure(); - syncRequest(request, procedure); - procedure.checkAndThrow(); - return procedure.result; - -// return syncRequest(request, new AsyncProcedureAdapter()); + return syncRequest(request, new AsyncProcedureAdapter<>() ); } @@ -2071,112 +2029,11 @@ public class ReadGraphImpl implements ReadGraph { assert (request != null); - // System.out.println("syncRequest " + request + " syncParent=" + - // syncParent); - - ListenerBase listener = getListenerBase(procedure); - - if (parent != null || listener != null || ((request.getFlags() & RequestFlags.SCHEDULE) > 0)) { - -// Object syncParent = request; - -// final ReadGraphImpl newGraph = newSync(); - - final ResultCallWrappedSingleQueryProcedure4 wrapper = new ResultCallWrappedSingleQueryProcedure4( - procedure, request); - - processor.query(this, request, parent, wrapper, listener); - -// newGraph.waitAsync(syncParent); - - Throwable e = wrapper.getException(); - if (e != null) { - // The request was async - produce meaningful stack trace by - // wrapping - if (e instanceof DatabaseException) - throw (DatabaseException) e; - else - throw new DatabaseException(e); - } - - return wrapper.getResult(); - - } else { - - // System.out.println("direct call " + request ); - - // Do not set the sync state.parent for external threads -// Object syncParent = request; - -// final ReadGraphImpl newGraph = newSync(); - - final ResultCallWrappedSingleQueryProcedure4 wrapper = new ResultCallWrappedSingleQueryProcedure4( - procedure, request); - - try { - - processor.tryQuery(this, request, wrapper); - - } catch (Throwable t) { - - wrapper.exception(this, t); - - } - - Throwable e = wrapper.getException(); - if (e != null) { - // The request was async - produce meaningful stack trace by - // wrapping - if (e instanceof DatabaseException) - throw (DatabaseException) e; - else - throw new DatabaseException(e); - } - - return wrapper.getResult(); - - } - - } - - final private void syncRequest(final AsyncRead request, final AsyncReadProcedure procedure) throws DatabaseException { - - assert (request != null); - - // System.out.println("syncRequest " + request + " syncParent=" + - // syncParent); - ListenerBase listener = getListenerBase(procedure); - if (parent != null || listener != null || ((request.getFlags() & RequestFlags.SCHEDULE) > 0)) { - -// final ReadGraphImpl newGraph = newSync(); - - final ResultCallWrappedSingleQueryProcedure4 wrapper = new ResultCallWrappedSingleQueryProcedure4( - procedure, request); - - processor.query(this, request, parent, wrapper, listener); - - } else { - - try { - -// final ReadGraphImpl newGraph = newSync(); - processor.tryQuery(this, request, procedure); -// newGraph.waitAsync(null); - waitAsyncProcedure(procedure); - - } catch (Throwable t) { - if(Development.DEVELOPMENT) { - if(Development.getProperty(DevelopmentKeys.WRITEGRAPH_EXCEPTION_STACKTRACES, Bindings.BOOLEAN)) { - t.printStackTrace(); - } - } - procedure.exception(this, t); - waitAsyncProcedure(procedure); - } - - } +// BlockingAsyncProcedure ap = new BlockingAsyncProcedure<>(this, procedure, request); + return (T)QueryCache.runnerAsyncReadEntry(this, request, parent, listener, procedure, true); +// return ap.get(); } @@ -2201,21 +2058,21 @@ public class ReadGraphImpl implements ReadGraph { 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); } @@ -2240,39 +2097,35 @@ public class ReadGraphImpl implements ReadGraph { } - @Override - public Collection syncRequest(MultiRead request, - AsyncMultiListener procedure) { - return syncRequest(request, (AsyncMultiProcedure) procedure); - } - @Override public Collection syncRequest(MultiRead request, SyncMultiListener procedure) { - return syncRequest(request, new SyncToAsyncMultiListener(procedure)); + return syncRequest(request, (SyncMultiProcedure)procedure); } @Override public Collection syncRequest(MultiRead request, MultiListener procedure) { - return syncRequest(request, new NoneToAsyncMultiListener(procedure)); + return syncRequest(request, new NoneToSyncMultiListener(procedure)); } @Override public Collection syncRequest(MultiRead request, - AsyncMultiProcedure procedure) { + SyncMultiProcedure procedure) { assert (request != null); ListenerBase listener = getListenerBase(procedure); + final ResultCallWrappedSyncQueryProcedure wrapper = new ResultCallWrappedSyncQueryProcedure(procedure); + if (parent != null || listener != null) { // Object syncParent = request; // final ReadGraphImpl newGraph = newSync(); - processor.query(this, request, parent, procedure, listener); + processor.query(this, request, parent, wrapper, listener); // newGraph.waitAsync(syncParent); @@ -2282,36 +2135,22 @@ public class ReadGraphImpl implements ReadGraph { // final ReadGraphImpl newGraph = newSync(); - final ResultCallWrappedQueryProcedure4 wrapper = new ResultCallWrappedQueryProcedure4(procedure); - try { - request.perform(this, wrapper); - } catch (Throwable t) { - wrapper.exception(this, t); -// newGraph.waitAsync(syncParent); - } } - // TODO - return null; - - } + return wrapper.get(); - @Override - public Collection syncRequest(MultiRead request, - SyncMultiProcedure procedure) { - return syncRequest(request, new SyncToAsyncMultiProcedure(procedure)); } @Override public Collection syncRequest(MultiRead request, MultiProcedure procedure) { - return syncRequest(request, new NoneToAsyncMultiProcedure(procedure)); + return syncRequest(request, new NoneToSyncMultiProcedure(procedure)); } static class AsyncMultiReadProcedure extends ArrayList implements AsyncMultiProcedure { @@ -2542,70 +2381,10 @@ public class ReadGraphImpl implements ReadGraph { final public T syncRequest(final ExternalRead request, final Procedure procedure) throws DatabaseException { - assert (request != null); - - ListenerBase listener = getListenerBase(procedure); - - final DataContainer exception = new DataContainer(); - final DataContainer result = new DataContainer(); - - if (parent != null || listener != null) { - -// final ReadGraphImpl newGraph = newSync(); - - processor.query(this, request, parent, new Procedure() { - - @Override - public void exception(Throwable throwable) { - exception.set(throwable); - procedure.exception(throwable); - } - - @Override - public void execute(T t) { - result.set(t); - procedure.execute(t); - } - - }, listener); + assert (request != null); -// newGraph.waitAsync(request); - - } else { - - try { - - T t = processor.tryQuery(this, request); - result.set(t); - procedure.execute(t); - - } catch (Throwable t) { - - if (t instanceof DatabaseException) { - exception.set((DatabaseException)t); - procedure.exception(exception.get()); - } else { - exception.set(new DatabaseException( - "Unexpected exception in ReadGraph.syncRequest(Read)", - t)); - procedure.exception(exception.get()); - } - - } - - } - - Throwable t = exception.get(); - if (t != null) { - if (t instanceof DatabaseException) - throw (DatabaseException) t; - else - throw new DatabaseException( - "Unexpected exception in ReadGraph.syncRequest(Read)", - t); - } - - return result.get(); + ListenerBase listener = procedure != null ? getListenerBase(procedure) : null; + return QueryCache.resultExternalReadEntry(this, request, parent, listener, procedure); } @@ -3252,42 +3031,19 @@ public class ReadGraphImpl implements ReadGraph { } -// @Override -// final public void forEachDirectObject(final Resource subject, -// final Resource relation, -// final AsyncMultiProcedure procedure) { -// -// processor.forEachDirectObject(this, subject, relation, procedure); -// -// } -// -// @Override -// public void forEachDirectObject(Resource subject, Resource relation, -// SyncMultiProcedure procedure) { -// forEachDirectObject(subject, relation, -// new SyncToAsyncMultiProcedure(procedure)); -// } -// -// @Override -// public void forEachDirectObject(Resource subject, Resource relation, -// MultiProcedure procedure) { -// forEachDirectObject(subject, relation, -// new NoneToAsyncMultiProcedure(procedure)); -// } - @Override - final public void forEachDirectPredicate(final Resource subject, final AsyncMultiProcedure procedure) { + final public void forEachDirectPredicate(final Resource subject, final AsyncProcedure> procedure) { processor.forEachDirectPredicate(this, subject, procedure); } @Override - public void forEachDirectPredicate(Resource subject, SyncMultiProcedure procedure) { - forEachDirectPredicate(subject, new SyncToAsyncMultiProcedure(procedure)); + final public void forEachDirectPredicate(final Resource subject, final SyncProcedure> procedure) { + forEachDirectPredicate(subject, new SyncToAsyncProcedure>(procedure)); } @Override - public void forEachDirectPredicate(Resource subject, MultiProcedure procedure) { - forEachDirectPredicate(subject, new NoneToAsyncMultiProcedure(procedure)); + public void forEachDirectPredicate(Resource subject, Procedure> procedure) { + forEachDirectPredicate(subject, new NoneToAsyncProcedure>(procedure)); } @Override @@ -3777,14 +3533,9 @@ public class ReadGraphImpl implements ReadGraph { // else procedure.execute(graph, (T) obj); - } catch (IOException e) { - procedure.exception(graph, e); - } catch (BufferUnderflowException e) { - procedure.exception(graph, e); } catch (Throwable t) { - procedure.exception(graph, t); + procedure.exception(graph, new ServiceException("Could not forValue for subject " + debugString(resource) + " and binding " + String.valueOf(binding) + " with bytes " + safeArrayToString(result), t)); } - } @Override @@ -3804,6 +3555,24 @@ public class ReadGraphImpl implements ReadGraph { }); } + + private static String safeArrayToString(byte[] a) { + if (a == null) + return "null"; + int iMax = a.length - 1; + if (iMax == -1) + return "[]"; + + StringBuilder b = new StringBuilder(); + b.append('['); + for (int i = 0; i < 100; i++) { // limit to first 100 items + b.append(a[i]); + if (i == iMax) + return b.append(']').toString(); + b.append(", "); + } + return b.append(", ... (" + a.length + ")]").toString(); + } @Override public void forValue(Resource subject, Binding binding, @@ -4498,14 +4267,9 @@ public class ReadGraphImpl implements ReadGraph { else procedure.execute(graph, (T) obj); - } catch (IOException e) { - procedure.exception(graph, e); - } catch (BufferUnderflowException e) { - procedure.exception(graph, e); } catch (Throwable t) { - procedure.exception(graph, t); + procedure.exception(graph, new ServiceException("Could not forValue for subject " + debugString(resource) + " and binding " + String.valueOf(binding) + " with bytes " + safeArrayToString(result), t)); } - } @Override @@ -4655,14 +4419,6 @@ public class ReadGraphImpl implements ReadGraph { listener); } -// @Override -// final public void forPossibleRelatedValue(final Resource subject, final Resource relation, final Binding binding, -// final AsyncProcedure procedure) { -// -// forPossibleRelatedValue(subject, relation, binding, procedure, false); -// -// } - final public void forPossibleRelatedValue(final Resource subject, final Resource relation, final Binding binding, final AsyncProcedure procedure) { @@ -5351,11 +5107,6 @@ public class ReadGraphImpl implements ReadGraph { return processor.getSession(); } -// @Override -// final public Builtins getBuiltins() { -// return processor.getSession().getBuiltins(); -// } - @Override public void asyncRequest(final Read request) { @@ -5400,55 +5151,25 @@ public class ReadGraphImpl implements ReadGraph { assert (request != null); assert (procedure != null); + + processor.schedule(new SessionTask(this) { - final ListenerBase listener = getListenerBase(procedure); - - if (parent != null || listener != null) { - - try { - processor.queryRead(this, request, parent, procedure, - listener); - } catch (Throwable e) { - // This throwable has already been transferred to procedure at this point - do nothing about it - //Logger.defaultLogError("Internal error ", e); - } - - } else { - -// final ReadGraphImpl newGraph = newSync(); - - try { - - T result = request.perform(this); - - try { - procedure.execute(this, result); - } catch (Throwable t) { - Logger.defaultLogError(t); - } - - } catch (Throwable t) { - + @Override + public void run0(int thread) { try { - procedure.exception(this, t); - } catch (Throwable t2) { - Logger.defaultLogError(t2); + final ListenerBase listener = getListenerBase(procedure); + QueryCache.runnerReadEntry(ReadGraphImpl.this, request, parent, listener, procedure, false); + } catch (DatabaseException e) { + Logger.defaultLogError(e); } - - } finally { - } - - } + + }); } public static ReadGraphImpl createAsync(QueryProcessor support) { - return new ReadGraphImpl(null, support); - } - - public static ReadGraphImpl forRecompute(CacheEntry entry, QueryProcessor support) { - return new ReadGraphImpl(entry, support); + return new ReadGraphImpl(null, null, support); } @Override @@ -5511,33 +5232,19 @@ public class ReadGraphImpl implements ReadGraph { assert (request != null); assert (procedure != null); - final ListenerBase listener = getListenerBase(procedure); - - if (parent != null || listener != null) { - - processor.query(this, request, parent, procedure, listener); - - } else { - - try { - - request.perform(this, new CallWrappedSingleQueryProcedure4(procedure, request)); - - } catch (Throwable t) { - - if (t instanceof DatabaseException) - procedure.exception(this, t); - else - procedure - .exception( - this, - new DatabaseException( - "Unexpected exception in ReadGraph.asyncRequest(SingleAsyncRead, SingleProcedure)", - t)); + processor.schedule(new SessionTask(this) { + @Override + public void run0(int thread) { + try { + final ListenerBase listener = getListenerBase(procedure); + QueryCache.runnerAsyncReadEntry(ReadGraphImpl.this, request, parent, listener, procedure, false); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } - - } + + }); } @@ -5558,9 +5265,9 @@ public class ReadGraphImpl implements ReadGraph { 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) { Logger.defaultLogError(t); } @@ -5572,27 +5279,22 @@ public class ReadGraphImpl implements ReadGraph { } - @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(final MultiRead request, - final AsyncMultiProcedure procedure) { + final SyncMultiProcedure procedure) { assert (request != null); assert (procedure != null); @@ -5614,7 +5316,11 @@ public class ReadGraphImpl implements ReadGraph { } catch (Throwable t) { - procedure.exception(this, t); + try { + procedure.exception(this, t); + } catch (DatabaseException e) { + LOGGER.error("Unexpected exception while handling exception", e); + } } @@ -5622,16 +5328,10 @@ public class ReadGraphImpl implements ReadGraph { } - @Override - public void asyncRequest(MultiRead request, - SyncMultiProcedure procedure) { - asyncRequest(request, new SyncToAsyncMultiProcedure(procedure)); - } - @Override public void asyncRequest(MultiRead request, MultiProcedure procedure) { - asyncRequest(request, new NoneToAsyncMultiProcedure(procedure)); + asyncRequest(request, new NoneToSyncMultiProcedure(procedure)); } @Override @@ -5771,51 +5471,41 @@ public class ReadGraphImpl implements ReadGraph { assert (request != null); assert (procedure != null); - ListenerBase listener = getListenerBase(procedure); + final ListenerBase listener = getListenerBase(procedure); if (parent != null || listener != null) { - processor.query(this, request, parent, procedure, listener); - - } else { - try { + QueryCacheBase.resultExternalReadEntry(this, request, parent, listener, procedure); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + // This throwable has already been transferred to procedure at this point - do nothing about it + } - request.register(this, new Listener() { - - @Override - public void execute(T result) { - procedure.execute(result); - } - - @Override - public void exception(Throwable t) { - procedure.exception(t); - } - - @Override - public String toString() { - return "asyncRequest(PrimitiveRead) -> " + request; - } + } else { - @Override - public boolean isDisposed() { - return true; - } + request.register(this, new Listener() { + @Override + public void execute(T result) { + procedure.execute(result); + } - }); + @Override + public void exception(Throwable t) { + procedure.exception(t); + } - } catch (Throwable t) { + @Override + public String toString() { + return "asyncRequest(PrimitiveRead) -> " + request; + } - if (t instanceof DatabaseException) - procedure.exception(t); - else - procedure - .exception(new DatabaseException( - "Unexpected exception in ReadGraph.asyncRequest(SingleAsyncRead, SingleProcedure)", - t)); + @Override + public boolean isDisposed() { + return true; + } - } + }); } @@ -5925,64 +5615,38 @@ public class ReadGraphImpl implements ReadGraph { * Internal routines */ - protected static String INTERNAL_ERROR_STRING = "Transaction aborted due to internal client error. Contact application support."; + protected static String INTERNAL_ERROR_STRING = "Transaction aborted due to internal client error."; /* * callerThread is the currently running thread state.syncThread is blocking for * this execution state.syncParent is the blocking request */ - final private boolean isExternal(int thread) { - return thread == Integer.MIN_VALUE; + ReadGraphImpl(ReadGraphImpl parentGraph, CacheEntry parent, QueryProcessor support) { + this.parentGraph = parentGraph; + this.parent = parent; + this.processor = support; + this.asyncBarrier = new AsyncBarrierImpl(parentGraph != null ? parentGraph.asyncBarrier : null, parent); } -// final private boolean isSync(int thread) { -// return thread < -1 && thread > Integer.MIN_VALUE; -// } + ReadGraphImpl(ReadGraphImpl graph, CacheEntry parent) { + this(graph, parent, graph.processor); + } ReadGraphImpl(ReadGraphImpl graph) { - this(graph.parent, graph.processor); + this(graph, graph.parent); } - ReadGraphImpl(CacheEntry parent, QueryProcessor support) { -// this.state = new ReadGraphState(barrier, support); - this.parent = parent; - this.processor = support; - } - - ReadGraphImpl(final QueryProcessor support) { - -// this.state = state; - this.processor = support; - this.parent = null; - + public ReadGraphImpl withParent(CacheEntry parent) { + return new ReadGraphImpl(this, parent); } -// public static ReadGraphImpl createSync(int syncThread, Object syncParent, -// ReadGraphSupportImpl support) { -// return new ReadGraphImpl(syncThread, syncThread, syncParent, null, -// support, new AsyncBarrierImpl(null)); -// } - - public static ReadGraphImpl create(QueryProcessor support) { - return new ReadGraphImpl(support); + public ReadGraphImpl forRecompute(CacheEntry parent) { + return new ReadGraphImpl(null, parent, processor); } -// public ReadGraphImpl newAsync() { -// return this; -//// if(!state.synchronizedExecution) { -//// return this; -//// } else { -//// return new ReadGraphImpl(false, parent, state.support, state.barrier); -//// } -// } - -// public ReadGraphImpl newSync() { -// return new ReadGraphImpl(parent, processor); -// } - - public ReadGraphImpl newSync(CacheEntry parentEntry) { - return new ReadGraphImpl(parentEntry, processor); + public static ReadGraphImpl create(QueryProcessor support) { + return new ReadGraphImpl(null, null, support); } public ReadGraphImpl newRestart(ReadGraphImpl impl) { @@ -5990,46 +5654,10 @@ public class ReadGraphImpl implements ReadGraph { WriteGraphImpl write = processor.getSession().getService( WriteGraphImpl.class); -// if (write.callerThread != impl.callerThread) -// return new WriteGraphImpl(impl.callerThread, parent, state.support, write.writeSupport, write.provider, write.state.barrier); return write; } -// public ReadGraphImpl newSync(Object parentRequest) { -// return new ReadGraphImpl(callerThread, state.parent, state.support, new AsyncBarrierImpl(state.barrier)); -// } - -// public ReadGraphImpl newSync(final int callerThread, Object parentRequest) { -// assert (state.syncThread == callerThread || (state.syncThread == Integer.MIN_VALUE && callerThread != Integer.MIN_VALUE)); -// return new ReadGraphImpl(callerThread, callerThread, parentRequest, -// state.parent, state.support, new AsyncBarrierImpl(state.barrier)); -// } -// -// public ReadGraphImpl newSyncAsync(Object parentRequest) { -//// assert (callerThread < 0); -// return new ReadGraphImpl(callerThread, state.syncThread, parentRequest, -// state.parent, state.support, new AsyncBarrierImpl(state.barrier)); -// } -// -// public ReadGraphImpl newSyncAsync(final int callerThread, -// Object parentRequest) { -//// assert (callerThread < 0); -// // assert(state.syncThread == callerThread || (state.syncThread == Integer.MIN_VALUE -// // && callerThread != Integer.MIN_VALUE) ); -// return new ReadGraphImpl(callerThread, callerThread, parentRequest, -// state.parent, state.support, new AsyncBarrierImpl(state.barrier)); -// } - - public ReadGraphImpl withAsyncParent(CacheEntry parent) { - return new ReadGraphImpl(parent, processor); - } - - public ReadGraphImpl withParent(CacheEntry parent) { - if(parent == this.parent) return this; - else return new ReadGraphImpl(parent, processor); - } - final private ListenerBase getListenerBase(final Object procedure) { if (procedure instanceof ListenerBase) return (ListenerBase) procedure; @@ -6041,56 +5669,13 @@ public class ReadGraphImpl implements ReadGraph { assert(procedure.done()); -// while (!procedure.done()) { -// -// boolean executed = processor.resumeTasks(callerThread, null, null); -// if (!executed) { -// try { -// Thread.sleep(1); -// // sema.tryAcquire(1, TimeUnit.MILLISECONDS); -// } catch (InterruptedException e) { -// e.printStackTrace(); -// } -// } -// -// } - } public void waitAsyncProcedure(AsyncReadProcedure procedure) { assert(procedure.done()); -// while (!procedure.done()) { -// -// boolean executed = processor.processor.resume(this); -// if (!executed) { -// try { -// Thread.sleep(1); -// // sema.tryAcquire(1, TimeUnit.MILLISECONDS); -// } catch (InterruptedException e) { -// e.printStackTrace(); -// } -// } -// -// } - } - -// public void waitAsync(Object request) { -// try { -// state.barrier.waitBarrier(request, this); -// } catch (Throwable t) { -// t.printStackTrace(); -// processor.scanPending(); -// processor.querySupport.checkTasks(); -// throw new RuntimeDatabaseException(t); -// } -// } - -// public void restart() { -// state.barrier.restart(); -// } public boolean resumeTasks() { return processor.resumeTasks(this); @@ -6329,6 +5914,8 @@ public class ReadGraphImpl implements ReadGraph { } Serializer serializer = binding.serializer(); byte[] bytes = serializer.serialize(initialValue); + // In case the file has been previously accessed and was larger we set the correct size now + rd.binaryFile.setLength(bytes.length); rd.binaryFile.write(bytes); ravs.put(resource, rd); return rd; @@ -6518,12 +6105,18 @@ public class ReadGraphImpl implements ReadGraph { return getValue(r, binding); } } else if(types.contains(L0.ExternalValue)) { - try { - return (T)ReflectionUtils.getValue(getURI(r)).getValue(); - } catch(ValueNotFoundException e) { - throw new DatabaseException(e); - } catch(ClassCastException e) { - throw new DatabaseException(e); + ComputationalValue cv = syncRequest(new PossibleAdapter(r, ComputationalValue.class), TransientCacheAsyncListener.instance()); + if(cv != null) { + return cv.getValue(this, r); + } else { + // This should not even be possible since L0 defines an adapter for all values + try { + return (T)ReflectionUtils.getValue(getURI(r)).getValue(); + } catch(ValueNotFoundException e) { + throw new DatabaseException(e); + } catch(ClassCastException e) { + throw new DatabaseException(e); + } } } else { @@ -6665,17 +6258,23 @@ public class ReadGraphImpl implements ReadGraph { throw new DatabaseException(e); } } - + + private static ThreadLocal syncGraph = new ThreadLocal() { + protected Boolean initialValue() { + return true; + } + }; + @Override public boolean setSynchronous(boolean value) { - boolean old = processor.synch; - processor.synch = value; + boolean old = getSynchronous(); + syncGraph.set(value); return old; } - + @Override public boolean getSynchronous() { - return processor.synch; + return syncGraph.get(); } public void ensureLoaded(int resource) { @@ -6706,5 +6305,20 @@ public class ReadGraphImpl implements ReadGraph { public Object getModificationCounter() { return processor.getSession().getModificationCounter(); } - + + @Override + public boolean performPending() { + return processor.performPending(this); + } + + public Set ancestorSet() { + HashSet result = new HashSet<>(); + ReadGraphImpl g = this; + while(g != null) { + result.add(g); + g = g.parentGraph; + } + return result; + } + }