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

   1 /*******************************************************************************\r
   2  * Copyright (c) 2007, 2010 Association for Decentralized Information Management\r
   3  * in Industry THTH ry.\r
   4  * All rights reserved. This program and the accompanying materials\r
   5  * are made available under the terms of the Eclipse Public License v1.0\r
   6  * which accompanies this distribution, and is available at\r
   7  * http://www.eclipse.org/legal/epl-v10.html\r
   8  *\r
   9  * Contributors:\r
  10  *     VTT Technical Research Centre of Finland - initial API and implementation\r
  11  *******************************************************************************/\r
  12 package org.simantics.db.impl.graph;\r
  13 \r
  14 import java.util.ArrayList;\r
  15 import java.util.Arrays;\r
  16 import java.util.Collection;\r
  17 import java.util.HashMap;\r
  18 import java.util.concurrent.atomic.AtomicInteger;\r
  19 \r
  20 import org.simantics.db.common.utils.Logger;\r
  21 import org.simantics.db.exception.RuntimeDatabaseException;\r
  22 import org.simantics.db.impl.query.QueryProcessor.AsyncBarrier;\r
  23 import org.simantics.db.impl.query.QueryProcessor.SessionTask;\r
  24 import org.simantics.utils.Development;\r
  25 \r
  26 final public class AsyncBarrierImpl extends AtomicInteger implements AsyncBarrier {\r
  27 \r
  28         private static final long serialVersionUID = 4724463372850048672L;\r
  29 \r
  30         static final HashMap<AsyncBarrierImpl, Collection<AsyncBarrierImpl>> reverseLookup = new HashMap<AsyncBarrierImpl, Collection<AsyncBarrierImpl>>();\r
  31         static final HashMap<AsyncBarrierImpl, Debugger> debuggerMap = new HashMap<AsyncBarrierImpl, Debugger>();\r
  32         static final HashMap<AsyncBarrierImpl, Boolean> restartMap = new HashMap<AsyncBarrierImpl, Boolean>();\r
  33 \r
  34         static final int WAIT_TIME = 600;\r
  35 \r
  36         public static final boolean BOOKKEEPING = false;\r
  37         public static final boolean PRINT = false;\r
  38         static final boolean RESTART_GUARD = false;\r
  39 \r
  40         final private AsyncBarrierImpl caller;\r
  41 \r
  42         public AsyncBarrierImpl(AsyncBarrierImpl caller) {\r
  43                 super(0);\r
  44                 if (BOOKKEEPING) {\r
  45                         synchronized (debuggerMap) {\r
  46                                 debuggerMap.put(this, new Debugger());\r
  47                         }\r
  48                         synchronized (reverseLookup) {\r
  49                                 Collection<AsyncBarrierImpl> barriers = reverseLookup\r
  50                                                 .get(caller);\r
  51                                 if (barriers == null) {\r
  52                                         barriers = new ArrayList<AsyncBarrierImpl>();\r
  53                                         reverseLookup.put(caller, barriers);\r
  54                                 }\r
  55                                 barriers.add(this);\r
  56                         }\r
  57                 }\r
  58                 this.caller = caller;\r
  59         }\r
  60 \r
  61         public class Debugger {\r
  62                 public HashMap<Object, ArrayList<String>> infos = new HashMap<Object, ArrayList<String>>();\r
  63 \r
  64                 public synchronized void inc(Object id, String info) {\r
  65                         if (id == null)\r
  66                                 return;\r
  67                         ArrayList<String> exist = infos.get(id);\r
  68                         if (exist == null) {\r
  69                                 exist = new ArrayList<String>();\r
  70                                 infos.put(id, exist);\r
  71                         } else {\r
  72                                 // System.err.println("Appending " + id + " += " + info);\r
  73                         }\r
  74                         exist.add(info);\r
  75                         // String exist = infos.put(id, info);\r
  76                         // if(exist != null) System.err.println("replacing " + exist +\r
  77                         // " => " + info + " for " + id);\r
  78                 }\r
  79 \r
  80                 public synchronized void dec(Object id) {\r
  81                         if (id == null)\r
  82                                 return;\r
  83                         ArrayList<String> exist = infos.get(id);\r
  84                         if (exist == null) {\r
  85                                 System.err.println("No data for " + id);\r
  86                         } else {\r
  87                                 exist.remove(0);\r
  88                                 if (exist.isEmpty())\r
  89                                         infos.remove(id);\r
  90                         }\r
  91                 }\r
  92 \r
  93                 @Override\r
  94                 public synchronized String toString() {\r
  95                         StringBuilder b = new StringBuilder();\r
  96                         for (ArrayList<String> ss : infos.values()) {\r
  97                                 for (String s : ss)\r
  98                                         b.append("info " + s + "\r\n");\r
  99                         }\r
 100                         return b.toString();\r
 101                 }\r
 102 \r
 103                 public synchronized void toErr(int indent) {\r
 104                         char[] spaces = new char[indent];\r
 105                         Arrays.fill(spaces, ' ');\r
 106                         for (ArrayList<String> ss : infos.values()) {\r
 107                                 for (String s : ss) {\r
 108                                         if (!s.startsWith("#"))\r
 109                                                 continue;\r
 110                                         StringBuilder b = new StringBuilder();\r
 111                                         b.append(spaces);\r
 112                                         b.append(s);\r
 113                                         System.err.println(b.toString());\r
 114                                 }\r
 115                         }\r
 116                 }\r
 117         }\r
 118 \r
 119         public void inc() {\r
 120 \r
 121                 if (BOOKKEEPING)\r
 122                         inc(new Object(), new Exception().getStackTrace()[1].toString());\r
 123                 else\r
 124                         inc(null, null);\r
 125 \r
 126         }\r
 127 \r
 128         public void inc(String debug) {\r
 129 \r
 130                 if (BOOKKEEPING)\r
 131                         inc(new Object(), new Exception().getStackTrace()[1].toString());\r
 132                 else\r
 133                         inc(null, null);\r
 134 \r
 135         }\r
 136 \r
 137         public void inc(Object id, String info) {\r
 138 \r
 139                 //              if (PRINT) {\r
 140                 //                      if (get() < 5)\r
 141                 //                              new Exception("inc " + get() + " " + this).printStackTrace();\r
 142                 //              }\r
 143 \r
 144                 if (BOOKKEEPING) {\r
 145                         Debugger debugger = debuggerMap.get(this);\r
 146                         if (debugger != null)\r
 147                                 debugger.inc(id, info);\r
 148                         // StackTraceElement[] tr = new Exception().getStackTrace();\r
 149                         // if(tr.length == 4)\r
 150                         // debugger.inc(new String[] { debug, tr[2].toString(),\r
 151                         // tr[3].toString() });\r
 152                         // else if(tr.length == 5)\r
 153                         // debugger.inc(new String[] { debug, tr[2].toString(),\r
 154                         // tr[3].toString(), tr[4].toString() });\r
 155                         // else if(tr.length == 6)\r
 156                         // debugger.inc(new String[] { debug, tr[2].toString(),\r
 157                         // tr[3].toString(), tr[4].toString(), tr[5].toString() });\r
 158                         // else\r
 159                         // debugger.inc(new String[] { debug, tr[2].toString(),\r
 160                         // tr[3].toString(), tr[4].toString(), tr[5].toString(),\r
 161                         // tr[6].toString() });\r
 162                 }\r
 163 \r
 164                 //              new Exception().printStackTrace();\r
 165 \r
 166                 if(PRINT) {\r
 167 \r
 168                         System.err.println("inc barrier[" + get() + "] " + this);\r
 169                         StackTraceElement[] elems = new Exception().getStackTrace();\r
 170                         for(int i=0;i<4;i++) System.err.println(elems[i]);\r
 171 \r
 172                 }\r
 173 \r
 174                 if (incrementAndGet() == 1) {\r
 175                         if (caller != null) {\r
 176                                 if (BOOKKEEPING)\r
 177                                         caller.inc(this, "Child");\r
 178                                 else\r
 179                                         caller.inc(null, null);\r
 180                         }\r
 181                 }\r
 182 \r
 183         }\r
 184 \r
 185         public void dec() {\r
 186 \r
 187                 if(PRINT) {\r
 188                         System.err.println("dec barrier[" + get() + "] " + this);\r
 189                         StackTraceElement[] elems = new Exception().getStackTrace();\r
 190                         for(int i=0;i<3;i++) System.err.println(elems[i]);\r
 191                 }\r
 192 \r
 193                 int count = decrementAndGet();\r
 194                 if (count < 1) {\r
 195                         if (count == 0) {\r
 196                                 if (caller != null)\r
 197                                         caller.dec(this);\r
 198                                 //                              sema.release();\r
 199                         }\r
 200                         if (count < 0) {\r
 201                                 Logger.defaultLogError(\r
 202                                                 "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.",\r
 203                                                 new Exception());\r
 204                         }\r
 205                         assert (count >= 0);\r
 206                 }\r
 207 \r
 208         }\r
 209 \r
 210         public void dec(Object id) {\r
 211 \r
 212                 if (PRINT) {\r
 213                         if (get() < 5)\r
 214                                 new Exception("dec" + get() + " " + this).printStackTrace();\r
 215                 }\r
 216 \r
 217                 if (BOOKKEEPING) {\r
 218                         Debugger debugger = debuggerMap.get(this);\r
 219                         if (debugger != null)\r
 220                                 debugger.dec(id);\r
 221                         // StackTraceElement[] tr = new Exception().getStackTrace();\r
 222                         // if(tr.length == 3)\r
 223                         // debugger.dec(new String[] { debug, tr[2].toString() });\r
 224                         // else if(tr.length == 4)\r
 225                         // debugger.dec(new String[] { debug, tr[2].toString(),\r
 226                         // tr[3].toString() });\r
 227                         // else\r
 228                         // debugger.dec(new String[] { debug, tr[2].toString(),\r
 229                         // tr[3].toString(), tr[4].toString() });\r
 230                 }\r
 231 \r
 232                 //              System.err.println("barrier " + this);\r
 233                 //              StackTraceElement[] elems = new Exception().getStackTrace();\r
 234                 //              for(int i=0;i<3;i++) System.err.println(elems[i]);\r
 235                 //              new Exception().printStackTrace();\r
 236 \r
 237                 int count = decrementAndGet();\r
 238                 if (count < 1) {\r
 239                         if (count == 0) {\r
 240                                 if (caller != null)\r
 241                                         caller.dec(this);\r
 242                                 if (RESTART_GUARD)\r
 243                                         restartMap.put(this, true);\r
 244                                 //                              sema.release();\r
 245                                 // if(DEBUGGER) {\r
 246                                 // debuggerMap.remove(this);\r
 247                                 // }\r
 248                                 // if(REVERSE_LOOKUP) {\r
 249                                 // reverseLookup.remove(this);\r
 250                                 // }\r
 251                         }\r
 252                         if (count < 0) {\r
 253                                 Logger.defaultLogError(\r
 254                                                 "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.",\r
 255                                                 new Exception());\r
 256                                 // String message = ;\r
 257                                 // System.out.println(message);\r
 258                                 // if (DEBUGGER) {\r
 259                                 // JOptionPane.showMessageDialog(null, message);\r
 260                                 // System.out.println(debugger);\r
 261                                 // }\r
 262                                 // sema.release();\r
 263                                 System.exit(-1);\r
 264                         }\r
 265                         assert (count >= 0);\r
 266                 }\r
 267         }\r
 268 \r
 269         private static void printReverse(AsyncBarrierImpl barrier, int indent) {\r
 270 \r
 271                 if (barrier.get() == 0)\r
 272                         return;\r
 273                 for (int i = 0; i < indent; i++)\r
 274                         System.err.print(" ");\r
 275                 System.err.println("[" + barrier.get() + " requests]: " + barrier);\r
 276                 if (BOOKKEEPING) {\r
 277                         Debugger debugger = debuggerMap.get(barrier);\r
 278                         debugger.toErr(indent + 2);\r
 279                 }\r
 280 \r
 281                 Collection<AsyncBarrierImpl> children = reverseLookup.get(barrier);\r
 282                 if (children != null) {\r
 283                         for (AsyncBarrierImpl child : children)\r
 284                                 printReverse(child, indent + 2);\r
 285                 }\r
 286 \r
 287         }\r
 288 \r
 289         public void waitBarrier(Object request, ReadGraphImpl impl) {\r
 290 \r
 291                 if (get() > 0) {\r
 292 \r
 293                         long waitCount = 0;\r
 294 \r
 295                         while (get() != 0) {\r
 296 \r
 297                                 boolean executed = impl.processor.resume(impl);\r
 298                                 if(executed) waitCount = 0;\r
 299                                 \r
 300                                 ++waitCount;\r
 301                                 if(waitCount > 100) Thread.yield();\r
 302                                 if(waitCount > 1000) {\r
 303                                         try {\r
 304                                                 Thread.sleep(1);\r
 305                                         } catch (InterruptedException e) {\r
 306                                                 e.printStackTrace();\r
 307                                         }\r
 308                                 }\r
 309                                 if(waitCount > WAIT_TIME*1000) {\r
 310 \r
 311                                         System.err.println("AsyncBarrierImpl.waitBarrier("\r
 312                                                         + request\r
 313                                                         + ") is taking long to execute, so far "\r
 314                                                         + (waitCount / 1000) + " s.");\r
 315 \r
 316                                         if (BOOKKEEPING) {\r
 317 \r
 318                                                 synchronized (reverseLookup) {\r
 319                                                         printReverse(this, 0);\r
 320                                                 }\r
 321 \r
 322                                         }\r
 323 \r
 324                                         if(Development.DEVELOPMENT) {\r
 325 \r
 326                                                 impl.processor.threadLocks[0].lock();\r
 327                                                 System.err.println("-queues=" + impl.processor.queues[0].size());\r
 328                                                 impl.processor.threadLocks[0].unlock();\r
 329                                                 System.err.println("-own=" + impl.processor.ownTasks[0].size());\r
 330                                                 System.err.println("-ownSync=" + impl.processor.ownSyncTasks[0].size());\r
 331                                                 for(SessionTask task : impl.processor.ownSyncTasks[0]) {\r
 332                                                         System.err.println("--" + task);\r
 333                                                 }\r
 334 \r
 335                                         }\r
 336 \r
 337                                         throw new RuntimeDatabaseException("Request timed out.");\r
 338                                         //waitCount = 0;\r
 339 \r
 340                                 }\r
 341 \r
 342                         }\r
 343 \r
 344                 }\r
 345 \r
 346         }\r
 347 \r
 348         public void restart() {\r
 349                 assertReady();\r
 350                 // log.clear();\r
 351                 //              sema.drainPermits();\r
 352                 if (RESTART_GUARD)\r
 353                         restartMap.remove(this);\r
 354                 if (BOOKKEEPING)\r
 355                         debuggerMap.put(this, new Debugger());\r
 356         }\r
 357 \r
 358         public void assertReady() {\r
 359                 int current = get();\r
 360                 if (current != 0)\r
 361                         throw new AssertionError("Barrier was not finished (pending="\r
 362                                         + current + ").");\r
 363         }\r
 364 \r
 365         public void report() {\r
 366                 // System.out.println("Barrier log:");\r
 367                 // for(Map.Entry<String, Integer> entry : sources.entrySet()) {\r
 368                 // System.out.println(entry.getKey() + " " + entry.getValue());\r
 369                 // }\r
 370                 // System.out.println("SyncIntProcedure log:");\r
 371                 // for(Map.Entry<String, Integer> entry :\r
 372                 // SyncIntProcedure.counters.entrySet()) {\r
 373                 // System.out.println(entry.getKey() + " " + entry.getValue());\r
 374                 // }\r
 375         }\r
 376 \r
 377         @Override\r
 378         public String toString() {\r
 379                 return "AsyncBarrierImpl@" + System.identityHashCode(this)\r
 380                                 + " - counter = " + get() + " - caller = " + caller;\r
 381         }\r
 382 \r
 383 }\r