bundles/org.simantics.db.impl/src/org/simantics/db/impl/graph/AsyncBarrierImpl.java

   1 /*******************************************************************************
   2  * Copyright (c) 2007, 2018 Association for Decentralized Information Management
   3  * in Industry THTH ry.
   4  * All rights reserved. This program and the accompanying materials
   5  * are made available under the terms of the Eclipse Public License v1.0
   6  * which accompanies this distribution, and is available at
   7  * http://www.eclipse.org/legal/epl-v10.html
   8  *
   9  * Contributors:
  10  *     VTT Technical Research Centre of Finland - initial API and implementation
  11  *******************************************************************************/
  12 package org.simantics.db.impl.graph;
  13
  14 import java.util.Collection;
  15 import java.util.concurrent.atomic.AtomicInteger;
  16
  17 import org.simantics.db.common.utils.Logger;
  18 import org.simantics.db.exception.RuntimeDatabaseException;
  19 import org.simantics.db.impl.query.CacheEntry;
  20 import org.simantics.db.impl.query.QueryProcessor.AsyncBarrier;
  21 import org.simantics.db.impl.query.QueryProcessor.SessionTask;
  22
  23 final public class AsyncBarrierImpl extends AtomicInteger implements AsyncBarrier {
  24
  25         private static final long serialVersionUID = 4724463372850048672L;
  26
  27         static final int WAIT_TIME = 60000;
  28
  29         public static final boolean PRINT = false;
  30
  31         final public AsyncBarrierImpl caller;
  32
  33         final public Runnable callback;
  34
  35         public AsyncBarrierImpl(AsyncBarrierImpl caller, CacheEntry<?> entry, Runnable callback) {
  36                 super(0);
  37                 this.caller = caller;
  38                 this.callback = callback;
  39         if (BarrierTracing.BOOKKEEPING) {
  40             BarrierTracing.trace(this, entry);
  41         }
  42         }
  43
  44
  45         public void inc() {
  46
  47             if(BarrierTracing.BOOKKEEPING) {
  48                 BarrierTracing.inc(this);
  49             } else {
  50                 inc(null, null);
  51             }
  52
  53         }
  54
  55         void inc(Object id, String info) {
  56
  57                 if(PRINT) {
  58                         System.err.println("inc barrier[" + get() + "] " + this);
  59                         StackTraceElement[] elems = new Exception().getStackTrace();
  60                         for(int i=0;i<4;i++) System.err.println(elems[i]);
  61                 }
  62
  63                 if (incrementAndGet() == 1) {
  64                         if (caller != null) {
  65                         if(BarrierTracing.BOOKKEEPING) {
  66                     caller.inc(this, "Child");
  67                         } else {
  68                     caller.inc(null, null);
  69                         }
  70                         }
  71                 }
  72
  73         }
  74
  75         public void dec() {
  76
  77                 if(PRINT) {
  78                         System.err.println("dec barrier[" + get() + "] " + this);
  79                         StackTraceElement[] elems = new Exception().getStackTrace();
  80                         for(int i=0;i<3;i++) System.err.println(elems[i]);
  81                 }
  82
  83                 int count = decrementAndGet();
  84                 if (count < 1) {
  85             if(BarrierTracing.BOOKKEEPING) {
  86                 BarrierTracing.dec(this, count);
  87             }
  88                         if (count == 0) {
  89                                 if (caller != null) {
  90                                         caller.dec();
  91                                 }
  92                         }
  93                         if (count < 0) {
  94                                 Logger.defaultLogError(
  95                                                 "Database request processing error. The application code has performed illegal actions (probably called multiple times the execute or exception method of a single result request.",
  96                                                 new Exception());
  97                         }
  98                         assert (count >= 0);
  99
 100                         if(callback != null)
 101                             callback.run();
 102
 103                 }
 104
 105         }
 106
 107         public static String report(AsyncBarrierImpl barrier) {
 108                 CacheEntry<?> e = BarrierTracing.entryMap.get(barrier);
 109                 if(e != null) return e.toString();
 110                 else return "Barrier@" + System.identityHashCode(barrier);
 111         }
 112
 113         public static void printReverse(AsyncBarrierImpl barrier, int indent) {
 114
 115                 if (barrier.get() == 0)
 116                         return;
 117                 for (int i = 0; i < indent; i++)
 118                         System.err.print(" ");
 119                 System.err.println("[" + barrier.get() + " requests]: " + report(barrier));
 120 //              if (BOOKKEEPING) {
 121 //                      Debugger debugger = debuggerMap.get(barrier);
 122 //                      debugger.toErr(indent + 2);
 123 //              }
 124
 125                 Collection<AsyncBarrierImpl> children = BarrierTracing.reverseLookup.get(barrier);
 126                 if (children != null) {
 127                         for (AsyncBarrierImpl child : children)
 128                                 printReverse(child, indent + 2);
 129                 }
 130
 131         }
 132
 133         public void waitBarrier(Object request, ReadGraphImpl impl) {
 134
 135                 if (get() > 0) {
 136
 137                         long waitCount = 0;
 138
 139                         while (get() != 0) {
 140
 141                                 boolean executed = impl.performPending();
 142                                 if(executed) waitCount = 0;
 143
 144                                 ++waitCount;
 145                                 if(waitCount > 100) Thread.yield();
 146                                 if(waitCount > 1000) {
 147                                         try {
 148                                                 Thread.sleep(1);
 149                                         } catch (InterruptedException e) {
 150                                                 e.printStackTrace();
 151                                         }
 152                                 }
 153                                 if(waitCount > WAIT_TIME) {
 154
 155                                         System.err.println("AsyncBarrierImpl.waitBarrier("
 156                                                         + request
 157                                                         + ") is taking long to execute, so far "
 158                                                         + (waitCount / 1000) + " s.");
 159
 160                                         if (BarrierTracing.BOOKKEEPING) {
 161                                                 synchronized (BarrierTracing.reverseLookup) {
 162                                                         printReverse(this, 0);
 163                                                 }
 164                                                 BarrierTracing.printBAPS();
 165                                         }
 166
 167                                         for(SessionTask t : impl.processor.freeScheduling) {
 168                                             System.err.println("Pending task:" + t);
 169                                         }
 170
 171 //                                      if(Development.DEVELOPMENT) {
 172 //                                              impl.processor.threadLocks[0].lock();
 173 //                                              System.err.println("-queues=" + impl.processor.queues[0].size());
 174 //                                              impl.processor.threadLocks[0].unlock();
 175 //                                              System.err.println("-own=" + impl.processor.ownTasks[0].size());
 176 //                                              System.err.println("-ownSync=" + impl.processor.ownSyncTasks[0].size());
 177 //                                              for(SessionTask task : impl.processor.ownSyncTasks[0]) {
 178 //                                                      System.err.println("--" + task);
 179 //                                              }
 180 //                                      }
 181
 182                                         throw new RuntimeDatabaseException("Request timed out.");
 183                                         //waitCount = 0;
 184
 185                                 }
 186
 187                         }
 188
 189                 }
 190
 191         }
 192
 193         public void restart() {
 194                 assertReady();
 195                 if(BarrierTracing.BOOKKEEPING) {
 196                     BarrierTracing.restart(this);
 197                 }
 198         }
 199
 200         public void assertReady() {
 201                 int current = get();
 202                 if (current != 0)
 203                         throw new AssertionError("Barrier was not finished (pending="
 204                                         + current + ").");
 205         }
 206
 207         public void report() {
 208                 // System.out.println("Barrier log:");
 209                 // for(Map.Entry<String, Integer> entry : sources.entrySet()) {
 210                 // System.out.println(entry.getKey() + " " + entry.getValue());
 211                 // }
 212                 // System.out.println("SyncIntProcedure log:");
 213                 // for(Map.Entry<String, Integer> entry :
 214                 // SyncIntProcedure.counters.entrySet()) {
 215                 // System.out.println(entry.getKey() + " " + entry.getValue());
 216                 // }
 217         }
 218
 219         @Override
 220         public String toString() {
 221                 return report(this);
 222 //              return "AsyncBarrierImpl@" + System.identityHashCode(this)
 223 //                              + " - counter = " + get() + " - caller = " + caller;
 224         }
 225
 226 }