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