bundles/org.simantics.utils.thread/src/org/simantics/utils/threads/ThreadUtils.java

   1 /*******************************************************************************
   2  * Copyright (c) 2007, 2010 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 /*
  13  *
  14  * @author Toni Kalajainen
  15  */
  16 package org.simantics.utils.threads;
  17
  18 import static java.util.concurrent.TimeUnit.NANOSECONDS;
  19
  20 import java.awt.EventQueue;
  21 import java.lang.reflect.InvocationTargetException;
  22 import java.util.Collection;
  23 import java.util.Collections;
  24 import java.util.HashMap;
  25 import java.util.HashSet;
  26 import java.util.LinkedList;
  27 import java.util.List;
  28 import java.util.Map;
  29 import java.util.Set;
  30 import java.util.concurrent.AbstractExecutorService;
  31 import java.util.concurrent.Executor;
  32 import java.util.concurrent.ExecutorService;
  33 import java.util.concurrent.ScheduledExecutorService;
  34 import java.util.concurrent.ScheduledThreadPoolExecutor;
  35 import java.util.concurrent.Semaphore;
  36 import java.util.concurrent.ThreadFactory;
  37 import java.util.concurrent.TimeUnit;
  38 import java.util.concurrent.atomic.AtomicInteger;
  39 import java.util.concurrent.locks.Lock;
  40
  41 /**
  42  * Utility for switching threads
  43  *
  44  * @see Executors2
  45  * @see IThreadWorkQueue
  46  */
  47 public class ThreadUtils {
  48
  49     public static final int CORES = Runtime.getRuntime().availableProcessors();
  50
  51     /**
  52      * Defines the maximum thread pool size of
  53      * {@link #getBlockingWorkExecutor()}. The system will enforce this value to
  54      * be at least Math.max(CORES, 8). It can be configured through the system
  55      * property <em>simantics.executor.blockingMaxThreads</em>.
  56      */
  57     public static final int MAX_BLOCKING_EXECUTOR_THREADS;
  58
  59     static {
  60         String blockingMaxThreadsProp = System.getProperty("simantics.executor.blockingMaxThreads", "" + CORES);
  61         int blockingMaxThreads = CORES;
  62         try {
  63             blockingMaxThreads = Integer.parseInt(blockingMaxThreadsProp);
  64         } catch (NumberFormatException e) {
  65         }
  66         MAX_BLOCKING_EXECUTOR_THREADS = Math.max(Math.max(blockingMaxThreads, 8), CORES);
  67     }
  68
  69     /** Never acquire non-blocking executor from this field, private use */
  70     public static ScheduledExecutorService NON_BLOCKING_EXECUTOR;
  71     /** Never acquire non-blocking executor from this field, private use */
  72     public static ExecutorService BLOCKING_EXECUTOR;
  73
  74     static ScheduledExecutorService TIMER;
  75
  76     /**
  77      * Get an {@link Executor} and {@link IThreadWorkQueue} for current thread
  78      *
  79      * @return an {@link Executor} and {@link IThreadWorkQueue} for current thread
  80      */
  81     public static CurrentThread getCurrentThread() {
  82         return CurrentThread.INSTANCE;
  83     }
  84
  85     /**
  86      * Get a timer executor. Never add blocking or long-wait runnables to this scheduler
  87      *
  88      * @return Executor that executes a non-blocking work
  89      */
  90     public static synchronized ScheduledExecutorService getTimer() {
  91         if (TIMER == null) {
  92             final ThreadGroup tg = new ThreadGroup("Timer");
  93             final AtomicInteger counter = new AtomicInteger(0);
  94             ThreadFactory tf = new ThreadFactory() {
  95                 @Override
  96                 public Thread newThread(Runnable r) {
  97                     Thread t = new Thread(tg, r, "Timer-"+(counter.incrementAndGet()));
  98                     if (!t.isDaemon())
  99                         t.setDaemon(true);
 100                     if (t.getPriority() != Thread.NORM_PRIORITY)
 101                         t.setPriority(Thread.NORM_PRIORITY);
 102                     return t;
 103                 }
 104             };
 105             TIMER = new ScheduledThreadPoolExecutor( 1, tf );
 106         }
 107         return TIMER;
 108     }
 109
 110     /**
 111      * Get Executor for work that doesn't lock or block. A non-blocking runnable never locks
 112      * anything, no semaphores, no synchronized() {} blocks. no Object.wait().
 113      *
 114      * @return Executor that executes a non-blocking work
 115      */
 116     public static synchronized ScheduledExecutorService getNonBlockingWorkExecutor() {
 117         if (NON_BLOCKING_EXECUTOR == null) {
 118             final ThreadGroup tg = new ThreadGroup("Non-Blocking-Worker-Group");
 119             final AtomicInteger counter = new AtomicInteger(0);
 120             ThreadFactory tf = new ThreadFactory() {
 121                 @Override
 122                 public Thread newThread(Runnable r) {
 123                     Thread t = new Thread(tg, r, "Non-Blocking-Worker-"+(counter.incrementAndGet()));
 124                     if (!t.isDaemon())
 125                         t.setDaemon(true);
 126                     if (t.getPriority() != Thread.NORM_PRIORITY)
 127                         t.setPriority(Thread.NORM_PRIORITY);
 128                     return t;
 129                 }
 130             };
 131             NON_BLOCKING_EXECUTOR =
 132                 new ScheduledThreadPoolExecutor( CORES, tf );
 133         }
 134         return NON_BLOCKING_EXECUTOR;
 135     }
 136
 137     /**
 138      * Get Executor executes work that may lock or block.
 139      *
 140      * @return executor for blocking operations
 141      */
 142     public static synchronized ExecutorService getBlockingWorkExecutor() {
 143         if (BLOCKING_EXECUTOR == null) {
 144             final ThreadGroup tg = new ThreadGroup("Blocking-Worker-Group");
 145             final AtomicInteger counter = new AtomicInteger(0);
 146             ThreadFactory tf = new ThreadFactory() {
 147                 @Override
 148                 public Thread newThread(Runnable r) {
 149                     Thread t = new Thread(tg, r, "Blocking-Worker-"+(counter.incrementAndGet()));
 150                     if (!t.isDaemon())
 151                         t.setDaemon(true);
 152                     if (t.getPriority() != Thread.NORM_PRIORITY)
 153                         t.setPriority(Thread.NORM_PRIORITY);
 154                     return t;
 155                 }
 156             };
 157             BLOCKING_EXECUTOR =
 158 //                new ThreadPoolExecutor(
 159 //                        0,
 160 //                        MAX_BLOCKING_EXECUTOR_THREADS,
 161 //                        3L, TimeUnit.SECONDS,
 162 //                        new SynchronousQueue<Runnable>(),
 163 //                        tf);
 164                     new ScheduledThreadPoolExecutor(MAX_BLOCKING_EXECUTOR_THREADS, tf);
 165         }
 166         return BLOCKING_EXECUTOR;
 167     }
 168
 169     /**
 170      * Better thread access allows thread context switching back to waiting
 171      * threads.
 172      *
 173      * @param access
 174      * @return an enhanced version of the specified queue
 175      */
 176     public static IThreadWorkQueue getBetterThreadAccess(IThreadWorkQueue access)
 177     {
 178         if (access instanceof BetterThreadAccess)
 179             return access;
 180         return new BetterThreadAccess(access);
 181     }
 182
 183     static Map<Thread, WaitingThread> map =
 184         new HashMap<Thread, WaitingThread>();
 185
 186     /**
 187      * Executes a runnable in thread synchronously.
 188      * If the thread locked and waits for this thread, then
 189      * the runnable is ran in that thread.
 190      *
 191      * This works only if all thread switching is done
 192      * using this method.
 193      *
 194      * @param threadAccess
 195      * @param runnable
 196      * @return true if thread accepted the runnable
 197      */
 198     public static boolean syncExec(IThreadWorkQueue threadAccess, final Runnable runnable)
 199     {
 200         if (threadAccess instanceof BetterThreadAccess)
 201             threadAccess = ((BetterThreadAccess) threadAccess).ta;
 202
 203         // Using current thread
 204         if (threadAccess.currentThreadAccess())
 205         {
 206             try {
 207                 runnable.run();
 208             } catch (RuntimeException e) {
 209                 handleRunnableError(e);
 210             }
 211             return true;
 212         }
 213
 214         final Thread senderThread = Thread.currentThread();
 215         final WaitingThread wt = new WaitingThread(senderThread);
 216         WaitingThread prevWt;
 217         Event e = new Event(runnable, new EventListener() {
 218             @Override
 219             public void eventDone(Event e) {
 220                 wt.completed(e);
 221             }}, null);
 222
 223         synchronized (ThreadUtils.class) {
 224             // Check if target has WaitingThread. Use it if it does
 225             WaitingThread targetWt = getWaitingThread(threadAccess.getThread());
 226             Thread waitingForThread = null;
 227             if (targetWt != null)
 228             {
 229                 // Check if it is allowed to use the target WT
 230                 if (isEventQueuingAllowed(senderThread, targetWt))
 231                 {
 232                     if (targetWt.addEvent(e)) {
 233                         synchronized(wt) {
 234                             waitingForThread = targetWt.thread;
 235                             e.setThread(waitingForThread);
 236                             wt.waitFor(e);
 237                         }
 238                     }
 239                 }
 240             }
 241
 242             if (waitingForThread == null) {
 243                 synchronized(wt) {
 244                     waitingForThread = threadAccess.asyncExec(e);
 245                     if (waitingForThread==null) return false;
 246                     e.setThread(waitingForThread);
 247                     wt.waitFor(e);
 248                 }
 249             }
 250
 251             prevWt = setWaitingThread(senderThread, wt);
 252         }
 253         // run errands in the mean time
 254         wt.waitAndProcessEvents();
 255         wt.stopAcceptingEvents();
 256         // Stop accepting errands
 257         removeWaitingThread(senderThread, prevWt);
 258         // Run last events
 259         wt.waitAndProcessEvents();
 260         return true;
 261     }
 262
 263     public static boolean multiSyncExec(Collection<Executable> executions)
 264     {
 265         if (executions.isEmpty()) return true;
 266         return multiSyncExec(executions.toArray(new Executable[executions.size()]));
 267     }
 268
 269     /**
 270      * Executes a list of executables in multiple threads and waits for all to complete.
 271      *
 272      * @param executions
 273      * @return <code>true</code> once execution has completed (is this necessary?)
 274      */
 275     public static boolean multiSyncExec(Executable ... executions)
 276     {
 277         if (executions.length==0) return true;
 278         if (executions.length==1) {
 279             return syncExec(executions[0].threadAccess, executions[0].runnable);
 280         }
 281         final Thread senderThread = Thread.currentThread();
 282         final WaitingThread wt = new WaitingThread(senderThread);
 283         WaitingThread prevWt = null;
 284         synchronized (ThreadUtils.class) {
 285             for (Executable pair : executions)
 286             {
 287                 IThreadWorkQueue threadAccess = pair.threadAccess;
 288                 if (threadAccess.currentThreadAccess())
 289                     continue;
 290                 if (threadAccess instanceof BetterThreadAccess)
 291                     threadAccess = ((BetterThreadAccess) threadAccess).ta;
 292                 Runnable runnable = pair.runnable;
 293
 294                 Event e = new Event(runnable, new EventListener() {
 295                     @Override
 296                     public void eventDone(Event e) {
 297                         wt.completed(e);
 298                     }}, null);
 299
 300                 // Check if target has WaitingThread. Use it if it does
 301                 WaitingThread targetWt = getWaitingThread(threadAccess.getThread());
 302                 Thread waitingForThread = null;
 303                 if (targetWt != null)
 304                 {
 305                     // Check if it is allowed to use the target WT
 306                     if (isEventQueuingAllowed(senderThread, targetWt))
 307                     {
 308                         if (targetWt.addEvent(e)) {
 309                             synchronized(wt) {
 310                                 waitingForThread = targetWt.thread;
 311                                 e.setThread(waitingForThread);
 312                                 wt.waitFor(e);
 313                             }
 314                         }
 315                     }
 316                 }
 317
 318                 if (waitingForThread == null) {
 319                     synchronized(wt) {
 320                         waitingForThread = threadAccess.asyncExec(e);
 321                         if (waitingForThread==null)
 322                             return false;
 323                         e.setThread(waitingForThread);
 324                         wt.waitFor(e);
 325                     }
 326                 }
 327             }
 328             prevWt = setWaitingThread(senderThread, wt);
 329         }
 330
 331         // Run local runnables
 332         for (Executable pair : executions)
 333         {
 334             IThreadWorkQueue threadAccess = pair.threadAccess;
 335             Runnable runnable = pair.runnable;
 336             if (threadAccess.currentThreadAccess())
 337                 try {
 338                     runnable.run();
 339                 } catch (RuntimeException e) {
 340                     handleRunnableError(e);
 341                 }
 342         }
 343
 344         // run errands in the mean time
 345         wt.waitAndProcessEvents();
 346         // Stop accepting errands
 347         wt.stopAcceptingEvents();
 348         removeWaitingThread(senderThread, prevWt);
 349         // Run last events
 350         wt.waitAndProcessEvents();
 351         return true;
 352     }
 353
 354     public static Thread asyncExec(IThreadWorkQueue threadAccess, final Runnable runnable)
 355     {
 356         if (threadAccess instanceof BetterThreadAccess)
 357             threadAccess = ((BetterThreadAccess) threadAccess).ta;
 358
 359         final Thread senderThread = Thread.currentThread();
 360         synchronized (ThreadUtils.class) {
 361
 362             Event e = new Event(runnable, null, null);
 363
 364             // Check if target has WaitingThread. Use it if it does
 365             WaitingThread targetWt = getWaitingThread(threadAccess.getThread());
 366             if (targetWt != null)
 367             {
 368                 //      Check if it is allowed to use the target WT
 369                 if (isEventQueuingAllowed(senderThread, targetWt))
 370                 {
 371                     if (targetWt.addEvent(e))
 372                         return targetWt.thread;
 373                 }
 374             }
 375             return threadAccess.asyncExec(runnable);
 376         }
 377     }
 378
 379     private static boolean _waitsFor(Thread sourceThread, Thread targetThread, Set<Thread> visitedTargetThreads)
 380     {
 381         assert(targetThread!=null);
 382         if (visitedTargetThreads.contains(targetThread))
 383             return false;
 384         visitedTargetThreads.add(targetThread);
 385         if (sourceThread == targetThread) return false;
 386         Set<Thread> waitsFor = getWaitsForThreads(targetThread);
 387         if (waitsFor==null||waitsFor.isEmpty()) return false;
 388         for (Thread aThreadTargetThreadWaitsFor : waitsFor)
 389         {
 390             if (aThreadTargetThreadWaitsFor==sourceThread) return true;
 391             if (visitedTargetThreads.contains(aThreadTargetThreadWaitsFor)) continue;
 392             if (_waitsFor(sourceThread, aThreadTargetThreadWaitsFor, visitedTargetThreads))
 393                 return true;
 394         }
 395         return false;
 396     }
 397
 398     static boolean waitsFor(Thread sourceThread, Thread targetThread)
 399     {
 400         return _waitsFor(sourceThread, targetThread, new HashSet<Thread>(3));
 401     }
 402
 403     static boolean isEventQueuingAllowed(Thread sourceThread, WaitingThread eventQueue)
 404     {
 405         if (!eventQueue.acceptEvents) return false;
 406         // queuing is allowed if target thread does wait for source thread
 407         return waitsFor(sourceThread, eventQueue.thread);
 408     }
 409
 410     private static void handleRunnableError(Throwable t)
 411     {
 412         t.printStackTrace();
 413     }
 414
 415     interface EventListener {
 416         void eventDone(Event e);
 417     }
 418
 419     public static class Event implements Runnable {
 420         Runnable r;
 421         EventListener l;
 422         Semaphore s;
 423         Thread t;
 424         public Event(Runnable r, EventListener l, Semaphore s) {
 425             this.r = r;
 426             this.l = l;
 427             this.s = s;
 428         }
 429         public void run() {
 430             setThread(Thread.currentThread());
 431             try {
 432                 r.run();
 433             } catch (RuntimeException e) {
 434                 handleRunnableError(e);
 435             } finally {
 436                 if (s!=null)
 437                     s.release(1);
 438                 if (l!=null)
 439                     l.eventDone(this);
 440             }
 441         }
 442         public synchronized Thread getThread()
 443         {
 444             while (t==null) {
 445                 try {
 446                     t.wait();
 447                 } catch (InterruptedException e) {
 448                 }
 449             }
 450             return t;
 451         }
 452         public synchronized void setThread(Thread t)
 453         {
 454             assert(t!=null);
 455             if (this.t!=null)
 456                 assert(this.t==t);
 457             this.t = t;
 458             notify();
 459         }
 460     }
 461
 462     /**
 463      * WaitingThread is a thread that waits for something and in the meantime
 464      * runs errands.
 465      */
 466     static class WaitingThread
 467     {
 468         final Thread thread;
 469         LinkedList<Event> queue = new LinkedList<Event>();
 470         boolean acceptEvents = true;
 471         Set<Event> waitingFor = new HashSet<Event>();
 472         Set<Event> completed = new HashSet<Event>();
 473         public WaitingThread(Thread thread) {
 474             this.thread = thread;
 475         }
 476         public synchronized void waitFor(Event event)
 477         {
 478             assert(thread!=null);
 479             waitingFor.add(event);
 480         }
 481         public synchronized void completed(Event event)
 482         {
 483             //assert(waitingFor.contains(event));
 484             completed.add(event);
 485             if (completed.size()==waitingFor.size())
 486                 notify();
 487         }
 488         synchronized boolean isEmpty() {
 489             return queue.isEmpty();
 490         }
 491         synchronized boolean keepWaiting() {
 492             if(waitingFor.size()!=completed.size())
 493                 return true;
 494             assert(waitingFor.equals(completed));
 495             return false;
 496         }
 497         public synchronized boolean addEvent(Event r)
 498         {
 499             if (!acceptEvents) return false;
 500             queue.add(r);
 501             notify();
 502             return true;
 503         }
 504
 505         static private int WAIT_MS = 10000;
 506         static private int WAIT_THRESHOLD_NS = 900000*WAIT_MS;
 507
 508         public void waitAndProcessEvents()
 509         {
 510             while (keepWaiting() || !isEmpty()) {
 511                 Event e = null;
 512                 synchronized(this)
 513                 {
 514                     if (!queue.isEmpty())
 515                         e = queue.pop();
 516                     if (e==null && keepWaiting())
 517                         try {
 518                                 long now = System.nanoTime();
 519                             wait(WAIT_MS);
 520                             long duration = System.nanoTime()-now;
 521                             if(duration > (WAIT_THRESHOLD_NS)) {
 522                                 for(Thread t : getWaitingForThreads())
 523                                     if(!t.isAlive())
 524                                         throw new IllegalStateException("Thread '" + thread + "' has died.");
 525                             }
 526                         } catch (InterruptedException e1) {
 527                         }
 528                 }
 529                 if (e!=null) {
 530                     try {
 531                         e.run();
 532                     } catch (RuntimeException e1) {
 533                         e1.printStackTrace();
 534                     }
 535                 }
 536             }
 537             while (!isEmpty())
 538             {
 539                 Event e = null;
 540                 synchronized(this)
 541                 {
 542                     if (!queue.isEmpty())
 543                         e = queue.pop();
 544                 }
 545                 if (e!=null)
 546                     try {
 547                         e.run();
 548                     } catch (RuntimeException e1) {
 549                         e1.printStackTrace();
 550                     }
 551             }
 552
 553         }
 554         public synchronized void stopAcceptingEvents()
 555         {
 556             acceptEvents = false;
 557         }
 558         public synchronized Set<Thread> getWaitingForThreads()
 559         {
 560             Set<Thread> result = new HashSet<Thread>(waitingFor.size());
 561             for (Event e : waitingFor)
 562             {
 563                 if (completed.contains(e)) continue;
 564                 result.add(e.getThread());
 565             }
 566             return result;
 567         }
 568     }
 569
 570     /**
 571      * Blocks until waiting thread has been set for thread t
 572      * @param t thread
 573      * @return waiting thread
 574      */
 575     synchronized static WaitingThread getWaitingThreadSync(Thread t)
 576     {
 577         WaitingThread result;
 578         do {
 579             result = map.get(t);
 580             if (result!=null) return result;
 581             try {
 582                 ThreadUtils.class.wait();
 583             } catch (InterruptedException e) {
 584             }
 585         } while (true);
 586     }
 587
 588     /**
 589      * Return an array which may have nulls
 590      * @param t
 591      * @return
 592      */
 593     static Set<Thread> getWaitsForThreads(Thread t)
 594     {
 595         WaitingThread wt = getWaitingThread(t);
 596         if (wt==null) return null;
 597         return wt.getWaitingForThreads();
 598     }
 599
 600     static synchronized WaitingThread getWaitingThread(Thread t)
 601     {
 602         return map.get(t);
 603     }
 604
 605     static synchronized WaitingThread setWaitingThread(Thread t, WaitingThread wt)
 606     {
 607         WaitingThread prev = map.put(t, wt);
 608         ThreadUtils.class.notifyAll();
 609         return prev;
 610     }
 611
 612     static synchronized void removeWaitingThread(Thread t, WaitingThread replaceWith)
 613     {
 614         assert(t == Thread.currentThread());
 615         map.remove(t);
 616         if (replaceWith!=null)
 617             map.put(t, replaceWith);
 618     }
 619
 620     static class BetterThreadAccess implements IThreadWorkQueue {
 621         IThreadWorkQueue ta;
 622         public BetterThreadAccess(IThreadWorkQueue ta)
 623         {
 624             if (ta instanceof BetterThreadAccess)
 625                 ta  = ((BetterThreadAccess) ta).ta;
 626             this.ta = ta;
 627         }
 628         @Override
 629         public Thread asyncExec(Runnable runnable) {
 630             return ThreadUtils.asyncExec(ta, runnable);
 631         }
 632         @Override
 633         public boolean currentThreadAccess() {
 634             return ta.currentThreadAccess();
 635         }
 636         @Override
 637         public Thread getThread() {
 638             return ta.getThread();
 639         }
 640         @Override
 641         public boolean syncExec(Runnable runnable) {
 642             return ThreadUtils.syncExec(ta, runnable);
 643         }
 644     }
 645
 646
 647
 648     /**
 649      * Executes command in the executor while maintaining possibility to
 650      * return back to caller.
 651      *
 652      * @param executor
 653      * @param command
 654      */
 655     public static void exec(Executor executor, final Runnable command)
 656     {
 657         final Thread[] calleeThread = new Thread[1];
 658         final Thread callerThread = Thread.currentThread();
 659         final Runnable wrappedCommand = new Runnable() {
 660             @Override
 661             public void run() {
 662                 calleeThread[0] = Thread.currentThread();
 663                 try {
 664                     command.run();
 665                 } finally {
 666                 }
 667             }
 668         };
 669
 670         // add dependency
 671         DEPENDENCIES.put(callerThread, calleeThread);
 672         executor.execute(wrappedCommand);
 673         DEPENDENCIES.remove(callerThread);
 674     }
 675     private static Map<Thread, Thread[]> DEPENDENCIES =
 676         Collections.synchronizedMap( new HashMap<Thread, Thread[]>() );
 677     @SuppressWarnings("unused")
 678     private static boolean hasDependency(Thread waiter, Thread worker)
 679     {
 680         // Follow dependency chain until worker is reached
 681         for (Thread t = waiter; t!=null;)
 682         {
 683             Thread[] potentialResult = DEPENDENCIES.get(t);
 684             if (potentialResult==null) break;
 685             t = potentialResult[0];
 686             if (t==worker) return true;
 687         }
 688         return false;
 689     }
 690
 691     /**
 692      * Lock multiple locks simultaneously. If all locks cannot be locked the
 693      * thread sleeps a moment and tries again.
 694      *
 695      * If all locks cannot be locked at once, this method doesn't prevent
 696      * other from locking them in the mean time.
 697      *
 698      * @param locks an array of locks. null values are ignored
 699      */
 700     public static void lock(Lock...locks) {
 701         if (locks.length==0) return;
 702         if (locks.length==1) {
 703             locks[0].lock();
 704             return;
 705         }
 706
 707         while (true) {
 708             int i = 0;
 709             for (;i<locks.length; i++) {
 710                 Lock l = locks[i];
 711                 if (l==null) continue;
 712                 if ( !locks[i].tryLock() ) break;
 713             }
 714             if (i==locks.length) return;
 715             for (int j=0; j<i; j++) {
 716                 Lock l = locks[j];
 717                 if (l==null) continue;
 718                 l.unlock();
 719             }
 720             try { NANOSECONDS.sleep(10000); } catch (InterruptedException e) {}
 721         }
 722     }
 723
 724     /**
 725      * Lock multiple locks simultaneously. If all locks cannot be locked the
 726      * thread sleeps a moment and tries again.
 727      *
 728      * If all locks cannot be locked at once, this method doesn't prevent
 729      * other from locking them in the mean time.
 730      *
 731      * [UNTESTED]
 732      *
 733      * @param locks1 an array of locks. null values are ignored
 734      * @param locks2 an array of locks. null values are ignored
 735      */
 736     public static void lock2(Lock[] locks1, Lock[] locks2) {
 737         int l1 = locks1.length;
 738         int l2 = locks2.length;
 739         int c = l1+l2;
 740         if (l1==0 && l2==0) return;
 741
 742         while (true) {
 743             int i = 0;
 744             // Attempt to lock everything in locks 1
 745             for (;i<l1; i++) {
 746                 Lock l = locks1[i];
 747                 if (l==null) continue;
 748                 if ( !locks1[i].tryLock() ) break;
 749             }
 750             // Attempt to lock everything in locks 2
 751             if (i==l1) {
 752                 for (;i<c; i++) {
 753                     Lock l = locks2[i];
 754                     if (l==null) continue;
 755                     if ( !locks2[i-l1].tryLock() ) break;
 756                 }
 757             }
 758             // All locked?
 759             if (i==c) return;
 760             // Cancel locks2
 761             if (i>l1) {
 762                 for (int j=l1; j<i; j++) {
 763                     Lock l = locks2[j-l1];
 764                     if (l==null) continue;
 765                     l.unlock();
 766                 }
 767             }
 768             // Cancel locks 1
 769             if (i>0) {
 770                 for (int j=0; j<i; j++) {
 771                     Lock l = locks1[j];
 772                     if (l==null) continue;
 773                     l.unlock();
 774                 }
 775             }
 776             try { NANOSECONDS.sleep(1); } catch (InterruptedException e) {}
 777         }
 778     }
 779
 780     /**
 781      * Try to lock multiple locks simultaneously. If all locks cannot be locked the
 782      * thread sleeps a moment and tries again.
 783      *
 784      * If all locks cannot be locked at once, this method doesn't prevent
 785      * other from locking them in the mean time.
 786      *
 787      * @param locks an array of locks. null values are ignored
 788      * @return
 789      */
 790     public static boolean tryLock(Lock...locks) {
 791         if (locks.length==0) return true;
 792         if (locks.length==1) {
 793             return locks[0].tryLock();
 794         }
 795
 796         int i = 0;
 797         for (;i<locks.length; i++) {
 798             Lock l = locks[i];
 799             if (l==null) continue;
 800             if ( !locks[i].tryLock() ) break;
 801         }
 802         if (i==locks.length) return true;
 803
 804         // Unlock & fail
 805         for (int j=0; j<i; j++) {
 806             Lock l = locks[j];
 807             if (l==null) continue;
 808             l.unlock();
 809         }
 810         return false;
 811     }
 812
 813
 814     /**
 815      * Unlock multiple locks. Use this after locking multiple locks.
 816      *
 817      * @param locks an array of locks. Null values are ignored
 818      */
 819     public static void unlock(Lock...locks) {
 820         for (Lock lock : locks) {
 821             if (lock != null) lock.unlock();
 822         }
 823     }
 824
 825     /**
 826      * Unlock multiple locks. Use this after locking multiple locks.
 827      *
 828      * @param locks1 an array of locks. Null values are ignored
 829      * @param locsk2 an array of locks. Null values are ignored
 830      */
 831     public static void unlock2(Lock[] locks1, Lock[] locks2) {
 832         for (Lock lock : locks1) {
 833             if (lock != null) lock.unlock();
 834         }
 835         for (Lock lock : locks2) {
 836             if (lock != null) lock.unlock();
 837         }
 838     }
 839
 840     /**
 841      * Join multiple arrays into a single array
 842      *
 843      * @param lockArrays
 844      * @return
 845      */
 846     public static Lock[] appendLockArrays(Lock[]...lockArrays) {
 847         int len = 0;
 848         for (Lock[] array : lockArrays) len +=array.length;
 849         Lock[] result = new Lock[ len ];
 850         int i = 0;
 851         for (Lock[] array : lockArrays) {
 852             System.arraycopy(array, 0, result, i, array.length);
 853             i += array.length;
 854         }
 855         return result;
 856     }
 857
 858     public static synchronized void shutdown() {
 859         if (TIMER != null) {
 860             //System.out.println("TIMERS");
 861             shutdownAndAwaitTermination(TIMER, 1000);
 862             TIMER = null;
 863         }
 864         if (NON_BLOCKING_EXECUTOR != null) {
 865             //System.out.println("NON_BLOCKING");
 866             shutdownAndAwaitTermination(NON_BLOCKING_EXECUTOR, 1000);
 867             NON_BLOCKING_EXECUTOR = null;
 868         }
 869         if (BLOCKING_EXECUTOR != null) {
 870             //System.out.println("BLOCKING");
 871             shutdownAndAwaitTermination(BLOCKING_EXECUTOR, 1000);
 872             BLOCKING_EXECUTOR = null;
 873         }
 874     }
 875
 876     /**
 877      * Grabbed from {@link ExecutorService} javadoc.
 878      *
 879      * @param pool {@link ExecutorService} to shut down
 880      */
 881     private static void shutdownAndAwaitTermination(ExecutorService pool, long timeoutMs) {
 882         //long t = System.currentTimeMillis();
 883         pool.shutdown(); // Disable new tasks from being submitted
 884         try {
 885             // Wait a while for existing tasks to terminate
 886             if (!pool.awaitTermination(timeoutMs, TimeUnit.MILLISECONDS)) {
 887                 List<Runnable> leftovers = pool.shutdownNow(); // Cancel currently executing tasks
 888                 if (!leftovers.isEmpty())
 889                     System.err.println("Thread pool '" + pool.toString()  + "' contained " + leftovers.size() + " tasks at forced shutdown: " + leftovers);
 890                 // Wait a while for tasks to respond to being cancelled
 891                 if (!pool.awaitTermination(timeoutMs, TimeUnit.MILLISECONDS))
 892                     System.err.println("Thread pool '" + pool.toString()  + "' did not terminate");
 893             }
 894         } catch (InterruptedException ie) {
 895             // (Re-)Cancel if current thread also interrupted
 896             pool.shutdownNow();
 897             // Preserve interrupt status
 898             Thread.currentThread().interrupt();
 899         }
 900         //long e = System.currentTimeMillis();
 901         //System.out.println("shutdown took: " + ((e-t)*1e-3) + " ms");
 902     }
 903
 904     @SuppressWarnings("unused")
 905     private static void uncheckedAwaitTermination(ExecutorService service, long time) {
 906         try {
 907             NON_BLOCKING_EXECUTOR.awaitTermination(time, TimeUnit.MILLISECONDS);
 908         } catch (InterruptedException e) {
 909         }
 910     }
 911
 912     // Executor support
 913
 914     // Executor that runs in current thread
 915     public static Executor CURRENT_THREAD = new CurrentThreadExecutor();
 916
 917     // Async executor queues the command into AWT event queue
 918     public static ExecutorService AWT_EDT = AWTThread.INSTANCE;
 919
 920     // Sync executor blocks the call until the command is finished
 921     public static ExecutorService AWT_EDT_SYNC = new AWTExecutorSync();
 922
 923 }
 924
 925 class AWTExecutorSync extends AbstractExecutorService {
 926
 927     @Override
 928     public void execute(Runnable command) {
 929         if (EventQueue.isDispatchThread())
 930         {
 931             command.run();
 932         } else {
 933             try {
 934                 EventQueue.invokeAndWait(command);
 935             } catch (InterruptedException e) {
 936                 throw new RuntimeException(e);
 937             } catch (InvocationTargetException e) {
 938                 throw new RuntimeException(e.getCause());
 939             }
 940         }
 941     }
 942
 943     @Override
 944     public void shutdown() {
 945     }
 946
 947     @Override
 948     public List<Runnable> shutdownNow() {
 949         return null;
 950     }
 951
 952     @Override
 953     public boolean isShutdown() {
 954         return false;
 955     }
 956
 957     @Override
 958     public boolean isTerminated() {
 959         return false;
 960     }
 961
 962     @Override
 963     public boolean awaitTermination(long timeout, TimeUnit unit)
 964             throws InterruptedException {
 965         return false;
 966     }
 967
 968     public static String getStackTrace(int levels){
 969         StringBuilder sb = new StringBuilder();
 970         sb.append( Thread.currentThread() );
 971         sb.append( "\n" );
 972                 Exception e3 = new Exception();
 973                 try { throw e3; } catch(Exception e2) {
 974
 975                         for (int i=1; i<Math.max(e2.getStackTrace().length, levels+1); i++) {
 976                                 StackTraceElement ste = e2.getStackTrace()[i];
 977                                 sb.append("  - ");
 978                                 sb.append(ste);
 979                         sb.append( "\n" );
 980                         }
 981                 }
 982         return sb.toString();
 983     }
 984 }
 985
 986 class CurrentThreadExecutor implements Executor {
 987     @Override
 988     public void execute(Runnable command) {
 989         command.run();
 990     }
 991 }