--- /dev/null
+/*******************************************************************************\r
+ * Copyright (c) 2007, 2010 Association for Decentralized Information Management\r
+ * in Industry THTH ry.\r
+ * All rights reserved. This program and the accompanying materials\r
+ * are made available under the terms of the Eclipse Public License v1.0\r
+ * which accompanies this distribution, and is available at\r
+ * http://www.eclipse.org/legal/epl-v10.html\r
+ *\r
+ * Contributors:\r
+ * VTT Technical Research Centre of Finland - initial API and implementation\r
+ *******************************************************************************/\r
+package org.simantics.db.impl.graph;\r
+\r
+import java.util.ArrayList;\r
+import java.util.Arrays;\r
+import java.util.Collection;\r
+import java.util.HashMap;\r
+import java.util.concurrent.atomic.AtomicInteger;\r
+\r
+import org.simantics.db.common.utils.Logger;\r
+import org.simantics.db.exception.RuntimeDatabaseException;\r
+import org.simantics.db.impl.query.QueryProcessor.AsyncBarrier;\r
+import org.simantics.db.impl.query.QueryProcessor.SessionTask;\r
+import org.simantics.utils.Development;\r
+\r
+final public class AsyncBarrierImpl extends AtomicInteger implements AsyncBarrier {\r
+\r
+ private static final long serialVersionUID = 4724463372850048672L;\r
+\r
+ static final HashMap<AsyncBarrierImpl, Collection<AsyncBarrierImpl>> reverseLookup = new HashMap<AsyncBarrierImpl, Collection<AsyncBarrierImpl>>();\r
+ static final HashMap<AsyncBarrierImpl, Debugger> debuggerMap = new HashMap<AsyncBarrierImpl, Debugger>();\r
+ static final HashMap<AsyncBarrierImpl, Boolean> restartMap = new HashMap<AsyncBarrierImpl, Boolean>();\r
+\r
+ static final int WAIT_TIME = 600;\r
+\r
+ public static final boolean BOOKKEEPING = false;\r
+ public static final boolean PRINT = false;\r
+ static final boolean RESTART_GUARD = false;\r
+\r
+ final private AsyncBarrierImpl caller;\r
+\r
+ public AsyncBarrierImpl(AsyncBarrierImpl caller) {\r
+ super(0);\r
+ if (BOOKKEEPING) {\r
+ synchronized (debuggerMap) {\r
+ debuggerMap.put(this, new Debugger());\r
+ }\r
+ synchronized (reverseLookup) {\r
+ Collection<AsyncBarrierImpl> barriers = reverseLookup\r
+ .get(caller);\r
+ if (barriers == null) {\r
+ barriers = new ArrayList<AsyncBarrierImpl>();\r
+ reverseLookup.put(caller, barriers);\r
+ }\r
+ barriers.add(this);\r
+ }\r
+ }\r
+ this.caller = caller;\r
+ }\r
+\r
+ public class Debugger {\r
+ public HashMap<Object, ArrayList<String>> infos = new HashMap<Object, ArrayList<String>>();\r
+\r
+ public synchronized void inc(Object id, String info) {\r
+ if (id == null)\r
+ return;\r
+ ArrayList<String> exist = infos.get(id);\r
+ if (exist == null) {\r
+ exist = new ArrayList<String>();\r
+ infos.put(id, exist);\r
+ } else {\r
+ // System.err.println("Appending " + id + " += " + info);\r
+ }\r
+ exist.add(info);\r
+ // String exist = infos.put(id, info);\r
+ // if(exist != null) System.err.println("replacing " + exist +\r
+ // " => " + info + " for " + id);\r
+ }\r
+\r
+ public synchronized void dec(Object id) {\r
+ if (id == null)\r
+ return;\r
+ ArrayList<String> exist = infos.get(id);\r
+ if (exist == null) {\r
+ System.err.println("No data for " + id);\r
+ } else {\r
+ exist.remove(0);\r
+ if (exist.isEmpty())\r
+ infos.remove(id);\r
+ }\r
+ }\r
+\r
+ @Override\r
+ public synchronized String toString() {\r
+ StringBuilder b = new StringBuilder();\r
+ for (ArrayList<String> ss : infos.values()) {\r
+ for (String s : ss)\r
+ b.append("info " + s + "\r\n");\r
+ }\r
+ return b.toString();\r
+ }\r
+\r
+ public synchronized void toErr(int indent) {\r
+ char[] spaces = new char[indent];\r
+ Arrays.fill(spaces, ' ');\r
+ for (ArrayList<String> ss : infos.values()) {\r
+ for (String s : ss) {\r
+ if (!s.startsWith("#"))\r
+ continue;\r
+ StringBuilder b = new StringBuilder();\r
+ b.append(spaces);\r
+ b.append(s);\r
+ System.err.println(b.toString());\r
+ }\r
+ }\r
+ }\r
+ }\r
+\r
+ public void inc() {\r
+\r
+ if (BOOKKEEPING)\r
+ inc(new Object(), new Exception().getStackTrace()[1].toString());\r
+ else\r
+ inc(null, null);\r
+\r
+ }\r
+\r
+ public void inc(String debug) {\r
+\r
+ if (BOOKKEEPING)\r
+ inc(new Object(), new Exception().getStackTrace()[1].toString());\r
+ else\r
+ inc(null, null);\r
+\r
+ }\r
+\r
+ public void inc(Object id, String info) {\r
+\r
+ // if (PRINT) {\r
+ // if (get() < 5)\r
+ // new Exception("inc " + get() + " " + this).printStackTrace();\r
+ // }\r
+\r
+ if (BOOKKEEPING) {\r
+ Debugger debugger = debuggerMap.get(this);\r
+ if (debugger != null)\r
+ debugger.inc(id, info);\r
+ // StackTraceElement[] tr = new Exception().getStackTrace();\r
+ // if(tr.length == 4)\r
+ // debugger.inc(new String[] { debug, tr[2].toString(),\r
+ // tr[3].toString() });\r
+ // else if(tr.length == 5)\r
+ // debugger.inc(new String[] { debug, tr[2].toString(),\r
+ // tr[3].toString(), tr[4].toString() });\r
+ // else if(tr.length == 6)\r
+ // debugger.inc(new String[] { debug, tr[2].toString(),\r
+ // tr[3].toString(), tr[4].toString(), tr[5].toString() });\r
+ // else\r
+ // debugger.inc(new String[] { debug, tr[2].toString(),\r
+ // tr[3].toString(), tr[4].toString(), tr[5].toString(),\r
+ // tr[6].toString() });\r
+ }\r
+\r
+ // new Exception().printStackTrace();\r
+\r
+ if(PRINT) {\r
+\r
+ System.err.println("inc barrier[" + get() + "] " + this);\r
+ StackTraceElement[] elems = new Exception().getStackTrace();\r
+ for(int i=0;i<4;i++) System.err.println(elems[i]);\r
+\r
+ }\r
+\r
+ if (incrementAndGet() == 1) {\r
+ if (caller != null) {\r
+ if (BOOKKEEPING)\r
+ caller.inc(this, "Child");\r
+ else\r
+ caller.inc(null, null);\r
+ }\r
+ }\r
+\r
+ }\r
+\r
+ public void dec() {\r
+\r
+ if(PRINT) {\r
+ System.err.println("dec barrier[" + get() + "] " + this);\r
+ StackTraceElement[] elems = new Exception().getStackTrace();\r
+ for(int i=0;i<3;i++) System.err.println(elems[i]);\r
+ }\r
+\r
+ int count = decrementAndGet();\r
+ if (count < 1) {\r
+ if (count == 0) {\r
+ if (caller != null)\r
+ caller.dec(this);\r
+ // sema.release();\r
+ }\r
+ if (count < 0) {\r
+ Logger.defaultLogError(\r
+ "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
+ new Exception());\r
+ }\r
+ assert (count >= 0);\r
+ }\r
+\r
+ }\r
+\r
+ public void dec(Object id) {\r
+\r
+ if (PRINT) {\r
+ if (get() < 5)\r
+ new Exception("dec" + get() + " " + this).printStackTrace();\r
+ }\r
+\r
+ if (BOOKKEEPING) {\r
+ Debugger debugger = debuggerMap.get(this);\r
+ if (debugger != null)\r
+ debugger.dec(id);\r
+ // StackTraceElement[] tr = new Exception().getStackTrace();\r
+ // if(tr.length == 3)\r
+ // debugger.dec(new String[] { debug, tr[2].toString() });\r
+ // else if(tr.length == 4)\r
+ // debugger.dec(new String[] { debug, tr[2].toString(),\r
+ // tr[3].toString() });\r
+ // else\r
+ // debugger.dec(new String[] { debug, tr[2].toString(),\r
+ // tr[3].toString(), tr[4].toString() });\r
+ }\r
+\r
+ // System.err.println("barrier " + this);\r
+ // StackTraceElement[] elems = new Exception().getStackTrace();\r
+ // for(int i=0;i<3;i++) System.err.println(elems[i]);\r
+ // new Exception().printStackTrace();\r
+\r
+ int count = decrementAndGet();\r
+ if (count < 1) {\r
+ if (count == 0) {\r
+ if (caller != null)\r
+ caller.dec(this);\r
+ if (RESTART_GUARD)\r
+ restartMap.put(this, true);\r
+ // sema.release();\r
+ // if(DEBUGGER) {\r
+ // debuggerMap.remove(this);\r
+ // }\r
+ // if(REVERSE_LOOKUP) {\r
+ // reverseLookup.remove(this);\r
+ // }\r
+ }\r
+ if (count < 0) {\r
+ Logger.defaultLogError(\r
+ "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
+ new Exception());\r
+ // String message = ;\r
+ // System.out.println(message);\r
+ // if (DEBUGGER) {\r
+ // JOptionPane.showMessageDialog(null, message);\r
+ // System.out.println(debugger);\r
+ // }\r
+ // sema.release();\r
+ System.exit(-1);\r
+ }\r
+ assert (count >= 0);\r
+ }\r
+ }\r
+\r
+ private static void printReverse(AsyncBarrierImpl barrier, int indent) {\r
+\r
+ if (barrier.get() == 0)\r
+ return;\r
+ for (int i = 0; i < indent; i++)\r
+ System.err.print(" ");\r
+ System.err.println("[" + barrier.get() + " requests]: " + barrier);\r
+ if (BOOKKEEPING) {\r
+ Debugger debugger = debuggerMap.get(barrier);\r
+ debugger.toErr(indent + 2);\r
+ }\r
+\r
+ Collection<AsyncBarrierImpl> children = reverseLookup.get(barrier);\r
+ if (children != null) {\r
+ for (AsyncBarrierImpl child : children)\r
+ printReverse(child, indent + 2);\r
+ }\r
+\r
+ }\r
+\r
+ public void waitBarrier(Object request, ReadGraphImpl impl) {\r
+\r
+ if (get() > 0) {\r
+\r
+ long waitCount = 0;\r
+\r
+ while (get() != 0) {\r
+\r
+ boolean executed = impl.processor.resume(impl);\r
+ if(executed) waitCount = 0;\r
+ \r
+ ++waitCount;\r
+ if(waitCount > 100) Thread.yield();\r
+ if(waitCount > 1000) {\r
+ try {\r
+ Thread.sleep(1);\r
+ } catch (InterruptedException e) {\r
+ e.printStackTrace();\r
+ }\r
+ }\r
+ if(waitCount > WAIT_TIME*1000) {\r
+\r
+ System.err.println("AsyncBarrierImpl.waitBarrier("\r
+ + request\r
+ + ") is taking long to execute, so far "\r
+ + (waitCount / 1000) + " s.");\r
+\r
+ if (BOOKKEEPING) {\r
+\r
+ synchronized (reverseLookup) {\r
+ printReverse(this, 0);\r
+ }\r
+\r
+ }\r
+\r
+ if(Development.DEVELOPMENT) {\r
+\r
+ impl.processor.threadLocks[0].lock();\r
+ System.err.println("-queues=" + impl.processor.queues[0].size());\r
+ impl.processor.threadLocks[0].unlock();\r
+ System.err.println("-own=" + impl.processor.ownTasks[0].size());\r
+ System.err.println("-ownSync=" + impl.processor.ownSyncTasks[0].size());\r
+ for(SessionTask task : impl.processor.ownSyncTasks[0]) {\r
+ System.err.println("--" + task);\r
+ }\r
+\r
+ }\r
+\r
+ throw new RuntimeDatabaseException("Request timed out.");\r
+ //waitCount = 0;\r
+\r
+ }\r
+\r
+ }\r
+\r
+ }\r
+\r
+ }\r
+\r
+ public void restart() {\r
+ assertReady();\r
+ // log.clear();\r
+ // sema.drainPermits();\r
+ if (RESTART_GUARD)\r
+ restartMap.remove(this);\r
+ if (BOOKKEEPING)\r
+ debuggerMap.put(this, new Debugger());\r
+ }\r
+\r
+ public void assertReady() {\r
+ int current = get();\r
+ if (current != 0)\r
+ throw new AssertionError("Barrier was not finished (pending="\r
+ + current + ").");\r
+ }\r
+\r
+ public void report() {\r
+ // System.out.println("Barrier log:");\r
+ // for(Map.Entry<String, Integer> entry : sources.entrySet()) {\r
+ // System.out.println(entry.getKey() + " " + entry.getValue());\r
+ // }\r
+ // System.out.println("SyncIntProcedure log:");\r
+ // for(Map.Entry<String, Integer> entry :\r
+ // SyncIntProcedure.counters.entrySet()) {\r
+ // System.out.println(entry.getKey() + " " + entry.getValue());\r
+ // }\r
+ }\r
+\r
+ @Override\r
+ public String toString() {\r
+ return "AsyncBarrierImpl@" + System.identityHashCode(this)\r
+ + " - counter = " + get() + " - caller = " + caller;\r
+ }\r
+\r
+}\r
Code Review / simantics / platform.git / blobdiff