bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/QueryCollectorImpl.java

   1 package org.simantics.db.impl.query;
   2
   3 import org.simantics.db.impl.query.QueryProcessor.QueryCollectorSupport;
   4
   5 class QueryCollectorImpl implements QueryProcessor.QueryCollector {
   6
   7         private static final boolean DEBUG = false;
   8         private static final boolean DEBUG_STATUS = false;
   9
  10         private final QueryProcessor queryProcessor;
  11
  12         final private QueryCollectorSupport support;
  13
  14         /*
  15          * Set to true: at end of iteration if moreAll is false
  16          * Set to false: upon gc start and whenever a prospect is added.
  17          *
  18          */
  19         boolean doneAll = false;
  20         /*
  21          * Set to true: upon gc start and whenever a prospect is added.
  22          * Set to false: at end of entry iteration
  23          * => if this is already false at end of entry iteration => no more work to be done
  24          */
  25         int moreAll = 0;
  26
  27         /*
  28          * Change propagation is in progress
  29          */
  30         boolean propagate = true;
  31
  32
  33         private static final int COLLECT_N = 1000;
  34
  35         private long spent = 0;
  36
  37         QueryCollectorImpl(QueryProcessor queryProcessor, QueryCollectorSupport support) {
  38                 this.queryProcessor = queryProcessor;
  39                 this.support = support;
  40         }
  41
  42         @Override
  43         public void collect(int youngTarget, int allowedTimeInMs) {
  44
  45                 // Flush listener registrations to prevent the collector from trashing
  46                 // listeners that are still queued up waiting to be registered.
  47                 queryProcessor.listening.sync();
  48
  49                 long start = System.nanoTime();
  50
  51                 // Refresh current size
  52                 int size = support.calculateCurrentSize();
  53                 int bound = queryProcessor.boundQueries;
  54                 int young = size - bound;
  55                 int youngPct = size > 0 ? 100*young / size : 0;
  56
  57                 // Initialize support for new run
  58                 // If support returns 0 we are starting from 0
  59                 if(support.start(youngTarget == 0)) {
  60
  61                         moreAll = 0;
  62
  63                         // We monitor all prospects here
  64                         CacheEntryBase prospect = support.iterate(0);
  65                         while(prospect != null) {
  66                                 if(prospect.isDiscarded()) {
  67                                         support.remove();
  68                                         propagate = true;
  69                                 } else {
  70                                         CacheEntry parent = prospect.getFirstParent(queryProcessor);
  71                                         if(parent == null) {
  72                                                 tryCollect(prospect);
  73                                         } else {
  74                                                 support.setLevel(prospect, parent.getLevel() + 1);
  75                                         }
  76                                 }
  77                                 prospect = support.iterate(0);
  78                         }
  79
  80                         // If no prospects were collected and most of the queries are old we can stop here
  81                         if(!propagate && youngPct < youngTarget) {
  82 //                              System.err.println("collect2 skipped");
  83 //                              System.err.println("-size=" + size);
  84 //                              System.err.println("-young=" + young);
  85                                 return;
  86                         }
  87
  88                 }
  89
  90                 long test = (long)allowedTimeInMs*1000000;
  91
  92                 start();
  93
  94                 while(true) {
  95
  96                         size = support.getCurrentSize();
  97                         bound = queryProcessor.boundQueries;
  98
  99                         long elapsed = System.nanoTime()-start;
 100                         boolean timeCondition = elapsed > test;
 101
 102                         if(doneAll || timeCondition) {
 103
 104                                 spent += elapsed;
 105
 106                                 if(DEBUG_STATUS)
 107                                         System.err.println("Query collector used " + 1e-9*elapsed + "s total queries: " + size + " bound queries: " + bound + " spent: " + (double)spent*1e-9);
 108
 109                                 return;
 110
 111                         }
 112
 113                         // Iterate all entries and update sets
 114                         if(!doneAll) {
 115 //                              long start2 = System.nanoTime();
 116                                 for(int i=0;i<COLLECT_N;i++) {
 117                                         CacheEntryBase entry = support.iterate(Integer.MAX_VALUE);
 118                                         if(entry == null) {
 119                                                 if(DEBUG) {
 120                                                         System.err.println("finished iteration");
 121                                                         System.err.println("-moreAll=" + moreAll);
 122                                                 }
 123                                                 if(moreAll < 1000) {
 124                                                         doneAll = true;
 125                                                         propagate = false;
 126                                                 }
 127                                                 moreAll = 0;
 128                                                 break;
 129                                         }
 130
 131                                         CacheEntry parent = entry.getFirstParent(queryProcessor);
 132                                         if(parent == null) {
 133
 134                                                 boolean collected = tryCollect(entry);
 135                                                 if(!collected) {
 136                                                         entry.setLevel((short)0);
 137                                                 }
 138
 139                                         } else {
 140
 141                                                 parent = entry.pruneFirstParents();
 142                                                 if(parent == null) {
 143
 144                                                         boolean collected = tryCollect(entry);
 145                                                         if(!collected) {
 146                                                                 entry.setLevel((short)0);
 147                                                         }
 148
 149                                                 } else {
 150
 151                                                         support.setLevel(entry, parent.getLevel() + 1);
 152
 153                                                 }
 154
 155                                         }
 156
 157                                         int status = entry.getGCStatus();
 158                                         if((status & CacheEntry.HAS_BEEN_BOUND) == 0) {
 159
 160                                                 if(parent != null && !parent.isDiscarded()) {
 161                                                         if((parent.getGCStatus() & CacheEntry.HAS_BEEN_BOUND) != 0) {
 162                                                                 queryProcessor.boundQueries++;
 163                                                                 entry.setGCStatusFlag(CacheEntry.HAS_BEEN_BOUND, true);
 164                                                         }
 165                                                 }
 166
 167                                                 if(queryProcessor.listening.hasListenerAfterDisposing(entry)) {
 168                                                         if((status & CacheEntry.HAS_BEEN_BOUND) == 0) {
 169                                                                 queryProcessor.boundQueries++;
 170                                                                 entry.setGCStatusFlag(CacheEntry.HAS_BEEN_BOUND, true);
 171                                                         }
 172                                                 }
 173                                         }
 174
 175                                 }
 176                         }
 177
 178                 }
 179
 180         }
 181
 182         private boolean tryCollect(CacheEntry entry) {
 183                 if (!queryProcessor.listening.hasListenerAfterDisposing(entry))
 184                         if(entry.shouldBeCollected()) {
 185                                 queryProcessor.removeQuery(entry);
 186                                 support.remove();
 187                                 propagate = true;
 188                                 moreAll++;
 189                                 doneAll = false;
 190                                 return true;
 191                         }
 192                 return false;
 193         }
 194
 195         private void start() {
 196                 moreAll = 1;
 197                 doneAll = false;
 198         }
 199
 200 }