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