X-Git-Url: https://gerrit.simantics.org/r/gitweb?a=blobdiff_plain;f=bundles%2Forg.simantics.db.impl%2Fsrc%2Forg%2Fsimantics%2Fdb%2Fimpl%2Fquery%2FQueryProcessor.java;h=2908bd43a83680fc7f24e3fe2faafb1b25bc1b53;hb=68ce0966a57f5153b133c6283fdbae10f683b745;hp=cca9d681edd8f88893a65d819afee6b290cf5d0a;hpb=e3290cc52342e783a2617f3009fff8e6f3e54560;p=simantics%2Fplatform.git diff --git a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryProcessor.java b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryProcessor.java index cca9d681e..2908bd43a 100644 --- a/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryProcessor.java +++ b/bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryProcessor.java @@ -16,6 +16,7 @@ import java.io.File; import java.io.FileOutputStream; import java.io.IOException; import java.io.PrintStream; +import java.lang.ref.WeakReference; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -65,7 +66,6 @@ import org.simantics.db.procedure.AsyncMultiListener; import org.simantics.db.procedure.AsyncMultiProcedure; import org.simantics.db.procedure.AsyncProcedure; import org.simantics.db.procedure.AsyncSetListener; -import org.simantics.db.procedure.Listener; import org.simantics.db.procedure.ListenerBase; import org.simantics.db.procedure.MultiProcedure; import org.simantics.db.procedure.Procedure; @@ -84,7 +84,6 @@ import org.simantics.utils.datastructures.Pair; import org.simantics.utils.datastructures.collections.CollectionUtils; import org.simantics.utils.datastructures.disposable.AbstractDisposable; -import gnu.trove.map.hash.THashMap; import gnu.trove.procedure.TIntProcedure; import gnu.trove.procedure.TLongProcedure; import gnu.trove.procedure.TObjectProcedure; @@ -94,23 +93,13 @@ import gnu.trove.set.hash.TIntHashSet; @SuppressWarnings({"rawtypes", "unchecked"}) final public class QueryProcessor extends AbstractDisposable implements ReadGraphSupport { - final public QueryCache cache = new QueryCache(); - public static int indent = 0; - public int size = 0; - // Garbage collection public int boundQueries = 0; - // Statistics - private int hits = 0; - - private int misses = 0; - - private int updates = 0; final private int functionalRelation; @@ -150,10 +139,10 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap private boolean updating = false; - static public boolean collecting = false; private boolean firingListeners = false; + final public QueryCache cache; final public QuerySupport querySupport; final public Session session; final public ResourceSupport resourceSupport; @@ -162,7 +151,9 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap QueryThread[] executors; - public ArrayList[] queues; +// public ArrayList[] queues; + + public LinkedList freeScheduling = new LinkedList(); enum ThreadState { @@ -174,14 +165,12 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap public ReentrantLock[] threadLocks; public Condition[] threadConditions; - public ArrayList[] ownTasks; + //public ArrayList[] ownTasks; - public ArrayList[] ownSyncTasks; + //public ArrayList[] ownSyncTasks; - ArrayList[] delayQueues; + //ArrayList[] delayQueues; - public boolean synch = true; - final Object querySupportLock; public Long modificationCounter = 0L; @@ -189,19 +178,21 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap public void close() { } - final public void scheduleOwn(int caller, SessionTask request) { - ownTasks[caller].add(request); - } +// final public void scheduleOwn(int caller, SessionTask request) { +// ownTasks[caller].add(request); +// } final public void scheduleAlways(int caller, SessionTask request) { - int performer = request.thread; - if(caller == performer) { - ownTasks[caller].add(request); - } else { - schedule(caller, request); - } +// int performer = request.thread; +// if(caller == performer) { +// ownTasks[caller].add(request); +// } else { +// schedule(caller, request); +// } + schedule(caller, request); + } final public void schedule(int caller, SessionTask request) { @@ -215,26 +206,50 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap assert(request != null); - if(caller == performer) { - request.run(caller); - } else { - ReentrantLock queueLock = threadLocks[performer]; - queueLock.lock(); - queues[performer].add(request); - // This thread could have been sleeping - if(queues[performer].size() == 1) { - if(ThreadState.SLEEP == threadStates[performer]) sleepers.decrementAndGet(); - threadConditions[performer].signalAll(); - } - queueLock.unlock(); - } +// if(caller == performer) { +// request.run(caller); +// } else { + +// if(performer == THREADS) { + + synchronized(querySupportLock) { + + freeScheduling.add(request); + + //System.err.println("schedule free task " + request + " => " + freeScheduling.size()); + + for(int i=0;i garbageTracker; + + private class GarbageTracker { + + @Override + protected void finalize() throws Throwable { + +// System.err.println("GarbageTracker"); +// +// garbageTracker = new WeakReference(new GarbageTracker()); + + super.finalize(); + + } + + } public QueryProcessor(final int threads, QuerySupport core, Set threadSet) throws DatabaseException { + //garbageTracker = new WeakReference(new GarbageTracker()); + THREADS = threads; THREAD_MASK = threads - 1; querySupport = core; + cache = new QueryCache(core, threads); session = querySupport.getSession(); resourceSupport = querySupport.getSupport(); querySupportLock = core.getLock(); executors = new QueryThread[THREADS]; - queues = new ArrayList[THREADS]; +// queues = new ArrayList[THREADS]; threadLocks = new ReentrantLock[THREADS]; threadConditions = new Condition[THREADS]; threadStates = new ThreadState[THREADS]; - ownTasks = new ArrayList[THREADS]; - ownSyncTasks = new ArrayList[THREADS]; - delayQueues = new ArrayList[THREADS * THREADS]; +// ownTasks = new ArrayList[THREADS]; +// ownSyncTasks = new ArrayList[THREADS]; +// delayQueues = new ArrayList[THREADS * THREADS]; // freeSchedule = new AtomicInteger(0); - for (int i = 0; i < THREADS * THREADS; i++) { - delayQueues[i] = new ArrayList(); - } +// for (int i = 0; i < THREADS * THREADS; i++) { +// delayQueues[i] = new ArrayList(); +// } for (int i = 0; i < THREADS; i++) { // tasks[i] = new ArrayList(); - ownTasks[i] = new ArrayList(); - ownSyncTasks[i] = new ArrayList(); - queues[i] = new ArrayList(); +// ownTasks[i] = new ArrayList(); +// ownSyncTasks[i] = new ArrayList(); +// queues[i] = new ArrayList(); threadLocks[i] = new ReentrantLock(); threadConditions[i] = threadLocks[i].newCondition(); // limits[i] = false; @@ -522,48 +557,63 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap public void forResource(ReadGraphImpl graph, final String id, CacheEntry parent, final InternalProcedure procedure) { - URIToResource.queryEach(graph, id, parent, null, new InternalProcedure() { + try { + + QueryCache.runnerURIToResource(graph, id, parent, null, new InternalProcedure() { - @Override - public void execute(ReadGraphImpl graph, Integer result) { + @Override + public void execute(ReadGraphImpl graph, Integer result) throws DatabaseException { - if (result != null && result != 0) { - procedure.execute(graph, result); - return; - } + if (result != null && result != 0) { + procedure.execute(graph, result); + return; + } - // Fall back to using the fixed builtins. - result = querySupport.getBuiltin(id); - if (result != 0) { - procedure.execute(graph, result); - return; - } + // Fall back to using the fixed builtins. +// result = querySupport.getBuiltin(id); +// if (result != 0) { +// procedure.execute(graph, result); +// return; +// } - try { - result = querySupport.getRandomAccessReference(id); - } catch (ResourceNotFoundException e) { - procedure.exception(graph, e); - return; - } +// try { +// result = querySupport.getRandomAccessReference(id); +// } catch (ResourceNotFoundException e) { +// procedure.exception(graph, e); +// return; +// } - if (result != 0) { - procedure.execute(graph, result); - } else { - procedure.exception(graph, new ResourceNotFoundException(id)); - } + if (result != 0) { + procedure.execute(graph, result); + } else { + procedure.exception(graph, new ResourceNotFoundException(id)); + } - } + } - @Override - public void exception(ReadGraphImpl graph, Throwable t) { - procedure.exception(graph, t); - } + @Override + public void exception(ReadGraphImpl graph, Throwable t) throws DatabaseException { + procedure.exception(graph, t); + } - }); + }); + } catch (DatabaseException e) { + + try { + + procedure.exception(graph, e); + + } catch (DatabaseException e1) { + + Logger.defaultLogError(e1); + + } + + } } - public void forBuiltin(ReadGraphImpl graph, final String id, CacheEntry parent, final InternalProcedure procedure) { + public void forBuiltin(ReadGraphImpl graph, final String id, CacheEntry parent, final InternalProcedure procedure) throws DatabaseException { Integer result = querySupport.getBuiltin(id); if (result != 0) { @@ -574,930 +624,120 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap } - public final void runAsyncRead(final ReadGraphImpl graph, final AsyncRead query, final CacheEntry parent, final ListenerBase listener, final AsyncProcedure procedure) { - - try { - cache.runQuery(graph, query, parent, listener, procedure); - } catch (DatabaseException e) { - throw new IllegalStateException(e); - } - -// int hash = requestHash(query); -// -// AsyncReadEntry entry = asyncReadMap.get(query, hash); -// -// if(parent == null && listener == null) { -// if(entry != null && (entry.isReady() || entry.isExcepted())) { -// System.out.println("ready " + query); -// entry.performFromCache(graph, this, procedure); -//// graph.state.barrier.dec(query); -// return; -// } else { -// query.perform(graph, procedure); -//// graph.state.barrier.dec(query); -// return; -// } -// } -// -// if(entry == null) { -// -// entry = new AsyncReadEntry(query); -// entry.setPending(); -// entry.clearResult(querySupport); -// asyncReadMap.put(query, entry, hash); -// -// performForEach(graph, query, entry, parent, listener, procedure, false); -// -// } else { -// -// if(entry.isPending()) { -// synchronized(entry) { -// if(entry.isPending()) { -// throw new IllegalStateException(); -// // final AsyncBarrierImpl parentBarrier = graph.state.barrier; -// // if(entry.procs == null) entry.procs = new ArrayList>(); -// // entry.procs.add(new AsyncProcedure() { -// // -// // @Override -// // public void execute(AsyncReadGraph graph, T result) { -// // procedure.execute(graph, result); -// // parentBarrier.dec(query); -// // } -// // -// // @Override -// // public void exception(AsyncReadGraph graph, Throwable throwable) { -// // procedure.exception(graph, throwable); -// // parentBarrier.dec(query); -// // } -// // -// // }); -//// if(graph.parent != null || listener != null) { -//// registerDependencies(graph, entry, parent, listener, procedure, false); -//// } -//// -//// query.perform(graph, procedure); -//// -//// return; -// -// } -// } -// } -// -// if(entry.isReady()) { -// entry.performFromCache(graph, this, procedure); -// registerDependencies(graph, entry, parent, listener, procedure, false); -// } else { -// performForEach(graph, query, entry, parent, listener, procedure, false); -// } -// -// } - - } - - final void runMultiRead(final ReadGraphImpl graph, MultiReadEntry cached, final MultiRead query, final CacheEntry parent, final QueryProcessor provider, final ListenerBase listener, final AsyncMultiProcedure procedure) { try { - cache.runQuery(graph, query, parent, listener, procedure); + QueryCache.runnerMultiReadEntry(graph, query, parent, listener, procedure); } catch (DatabaseException e) { throw new IllegalStateException(e); } -// MultiReadEntry entry = cached != null ? cached : provider.multiReadMap.get(query); -// if(entry == null) { -// -// entry = new MultiReadEntry(query); -// entry.setPending(); -// entry.clearResult(provider.querySupport); -// -// provider.multiReadMap.put(query, entry); -// -// provider.performForEach(graph, query, entry, parent, listener, procedure, false); -// -// } else { -// -// if(entry.isPending()) { -// -// synchronized(entry) { -// -// if(entry.isPending()) { -// throw new IllegalStateException(); -// -//// if(entry.procs == null) entry.procs = new ArrayList, AsyncBarrier>>(); -//// entry.procs.add(new Pair(procedure, parentBarrier)); -//// if(graph.parent != null || listener != null) { -//// provider.registerDependencies(graph, entry, parent, listener, procedure, false); -//// } -// -// // If this was synchronized we must wait here until completion -// // if(graph.state.synchronizedExecution) { -// // while(entry.isPending()) { -// // graph.resumeTasks(graph.callerThread, null, null); -// // } -// // } -//// -//// return; -// -// } -// } -// -// entry.performFromCache(graph, provider, procedure); -//// graph.state.barrier.dec(query); -// return; -// -// } else { -// -// provider.performForEach(graph, query, entry, parent, listener, procedure, false); -// -// } -// -// } - } public final void runAsyncMultiRead(final ReadGraphImpl graph, final AsyncMultiRead query, final CacheEntry parent, final ListenerBase listener, final AsyncMultiProcedure procedure) { try { - cache.runQuery(graph, query, parent, listener, procedure); + QueryCache.runnerAsyncMultiReadEntry(graph, query, parent, listener, procedure); } catch (DatabaseException e) { throw new IllegalStateException(e); } + } -// int hash = requestHash(query); -// -// AsyncMultiReadEntry entry = asyncMultiReadMap.get(query, hash); -// -// if(parent == null && listener == null) { -// if(entry != null && (entry.isReady() || entry.isExcepted())) { -// System.out.println("ready " + query); -// entry.performFromCache(graph, this, procedure); -// return; -// } else { -// query.perform(graph, procedure); -// return; -// } -// } -// -// if(entry == null) { -// -// entry = new AsyncMultiReadEntry(query); -// entry.setPending(); -// entry.clearResult(querySupport); -// -// asyncMultiReadMap.put(query, entry, hash); -// -// performForEach(graph, query, entry, parent, listener, procedure, false); -// -// } else { -// -// if(entry.isPending()) { -// -// synchronized(entry) { -// if(entry.isPending()) { -// throw new IllegalStateException(); -//// if(entry.procs == null) entry.procs = new ArrayList>(); -//// entry.procs.add(procedure); -//// if(graph.parent != null || listener != null) { -//// registerDependencies(graph, entry, parent, listener, procedure, false); -//// } -//// return; -// } -// } -// } -// -// performForEach(graph, query, entry, parent, listener, procedure, false); + final void runPrimitiveRead(ReadGraphImpl graph, ExternalReadEntry cached, final ExternalRead query, final CacheEntry parent, final QueryProcessor provider, final ListenerBase listener, final AsyncProcedure procedure) throws DatabaseException { + QueryCache.runnerExternalReadEntry(graph, query, parent, listener, procedure); + } + +// @Override +// public T query(final ReadGraphImpl graph, final Read query, final CacheEntry parent, final AsyncProcedure procedure, final ListenerBase listener) throws DatabaseException { +// +// return QueryCache.resultReadEntry(graph, query, parent, listener, procedure); // -// } +// } + + public void queryMultiRead(final ReadGraphImpl graph, final MultiRead query, final CacheEntry parent, final ListenerBase listener, final AsyncMultiProcedure procedure) throws DatabaseException { + + QueryCache.runnerMultiReadEntry(graph, query, parent, listener, procedure); } - final void runPrimitiveRead(ReadGraphImpl graph, ExternalReadEntry cached, final ExternalRead query, final CacheEntry parent, final QueryProcessor provider, final ListenerBase listener, final Procedure procedure) throws DatabaseException { + public void queryPrimitiveRead(final ReadGraphImpl graph, final ExternalRead query, final CacheEntry parent, final ListenerBase listener, final AsyncProcedure procedure) throws DatabaseException { - cache.runQuery(graph, query, parent, listener, procedure); + QueryCache.runnerExternalReadEntry(graph, query, parent, listener, procedure); -// final ExternalReadEntry entry = cached != null ? cached : provider.externalReadMap.get(query); -// if(entry == null) { -// provider.performForEach(graph, query, new ExternalReadEntry(query), parent, listener, procedure, false); -// } else { -// if(entry.isPending()) { -// synchronized(entry) { -// if(entry.isPending()) { -// throw new IllegalStateException(); -//// if(entry.procs == null) entry.procs = new ArrayList>(); -//// entry.procs.add(procedure); -//// return; -// } -// } -// } -// provider.performForEach(graph, query, entry, parent, listener, procedure, false); -// } + } + boolean isBound(ExternalReadEntry entry) { + if(entry.hasParents()) return true; + else if(hasListener(entry)) return true; + else return false; } - public int requestHash(Object object) { - try { - return object.hashCode(); - } catch (Throwable t) { - Logger.defaultLogError(t); - return 0; + synchronized public ListenerEntry registerDependencies(ReadGraphImpl graph, CacheEntry child, CacheEntry parent, ListenerBase listener, Object procedure, boolean inferred) { + + if (parent != null && !inferred) { + try { + if(!child.isImmutable(graph)) + 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; } + } + - @Override - public T queryRead(final ReadGraphImpl graph, final Read query, final CacheEntry parent, final AsyncProcedure procedure, final ListenerBase listener) throws Throwable { + static class Dummy implements InternalProcedure, IntProcedure { - return (T)cache.runQuery(graph, query, parent, listener, procedure); + @Override + public void execute(ReadGraphImpl graph, int i) { + } -// assert(query != null); -// -// ReadEntry entry = readMap.get(query); -// -// if(entry != null) { -// if(parent == null && (listener == null || listener.isDisposed()) && entry.isReady()) { -// return (T)entry.get(graph, this, procedure); -// } else if (entry.isPending()) { -// throw new IllegalStateException(); -// } -// } -// -// if(entry == null) { -// -// entry = new ReadEntry(query); -// entry.setPending(); -// entry.clearResult(querySupport); -// -// readMap.put(query, entry); -// -// return (T)performForEach(graph, query, entry, parent, listener, procedure, false); -// -// } else { -// -// if(entry.isPending()) { -// throw new IllegalStateException(); -// } else { -// return (T)performForEach(graph, query, entry, parent, listener, procedure, false); -// } -// -// } + @Override + public void finished(ReadGraphImpl graph) { + } + + @Override + public void execute(ReadGraphImpl graph, Object result) { + } + @Override + public void exception(ReadGraphImpl graph, Throwable throwable) { + } + } + + private static final Dummy dummy = new Dummy(); - public void queryMultiRead(final ReadGraphImpl graph, final MultiRead query, final CacheEntry parent, final ListenerBase listener, final AsyncMultiProcedure procedure) throws DatabaseException { + /* + public Object performForEach2(ReadGraphImpl graph, UnaryQuery query, CacheEntry parent, ListenerBase listener, Procedure procedure) throws Throwable { - cache.runQuery(graph, query, parent, listener, procedure); + if (DebugPolicy.PERFORM) + System.out.println("PE[ " + (query.hashCode() & THREAD_MASK) + "] " + query); -// assert(query != null); -// assert(procedure != null); -// -// final MultiReadEntry entry = multiReadMap.get(query); -// -// if(parent == null && !(listener != null)) { -// if(entry != null && entry.isReady()) { -// entry.performFromCache(graph, this, procedure); -// return; -// } -// } -// -// runMultiRead(graph, entry, query, parent, this, listener, procedure); + assert (!dirty); + assert (!collecting); - } + assert(query.assertNotDiscarded()); - public void queryPrimitiveRead(final ReadGraphImpl graph, final ExternalRead query, final CacheEntry parent, final ListenerBase listener, final Procedure procedure) throws DatabaseException { + registerDependencies(graph, query, parent, listener, procedure, false); - cache.runQuery(graph, query, parent, listener, procedure); + // FRESH, REFUTED, EXCEPTED go here + if (!query.isReady()) { -// assert(query != null); -// assert(procedure != null); -// -// final ExternalReadEntry entry = externalReadMap.get(query); -// -// if(parent == null && !(listener != null)) { -// if(entry != null && entry.isReady()) { -// entry.performFromCache(procedure); -// return; -// } -// } -// -// runPrimitiveRead(graph, entry, query, parent, this, listener, procedure); + size++; + misses++; - } - -// public void performForEach(ReadGraphImpl parentGraph, final AsyncRead query, final AsyncReadEntry entry, final CacheEntry parent, final ListenerBase base, final AsyncProcedure procedure, -// boolean inferredDependency) { -// -// if (DebugPolicy.PERFORM) -// System.out.println("PE[ " + (query.hashCode() & THREAD_MASK) + "] " + query); -// -// assert (!dirty); -// assert (!collecting); -// -// assert(!entry.isDiscarded()); -// -// final ListenerEntry listenerEntry = registerDependencies(parentGraph, entry, parent, base, procedure, inferredDependency); -// -// // FRESH, REFUTED, EXCEPTED go here -// if (!entry.isReady()) { -// -// entry.setPending(); -// -// size++; -// -// try { -// -// final ReadGraphImpl finalParentGraph = parentGraph; -// -// query.perform(parentGraph.withParent(entry), new AsyncProcedure() { -// -// @Override -// public void execute(AsyncReadGraph returnGraph, T result) { -// ReadGraphImpl impl = (ReadGraphImpl)returnGraph; -// //AsyncReadGraph resumeGraph = finalParentGraph.newAsync(); -// entry.addOrSet(finalParentGraph, result); -// if(listenerEntry != null) { -// primeListenerEntry(listenerEntry, result); -// } -// try { -// procedure.execute(finalParentGraph, 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(finalParentGraph, t); -// try { -// procedure.exception(finalParentGraph, 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); -// -// } -// -// misses++; -// -// } else { -// -// entry.performFromCache(parentGraph, this, new AsyncProcedure() { -// -// @Override -// public void exception(AsyncReadGraph graph, Throwable throwable) { -// procedure.exception(graph, throwable); -// } -// -// @Override -// public void execute(AsyncReadGraph graph, T result) { -// procedure.execute(graph, result); -// if(listenerEntry != null) { -// primeListenerEntry(listenerEntry, result); -// } -// } -// -// }); -// -//// parentBarrier.dec(query); -// -// hits++; -// -// } -// -// assert (!entry.isDiscarded()); -// -// } - -// public T performForEach(final ReadGraphImpl graph, final Read query, final ReadEntry entry, final CacheEntry parent, final ListenerBase listener, final AsyncProcedure procedure, -// boolean inferredDependency) throws Throwable { -// -// if (DebugPolicy.PERFORM) -// System.out.println("PE[ " + (query.hashCode() & THREAD_MASK) + "] " + query); -// -// assert (!dirty); -// assert (!collecting); -// -// entry.assertNotDiscarded(); -// -// if(entry.isReady()) { -// -// // EXCEPTED goes here -// -//// if(procedure != null) entry.performFromCache(graph, this, procedure); -//// parentBarrier.dec(query); -// hits++; -// -// ListenerEntry listenerEntry = registerDependencies(graph, entry, parent, listener, procedure, inferredDependency); -// -// T result = (T)entry.get(graph, this, procedure); -// -// if(listenerEntry != null) primeListenerEntry(listenerEntry, result); -// -// return result; -// -// } else { -// -// // FRESH, REFUTED, PENDING go here -// -// entry.setPending(); -// -// size++; -// misses++; -// -// ListenerEntry listenerEntry = registerDependencies(graph, entry, parent, listener, procedure, inferredDependency); -// -// final ReadGraphImpl performGraph = graph.newSync(entry); -// -// try { -// -// if(Development.DEVELOPMENT) -// Development.recordHistogram("run " + query); -// -// T result = query.perform(performGraph); -// entry.addOrSet(performGraph, result); -// -// if(listenerEntry != null) primeListenerEntry(listenerEntry, result); -// -// return (T)entry.get(graph, this, procedure); -// -// } catch (Throwable t) { -// -// entry.except(t); -// return (T)entry.get(graph, this, procedure); -// -// } -// -// } -// -// } - -// public void performForEach(final ReadGraphImpl graph, final MultiRead query, final MultiReadEntry entry, CacheEntry parent, final ListenerBase listener, final AsyncMultiProcedure procedure, -// boolean inferredDependency) { -// -// if (DebugPolicy.PERFORM) -// System.out.println("PE[ " + (query.hashCode() & THREAD_MASK) + "] " + query); -// -// assert (!dirty); -// assert (!collecting); -// -// assert(!entry.isPending()); -// assert(!entry.isDiscarded()); -// -// // FRESH, REFUTED, EXCEPTED go here -// if (!entry.isReady()) { -// -// entry.setPending(); -// entry.clearResult(); -// -// multiReadMap.put(query, entry); -// size++; -// -// final ReadGraphImpl newGraph = graph.newSync(entry); -//// newGraph.state.barrier.inc(); -// -// try { -// -// query.perform(newGraph, new AsyncMultiProcedure() { -// -// @Override -// public void execute(AsyncReadGraph graph, T result) { -// entry.addOrSet(result); -// try { -// procedure.execute(graph, result); -// } catch (Throwable t) { -// t.printStackTrace(); -// } -// } -// -// @Override -// public void finished(AsyncReadGraph graph) { -// entry.finish(graph); -// try { -// procedure.finished(graph); -// } catch (Throwable t) { -// t.printStackTrace(); -// } -//// newGraph.state.barrier.dec(); -//// parentBarrier.dec(); -// } -// -// @Override -// public void exception(AsyncReadGraph graph, Throwable t) { -// entry.except(t); -// try { -// procedure.exception(graph, t); -// } catch (Throwable t2) { -// t2.printStackTrace(); -// } -//// newGraph.state.barrier.dec(); -//// parentBarrier.dec(); -// } -// -// }); -// -// } catch (DatabaseException e) { -// -// entry.except(e); -// try { -// procedure.exception(graph, e); -// } catch (Throwable t2) { -// t2.printStackTrace(); -// } -//// newGraph.state.barrier.dec(); -//// parentBarrier.dec(); -// -// } catch (Throwable t) { -// -// DatabaseException e = new DatabaseException(t); -// -// entry.except(e); -// try { -// procedure.exception(graph, e); -// } catch (Throwable t2) { -// t2.printStackTrace(); -// } -//// newGraph.state.barrier.dec(); -//// parentBarrier.dec(); -// -// } -// -// misses++; -// -// } else { -// -// entry.performFromCache(graph, this, procedure); -// hits++; -// -// -// } -// -// assert (!entry.isDiscarded()); -// -// registerDependencies(graph, entry, parent, listener, procedure, inferredDependency); -// -// } -// -// -// public void performForEach(final ReadGraphImpl callerGraph, AsyncMultiRead query, final AsyncMultiReadEntry entry, final CacheEntry parent, final ListenerBase listener, final AsyncMultiProcedure procedure, -// boolean inferredDependency) { -// -// if (DebugPolicy.PERFORM) -// System.out.println("PE[ " + (query.hashCode() & THREAD_MASK) + "] " + query); -// -// assert (!dirty); -// assert (!collecting); -// -// try { -// -// assert(!entry.isDiscarded()); -// -// // FRESH, REFUTED, EXCEPTED go here -// if (!entry.isReady()) { -// -// size++; -// -// try { -// -// ReadGraphImpl performGraph = callerGraph.withAsyncParent(entry); -// -// query.perform(performGraph, new AsyncMultiProcedure() { -// -// @Override -// public void execute(AsyncReadGraph graph, T result) { -// ReadGraphImpl impl = (ReadGraphImpl)graph; -//// ReadGraphImpl executeGraph = callerGraph.newAsync(); -// entry.addOrSet(result); -// try { -// procedure.execute(callerGraph, result); -// } catch (Throwable t) { -// t.printStackTrace(); -// } -// } -// -// @Override -// public void finished(AsyncReadGraph graph) { -// ReadGraphImpl impl = (ReadGraphImpl)graph; -//// ReadGraphImpl executeGraph = callerGraph.newAsync(); -// entry.finish(callerGraph); -// try { -// procedure.finished(callerGraph); -// } catch (Throwable t) { -// t.printStackTrace(); -// } -// } -// -// @Override -// public void exception(AsyncReadGraph graph, Throwable t) { -// ReadGraphImpl impl = (ReadGraphImpl)graph; -//// ReadGraphImpl executeGraph = callerGraph.newAsync(); -// entry.except(callerGraph, t); -// try { -// procedure.exception(callerGraph, t); -// } catch (Throwable t2) { -// t2.printStackTrace(); -// } -// } -// -// }); -// -// } catch (Throwable t) { -// -// entry.except(t); -// try { -// procedure.exception(callerGraph, t); -// } catch (Throwable t2) { -// t2.printStackTrace(); -// } -// -// } -// -// -// misses++; -// -// } else { -// -// entry.performFromCache(callerGraph, this, procedure); -// -// hits++; -// -// } -// -// assert (!entry.isDiscarded()); -// -// registerDependencies(callerGraph, entry, parent, listener, procedure, inferredDependency); -// -// } catch (Throwable t) { -// -// Logger.defaultLogError(t); -// -// } finally { -// -// } -// -// } -// -// public void performForEach(ReadGraphImpl graph, final ExternalRead query, final ExternalReadEntry entry, final CacheEntry parent, final ListenerBase base, final Procedure procedure, -// boolean inferredDependency) { -// -// if (DebugPolicy.PERFORM) -// System.out.println("PE[ " + (query.hashCode() & THREAD_MASK) + "] " + query); -// -// assert (!dirty); -// assert (!collecting); -// -// assert(!entry.isPending()); -// assert(!entry.isDiscarded()); -// -// registerDependencies(graph, entry, parent, base, procedure, inferredDependency); -// -// // FRESH, REFUTED, EXCEPTED go here -// if (!entry.isReady()) { -// -// entry.setPending(); -// entry.clearResult(querySupport); -// -// externalReadMap.put(query, entry); -// size++; -// -// try { -// -// query.register(graph, new Listener() { -// -// AtomicBoolean used = new AtomicBoolean(false); -// -// @Override -// public void execute(T result) { -// -// // Just for safety -// if(entry.isDiscarded()) return; -// if(entry.isExcepted()) entry.setPending(); -// -// if(used.compareAndSet(false, true)) { -// entry.addOrSet(QueryProcessor.this, result); -// procedure.execute(result); -// } else { -// entry.queue(result); -// updatePrimitive(query); -// } -// -// } -// -// @Override -// public void exception(Throwable t) { -// -// entry.except(t); -// -// if(used.compareAndSet(false, true)) { -// procedure.exception(t); -// } else { -//// entry.queue(result); -// updatePrimitive(query); -// } -// -// } -// -// @Override -// public String toString() { -// return procedure.toString(); -// } -// -// @Override -// public boolean isDisposed() { -// return entry.isDiscarded() || !isBound(entry); -// } -// -// }); -// -// } catch (Throwable t) { -// -// entry.except(t); -// procedure.exception(t); -// -// } -// -// misses++; -// -// } else { -// -// entry.performFromCache(procedure); -// -// hits++; -// -// } -// -// assert (!entry.isDiscarded()); -// -// } - - boolean isBound(ExternalReadEntry entry) { - if(entry.hasParents()) return true; - else if(hasListener(entry)) return true; - else return false; - } - - synchronized public ListenerEntry registerDependencies(ReadGraphImpl graph, CacheEntry child, CacheEntry parent, ListenerBase listener, Object procedure, boolean inferred) { - - if (parent != null && !inferred) { - try { - if(!child.isImmutable(graph)) - 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 void performForEach(ReadGraphImpl graph, BinaryQuery query, CacheEntry parent, ListenerBase listener, Procedure procedure) { - - if (DebugPolicy.PERFORM) - System.out.println("PE[ " + (query.hashCode() & THREAD_MASK) + "] " + query); - - assert (!dirty); - assert (!collecting); - - try { - - registerDependencies(graph, query, parent, listener, procedure, false); - - // FRESH, REFUTED, EXCEPTED go here - if (!query.isReady()) { - - boolean fresh = query.isFresh(); - - if(fresh) { - size++; - } - - query.computeForEach(graph, this, procedure, true); - - misses++; - - } else { - - query.performFromCache(graph, this, procedure); - - hits++; - - } - - } catch (Throwable t) { - - Logger.defaultLogError(t); - - } - } - - public Object performForEach(ReadGraphImpl graph, UnaryQuery query, CacheEntry parent, ListenerBase listener, Procedure procedure) { - - if (DebugPolicy.PERFORM) - System.out.println("PE[ " + (query.hashCode() & THREAD_MASK) + "] " + query); - - assert (!dirty); - assert (!collecting); - - try { - - assert(query.assertNotDiscarded()); - - registerDependencies(graph, query, parent, listener, procedure, false); - - // FRESH, REFUTED, EXCEPTED go here - if (!query.isReady()) { - - size++; - misses++; - - return query.computeForEach(graph, this, procedure, true); - - - } else { - - hits++; - - return query.performFromCache(graph, this, procedure); - - } - - } catch (Throwable t) { - - Logger.defaultLogError(t); - return null; - - } - - } - - static class Dummy implements InternalProcedure, IntProcedure { - - @Override - public void execute(ReadGraphImpl graph, int i) { - } - - @Override - public void finished(ReadGraphImpl graph) { - } - - @Override - public void execute(ReadGraphImpl graph, Object result) { - } - - @Override - public void exception(ReadGraphImpl graph, Throwable throwable) { - } - - } - - private static final Dummy dummy = new Dummy(); - - public Object performForEach2(ReadGraphImpl graph, UnaryQuery query, CacheEntry parent, ListenerBase listener, Procedure procedure) throws Throwable { - - if (DebugPolicy.PERFORM) - System.out.println("PE[ " + (query.hashCode() & THREAD_MASK) + "] " + query); - - assert (!dirty); - assert (!collecting); - - assert(query.assertNotDiscarded()); - - registerDependencies(graph, query, parent, listener, procedure, false); - - // FRESH, REFUTED, EXCEPTED go here - if (!query.isReady()) { - - size++; - misses++; - - query.computeForEach(graph, this, (Procedure)dummy, true); - return query.get(graph, this, null); + query.computeForEach(graph, this, (Procedure)dummy, true); + return query.get(graph, this, null); } else { @@ -1508,50 +748,8 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap } } + */ - public void performForEach(ReadGraphImpl graph, StringQuery query, CacheEntry parent, final ListenerBase listener, Procedure procedure) { - - if (DebugPolicy.PERFORM) - System.out.println("PE[ " + (query.hashCode() & THREAD_MASK) + "] " + query); - - assert (!dirty); - assert (!collecting); - - try { - - if(query.isDiscarded()) { - System.err.println("aff"); - } - assert(!query.isDiscarded()); - - // FRESH, REFUTED, EXCEPTED go here - if (!query.isReady()) { - - query.computeForEach(graph.withAsyncParent(query), this, procedure); - - size++; - misses++; - - } else { - - query.performFromCache(graph, this, procedure); - - hits++; - - } - - assert (!query.isDiscarded()); - - registerDependencies(graph, query, parent, listener, procedure, false); - - } catch (Throwable t) { - - t.printStackTrace(); - Logger.defaultLogError(t); - - } - - } interface QueryCollectorSupport { public CacheCollectionResult allCaches(); @@ -1679,7 +877,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap @Override public int getCurrentSize() { - return size; + return cache.size; } } @@ -1689,7 +887,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap private QueryCollector collector = new QueryCollectorImpl(this, collectorSupport); public int querySize() { - return size; + return cache.size; } public void gc(int youngTarget, int allowedTimeInMs) { @@ -1797,7 +995,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap return Collections.emptyList(); } - void processListenerReport(CacheEntry entry, Map> workarea) { + void processListenerReport(CacheEntry entry, Map> workarea) { if(!workarea.containsKey(entry)) { @@ -1968,7 +1166,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap int unboundCounter = 0; int unknownCounter = 0; - for(CacheEntry entry : workarea.keySet()) { + for(CacheEntry entry : workarea.keySet()) { //System.err.println("process " + entry); @@ -2176,8 +1374,8 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap query.removeEntry(this); - updates++; - size--; + cache.updates++; + cache.size--; if((entry.getGCStatus() & CacheEntry.HAS_BEEN_BOUND) != 0) boundQueries--; @@ -2198,7 +1396,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap CacheEntry entry = e.entry; -// System.err.println("updateQuery " + entry); + //System.err.println("updateQuery " + entry); /* * If the dependency graph forms a DAG, some entries are inserted in the @@ -2245,7 +1443,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap } } - updates++; + cache.updates++; if (Development.DEVELOPMENT) { if(Development.getProperty(DevelopmentKeys.QUERYPROCESSOR_UPDATE, Bindings.BOOLEAN)) { @@ -2394,7 +1592,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap ReadGraphImpl parentGraph = graph.withParent(entry); - query.recompute(parentGraph, this, entry); + query.recompute(parentGraph); if(entry.isExcepted()) return ListenerEntry.NO_VALUE; @@ -2445,7 +1643,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap public void performScheduledUpdates(WriteGraphImpl graph) { assert (!updating); - assert (!collecting); + assert (!cache.collecting); assert (!firingListeners); firingListeners = true; @@ -2493,7 +1691,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap try { if(DebugPolicy.LISTENER) System.out.println("Firing " + listenerEntry.procedure + " for " + listenerEntry.entry); - entry.performFromCache(graph, this, listenerEntry.procedure); + entry.performFromCache(graph, listenerEntry.procedure); } catch (Throwable t) { t.printStackTrace(); } @@ -2517,7 +1715,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap */ public boolean update(final ReadGraphImpl graph, final CacheEntry entry) { - assert (!collecting); + assert (!cache.collecting); assert (!updating); updating = true; @@ -2585,8 +1783,6 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap } - volatile public boolean dirty = false; - private ObjectUpdateSet scheduledObjectUpdates = new ObjectUpdateSet(); private ValueUpdateSet scheduledValueUpdates = new ValueUpdateSet(); private ValueUpdateSet scheduledInvalidates = new ValueUpdateSet(); @@ -2596,7 +1792,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap public void performDirtyUpdates(final ReadGraphImpl graph) { - dirty = false; + cache.dirty = false; lastInvalidate = 0; if (Development.DEVELOPMENT) { @@ -2613,14 +1809,14 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap final int subject = (int)(arg0 >>> 32); final int predicate = (int)(arg0 & 0xffffffff); - for(Objects o : Objects.entries(QueryProcessor.this, subject)) update(graph, o); - for(DirectObjects o : DirectObjects.entries(QueryProcessor.this, subject)) update(graph, o); - for(Statements o : Statements.entries(QueryProcessor.this, subject)) update(graph, o); + for(Objects o : QueryCache.entriesObjects(QueryProcessor.this, subject)) update(graph, o); + for(DirectObjects o : QueryCache.entriesDirectObjects(QueryProcessor.this, subject)) update(graph, o); + for(Statements o : QueryCache.entriesStatements(QueryProcessor.this, subject)) update(graph, o); if(predicate == instanceOf || predicate == inherits || predicate == subrelationOf) { - PrincipalTypes principalTypes = PrincipalTypes.entry(QueryProcessor.this, subject); + PrincipalTypes principalTypes = QueryCache.entryPrincipalTypes(QueryProcessor.this, subject); if(principalTypes != null) update(graph, principalTypes); - Types types = Types.entry(QueryProcessor.this, subject); + Types types = QueryCache.entryTypes(QueryProcessor.this, subject); if(types != null) update(graph, types); } @@ -2629,9 +1825,9 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap if(superRelations != null) update(graph, superRelations); } - DirectPredicates dp = DirectPredicates.entry(QueryProcessor.this, subject); + DirectPredicates dp = QueryCache.entryDirectPredicates(QueryProcessor.this, subject); if(dp != null) update(graph, dp); - OrderedSet os = OrderedSet.entry(QueryProcessor.this, predicate); + OrderedSet os = QueryCache.entryOrderedSet(QueryProcessor.this, predicate); if(os != null) update(graph, os); scheduledObjectUpdates.clear(); @@ -2644,7 +1840,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap int arg0 = scheduledValueUpdates.getFirst(); - ValueQuery valueQuery = ValueQuery.entry(QueryProcessor.this, arg0); + ValueQuery valueQuery = QueryCache.entryValueQuery(QueryProcessor.this, arg0); if(valueQuery != null) update(graph, valueQuery); scheduledValueUpdates.clear(); @@ -2666,11 +1862,11 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap @Override public boolean execute(Object arg0) { - ExternalReadEntry query = (ExternalReadEntry)cache.externalReadMap.get(arg0); + ExternalReadEntry query = (ExternalReadEntry)cache.externalReadEntryMap.get(arg0); if (query != null) { boolean listening = update(graph, query); if (!listening && !query.hasParents()) { - cache.externalReadMap.remove(arg0); + cache.externalReadEntryMap.remove(arg0); query.discard(); } } @@ -2683,7 +1879,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap @Override public boolean execute(int arg0) { - ValueQuery valueQuery = ValueQuery.entry(QueryProcessor.this, arg0); + ValueQuery valueQuery = QueryCache.entryValueQuery(QueryProcessor.this, arg0); if(valueQuery != null) update(graph, valueQuery); return true; } @@ -2695,12 +1891,12 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap @Override public boolean execute(int resource) { - ValueQuery valueQuery = ValueQuery.entry(QueryProcessor.this, resource); + ValueQuery valueQuery = QueryCache.entryValueQuery(QueryProcessor.this, resource); if(valueQuery != null) update(graph, valueQuery); - PrincipalTypes principalTypes = PrincipalTypes.entry(QueryProcessor.this, resource); + PrincipalTypes principalTypes = QueryCache.entryPrincipalTypes(QueryProcessor.this, resource); if(principalTypes != null) update(graph, principalTypes); - Types types = Types.entry(QueryProcessor.this, resource); + Types types = QueryCache.entryTypes(QueryProcessor.this, resource); if(types != null) update(graph, types); SuperRelations superRelations = SuperRelations.entry(QueryProcessor.this, resource); @@ -2722,9 +1918,9 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap final int predicate = (int)(arg0 & 0xffffffff); if(predicate == instanceOf || predicate == inherits || predicate == subrelationOf) { - PrincipalTypes principalTypes = PrincipalTypes.entry(QueryProcessor.this, subject); + PrincipalTypes principalTypes = QueryCache.entryPrincipalTypes(QueryProcessor.this, subject); if(principalTypes != null) update(graph, principalTypes); - Types types = Types.entry(QueryProcessor.this, subject); + Types types = QueryCache.entryTypes(QueryProcessor.this, subject); if(types != null) update(graph, types); } @@ -2747,11 +1943,11 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap @Override public boolean execute(final int subject) { - for(Objects o : Objects.entries(QueryProcessor.this, subject)) update(graph, o); - for(DirectObjects o : DirectObjects.entries(QueryProcessor.this, subject)) update(graph, o); - for(Statements o : Statements.entries(QueryProcessor.this, subject)) update(graph, o); + for(Objects o : QueryCache.entriesObjects(QueryProcessor.this, subject)) update(graph, o); + for(DirectObjects o : QueryCache.entriesDirectObjects(QueryProcessor.this, subject)) update(graph, o); + for(Statements o : QueryCache.entriesStatements(QueryProcessor.this, subject)) update(graph, o); - DirectPredicates entry = DirectPredicates.entry(QueryProcessor.this, subject); + DirectPredicates entry = QueryCache.entryDirectPredicates(QueryProcessor.this, subject); if(entry != null) update(graph, entry); return true; @@ -2765,7 +1961,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap @Override public boolean execute(int orderedSet) { - OrderedSet entry = OrderedSet.entry(QueryProcessor.this, orderedSet); + OrderedSet entry = QueryCache.entryOrderedSet(QueryProcessor.this, orderedSet); if(entry != null) update(graph, entry); return true; @@ -2794,12 +1990,12 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap public void updateValue(final int resource) { scheduledValueUpdates.add(resource); - dirty = true; + cache.dirty = true; } public void updateStatements(final int resource, final int predicate) { scheduledObjectUpdates.add((((long)resource) << 32) + predicate); - dirty = true; + cache.dirty = true; } private int lastInvalidate = 0; @@ -2808,7 +2004,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap if(lastInvalidate == resource) return; scheduledValueUpdates.add(resource); lastInvalidate = resource; - dirty = true; + cache.dirty = true; } public void updatePrimitive(final ExternalRead primitive) { @@ -2824,7 +2020,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap @Override public synchronized String toString() { - return "QueryProvider [size = " + size + ", hits = " + hits + " misses = " + misses + ", updates = " + updates + "]"; + return "QueryProvider [size = " + cache.size + ", hits = " + cache.hits + " misses = " + cache.misses + ", updates = " + cache.updates + "]"; } @Override @@ -2880,28 +2076,28 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap } public int getHits() { - return hits; + return cache.hits; } public int getMisses() { - return misses; + return cache.misses; } public int getSize() { - return size; + return cache.size; } public Set getReferencedClusters() { HashSet result = new HashSet(); - for (CacheEntry entry : cache.objectsMap.values()) { + for (CacheEntry entry : QueryCache.entriesObjects(this)) { Objects query = (Objects) entry.getQuery(); result.add(querySupport.getClusterId(query.r1())); } - for (CacheEntry entry : cache.directPredicatesMap.values()) { + for (CacheEntry entry : QueryCache.entriesDirectPredicates(this)) { DirectPredicates query = (DirectPredicates) entry.getQuery(); result.add(querySupport.getClusterId(query.id)); } - for (CacheEntry entry : cache.valueMap.values()) { + for (CacheEntry entry : cache.valueQueryMap.values()) { ValueQuery query = (ValueQuery) entry.getQuery(); result.add(querySupport.getClusterId(query.id)); } @@ -2926,7 +2122,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap public int clean() { collector.collect(0, Integer.MAX_VALUE); - return size; + return cache.size; } public void clean(final Collection> requests) { @@ -2940,7 +2136,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap public CacheEntryBase iterate(int level) { if(iterator.hasNext()) { ExternalRead request = iterator.next(); - ExternalReadEntry entry = cache.externalReadMap.get(request); + ExternalReadEntry entry = cache.externalReadEntryMap.get(request); if (entry != null) return entry; else return iterate(level); } else { @@ -2960,7 +2156,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap public Collection getRootList() { ArrayList result = new ArrayList(requests.size()); for (ExternalRead request : requests) { - ExternalReadEntry entry = cache.externalReadMap.get(request); + ExternalReadEntry entry = cache.externalReadEntryMap.get(request); if (entry != null) result.add(entry); } @@ -2968,7 +2164,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap } @Override public int getCurrentSize() { - return size; + return cache.size; } @Override public int calculateCurrentSize() { @@ -3069,122 +2265,123 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap @Override final public void forEachPredicate(final ReadGraphImpl impl, final Resource subject, final AsyncMultiProcedure procedure) { - assert(subject != null); - assert(procedure != null); - - final ListenerBase listener = getListenerBase(procedure); - - IntProcedure ip = new IntProcedure() { - - AtomicBoolean first = new AtomicBoolean(true); - - @Override - public void execute(ReadGraphImpl graph, int i) { - try { - if(first.get()) { - procedure.execute(graph, querySupport.getResource(i)); - } else { - procedure.execute(impl.newRestart(graph), querySupport.getResource(i)); - } - } catch (Throwable t2) { - Logger.defaultLogError(t2); - } - } - - @Override - public void finished(ReadGraphImpl graph) { - try { - if(first.compareAndSet(true, false)) { - procedure.finished(graph); -// impl.state.barrier.dec(this); - } else { - procedure.finished(impl.newRestart(graph)); - } - - } catch (Throwable t2) { - Logger.defaultLogError(t2); - } - } + throw new UnsupportedOperationException(); - @Override - public void exception(ReadGraphImpl graph, Throwable t) { - try { - if(first.compareAndSet(true, false)) { - procedure.exception(graph, t); -// impl.state.barrier.dec(this); - } else { - procedure.exception(impl.newRestart(graph), t); - } - } catch (Throwable t2) { - Logger.defaultLogError(t2); - } - } - - }; - - int sId = querySupport.getId(subject); - -// if(AsyncBarrierImpl.BOOKKEEPING) impl.state.barrier.inc(ip, "#Predicates#" + sId); -// else impl.state.barrier.inc(null, null); - - Predicates.queryEach(impl, sId, this, impl.parent, listener, ip); +// assert(subject != null); +// assert(procedure != null); +// +// final ListenerBase listener = getListenerBase(procedure); +// +// IntProcedure ip = new IntProcedure() { +// +// AtomicBoolean first = new AtomicBoolean(true); +// +// @Override +// public void execute(ReadGraphImpl graph, int i) { +// try { +// if(first.get()) { +// procedure.execute(graph, querySupport.getResource(i)); +// } else { +// procedure.execute(impl.newRestart(graph), querySupport.getResource(i)); +// } +// } catch (Throwable t2) { +// Logger.defaultLogError(t2); +// } +// } +// +// @Override +// public void finished(ReadGraphImpl graph) { +// try { +// if(first.compareAndSet(true, false)) { +// procedure.finished(graph); +//// impl.state.barrier.dec(this); +// } else { +// procedure.finished(impl.newRestart(graph)); +// } +// +// } catch (Throwable t2) { +// Logger.defaultLogError(t2); +// } +// } +// +// @Override +// public void exception(ReadGraphImpl graph, Throwable t) { +// try { +// if(first.compareAndSet(true, false)) { +// procedure.exception(graph, t); +// } else { +// procedure.exception(impl.newRestart(graph), t); +// } +// } catch (Throwable t2) { +// Logger.defaultLogError(t2); +// } +// } +// +// }; +// +// int sId = querySupport.getId(subject); +// +// try { +// QueryCache.runnerPredicates(impl, sId, impl.parent, listener, ip); +// } catch (DatabaseException e) { +// Logger.defaultLogError(e); +// } } @Override final public void forEachPredicate(final ReadGraphImpl impl, final Resource subject, final MultiProcedure procedure) { + + throw new UnsupportedOperationException(); - assert(subject != null); - assert(procedure != null); - - final ListenerBase listener = getListenerBase(procedure); - -// impl.state.barrier.inc(); - - Predicates.queryEach(impl, querySupport.getId(subject), this, impl.parent, listener, new IntProcedure() { - - @Override - public void execute(ReadGraphImpl graph, int i) { - try { - procedure.execute(querySupport.getResource(i)); - } catch (Throwable t2) { - Logger.defaultLogError(t2); - } - } - - @Override - public void finished(ReadGraphImpl graph) { - try { - procedure.finished(); - } catch (Throwable t2) { - Logger.defaultLogError(t2); - } -// impl.state.barrier.dec(); - } - - @Override - public void exception(ReadGraphImpl graph, Throwable t) { - try { - procedure.exception(t); - } catch (Throwable t2) { - Logger.defaultLogError(t2); - } -// impl.state.barrier.dec(); - } - - }); +// assert(subject != null); +// assert(procedure != null); +// +// final ListenerBase listener = getListenerBase(procedure); +// +// try { +// QueryCache.runnerPredicates(impl, querySupport.getId(subject), impl.parent, listener, new IntProcedure() { +// +// @Override +// public void execute(ReadGraphImpl graph, int i) { +// try { +// procedure.execute(querySupport.getResource(i)); +// } catch (Throwable t2) { +// Logger.defaultLogError(t2); +// } +// } +// +// @Override +// public void finished(ReadGraphImpl graph) { +// try { +// procedure.finished(); +// } catch (Throwable t2) { +// Logger.defaultLogError(t2); +// } +//// impl.state.barrier.dec(); +// } +// +// @Override +// public void exception(ReadGraphImpl graph, Throwable t) { +// try { +// procedure.exception(t); +// } catch (Throwable t2) { +// Logger.defaultLogError(t2); +// } +//// impl.state.barrier.dec(); +// } +// +// }); +// } catch (DatabaseException e) { +// Logger.defaultLogError(e); +// } } @Override final public IntSet getPredicates(final ReadGraphImpl impl, final Resource subject) throws Throwable { - - assert(subject != null); - - return Predicates.queryEach2(impl, querySupport.getId(subject), this, impl.parent); - + return QueryCacheBase.resultPredicates(impl, querySupport.getId(subject), impl.parent, null); } - @Override final public void forEachStatement(final ReadGraphImpl impl, final Resource subject, @@ -3198,38 +2395,42 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap // impl.state.barrier.inc(); - Statements.queryEach(impl, querySupport.getId(subject), querySupport.getId(predicate), this, impl.parent, listener, new TripleIntProcedureAdapter() { + try { + Statements.queryEach(impl, querySupport.getId(subject), querySupport.getId(predicate), this, impl.parent, listener, new TripleIntProcedureAdapter() { - @Override - public void execute(ReadGraphImpl graph, int s, int p, int o) { - try { - procedure.execute(querySupport.getStatement(s, p, o)); - } catch (Throwable t2) { - Logger.defaultLogError(t2); + @Override + public void execute(ReadGraphImpl graph, int s, int p, int o) { + try { + procedure.execute(querySupport.getStatement(s, p, o)); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } } - } - @Override - public void finished(ReadGraphImpl graph) { - try { - procedure.finished(); - } catch (Throwable t2) { - Logger.defaultLogError(t2); - } + @Override + public void finished(ReadGraphImpl graph) { + try { + procedure.finished(); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } // impl.state.barrier.dec(); - } - - @Override - public void exception(ReadGraphImpl graph, Throwable t) { - try { - procedure.exception(t); - } catch (Throwable t2) { - Logger.defaultLogError(t2); } + + @Override + public void exception(ReadGraphImpl graph, Throwable t) { + try { + procedure.exception(t); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } // impl.state.barrier.dec(); - } + } - }); + }); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } @@ -3302,7 +2503,11 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap // if(AsyncBarrierImpl.BOOKKEEPING) impl.state.barrier.inc(proc, "#Statements" + sId + "#" + pId); // else impl.state.barrier.inc(null, null); - Statements.queryEach(impl, sId, pId, this, impl.parent, listener, proc); + try { + Statements.queryEach(impl, sId, pId, this, impl.parent, listener, proc); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } @@ -3375,7 +2580,11 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap // if(AsyncBarrierImpl.BOOKKEEPING) impl.state.barrier.inc(proc, "#Statements" + sId + "#" + pId); // else impl.state.barrier.inc(null, null); - Statements.queryEach(impl, sId, pId, this, impl.parent, listener, proc); + try { + Statements.queryEach(impl, sId, pId, this, impl.parent, listener, proc); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } @@ -3447,38 +2656,42 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap // impl.state.barrier.inc(); - AssertedStatements.queryEach(impl, querySupport.getId(subject), querySupport.getId(predicate), this, impl.parent, listener, new TripleIntProcedureAdapter() { + try { + QueryCache.runnerAssertedStatements(impl, querySupport.getId(subject), querySupport.getId(predicate), impl.parent, listener, new TripleIntProcedureAdapter() { - @Override - public void execute(ReadGraphImpl graph, int s, int p, int o) { - try { - procedure.execute(graph, querySupport.getStatement(s, p, o)); - } catch (Throwable t2) { - Logger.defaultLogError(t2); + @Override + public void execute(ReadGraphImpl graph, int s, int p, int o) { + try { + procedure.execute(graph, querySupport.getStatement(s, p, o)); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } } - } - @Override - public void finished(ReadGraphImpl graph) { - try { - procedure.finished(graph); - } catch (Throwable t2) { - Logger.defaultLogError(t2); - } + @Override + public void finished(ReadGraphImpl graph) { + try { + procedure.finished(graph); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } // impl.state.barrier.dec(); - } - - @Override - public void exception(ReadGraphImpl graph, Throwable t) { - try { - procedure.exception(graph, t); - } catch (Throwable t2) { - Logger.defaultLogError(t2); } + + @Override + public void exception(ReadGraphImpl graph, Throwable t) { + try { + procedure.exception(graph, t); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } // impl.state.barrier.dec(); - } + } - }); + }); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } @@ -3498,84 +2711,65 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap // impl.state.barrier.inc(); - Objects.runner(impl, querySupport.getId(subject), querySupport.getId(predicate), impl.parent, listener, new IntProcedure() { + try { + QueryCache.runnerObjects(impl, querySupport.getId(subject), querySupport.getId(predicate), impl.parent, listener, new IntProcedure() { - @Override - public void execute(ReadGraphImpl graph, int i) { - try { - procedure.execute(querySupport.getResource(i)); - } catch (Throwable t2) { - Logger.defaultLogError(t2); + @Override + public void execute(ReadGraphImpl graph, int i) { + try { + procedure.execute(querySupport.getResource(i)); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } } - } - @Override - public void finished(ReadGraphImpl graph) { - try { - procedure.finished(); - } catch (Throwable t2) { - Logger.defaultLogError(t2); - } + @Override + public void finished(ReadGraphImpl graph) { + try { + procedure.finished(); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } // impl.state.barrier.dec(); - } - - @Override - public void exception(ReadGraphImpl graph, Throwable t) { - System.out.println("forEachObject exception " + t); - try { - procedure.exception(t); - } catch (Throwable t2) { - Logger.defaultLogError(t2); } + + @Override + public void exception(ReadGraphImpl graph, Throwable t) { + System.out.println("forEachObject exception " + t); + try { + procedure.exception(t); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } // impl.state.barrier.dec(); - } + } - }); + }); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } + @Override + final public void forEachDirectPredicate(final ReadGraphImpl impl, final Resource subject, final AsyncMultiProcedure procedure) { -// @Override -// final public void forEachDirectObject(final ReadGraphImpl impl, final Resource subject, final Resource predicate, final AsyncMultiProcedure procedure) { -// + throw new UnsupportedOperationException(); + // assert(subject != null); -// assert(predicate != null); // assert(procedure != null); // // final ListenerBase listener = getListenerBase(procedure); // -// int sId = querySupport.getId(subject); -// int pId = querySupport.getId(predicate); -// -// MultiIntProcedure proc = new MultiIntProcedure(procedure, impl, support); -// -// if(AsyncBarrierImpl.BOOKKEEPING) impl.state.barrier.inc(proc, "#DirectObjects" + sId + "#" + pId); -// else impl.state.barrier.inc(null, null); -// -// // final Exception caller = new Exception(); -// -// // final Pair exceptions = Pair.make(callerException, new Exception()); +// MultiIntProcedure proc = new MultiIntProcedure(procedure, impl, querySupport); // -// DirectObjects.queryEach(impl, sId, pId, processor, impl.parent, listener, proc); +// int sId = querySupport.getId(subject); // -// } - - @Override - final public void forEachDirectPredicate(final ReadGraphImpl impl, final Resource subject, final AsyncMultiProcedure procedure) { - - assert(subject != null); - assert(procedure != null); - - final ListenerBase listener = getListenerBase(procedure); - - MultiIntProcedure proc = new MultiIntProcedure(procedure, impl, querySupport); - - int sId = querySupport.getId(subject); - -// if(AsyncBarrierImpl.BOOKKEEPING) impl.state.barrier.inc(proc, "#DirectPredicates" + sId); -// else impl.state.barrier.inc(null, null); - - DirectPredicates.queryEach(impl, sId, this, impl.parent, listener, proc); +// try { +// QueryCache.runnerDirectPredicates(impl, sId, impl.parent, listener, proc); +// } catch (DatabaseException e) { +// Logger.defaultLogError(e); +// } } @@ -3641,16 +2835,59 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap final int sId = querySupport.getId(subject); final int pId = querySupport.getId(predicate); - Objects.runner(impl, sId, pId, impl.parent, listener, procedure); + try { + QueryCache.runnerObjects(impl, sId, pId, impl.parent, listener, procedure); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } - final public int getSingleObject(final ReadGraphImpl impl, final Resource subject, final Resource predicate) throws DatabaseException { + static class Runner2Procedure implements IntProcedure { + + public int single = 0; + public Throwable t = null; + + public void clear() { + single = 0; + t = null; + } + + @Override + public void execute(ReadGraphImpl graph, int i) { + if(single == 0) single = i; + else single = -1; + } + + @Override + public void finished(ReadGraphImpl graph) { + if(single == -1) single = 0; + } - final int sId = querySupport.getId(subject); - final int pId = querySupport.getId(predicate); + @Override + public void exception(ReadGraphImpl graph, Throwable throwable) { + single = 0; + this.t = throwable; + } + + public int get() throws DatabaseException { + if(t != null) { + if(t instanceof DatabaseException) throw (DatabaseException)t; + else throw new DatabaseException(t); + } + return single; + } + + } + + final public int getSingleObject(final ReadGraphImpl impl, final Resource subject, final Resource predicate) throws DatabaseException { + + final int sId = querySupport.getId(subject); + final int pId = querySupport.getId(predicate); - return Objects.runner2(impl, sId, pId, impl.parent); + Runner2Procedure proc = new Runner2Procedure(); + QueryCache.runnerObjects(impl, sId, pId, impl.parent, null, proc); + return proc.get(); } @@ -4062,40 +3299,42 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap final ListenerBase listener = getListenerBase(procedure); -// impl.state.barrier.inc(); - - AssertedStatements.queryEach(impl, querySupport.getId(subject), querySupport.getId(predicate), this, impl.parent, listener, new TripleIntProcedure() { + try { + QueryCache.runnerAssertedStatements(impl, querySupport.getId(subject), querySupport.getId(predicate), impl.parent, listener, new TripleIntProcedure() { - @Override - public void execute(ReadGraphImpl graph, int s, int p, int o) { - try { - procedure.execute(graph, querySupport.getResource(o)); - } catch (Throwable t2) { - Logger.defaultLogError(t2); + @Override + public void execute(ReadGraphImpl graph, int s, int p, int o) { + try { + procedure.execute(graph, querySupport.getResource(o)); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } } - } - @Override - public void finished(ReadGraphImpl graph) { - try { - procedure.finished(graph); - } catch (Throwable t2) { - Logger.defaultLogError(t2); - } + @Override + public void finished(ReadGraphImpl graph) { + try { + procedure.finished(graph); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } // impl.state.barrier.dec(); - } - - @Override - public void exception(ReadGraphImpl graph, Throwable t) { - try { - procedure.exception(graph, t); - } catch (Throwable t2) { - Logger.defaultLogError(t2); } + + @Override + public void exception(ReadGraphImpl graph, Throwable t) { + try { + procedure.exception(graph, t); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } // impl.state.barrier.dec(); - } + } - }); + }); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } @@ -4145,7 +3384,11 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap // if(AsyncBarrierImpl.BOOKKEEPING) impl.state.barrier.inc(ip, "#PrincipalTypes#" + sId); // else impl.state.barrier.inc(null, null); - PrincipalTypes.queryEach(impl, sId, this, impl.parent, listener, ip); + try { + QueryCache.runnerPrincipalTypes(impl, sId, impl.parent, listener, ip); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } @@ -4159,39 +3402,42 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap // impl.state.barrier.inc(); - PrincipalTypes.queryEach(impl, querySupport.getId(subject), this, impl.parent, listener, new IntProcedure() { + try { + QueryCache.runnerPrincipalTypes(impl, querySupport.getId(subject), impl.parent, listener, new IntProcedure() { - @Override - public void execute(ReadGraphImpl graph, int i) { - try { - procedure.execute(querySupport.getResource(i)); - } catch (Throwable t2) { - Logger.defaultLogError(t2); + @Override + public void execute(ReadGraphImpl graph, int i) { + try { + procedure.execute(querySupport.getResource(i)); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } } - } - @Override - public void finished(ReadGraphImpl graph) { - try { - procedure.finished(); - } catch (Throwable t2) { - Logger.defaultLogError(t2); - } + @Override + public void finished(ReadGraphImpl graph) { + try { + procedure.finished(); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } // impl.state.barrier.dec(); - } - - @Override - public void exception(ReadGraphImpl graph, Throwable t) { - try { - procedure.exception(t); - } catch (Throwable t2) { - Logger.defaultLogError(t2); } -// impl.state.barrier.dec(); - } - }); + @Override + public void exception(ReadGraphImpl graph, Throwable t) { + try { + procedure.exception(t); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } +// impl.state.barrier.dec(); + } + }); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } final public void forTypes(final ReadGraphImpl impl, final Resource subject, final AsyncProcedure> procedure) { @@ -4200,47 +3446,29 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap assert(procedure != null); final ListenerBase listener = getListenerBase(procedure); + assert(listener == null); InternalProcedure ip = new InternalProcedure() { - AtomicBoolean first = new AtomicBoolean(true); - @Override public void execute(final ReadGraphImpl graph, IntSet set) { - try { - if(first.compareAndSet(true, false)) { - procedure.execute(graph, set); -// impl.state.barrier.dec(this); - } else { - procedure.execute(impl.newRestart(graph), set); - } - } catch (Throwable t2) { - Logger.defaultLogError(t2); - } + procedure.execute(graph, set); } @Override public void exception(ReadGraphImpl graph, Throwable t) { - try { - if(first.compareAndSet(true, false)) { - procedure.exception(graph, t); -// impl.state.barrier.dec(this); - } else { - procedure.exception(impl.newRestart(graph), t); - } - } catch (Throwable t2) { - Logger.defaultLogError(t2); - } + procedure.exception(graph, t); } }; int sId = querySupport.getId(subject); -// if(AsyncBarrierImpl.BOOKKEEPING) impl.state.barrier.inc(ip, "#Types" + sId); -// else impl.state.barrier.inc(null, null); - - Types.queryEach(impl, sId, this, impl.parent, listener, ip); + try { + QueryCache.runnerTypes(impl, sId, impl.parent, listener, ip); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } @@ -4249,53 +3477,37 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap assert(subject != null); - return Types.queryEach2(impl, querySupport.getId(subject), this, impl.parent); + return QueryCacheBase.resultTypes(impl, querySupport.getId(subject), impl.parent, null); } @Override final public void forRelationInfo(final ReadGraphImpl impl, final Resource subject, final AsyncProcedure procedure) { - + assert(subject != null); assert(procedure != null); final ListenerBase listener = getListenerBase(procedure); + assert(listener == null); -// impl.state.barrier.inc(); - - RelationInfoQuery.queryEach(impl, querySupport.getId(subject), this, impl.parent, listener, new InternalProcedure() { - - AtomicBoolean first = new AtomicBoolean(true); + try { + + QueryCache.runnerRelationInfoQuery(impl, querySupport.getId(subject), impl.parent, listener, new InternalProcedure() { - @Override - public void execute(final ReadGraphImpl graph, RelationInfo set) { - try { - if(first.compareAndSet(true, false)) { - procedure.execute(graph, set); -// impl.state.barrier.dec(); - } else { - procedure.execute(impl.newRestart(graph), set); - } - } catch (Throwable t2) { - Logger.defaultLogError(t2); + @Override + public void execute(final ReadGraphImpl graph, RelationInfo set) { + procedure.execute(graph, set); } - } - @Override - public void exception(ReadGraphImpl graph, Throwable t) { - try { - if(first.compareAndSet(true, false)) { - procedure.exception(graph, t); -// impl.state.barrier.dec("ReadGraphSupportImpl.1353"); - } else { - procedure.exception(impl.newRestart(graph), t); - } - } catch (Throwable t2) { - Logger.defaultLogError(t2); + @Override + public void exception(ReadGraphImpl graph, Throwable t) { + procedure.exception(graph, t); } - } - }); + }); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } @@ -4307,14 +3519,13 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap final ListenerBase listener = getListenerBase(procedure); -// impl.state.barrier.inc(); - - SuperTypes.queryEach(impl, querySupport.getId(subject), this, impl.parent, listener, new InternalProcedure() { + try { + QueryCache.runnerSuperTypes(impl, querySupport.getId(subject), impl.parent, listener, new InternalProcedure() { - AtomicBoolean first = new AtomicBoolean(true); + AtomicBoolean first = new AtomicBoolean(true); - @Override - public void execute(final ReadGraphImpl graph, IntSet set) { + @Override + public void execute(final ReadGraphImpl graph, IntSet set) { // final HashSet result = new HashSet(); // set.forEach(new TIntProcedure() { // @@ -4325,33 +3536,36 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap // } // // }); - try { - if(first.compareAndSet(true, false)) { - procedure.execute(graph, set); + try { + if(first.compareAndSet(true, false)) { + procedure.execute(graph, set); // impl.state.barrier.dec(); - } else { - procedure.execute(impl.newRestart(graph), set); + } else { + procedure.execute(impl.newRestart(graph), set); + } + } catch (Throwable t2) { + Logger.defaultLogError(t2); } - } catch (Throwable t2) { - Logger.defaultLogError(t2); } - } - @Override - public void exception(ReadGraphImpl graph, Throwable t) { - try { - if(first.compareAndSet(true, false)) { - procedure.exception(graph, t); + @Override + public void exception(ReadGraphImpl graph, Throwable t) { + try { + if(first.compareAndSet(true, false)) { + procedure.exception(graph, t); // impl.state.barrier.dec(); - } else { - procedure.exception(impl.newRestart(graph), t); + } else { + procedure.exception(impl.newRestart(graph), t); + } + } catch (Throwable t2) { + Logger.defaultLogError(t2); } - } catch (Throwable t2) { - Logger.defaultLogError(t2); } - } - }); + }); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } @@ -4402,7 +3616,13 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap // if(AsyncBarrierImpl.BOOKKEEPING) impl.state.barrier.inc(ip, "#DirectSuperRelations#" + sId); // else impl.state.barrier.inc(null, null); - DirectSuperRelations.queryEach(impl, sId, this, impl.parent, listener, ip); + try { + QueryCache.runnerDirectSuperRelations(impl, sId, impl.parent, listener, ip); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } + +// DirectSuperRelations.queryEach(impl, sId, this, impl.parent, listener, ip); } @@ -4467,31 +3687,31 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap // if(AsyncBarrierImpl.BOOKKEEPING) impl.state.barrier.inc(ip, "#SuperRelations#" + sId); // else impl.state.barrier.inc(null, null); - SuperRelations.queryEach(impl, sId, this, impl.parent, listener, ip); + try { + QueryCache.runnerSuperRelations(impl, sId, impl.parent, listener, ip); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } final public byte[] getValue(final ReadGraphImpl impl, final Resource subject) throws DatabaseException { - - int sId = querySupport.getId(subject); - return ValueQuery.queryEach(impl, sId, impl.parent); - + return getValue(impl, querySupport.getId(subject)); } final public byte[] getValue(final ReadGraphImpl impl, final int subject) throws DatabaseException { - - return ValueQuery.queryEach(impl, subject, impl.parent); - + return QueryCache.resultValueQuery(impl, subject, impl.parent, null); } @Override - final public byte[] forValue(final ReadGraphImpl impl, final Resource subject, final AsyncProcedure procedure) { + final public void forValue(final ReadGraphImpl impl, final Resource subject, final AsyncProcedure procedure) { assert(subject != null); + assert(procedure != null); int sId = querySupport.getId(subject); - if(procedure != null) { +// if(procedure != null) { final ListenerBase listener = getListenerBase(procedure); @@ -4532,13 +3752,19 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap // if(AsyncBarrierImpl.BOOKKEEPING) impl.state.barrier.inc(ip, "#Value" + sId); // else impl.state.barrier.inc(null, null); - return ValueQuery.queryEach(impl, sId, impl.parent, listener, ip); - - } else { + try { + QueryCache.runnerValueQuery(impl, sId, impl.parent, listener, ip); + } catch (DatabaseException e) { + throw new IllegalStateException("Internal error"); + } - return ValueQuery.queryEach(impl, sId, impl.parent, null, null); - - } +// } else { +// +// return QueryCacheBase.runnerValueQuery(impl, sId, impl.parent, null, null); +// +// } +// +// throw new IllegalStateException("Internal error"); } @@ -4591,7 +3817,11 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap // if(AsyncBarrierImpl.BOOKKEEPING) impl.state.barrier.inc(ip, "#Value" + sId); // else impl.state.barrier.inc(null, null); - ValueQuery.queryEach(impl, sId, impl.parent, listener, ip); + try { + QueryCache.runnerValueQuery(impl, sId, impl.parent, listener, ip); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } else { @@ -4615,7 +3845,11 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap int sId = querySupport.getId(subject); - ValueQuery.queryEach(impl, sId, impl.parent, listener, ip); + try { + QueryCache.runnerValueQuery(impl, sId, impl.parent, listener, ip); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } @@ -4695,7 +3929,11 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap // if(AsyncBarrierImpl.BOOKKEEPING) impl.state.barrier.inc(ip, "#DirectObjects#" + sId); // else impl.state.barrier.inc(null, null); - Objects.runner(impl, sId, getInverseOf(), impl.parent, listener, ip); + try { + QueryCache.runnerObjects(impl, sId, getInverseOf(), impl.parent, listener, ip); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } @@ -4745,29 +3983,33 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap // impl.state.barrier.inc(); - forBuiltin(impl, id, impl.parent, new InternalProcedure() { + try { + forBuiltin(impl, id, impl.parent, new InternalProcedure() { - @Override - public void execute(ReadGraphImpl graph, Integer result) { - try { - procedure.execute(graph, querySupport.getResource(result)); - } catch (Throwable t2) { - Logger.defaultLogError(t2); - } + @Override + public void execute(ReadGraphImpl graph, Integer result) { + try { + procedure.execute(graph, querySupport.getResource(result)); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } // impl.state.barrier.dec(); - } + } - @Override - public void exception(ReadGraphImpl graph, Throwable t) { - try { - procedure.exception(graph, t); - } catch (Throwable t2) { - Logger.defaultLogError(t2); - } + @Override + public void exception(ReadGraphImpl graph, Throwable t) { + try { + procedure.exception(graph, t); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } // impl.state.barrier.dec(); - } + } - }); + }); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } @@ -4779,39 +4021,13 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap final ListenerBase listener = getListenerBase(procedure); -// impl.state.barrier.inc(); - - DirectPredicates.queryEach(impl, querySupport.getId(subject), this, impl.parent, listener, new IntProcedure() { - - boolean found = false; - - @Override - public void execute(ReadGraphImpl graph, int object) { - found = true; - } - - @Override - public void finished(ReadGraphImpl graph) { - try { - procedure.execute(graph, found); - } catch (Throwable t2) { - Logger.defaultLogError(t2); - } -// impl.state.barrier.dec(); - } - - @Override - public void exception(ReadGraphImpl graph, Throwable t) { - try { - procedure.exception(graph, t); - } catch (Throwable t2) { - Logger.defaultLogError(t2); - } -// impl.state.barrier.dec(); - } - - }); - + try { + IntSet result = QueryCache.resultDirectPredicates(impl, querySupport.getId(subject), impl.parent, listener); + procedure.execute(impl, !result.isEmpty()); + } catch (DatabaseException e) { + procedure.exception(impl, e); + } + } @Override @@ -4911,30 +4127,34 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap // impl.state.barrier.inc(); - ValueQuery.queryEach(impl, querySupport.getId(subject), impl.parent, listener, new InternalProcedure() { + try { + QueryCache.runnerValueQuery(impl, querySupport.getId(subject), impl.parent, listener, new InternalProcedure() { - @Override - public void execute(ReadGraphImpl graph, byte[] object) { - boolean result = object != null; - try { - procedure.execute(graph, result); - } catch (Throwable t2) { - Logger.defaultLogError(t2); - } + @Override + public void execute(ReadGraphImpl graph, byte[] object) { + boolean result = object != null; + try { + procedure.execute(graph, result); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } // impl.state.barrier.dec(); - } - - @Override - public void exception(ReadGraphImpl graph, Throwable t) { - try { - procedure.exception(graph, t); - } catch (Throwable t2) { - Logger.defaultLogError(t2); } + + @Override + public void exception(ReadGraphImpl graph, Throwable t) { + try { + procedure.exception(graph, t); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } // impl.state.barrier.dec(); - } + } - }); + }); + } catch (DatabaseException e) { + Logger.defaultLogError(e); + } } @@ -4946,139 +4166,139 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap final ListenerBase listener = getListenerBase(procedure); -// impl.state.barrier.inc(); - - OrderedSet.queryEach(impl, querySupport.getId(subject), this, impl.parent, listener, new IntProcedure() { + try { + + QueryCache.runnerOrderedSet(impl, querySupport.getId(subject), impl.parent, listener, new IntProcedure() { - @Override - public void exception(ReadGraphImpl graph, Throwable t) { - try { - procedure.exception(graph, t); - } catch (Throwable t2) { - Logger.defaultLogError(t2); - } + @Override + public void exception(ReadGraphImpl graph, Throwable t) { + try { + procedure.exception(graph, t); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } // impl.state.barrier.dec(); - } - - @Override - public void execute(ReadGraphImpl graph, int i) { - try { - procedure.execute(graph, querySupport.getResource(i)); - } catch (Throwable t2) { - Logger.defaultLogError(t2); } - } - @Override - public void finished(ReadGraphImpl graph) { - try { - procedure.finished(graph); - } catch (Throwable t2) { - Logger.defaultLogError(t2); + @Override + public void execute(ReadGraphImpl graph, int i) { + try { + procedure.execute(graph, querySupport.getResource(i)); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } } -// impl.state.barrier.dec(); - } - - }); - - } - - @Override - final public void query(final ReadGraphImpl impl, final AsyncRead request, final CacheEntry parent, final AsyncProcedure procedure, ListenerBase listener) { - - assert(request != null); - assert(procedure != null); -// if(AsyncBarrierImpl.BOOKKEEPING) impl.state.barrier.inc(request, "#" + request.toString() + ".1999"); -// else impl.state.barrier.inc(null, null); - - runAsyncRead(impl, request, parent, listener, procedure); - - } - - @Override - final public T tryQuery(final ReadGraphImpl graph, final Read request) throws DatabaseException { - - assert(graph != null); - assert(request != null); + @Override + public void finished(ReadGraphImpl graph) { + try { + procedure.finished(graph); + } catch (Throwable t2) { + Logger.defaultLogError(t2); + } +// impl.state.barrier.dec(); + } - final ReadEntry entry = cache.readMap.get(request); - if(entry != null && entry.isReady()) { - return (T)entry.get(graph, this, null); - } else { - return request.perform(graph); + }); + } catch (DatabaseException e) { + Logger.defaultLogError(e); } } - final public T tryQuery(final ReadGraphImpl graph, final ExternalRead request) throws DatabaseException { - - assert(graph != null); - assert(request != null); - - final ExternalReadEntry entry = cache.externalReadMap.get(request); - if(entry != null && entry.isReady()) { - if(entry.isExcepted()) { - Throwable t = (Throwable)entry.getResult(); - if(t instanceof DatabaseException) throw (DatabaseException)t; - else throw new DatabaseException(t); - } else { - return (T)entry.getResult(); - } - } else { - - final DataContainer result = new DataContainer(); - final DataContainer exception = new DataContainer(); - - request.register(graph, new Listener() { - - @Override - public void exception(Throwable t) { - exception.set(t); - } - - @Override - public void execute(T t) { - result.set(t); - } - - @Override - public boolean isDisposed() { - return true; - } - - }); - - Throwable t = exception.get(); - if(t != null) { - if(t instanceof DatabaseException) throw (DatabaseException)t; - else throw new DatabaseException(t); - } - - return result.get(); +// @Override +// final public void query(final ReadGraphImpl impl, final AsyncRead request, final CacheEntry parent, final AsyncProcedure procedure, ListenerBase listener) throws DatabaseException { +// +// assert(request != null); +// assert(procedure != null); +// +// QueryCache.runnerAsyncReadEntry(impl, request, parent, listener, procedure); +// +// } - } +// @Override +// final public T tryQuery(final ReadGraphImpl graph, final Read request) throws DatabaseException { +// +// assert(graph != null); +// assert(request != null); +// +// final ReadEntry entry = (ReadEntry)cache.getCached(request); +// if(entry != null && entry.isReady()) { +// return (T)entry.get(graph, this, null); +// } else { +// return request.perform(graph); +// } +// +// } - } +// final public T tryQuery(final ReadGraphImpl graph, final ExternalRead request) throws DatabaseException { +// +// assert(graph != null); +// assert(request != null); +// +// final ExternalReadEntry entry = cache.externalReadMap.get(request); +// if(entry != null && entry.isReady()) { +// if(entry.isExcepted()) { +// Throwable t = (Throwable)entry.getResult(); +// if(t instanceof DatabaseException) throw (DatabaseException)t; +// else throw new DatabaseException(t); +// } else { +// return (T)entry.getResult(); +// } +// } else { +// +// final DataContainer result = new DataContainer(); +// final DataContainer exception = new DataContainer(); +// +// request.register(graph, new Listener() { +// +// @Override +// public void exception(Throwable t) { +// exception.set(t); +// } +// +// @Override +// public void execute(T t) { +// result.set(t); +// } +// +// @Override +// public boolean isDisposed() { +// return true; +// } +// +// }); +// +// Throwable t = exception.get(); +// if(t != null) { +// if(t instanceof DatabaseException) throw (DatabaseException)t; +// else throw new DatabaseException(t); +// } +// +// return result.get(); +// +// } +// +// } - @Override - final public void tryQuery(final ReadGraphImpl graph, final AsyncRead request, AsyncProcedure procedure) { - - assert(graph != null); - assert(request != null); - - final AsyncReadEntry entry = cache.asyncReadMap.get(request); - if(entry != null && entry.isReady()) { - if(entry.isExcepted()) { - procedure.exception(graph, (Throwable)entry.getResult()); - } else { - procedure.execute(graph, (T)entry.getResult()); - } - } else { - request.perform(graph, procedure); - } - - } +// @Override +// final public void tryQuery(final ReadGraphImpl graph, final AsyncRead request, AsyncProcedure procedure) { +// +// assert(graph != null); +// assert(request != null); +// +// final AsyncReadEntry entry = cache.asyncReadMap.get(request); +// if(entry != null && entry.isReady()) { +// if(entry.isExcepted()) { +// procedure.exception(graph, (Throwable)entry.getResult()); +// } else { +// procedure.execute(graph, (T)entry.getResult()); +// } +// } else { +// request.perform(graph, procedure); +// } +// +// } @Override final public void query(final ReadGraphImpl impl, final MultiRead request, final CacheEntry parent, final AsyncMultiProcedure procedure, ListenerBase listener) { @@ -5152,48 +4372,48 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap } - @Override - final public void query(final ReadGraphImpl impl, final ExternalRead request, final CacheEntry parent, final Procedure procedure, ListenerBase listener) { - - assert(request != null); - assert(procedure != null); - - try { - - queryPrimitiveRead(impl, request, parent, listener, new Procedure() { - - @Override - public void execute(T result) { - try { - procedure.execute(result); - } catch (Throwable t2) { - Logger.defaultLogError(t2); - } - } - - @Override - public String toString() { - return procedure.toString(); - } - - @Override - public void exception(Throwable t) { - try { - procedure.exception(t); - } catch (Throwable t2) { - Logger.defaultLogError(t2); - } - } - - }); - - } catch (DatabaseException e) { - - throw new IllegalStateException(e); - - } - - } +// @Override +// final public void query(final ReadGraphImpl impl, final ExternalRead request, final CacheEntry parent, final Procedure procedure, ListenerBase listener) throws DatabaseException { +// +// assert(request != null); +// assert(procedure != null); +// +// try { +// +// queryPrimitiveRead(impl, request, parent, listener, new AsyncProcedure() { +// +// @Override +// public String toString() { +// return procedure.toString(); +// } +// +// @Override +// public void execute(AsyncReadGraph graph, T result) { +// try { +// procedure.execute(result); +// } catch (Throwable t2) { +// Logger.defaultLogError(t2); +// } +// } +// +// @Override +// public void exception(AsyncReadGraph graph, Throwable throwable) { +// try { +// procedure.exception(throwable); +// } catch (Throwable t2) { +// Logger.defaultLogError(t2); +// } +// } +// +// }); +// +// } catch (DatabaseException e) { +// +// throw new IllegalStateException(e); +// +// } +// +// } @Override public VirtualGraph getProvider(Resource subject, Resource predicate, Resource object) {