/*******************************************************************************
- * 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
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;
import org.simantics.db.common.primitiverequest.VariantValueImplied;
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.adapter.TransientCacheAsyncListener;
import org.simantics.db.common.procedure.adapter.TransientCacheListener;
import org.simantics.db.common.procedure.single.SyncReadProcedure;
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;
import org.simantics.db.exception.ResourceNotFoundException;
import org.simantics.db.exception.ServiceException;
import org.simantics.db.exception.ValidationException;
+import org.simantics.db.impl.BlockingAsyncProcedure;
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;
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;
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;
final public QueryProcessor processor;
+ public AsyncBarrierImpl asyncBarrier = null;
+
final static Binding DATA_TYPE_BINDING_INTERNAL = Bindings.getBindingUnchecked(Datatype.class);
final static Serializer DATA_TYPE_SERIALIZER = Bindings.getSerializerUnchecked(DATA_TYPE_BINDING_INTERNAL);
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) {
try {
- return syncRequest(new org.simantics.db.common.primitiverequest.Resource(
- id));
+ return getResource(id);
} catch (ResourceNotFoundException e) {
}
}
+
+ @Override
+ public Map<String, Resource> 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();
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);
+ return QueryCache.resultReadEntry(this, request, parent, null, null);
- } 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);
-
- }
-
- }
-
}
@Override
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);
}
throw (DatabaseException) exception;
else
throw new DatabaseException(
- "Unexpected exception in ReadGraph.syncRequest(AsyncMultiRead)",
+ "Unexpected exception in ReadGraph.syncRequest(AsyncRead)",
exception);
}
}
throws DatabaseException {
assert (request != null);
- AsyncReadProcedure<T> procedure = new AsyncReadProcedure<T>();
- syncRequest(request, procedure);
- procedure.checkAndThrow();
- return procedure.result;
-
-// return syncRequest(request, new AsyncProcedureAdapter<T>());
+ asyncBarrier = new AsyncBarrierImpl(null);
+ BlockingAsyncProcedure<T> ap = new BlockingAsyncProcedure<>(this, null, request);
+ syncRequest(request, ap);
+ return ap.get();
}
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<T> wrapper = new ResultCallWrappedSingleQueryProcedure4<T>(
- 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<T> wrapper = new ResultCallWrappedSingleQueryProcedure4<T>(
- procedure, request);
-
- try {
-
- processor.tryQuery(this, request, wrapper);
-
- } catch (Throwable t) {
-
- wrapper.exception(this, t);
-
- }
+ BlockingAsyncProcedure<T> ap = new BlockingAsyncProcedure<>(this, procedure, request);
- 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();
+ QueryCache.runnerAsyncReadEntry(this, request, parent, listener, ap, true);
- }
+ return ap.get();
}
assert (request != null);
- // System.out.println("syncRequest " + request + " syncParent=" +
- // syncParent);
-
ListenerBase listener = getListenerBase(procedure);
+ assert(listener == null);
- if (parent != null || listener != null || ((request.getFlags() & RequestFlags.SCHEDULE) > 0)) {
+ BlockingAsyncProcedure<T> ap = new BlockingAsyncProcedure<>(this, procedure, request);
-// final ReadGraphImpl newGraph = newSync();
-
- final ResultCallWrappedSingleQueryProcedure4<T> wrapper = new ResultCallWrappedSingleQueryProcedure4<T>(
- procedure, request);
+ QueryCache.runnerAsyncReadEntry(this, request, parent, listener, ap, true);
- 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.<Boolean>getProperty(DevelopmentKeys.WRITEGRAPH_EXCEPTION_STACKTRACES, Bindings.BOOLEAN)) {
- t.printStackTrace();
- }
- }
- procedure.exception(this, t);
- waitAsyncProcedure(procedure);
- }
-
- }
+ ap.get();
}
final ArrayList<T> result = new ArrayList<T>();
final DataContainer<Throwable> exception = new DataContainer<Throwable>();
- syncRequest(request, new AsyncMultiProcedure<T>() {
+ syncRequest(request, new SyncMultiProcedure<T>() {
@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);
}
}
- @Override
- public <T> Collection<T> syncRequest(MultiRead<T> request,
- AsyncMultiListener<T> procedure) {
- return syncRequest(request, (AsyncMultiProcedure<T>) procedure);
- }
-
@Override
public <T> Collection<T> syncRequest(MultiRead<T> request,
SyncMultiListener<T> procedure) {
- return syncRequest(request, new SyncToAsyncMultiListener<T>(procedure));
+ return syncRequest(request, (SyncMultiProcedure<T>)procedure);
}
@Override
public <T> Collection<T> syncRequest(MultiRead<T> request,
MultiListener<T> procedure) {
- return syncRequest(request, new NoneToAsyncMultiListener<T>(procedure));
+ return syncRequest(request, new NoneToSyncMultiListener<T>(procedure));
}
@Override
public <T> Collection<T> syncRequest(MultiRead<T> request,
- AsyncMultiProcedure<T> procedure) {
+ SyncMultiProcedure<T> procedure) {
assert (request != null);
ListenerBase listener = getListenerBase(procedure);
+ final ResultCallWrappedSyncQueryProcedure<T> wrapper = new ResultCallWrappedSyncQueryProcedure<T>(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);
// final ReadGraphImpl newGraph = newSync();
- final ResultCallWrappedQueryProcedure4<T> wrapper = new ResultCallWrappedQueryProcedure4<T>(procedure);
-
try {
-
request.perform(this, wrapper);
-
} catch (Throwable t) {
-
wrapper.exception(this, t);
-// newGraph.waitAsync(syncParent);
-
}
}
- // TODO
- return null;
-
- }
+ return wrapper.get();
- @Override
- public <T> Collection<T> syncRequest(MultiRead<T> request,
- SyncMultiProcedure<T> procedure) {
- return syncRequest(request, new SyncToAsyncMultiProcedure<T>(procedure));
}
@Override
public <T> Collection<T> syncRequest(MultiRead<T> request,
MultiProcedure<T> procedure) {
- return syncRequest(request, new NoneToAsyncMultiProcedure<T>(procedure));
+ return syncRequest(request, new NoneToSyncMultiProcedure<T>(procedure));
}
static class AsyncMultiReadProcedure<T> extends ArrayList<T> implements AsyncMultiProcedure<T> {
final public <T> T syncRequest(final ExternalRead<T> request,
final Procedure<T> procedure) throws DatabaseException {
- assert (request != null);
-
- ListenerBase listener = getListenerBase(procedure);
-
- final DataContainer<Throwable> exception = new DataContainer<Throwable>();
- final DataContainer<T> result = new DataContainer<T>();
-
- if (parent != null || listener != null) {
-
-// final ReadGraphImpl newGraph = newSync();
-
- processor.query(this, request, parent, new Procedure<T>() {
-
- @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);
-
-// 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);
- }
+ assert (request != null);
- return result.get();
+ ListenerBase listener = procedure != null ? getListenerBase(procedure) : null;
+ return QueryCache.resultExternalReadEntry(this, request, parent, listener, procedure);
}
}
-// @Override
-// final public void forEachDirectObject(final Resource subject,
-// final Resource relation,
-// final AsyncMultiProcedure<Resource> procedure) {
-//
-// processor.forEachDirectObject(this, subject, relation, procedure);
-//
-// }
-//
-// @Override
-// public void forEachDirectObject(Resource subject, Resource relation,
-// SyncMultiProcedure<Resource> procedure) {
-// forEachDirectObject(subject, relation,
-// new SyncToAsyncMultiProcedure<Resource>(procedure));
-// }
-//
-// @Override
-// public void forEachDirectObject(Resource subject, Resource relation,
-// MultiProcedure<Resource> procedure) {
-// forEachDirectObject(subject, relation,
-// new NoneToAsyncMultiProcedure<Resource>(procedure));
-// }
-
@Override
- final public void forEachDirectPredicate(final Resource subject, final AsyncMultiProcedure<Resource> procedure) {
+ final public void forEachDirectPredicate(final Resource subject, final AsyncProcedure<Set<Resource>> procedure) {
processor.forEachDirectPredicate(this, subject, procedure);
}
@Override
- public void forEachDirectPredicate(Resource subject, SyncMultiProcedure<Resource> procedure) {
- forEachDirectPredicate(subject, new SyncToAsyncMultiProcedure<Resource>(procedure));
+ final public void forEachDirectPredicate(final Resource subject, final SyncProcedure<Set<Resource>> procedure) {
+ forEachDirectPredicate(subject, new SyncToAsyncProcedure<Set<Resource>>(procedure));
}
@Override
- public void forEachDirectPredicate(Resource subject, MultiProcedure<Resource> procedure) {
- forEachDirectPredicate(subject, new NoneToAsyncMultiProcedure<Resource>(procedure));
+ public void forEachDirectPredicate(Resource subject, Procedure<Set<Resource>> procedure) {
+ forEachDirectPredicate(subject, new NoneToAsyncProcedure<Set<Resource>>(procedure));
}
@Override
listener);
}
-// @Override
-// final public <T> void forPossibleRelatedValue(final Resource subject, final Resource relation, final Binding binding,
-// final AsyncProcedure<T> procedure) {
-//
-// forPossibleRelatedValue(subject, relation, binding, procedure, false);
-//
-// }
-
final public <T> void forPossibleRelatedValue(final Resource subject, final Resource relation, final Binding binding,
final AsyncProcedure<T> procedure) {
return processor.getSession();
}
-// @Override
-// final public Builtins getBuiltins() {
-// return processor.getSession().getBuiltins();
-// }
-
@Override
public <T> void asyncRequest(final Read<T> request) {
assert (request != null);
assert (procedure != null);
+
+ processor.schedule(new SessionTask(false) {
- 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 run(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 {
-
}
-
- }
+
+ });
}
return new ReadGraphImpl(null, support);
}
- public static ReadGraphImpl forRecompute(CacheEntry entry, QueryProcessor support) {
- return new ReadGraphImpl(entry, support);
- }
-
@Override
public <T> void asyncRequest(Read<T> request, SyncProcedure<T> procedure) {
asyncRequest(request, new SyncToAsyncProcedure<T>(procedure));
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<T>(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(false) {
+ @Override
+ public void run(int thread) {
+ try {
+ final ListenerBase listener = getListenerBase(procedure);
+ QueryCache.runnerAsyncReadEntry(ReadGraphImpl.this, request, parent, listener, procedure, false);
+ } catch (DatabaseException e) {
+ Logger.defaultLogError(e);
+ }
}
-
- }
+
+ });
}
assert (request != null);
- asyncRequest(request, new AsyncMultiProcedureAdapter<T>() {
+ asyncRequest(request, new SyncMultiProcedureAdapter<T>() {
@Override
- public void exception(AsyncReadGraph graph, Throwable t) {
+ public void exception(ReadGraph graph, Throwable t) {
Logger.defaultLogError(t);
}
}
- @Override
- public <T> void asyncRequest(MultiRead<T> request,
- AsyncMultiListener<T> procedure) {
- asyncRequest(request, (AsyncMultiProcedure<T>) procedure);
- }
-
@Override
public <T> void asyncRequest(MultiRead<T> request,
SyncMultiListener<T> procedure) {
- asyncRequest(request, new SyncToAsyncMultiListener<T>(procedure));
+ asyncRequest(request, (SyncMultiProcedure<T>)procedure);
}
@Override
public <T> void asyncRequest(MultiRead<T> request,
MultiListener<T> procedure) {
- asyncRequest(request, new NoneToAsyncMultiListener<T>(procedure));
+ asyncRequest(request, new NoneToSyncMultiListener<T>(procedure));
}
+
@Override
public <T> void asyncRequest(final MultiRead<T> request,
- final AsyncMultiProcedure<T> procedure) {
+ final SyncMultiProcedure<T> procedure) {
assert (request != null);
assert (procedure != null);
} catch (Throwable t) {
- procedure.exception(this, t);
+ try {
+ procedure.exception(this, t);
+ } catch (DatabaseException e) {
+ LOGGER.error("Unexpected exception while handling exception", e);
+ }
}
}
- @Override
- public <T> void asyncRequest(MultiRead<T> request,
- SyncMultiProcedure<T> procedure) {
- asyncRequest(request, new SyncToAsyncMultiProcedure<T>(procedure));
- }
-
@Override
public <T> void asyncRequest(MultiRead<T> request,
MultiProcedure<T> procedure) {
- asyncRequest(request, new NoneToAsyncMultiProcedure<T>(procedure));
+ asyncRequest(request, new NoneToSyncMultiProcedure<T>(procedure));
}
@Override
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<T>() {
-
- @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<T>() {
+ @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;
+ }
- }
+ });
}
return thread == Integer.MIN_VALUE;
}
-// final private boolean isSync(int thread) {
-// return thread < -1 && thread > Integer.MIN_VALUE;
-// }
-
ReadGraphImpl(ReadGraphImpl graph) {
this(graph.parent, graph.processor);
}
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 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);
+ return new ReadGraphImpl(null, support);
}
-// 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 newAsync(ReadGraphImpl parent) {
+ ReadGraphImpl result = new ReadGraphImpl(parent);
+ result.asyncBarrier = new AsyncBarrierImpl(parent.asyncBarrier);
+ return result;
}
public ReadGraphImpl newRestart(ReadGraphImpl impl) {
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);
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 <T> void waitAsyncProcedure(AsyncReadProcedure<T> 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);
throw new DatabaseException(e);
}
}
-
+
+ private static ThreadLocal<Boolean> syncGraph = new ThreadLocal<Boolean>() {
+ 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) {
public Object getModificationCounter() {
return processor.getSession().getModificationCounter();
}
-
+
+ @Override
+ public boolean performPending() {
+ return processor.performPending(processor.thread.get());
+ }
+
}
Code Review / simantics / platform.git / blobdiff