-package fi.vtt.simantics.procore.internal;\r
-\r
-import java.util.Collection;\r
-\r
-import org.simantics.db.AsyncReadGraph;\r
-import org.simantics.db.ReadGraph;\r
-import org.simantics.db.Resource;\r
-import org.simantics.db.WriteGraph;\r
-import org.simantics.db.common.request.WriteRequest;\r
-import org.simantics.db.exception.DatabaseException;\r
-import org.simantics.db.impl.ResourceImpl;\r
-import org.simantics.db.impl.graph.ReadGraphImpl;\r
-import org.simantics.db.procedure.AsyncContextMultiProcedure;\r
-import org.simantics.db.procedure.AsyncMultiProcedure;\r
-import org.simantics.db.request.ExternalRead;\r
-import org.simantics.db.service.QueryControl;\r
-import org.simantics.utils.DataContainer;\r
-\r
-public class QueryControlImpl implements QueryControl {\r
- \r
- final private SessionImplSocket session;\r
- \r
- QueryControlImpl(SessionImplSocket session) {\r
- this.session = session;\r
- }\r
-\r
- @Override\r
- public int getAmountOfQueryThreads() {\r
- return session.getAmountOfQueryThreads();\r
- }\r
- \r
- @Override\r
- public int getGraphThread(AsyncReadGraph graph) {\r
- return 0;\r
- }\r
-\r
- @Override\r
- public int flush() {\r
- final DataContainer<Integer> result = new DataContainer<Integer>(); \r
- try {\r
- session.syncRequest(new WriteRequest() {\r
-\r
- @Override\r
- public void perform(WriteGraph graph) throws DatabaseException {\r
- result.set(session.queryProvider2.clean());\r
- }\r
- \r
- });\r
- } catch (DatabaseException e) {\r
- e.printStackTrace();\r
- }\r
- return result.get();\r
- }\r
-\r
- @Override\r
- public int flush(ReadGraph graph) {\r
- return session.queryProvider2.clean();\r
- }\r
- \r
- @Override\r
- public int count() {\r
- return session.queryProvider2.querySize();\r
- }\r
- \r
- @Override\r
- public void gc(ReadGraph graph, int allowedTimeInMs) {\r
- // 20% young target\r
- session.queryProvider2.gc(20, allowedTimeInMs);\r
- }\r
- \r
- @Override\r
- public void gc(final Collection<ExternalRead<?>> requests) {\r
- try {\r
- session.syncRequest(new WriteRequest() {\r
- @Override\r
- public void perform(WriteGraph graph) throws DatabaseException {\r
- gc(graph, requests);\r
- }\r
- });\r
- } catch (DatabaseException e) {\r
- e.printStackTrace();\r
- }\r
- }\r
-\r
- @Override\r
- public void gc(WriteGraph graph, final Collection<ExternalRead<?>> requests) {\r
- if (graph == null)\r
- throw new IllegalArgumentException("null WriteGraph");\r
- if (requests == null)\r
- throw new IllegalArgumentException("null requests");\r
- session.queryProvider2.clean(requests);\r
- }\r
- \r
- @Override\r
- public boolean scheduleByCluster(AsyncReadGraph graph, final Resource resource, final AsyncMultiProcedure<Resource> procedure) {\r
- final ReadGraphImpl impl = (ReadGraphImpl)graph;\r
- ResourceImpl res = (ResourceImpl)resource;\r
- int targetThread = ((res.id >>> 16) & session.queryProvider2.THREAD_MASK);\r
- if(0 == targetThread) return true;\r
- //System.err.println("scheduleByCluster[" + res.id + "|" + (res.id>>>16) + "] " + impl.callerThread + " -> " + targetThread);\r
-// impl.state.barrier.inc();\r
-// \r
-// AsyncReadGraph targetGraph = impl.newAsync();\r
- procedure.execute(impl, resource);\r
-// impl.state.barrier.dec();\r
- \r
- return false;\r
- \r
- }\r
-\r
- @Override\r
- public <C> boolean scheduleByCluster(AsyncReadGraph graph, final Resource resource, final C context, final AsyncContextMultiProcedure<C, Resource> procedure) {\r
- final ReadGraphImpl impl = (ReadGraphImpl)graph;\r
- ResourceImpl res = (ResourceImpl)resource;\r
- int targetThread = ((res.id >>> 16) & session.queryProvider2.THREAD_MASK);\r
- if(0 == targetThread) return true;\r
- //System.err.println("scheduleByCluster[" + res.id + "|" + (res.id>>>16) + "] " + impl.callerThread + " -> " + targetThread);\r
-// impl.state.barrier.inc();\r
-// \r
-// AsyncReadGraph targetGraph = impl.newAsync();\r
- procedure.execute(impl, context, resource);\r
-// impl.state.barrier.dec();\r
- \r
- return false;\r
- }\r
-\r
- @Override\r
- public void schedule(AsyncReadGraph graph, int targetThread, final ControlProcedure procedure) {\r
- final ReadGraphImpl impl = (ReadGraphImpl)graph;\r
-\r
-// impl.state.barrier.inc();\r
-//\r
-// AsyncReadGraph targetGraph = impl.newAsync();\r
- procedure.execute(impl);\r
-// impl.state.barrier.dec();\r
-\r
- }\r
- \r
- @Override\r
- public ReadGraph getIndependentGraph(ReadGraph graph) {\r
- ReadGraphImpl impl = (ReadGraphImpl)graph;\r
- return impl.withParent(null);\r
- }\r
-\r
- @Override\r
- public boolean hasParentRequest(ReadGraph graph) {\r
- ReadGraphImpl impl = (ReadGraphImpl)graph;\r
- return impl.parent != null;\r
- }\r
- \r
- @Override\r
- public boolean resume(AsyncReadGraph graph) {\r
- ReadGraphImpl impl = (ReadGraphImpl)graph;\r
- return impl.processor.querySupport.resume(impl);\r
- }\r
-\r
-}\r
+package fi.vtt.simantics.procore.internal;
+
+import java.util.Collection;
+
+import org.simantics.db.AsyncReadGraph;
+import org.simantics.db.ReadGraph;
+import org.simantics.db.Resource;
+import org.simantics.db.WriteGraph;
+import org.simantics.db.common.request.WriteRequest;
+import org.simantics.db.exception.DatabaseException;
+import org.simantics.db.impl.ResourceImpl;
+import org.simantics.db.impl.graph.ReadGraphImpl;
+import org.simantics.db.procedure.AsyncContextMultiProcedure;
+import org.simantics.db.procedure.AsyncMultiProcedure;
+import org.simantics.db.request.ExternalRead;
+import org.simantics.db.request.Read;
+import org.simantics.db.service.QueryControl;
+import org.simantics.utils.DataContainer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class QueryControlImpl implements QueryControl {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(QueryControlImpl.class);
+
+ final private SessionImplSocket session;
+
+ QueryControlImpl(SessionImplSocket session) {
+ this.session = session;
+ }
+
+ @Override
+ public int getAmountOfQueryThreads() {
+ return session.getAmountOfQueryThreads();
+ }
+
+ @Override
+ public int getGraphThread(AsyncReadGraph graph) {
+ return 0;
+ }
+
+ @Override
+ public int flush() {
+ final DataContainer<Integer> result = new DataContainer<Integer>();
+ try {
+ session.syncRequest(new WriteRequest() {
+
+ @Override
+ public void perform(WriteGraph graph) throws DatabaseException {
+ result.set(session.queryProvider2.clean());
+ }
+
+ });
+ } catch (DatabaseException e) {
+ LOGGER.error("query flush failed", e);
+ }
+ return result.get();
+ }
+
+ @Override
+ public int flush(ReadGraph graph) {
+ return session.queryProvider2.clean();
+ }
+
+ @Override
+ public int count() {
+ return session.queryProvider2.querySize();
+ }
+
+ @Override
+ public void gc(ReadGraph graph, int allowedTimeInMs) {
+ // 20% young target
+ session.queryProvider2.gc(20, allowedTimeInMs);
+ }
+
+ @Override
+ public void gc(final Collection<ExternalRead<?>> requests) {
+ try {
+ session.syncRequest(new WriteRequest() {
+ @Override
+ public void perform(WriteGraph graph) throws DatabaseException {
+ gc(graph, requests);
+ }
+ });
+ } catch (DatabaseException e) {
+ LOGGER.error("query gc failed", e);
+ }
+ }
+
+ @Override
+ public void gc(WriteGraph graph, final Collection<ExternalRead<?>> requests) {
+ if (graph == null)
+ throw new IllegalArgumentException("null WriteGraph");
+ if (requests == null)
+ throw new IllegalArgumentException("null requests");
+ session.queryProvider2.clean(requests);
+ }
+
+ @Override
+ public boolean scheduleByCluster(AsyncReadGraph graph, final Resource resource, final AsyncMultiProcedure<Resource> procedure) {
+ final ReadGraphImpl impl = (ReadGraphImpl)graph;
+ ResourceImpl res = (ResourceImpl)resource;
+ int targetThread = ((res.id >>> 16) & session.queryProvider2.THREAD_MASK);
+ if(0 == targetThread) return true;
+ //System.err.println("scheduleByCluster[" + res.id + "|" + (res.id>>>16) + "] " + impl.callerThread + " -> " + targetThread);
+// impl.state.barrier.inc();
+//
+// AsyncReadGraph targetGraph = impl.newAsync();
+ procedure.execute(impl, resource);
+// impl.state.barrier.dec();
+
+ return false;
+
+ }
+
+ @Override
+ public <C> boolean scheduleByCluster(AsyncReadGraph graph, final Resource resource, final C context, final AsyncContextMultiProcedure<C, Resource> procedure) {
+ final ReadGraphImpl impl = (ReadGraphImpl)graph;
+ ResourceImpl res = (ResourceImpl)resource;
+ int targetThread = ((res.id >>> 16) & session.queryProvider2.THREAD_MASK);
+ if(0 == targetThread) return true;
+ //System.err.println("scheduleByCluster[" + res.id + "|" + (res.id>>>16) + "] " + impl.callerThread + " -> " + targetThread);
+// impl.state.barrier.inc();
+//
+// AsyncReadGraph targetGraph = impl.newAsync();
+ procedure.execute(impl, context, resource);
+// impl.state.barrier.dec();
+
+ return false;
+ }
+
+ @Override
+ public void schedule(AsyncReadGraph graph, int targetThread, final ControlProcedure procedure) {
+ final ReadGraphImpl impl = (ReadGraphImpl)graph;
+
+// impl.state.barrier.inc();
+//
+// AsyncReadGraph targetGraph = impl.newAsync();
+ procedure.execute(impl);
+// impl.state.barrier.dec();
+
+ }
+
+ @Override
+ public ReadGraph getIndependentGraph(ReadGraph graph) {
+ ReadGraphImpl impl = (ReadGraphImpl)graph;
+ return impl.withParent(null, null, false);
+ }
+
+ @Override
+ public <T> T syncRequestIndependent(ReadGraph graph, Read<T> request) throws DatabaseException {
+ ReadGraphImpl independent = ((ReadGraphImpl)graph).withParent(null, null, false);
+ independent.asyncBarrier.inc();
+ try {
+ return independent.syncRequest(request);
+ } finally {
+ independent.asyncBarrier.dec();
+ }
+ }
+
+ @Override
+ public boolean hasParentRequest(ReadGraph graph) {
+ ReadGraphImpl impl = (ReadGraphImpl)graph;
+ return impl.parent != null;
+ }
+
+ @Override
+ public boolean resume(ReadGraph graph) {
+ ReadGraphImpl impl = (ReadGraphImpl)graph;
+ return impl.processor.querySupport.resume(impl);
+ }
+
+}
Code Review / simantics / platform.git / blobdiff