-package org.simantics.db.impl.query;\r
-\r
-import org.simantics.db.impl.query.QueryProcessor.QueryCollectorSupport;\r
-\r
-class QueryCollectorImpl implements QueryProcessor.QueryCollector {\r
- \r
- private static final boolean DEBUG = false;\r
- private static final boolean DEBUG_STATUS = false;\r
-\r
- private final QueryProcessor queryProcessor;\r
-\r
- final private QueryCollectorSupport support;\r
-\r
- /*\r
- * Set to true: at end of iteration if moreAll is false\r
- * Set to false: upon gc start and whenever a prospect is added.\r
- * \r
- */\r
- boolean doneAll = false;\r
- /*\r
- * Set to true: upon gc start and whenever a prospect is added.\r
- * Set to false: at end of entry iteration\r
- * => if this is already false at end of entry iteration => no more work to be done\r
- */\r
- int moreAll = 0;\r
- \r
- /*\r
- * Change propagation is in progress\r
- */\r
- boolean propagate = true;\r
- \r
- \r
- private static final int COLLECT_N = 1000;\r
- \r
- private long spent = 0;\r
- \r
- QueryCollectorImpl(QueryProcessor queryProcessor, QueryCollectorSupport support) {\r
- this.queryProcessor = queryProcessor;\r
- this.support = support;\r
- }\r
- \r
- @Override\r
- public void collect(int youngTarget, int allowedTimeInMs) {\r
-\r
- long start = System.nanoTime();\r
-\r
- // Refresh current size\r
- int size = support.calculateCurrentSize();\r
- int bound = queryProcessor.boundQueries;\r
- int young = size - bound;\r
- int youngPct = size > 0 ? 100*young / size : 0;\r
- \r
- // Initialize support for new run\r
- // If support returns 0 we are starting from 0\r
- if(support.start(youngTarget == 0)) {\r
- \r
- moreAll = 0;\r
-\r
- // We monitor all prospects here\r
- CacheEntryBase prospect = support.iterate(0);\r
- while(prospect != null) {\r
- if(prospect.isDiscarded()) {\r
- support.remove();\r
- propagate = true;\r
- } else {\r
- CacheEntry parent = prospect.getFirstParent(queryProcessor);\r
- if(parent == null) {\r
- tryCollect(prospect);\r
- } else {\r
- support.setLevel(prospect, parent.getLevel() + 1);\r
- }\r
- }\r
- prospect = support.iterate(0);\r
- }\r
-\r
- // If no prospects were collected and most of the queries are old we can stop here\r
- if(!propagate && youngPct < youngTarget) {\r
-// System.err.println("collect2 skipped");\r
-// System.err.println("-size=" + size);\r
-// System.err.println("-young=" + young);\r
- return; \r
- }\r
- \r
- }\r
- \r
- long test = (long)allowedTimeInMs*1000000;\r
-\r
- start();\r
- \r
- while(true) {\r
- \r
- size = support.getCurrentSize();\r
- bound = queryProcessor.boundQueries;\r
- \r
- long elapsed = System.nanoTime()-start;\r
- boolean timeCondition = elapsed > test; \r
-\r
- if(doneAll || timeCondition) {\r
- \r
- spent += elapsed;\r
- \r
- if(DEBUG_STATUS)\r
- System.err.println("Query collector used " + 1e-9*elapsed + "s total queries: " + size + " bound queries: " + bound + " spent: " + (double)spent*1e-9);\r
- \r
- return;\r
- \r
- }\r
-\r
- // Iterate all entries and update sets\r
- if(!doneAll) {\r
-// long start2 = System.nanoTime();\r
- for(int i=0;i<COLLECT_N;i++) {\r
- CacheEntryBase entry = support.iterate(Integer.MAX_VALUE);\r
- if(entry == null) {\r
- if(DEBUG) {\r
- System.err.println("finished iteration");\r
- System.err.println("-moreAll=" + moreAll);\r
- }\r
- if(moreAll < 1000) {\r
- doneAll = true;\r
- propagate = false;\r
- }\r
- moreAll = 0;\r
- break;\r
- }\r
- \r
- CacheEntry parent = entry.getFirstParent(queryProcessor);\r
- if(parent == null) {\r
- \r
- boolean collected = tryCollect(entry);\r
- if(!collected) {\r
- entry.setLevel((short)0);\r
- }\r
- \r
- } else {\r
-\r
- parent = entry.pruneFirstParents();\r
- if(parent == null) {\r
- \r
- boolean collected = tryCollect(entry);\r
- if(!collected) {\r
- entry.setLevel((short)0);\r
- }\r
- \r
- } else {\r
- \r
- support.setLevel(entry, parent.getLevel() + 1);\r
- \r
- }\r
- \r
- }\r
- \r
- int status = entry.getGCStatus(); \r
- if((status & CacheEntry.HAS_BEEN_BOUND) == 0) {\r
- \r
- if(parent != null && !parent.isDiscarded()) {\r
- if((parent.getGCStatus() & CacheEntry.HAS_BEEN_BOUND) != 0) {\r
- queryProcessor.boundQueries++;\r
- entry.setGCStatusFlag(CacheEntry.HAS_BEEN_BOUND, true);\r
- }\r
- }\r
-\r
- if(queryProcessor.hasListenerAfterDisposing(entry)) {\r
- if((status & CacheEntry.HAS_BEEN_BOUND) == 0) {\r
- queryProcessor.boundQueries++;\r
- entry.setGCStatusFlag(CacheEntry.HAS_BEEN_BOUND, true);\r
- }\r
- }\r
- }\r
- \r
- }\r
- }\r
-\r
- }\r
-\r
- }\r
-\r
- private boolean tryCollect(CacheEntry entry) {\r
- if (!queryProcessor.hasListenerAfterDisposing(entry))\r
- if(entry.shouldBeCollected()) {\r
- queryProcessor.removeQuery(entry);\r
- support.remove();\r
- propagate = true;\r
- moreAll++;\r
- doneAll = false;\r
- return true;\r
- }\r
- return false;\r
- }\r
-\r
- private void start() {\r
- moreAll = 1;\r
- doneAll = false;\r
- }\r
-\r
+package org.simantics.db.impl.query;
+
+import org.simantics.db.impl.query.QueryProcessor.QueryCollectorSupport;
+
+class QueryCollectorImpl implements QueryProcessor.QueryCollector {
+
+ private static final boolean DEBUG = false;
+ private static final boolean DEBUG_STATUS = false;
+
+ private final QueryProcessor queryProcessor;
+
+ final private QueryCollectorSupport support;
+
+ /*
+ * Set to true: at end of iteration if moreAll is false
+ * Set to false: upon gc start and whenever a prospect is added.
+ *
+ */
+ boolean doneAll = false;
+ /*
+ * Set to true: upon gc start and whenever a prospect is added.
+ * Set to false: at end of entry iteration
+ * => if this is already false at end of entry iteration => no more work to be done
+ */
+ int moreAll = 0;
+
+ /*
+ * Change propagation is in progress
+ */
+ boolean propagate = true;
+
+
+ private static final int COLLECT_N = 1000;
+
+ private long spent = 0;
+
+ QueryCollectorImpl(QueryProcessor queryProcessor, QueryCollectorSupport support) {
+ this.queryProcessor = queryProcessor;
+ this.support = support;
+ }
+
+ @Override
+ public void collect(int youngTarget, int allowedTimeInMs) {
+
+ // Flush listener registrations to prevent the collector from trashing
+ // listeners that are still queued up waiting to be registered.
+ queryProcessor.listening.sync();
+
+ long start = System.nanoTime();
+
+ // Refresh current size
+ int size = support.calculateCurrentSize();
+ int bound = queryProcessor.boundQueries;
+ int young = size - bound;
+ int youngPct = size > 0 ? 100*young / size : 0;
+
+ // Initialize support for new run
+ // If support returns 0 we are starting from 0
+ if(support.start(youngTarget == 0)) {
+
+ moreAll = 0;
+
+ // We monitor all prospects here
+ CacheEntryBase prospect = support.iterate(0);
+ while(prospect != null) {
+ if(prospect.isDiscarded()) {
+ support.remove();
+ propagate = true;
+ } else {
+ CacheEntry parent = prospect.getFirstParent(queryProcessor);
+ if(parent == null) {
+ tryCollect(prospect);
+ } else {
+ support.setLevel(prospect, parent.getLevel() + 1);
+ }
+ }
+ prospect = support.iterate(0);
+ }
+
+ // If no prospects were collected and most of the queries are old we can stop here
+ if(!propagate && youngPct < youngTarget) {
+// System.err.println("collect2 skipped");
+// System.err.println("-size=" + size);
+// System.err.println("-young=" + young);
+ return;
+ }
+
+ }
+
+ long test = (long)allowedTimeInMs*1000000;
+
+ start();
+
+ while(true) {
+
+ size = support.getCurrentSize();
+ bound = queryProcessor.boundQueries;
+
+ long elapsed = System.nanoTime()-start;
+ boolean timeCondition = elapsed > test;
+
+ if(doneAll || timeCondition) {
+
+ spent += elapsed;
+
+ if(DEBUG_STATUS)
+ System.err.println("Query collector used " + 1e-9*elapsed + "s total queries: " + size + " bound queries: " + bound + " spent: " + (double)spent*1e-9);
+
+ return;
+
+ }
+
+ // Iterate all entries and update sets
+ if(!doneAll) {
+// long start2 = System.nanoTime();
+ for(int i=0;i<COLLECT_N;i++) {
+ CacheEntryBase entry = support.iterate(Integer.MAX_VALUE);
+ if(entry == null) {
+ if(DEBUG) {
+ System.err.println("finished iteration");
+ System.err.println("-moreAll=" + moreAll);
+ }
+ if(moreAll < 1000) {
+ doneAll = true;
+ propagate = false;
+ }
+ moreAll = 0;
+ break;
+ }
+
+ CacheEntry parent = entry.getFirstParent(queryProcessor);
+ if(parent == null) {
+
+ boolean collected = tryCollect(entry);
+ if(!collected) {
+ entry.setLevel((short)0);
+ }
+
+ } else {
+
+ parent = entry.pruneFirstParents();
+ if(parent == null) {
+
+ boolean collected = tryCollect(entry);
+ if(!collected) {
+ entry.setLevel((short)0);
+ }
+
+ } else {
+
+ support.setLevel(entry, parent.getLevel() + 1);
+
+ }
+
+ }
+
+ int status = entry.getGCStatus();
+ if((status & CacheEntry.HAS_BEEN_BOUND) == 0) {
+
+ if(parent != null && !parent.isDiscarded()) {
+ if((parent.getGCStatus() & CacheEntry.HAS_BEEN_BOUND) != 0) {
+ queryProcessor.boundQueries++;
+ entry.setGCStatusFlag(CacheEntry.HAS_BEEN_BOUND, true);
+ }
+ }
+
+ if(queryProcessor.listening.hasListenerAfterDisposing(entry)) {
+ if((status & CacheEntry.HAS_BEEN_BOUND) == 0) {
+ queryProcessor.boundQueries++;
+ entry.setGCStatusFlag(CacheEntry.HAS_BEEN_BOUND, true);
+ }
+ }
+ }
+
+ }
+ }
+
+ }
+
+ }
+
+ private boolean tryCollect(CacheEntry entry) {
+ if (!queryProcessor.listening.hasListenerAfterDisposing(entry))
+ if(entry.shouldBeCollected()) {
+ queryProcessor.removeQuery(entry);
+ support.remove();
+ propagate = true;
+ moreAll++;
+ doneAll = false;
+ return true;
+ }
+ return false;
+ }
+
+ private void start() {
+ moreAll = 1;
+ doneAll = false;
+ }
+
}
\ No newline at end of file
Code Review / simantics / platform.git / blobdiff