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