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                 long start = System.nanoTime();
  46
  47                 // Refresh current size
  48                 int size = support.calculateCurrentSize();
  49                 int bound = queryProcessor.boundQueries;
  50                 int young = size - bound;
  51                 int youngPct = size > 0 ? 100*young / size : 0;
  52
  53                 // Initialize support for new run
  54                 // If support returns 0 we are starting from 0
  55                 if(support.start(youngTarget == 0)) {
  56
  57                         moreAll = 0;
  58
  59                         // We monitor all prospects here
  60                         CacheEntryBase prospect = support.iterate(0);
  61                         while(prospect != null) {
  62                                 if(prospect.isDiscarded()) {
  63                                         support.remove();
  64                                         propagate = true;
  65                                 } else {
  66                                         CacheEntry parent = prospect.getFirstParent(queryProcessor);
  67                                         if(parent == null) {
  68                                                 tryCollect(prospect);
  69                                         } else {
  70                                                 support.setLevel(prospect, parent.getLevel() + 1);
  71                                         }
  72                                 }
  73                                 prospect = support.iterate(0);
  74                         }
  75
  76                         // If no prospects were collected and most of the queries are old we can stop here
  77                         if(!propagate && youngPct < youngTarget) {
  78 //                              System.err.println("collect2 skipped");
  79 //                              System.err.println("-size=" + size);
  80 //                              System.err.println("-young=" + young);
  81                                 return;
  82                         }
  83
  84                 }
  85
  86                 long test = (long)allowedTimeInMs*1000000;
  87
  88                 start();
  89
  90                 while(true) {
  91
  92                         size = support.getCurrentSize();
  93                         bound = queryProcessor.boundQueries;
  94
  95                         long elapsed = System.nanoTime()-start;
  96                         boolean timeCondition = elapsed > test;
  97
  98                         if(doneAll || timeCondition) {
  99
 100                                 spent += elapsed;
 101
 102                                 if(DEBUG_STATUS)
 103                                         System.err.println("Query collector used " + 1e-9*elapsed + "s total queries: " + size + " bound queries: " + bound + " spent: " + (double)spent*1e-9);
 104
 105                                 return;
 106
 107                         }
 108
 109                         // Iterate all entries and update sets
 110                         if(!doneAll) {
 111 //                              long start2 = System.nanoTime();
 112                                 for(int i=0;i<COLLECT_N;i++) {
 113                                         CacheEntryBase entry = support.iterate(Integer.MAX_VALUE);
 114                                         if(entry == null) {
 115                                                 if(DEBUG) {
 116                                                         System.err.println("finished iteration");
 117                                                         System.err.println("-moreAll=" + moreAll);
 118                                                 }
 119                                                 if(moreAll < 1000) {
 120                                                         doneAll = true;
 121                                                         propagate = false;
 122                                                 }
 123                                                 moreAll = 0;
 124                                                 break;
 125                                         }
 126
 127                                         CacheEntry parent = entry.getFirstParent(queryProcessor);
 128                                         if(parent == null) {
 129
 130                                                 boolean collected = tryCollect(entry);
 131                                                 if(!collected) {
 132                                                         entry.setLevel((short)0);
 133                                                 }
 134
 135                                         } else {
 136
 137                                                 parent = entry.pruneFirstParents();
 138                                                 if(parent == null) {
 139
 140                                                         boolean collected = tryCollect(entry);
 141                                                         if(!collected) {
 142                                                                 entry.setLevel((short)0);
 143                                                         }
 144
 145                                                 } else {
 146
 147                                                         support.setLevel(entry, parent.getLevel() + 1);
 148
 149                                                 }
 150
 151                                         }
 152
 153                                         int status = entry.getGCStatus();
 154                                         if((status & CacheEntry.HAS_BEEN_BOUND) == 0) {
 155
 156                                                 if(parent != null && !parent.isDiscarded()) {
 157                                                         if((parent.getGCStatus() & CacheEntry.HAS_BEEN_BOUND) != 0) {
 158                                                                 queryProcessor.boundQueries++;
 159                                                                 entry.setGCStatusFlag(CacheEntry.HAS_BEEN_BOUND, true);
 160                                                         }
 161                                                 }
 162
 163                                                 if(queryProcessor.hasListenerAfterDisposing(entry)) {
 164                                                         if((status & CacheEntry.HAS_BEEN_BOUND) == 0) {
 165                                                                 queryProcessor.boundQueries++;
 166                                                                 entry.setGCStatusFlag(CacheEntry.HAS_BEEN_BOUND, true);
 167                                                         }
 168                                                 }
 169                                         }
 170
 171                                 }
 172                         }
 173
 174                 }
 175
 176         }
 177
 178         private boolean tryCollect(CacheEntry entry) {
 179                 if (!queryProcessor.hasListenerAfterDisposing(entry))
 180                         if(entry.shouldBeCollected()) {
 181                                 queryProcessor.removeQuery(entry);
 182                                 support.remove();
 183                                 propagate = true;
 184                                 moreAll++;
 185                                 doneAll = false;
 186                                 return true;
 187                         }
 188                 return false;
 189         }
 190
 191         private void start() {
 192                 moreAll = 1;
 193                 doneAll = false;
 194         }
 195
 196 }