import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.Semaphore;
-import java.util.concurrent.atomic.AtomicBoolean;
+import org.simantics.databoard.Bindings;
import org.simantics.db.AsyncReadGraph;
+import org.simantics.db.DevelopmentKeys;
import org.simantics.db.ObjectResourceIdMap;
import org.simantics.db.ReadGraph;
import org.simantics.db.RelationInfo;
import org.simantics.db.common.utils.Logger;
import org.simantics.db.exception.DatabaseException;
-import org.simantics.db.impl.DebugPolicy;
import org.simantics.db.impl.graph.ReadGraphImpl;
import org.simantics.db.impl.procedure.InternalProcedure;
-import org.simantics.db.impl.query.QueryProcessor.SessionTask;
import org.simantics.db.procedure.AsyncMultiProcedure;
import org.simantics.db.procedure.AsyncProcedure;
-import org.simantics.db.procedure.Listener;
import org.simantics.db.procedure.ListenerBase;
import org.simantics.db.procedure.Procedure;
import org.simantics.db.procedure.SyncMultiProcedure;
import org.simantics.db.request.ExternalRead;
import org.simantics.db.request.MultiRead;
import org.simantics.db.request.Read;
+import org.simantics.utils.Development;
import gnu.trove.map.hash.THashMap;
import gnu.trove.map.hash.TObjectIntHashMap;
protected final StableHashMap<AsyncMultiRead, AsyncMultiReadEntry> asyncMultiReadEntryMap;
protected final StableHashMap<ExternalRead, ExternalReadEntry> externalReadEntryMap;
- final THashMap<CacheEntry, ArrayList<ListenerEntry>> listeners;
-
public final QuerySupport querySupport;
public QueryCacheBase(QuerySupport querySupport, int threads) {
asyncMultiReadEntryMap = new StableHashMap<AsyncMultiRead, AsyncMultiReadEntry>();
multiReadEntryMap = new StableHashMap<MultiRead, MultiReadEntry>();
externalReadEntryMap = new StableHashMap<ExternalRead, ExternalReadEntry>();
- listeners = new THashMap<CacheEntry, ArrayList<ListenerEntry>>(10, 0.75f);
- }
-
-// public <T> Object performQuery(ReadGraphImpl parentGraph, final AsyncRead<T> query, final CacheEntryBase entry_, AsyncProcedure procedure_) throws DatabaseException {
-//
-// AsyncReadEntry<T> entry = (AsyncReadEntry<T>)entry_;
-// AsyncProcedure<T> procedure = (AsyncProcedure<T>)procedure_;
-//
-// ReadGraphImpl queryGraph = parentGraph.withParent(entry_);
-//
-// try {
-//
-// query.perform(queryGraph, new AsyncProcedure<T>() {
-//
-// @Override
-// public void execute(AsyncReadGraph returnGraph, T result) {
-// ReadGraphImpl impl = (ReadGraphImpl)returnGraph;
-// entry.addOrSet(parentGraph, result);
-// try {
-// procedure.execute(parentGraph, result);
-// } catch (Throwable t) {
-// t.printStackTrace();
-// }
-//// parentBarrier.dec(query);
-// }
-//
-// @Override
-// public void exception(AsyncReadGraph returnGraph, Throwable t) {
-// ReadGraphImpl impl = (ReadGraphImpl)returnGraph;
-//// AsyncReadGraph resumeGraph = finalParentGraph.newAsync();
-// entry.except(parentGraph, t);
-// try {
-// procedure.exception(parentGraph, t);
-// } catch (Throwable t2) {
-// t2.printStackTrace();
-// }
-//// parentBarrier.dec(query);
-// }
-//
-// @Override
-// public String toString() {
-// return procedure.toString();
-// }
-//
-// });
-//
-// } catch (Throwable t) {
-//
-// entry.except(t);
-// try {
-// procedure.exception(parentGraph, t);
-// } catch (Throwable t2) {
-// t2.printStackTrace();
-// }
-//// parentBarrier.dec(query);
-//
-// }
-//
-// return null;
-//
-// }
-
-// public <T> Object performQuery(ReadGraphImpl parentGraph, final Read<T> query, final CacheEntryBase entry_, AsyncProcedure procedure_) throws DatabaseException {
-//
-// ReadGraphImpl queryGraph = parentGraph.withParent(entry_);
-//
-// ReadEntry entry = (ReadEntry)entry_;
-//
-// try {
-//
-// T result = (T)query.perform(queryGraph);
-// entry.addOrSet(queryGraph, result);
-//
-// return (T)entry.get(parentGraph, procedure_);
-//
-// } catch (Throwable t) {
-//
-// entry.except(t);
-// return (T)entry.get(parentGraph, procedure_);
-//
-// }
-//
-// }
-
- public <T> Object performQuery(ReadGraphImpl parentGraph, final ExternalRead<T> query, final CacheEntryBase entry_, AsyncProcedure procedure_) throws DatabaseException {
-
- ExternalReadEntry entry = (ExternalReadEntry)entry_;
- AsyncProcedure<T> procedure = (AsyncProcedure<T>)procedure_;
-
- try {
-
- query.register(parentGraph, new Listener<T>() {
-
- AtomicBoolean used = new AtomicBoolean(false);
-
- @Override
- public void execute(T result) {
-
- // Just for safety
- if(entry.isDiscarded()) return;
-
- if(used.compareAndSet(false, true)) {
- //entry.setPending();
- entry.addOrSet(parentGraph.processor, result);
- procedure.execute(parentGraph, result);
- } else {
- entry.queue(result);
- parentGraph.processor.updatePrimitive(query);
- }
-
- }
-
- @Override
- public void exception(Throwable t) {
-
- entry.except(t);
-
- if(used.compareAndSet(false, true)) {
- procedure.exception(parentGraph, t);
- } else {
-// entry.queue(result);
- parentGraph.processor.updatePrimitive(query);
- }
-
- }
-
- @Override
- public String toString() {
- return procedure.toString();
- }
-
- @Override
- public boolean isDisposed() {
- return entry.isDiscarded() || !parentGraph.processor.isBound(entry);
- }
-
- });
-
- return entry.getResult();
-
- } catch (Throwable t) {
-
- entry.except(t);
- procedure.exception(parentGraph, t);
- return entry.getResult();
-
- }
-
}
public <T> Object performQuery(ReadGraphImpl parentGraph, final AsyncMultiRead<T> query, final CacheEntryBase entry_, Object procedure_) throws DatabaseException {
- ReadGraphImpl queryGraph = parentGraph.withParent(entry_);
+ ReadGraphImpl queryGraph = parentGraph.withParent(entry_, null, false);
AsyncMultiReadEntry entry = (AsyncMultiReadEntry)entry_;
AsyncMultiProcedure<T> procedure = (AsyncMultiProcedure<T>)procedure_;
public <T> Object performQuery(ReadGraphImpl parentGraph, final MultiRead<T> query, final CacheEntryBase entry_, Object procedure_) throws DatabaseException {
- ReadGraphImpl queryGraph = parentGraph.withParent(entry_);
+ ReadGraphImpl queryGraph = parentGraph.withParent(entry_, null, true);
MultiReadEntry entry = (MultiReadEntry)entry_;
SyncMultiProcedure<T> procedure = (SyncMultiProcedure<T>)procedure_;
}
- public ListenerEntry registerDependencies(ReadGraphImpl graph, CacheEntry child, CacheEntry parent, ListenerBase listener, Object procedure, boolean inferred) {
-
- if (parent != null && !inferred) {
- try {
- if(!child.isImmutable(graph)) {
- synchronized(child) {
- child.addParent(parent);
- }
- }
- } catch (DatabaseException e) {
- Logger.defaultLogError(e);
- }
- if(DebugPolicy.DEPENDENCIES) System.out.println(child + " -> " + parent);
- }
-
- if (listener != null) {
- return registerListener(child, listener, procedure);
- } else {
- return null;
- }
-
- }
-
- public synchronized ListenerEntry registerListener(final CacheEntry entry, final ListenerBase base, final Object procedure) {
-
- assert (entry != null);
-
- if (base.isDisposed())
- return null;
-
- return addListener(entry, base, procedure);
-
- }
-
- protected void primeListenerEntry(final ListenerEntry entry, final Object result) {
- entry.setLastKnown(result);
- }
-
- private ListenerEntry addListener(CacheEntry entry, ListenerBase base, Object procedure) {
-
- assert (entry != null);
- assert (procedure != null);
-
- ArrayList<ListenerEntry> list = listeners.get(entry);
- if (list == null) {
- list = new ArrayList<ListenerEntry>(1);
- listeners.put(entry, list);
- }
-
- ListenerEntry result = new ListenerEntry(entry, base, procedure);
- int currentIndex = list.indexOf(result);
- // There was already a listener
- if(currentIndex > -1) {
- ListenerEntry current = list.get(currentIndex);
- if(!current.base.isDisposed()) return null;
- list.set(currentIndex, result);
- } else {
- list.add(result);
- }
-
- if(DebugPolicy.LISTENER) {
- new Exception().printStackTrace();
- System.out.println("addListener -> " + list.size() + " " + entry + " " + base + " " + procedure);
- }
-
- return result;
-
- }
-
-
public Collection<CacheEntry> getRootList() {
ArrayList<CacheEntry> result = new ArrayList<CacheEntry>();
int counter = 0;
while(entry.isPending()) {
try {
- boolean performed = graph.performPending();
+ boolean performed = false;//graph.performPending();
if(!performed) {
Thread.sleep(1);
counter++;
return wrap.get();
}
- public static <T> T resultReadEntry(ReadGraphImpl graph, Read r, CacheEntry parent, ListenerBase listener, AsyncProcedure<T> procedure) throws DatabaseException {
- return (T)QueryCache.runnerReadEntry(graph, r, parent, listener, procedure, true);
- }
-
public static <T> T resultAsyncReadEntry(ReadGraphImpl graph, AsyncRead r, CacheEntry parent, ListenerBase listener, AsyncProcedure<T> procedure) throws DatabaseException {
return (T)QueryCache.runnerAsyncReadEntry(graph, r, parent, listener, procedure, true);
}
Code Review / simantics / platform.git / blobdiff