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=a2fff0515b1975f98484ca969005ff40b5c32113;hpb=ad18bf76b19bec530bc68c6041ab5fb245b8ee4d;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 a2fff0515..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; @@ -150,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 { @@ -162,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; @@ -177,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) { @@ -203,19 +206,43 @@ 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; @@ -293,26 +339,26 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap 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; @@ -524,18 +570,18 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap } // Fall back to using the fixed builtins. - result = querySupport.getBuiltin(id); - if (result != 0) { - procedure.execute(graph, result); - return; - } +// 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); @@ -552,7 +598,17 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap }); } catch (DatabaseException e) { - Logger.defaultLogError(e); + + try { + + procedure.exception(graph, e); + + } catch (DatabaseException e1) { + + Logger.defaultLogError(e1); + + } + } } @@ -593,12 +649,12 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap 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); - - } +// @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 { @@ -1340,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 @@ -2324,11 +2380,8 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap @Override final public IntSet getPredicates(final ReadGraphImpl impl, final Resource subject) throws Throwable { - - return QueryCacheBase.resultPredicates(impl, querySupport.getId(subject), impl.parent, null, null); - + return QueryCacheBase.resultPredicates(impl, querySupport.getId(subject), impl.parent, null); } - @Override final public void forEachStatement(final ReadGraphImpl impl, final Resource subject, @@ -3393,37 +3446,18 @@ 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); } }; @@ -3443,50 +3477,31 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap assert(subject != null); - return QueryCacheBase.resultTypes(impl, querySupport.getId(subject), impl.parent, null, null); + 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); try { QueryCache.runnerRelationInfoQuery(impl, querySupport.getId(subject), impl.parent, listener, new InternalProcedure() { - AtomicBoolean first = new AtomicBoolean(true); - @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); - } + 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); - } + procedure.exception(graph, t); } }); @@ -3685,17 +3700,18 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap } final public byte[] getValue(final ReadGraphImpl impl, final int subject) throws DatabaseException { - return QueryCache.resultValueQuery(impl, subject, impl.parent, null, null); + 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); @@ -3737,23 +3753,18 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap // else impl.state.barrier.inc(null, null); try { - return QueryCacheBase.resultValueQuery(impl, sId, impl.parent, listener, ip); + QueryCache.runnerValueQuery(impl, sId, impl.parent, listener, ip); } catch (DatabaseException e) { - Logger.defaultLogError(e); - } - - - } else { + throw new IllegalStateException("Internal error"); + } - try { - return QueryCacheBase.resultValueQuery(impl, sId, impl.parent, null, null); - } catch (DatabaseException e) { - Logger.defaultLogError(e); - } - - } - - throw new IllegalStateException("Internal error"); +// } else { +// +// return QueryCacheBase.runnerValueQuery(impl, sId, impl.parent, null, null); +// +// } +// +// throw new IllegalStateException("Internal error"); } @@ -4011,7 +4022,7 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap final ListenerBase listener = getListenerBase(procedure); try { - IntSet result = QueryCache.resultDirectPredicates(impl, querySupport.getId(subject), impl.parent, listener,QueryCache.emptyProcedureDirectPredicates); + IntSet result = QueryCache.resultDirectPredicates(impl, querySupport.getId(subject), impl.parent, listener); procedure.execute(impl, !result.isEmpty()); } catch (DatabaseException e) { procedure.exception(impl, e); @@ -4195,15 +4206,15 @@ final public class QueryProcessor extends AbstractDisposable implements ReadGrap } - @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 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 { @@ -4361,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 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 +// 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) {