Find Usages Diff Raw Download HTML Widget
Start line:  
End line:  

Snippet Preview

Snippet HTML Code

Stack Overflow Questions 
1
   // Revision 1.12
2
   
3
   /*
4
    * Written by Doug Lea with assistance from members of JCP JSR-166
5
    * Expert Group and released to the public domain, as explained at
6
    * http://creativecommons.org/publicdomain/zero/1.0/
7
    */
8
   
9
   package org.infinispan.commons.util.concurrent.jdk8backported;
10
  
13
  import java.util.List;
Abstract base class for tasks that run within a ForkJoinPool. A ForkJoinTask is a thread-like entity that is much lighter weight than a normal thread. Huge numbers of tasks and subtasks may be hosted by a small number of actual threads in a ForkJoinPool, at the price of some usage limitations.

A "main" ForkJoinTask begins execution when it is explicitly submitted to a ForkJoinPool, or, if not already engaged in a ForkJoin computation, commenced in the ForkJoinPool.commonPool() via fork(), invoke(), or related methods. Once started, it will usually in turn start other subtasks. As indicated by the name of this class, many programs using ForkJoinTask employ only methods fork() and join(), or derivatives such as invokeAll(org.infinispan.commons.util.concurrent.jdk8backported.ForkJoinTask[]). However, this class also provides a number of other methods that can come into play in advanced usages, as well as extension mechanics that allow support of new forms of fork/join processing.

A ForkJoinTask is a lightweight form of java.util.concurrent.Future. The efficiency of ForkJoinTasks stems from a set of restrictions (that are only partially statically enforceable) reflecting their main use as computational tasks calculating pure functions or operating on purely isolated objects. The primary coordination mechanisms are fork(), that arranges asynchronous execution, and join(), that doesn't proceed until the task's result has been computed. Computations should ideally avoid synchronized methods or blocks, and should minimize other blocking synchronization apart from joining other tasks or using synchronizers such as Phasers that are advertised to cooperate with fork/join scheduling. Subdividable tasks should also not perform blocking I/O, and should ideally access variables that are completely independent of those accessed by other running tasks. These guidelines are loosely enforced by not permitting checked exceptions such as IOExceptions to be thrown. However, computations may still encounter unchecked exceptions, that are rethrown to callers attempting to join them. These exceptions may additionally include java.util.concurrent.RejectedExecutionException stemming from internal resource exhaustion, such as failure to allocate internal task queues. Rethrown exceptions behave in the same way as regular exceptions, but, when possible, contain stack traces (as displayed for example using ex.printStackTrace()) of both the thread that initiated the computation as well as the thread actually encountering the exception; minimally only the latter.

It is possible to define and use ForkJoinTasks that may block, but doing do requires three further considerations: (1) Completion of few if any other tasks should be dependent on a task that blocks on external synchronization or I/O. Event-style async tasks that are never joined (for example, those subclassing CountedCompleter) often fall into this category. (2) To minimize resource impact, tasks should be small; ideally performing only the (possibly) blocking action. (3) Unless the ForkJoinPool.ManagedBlocker API is used, or the number of possibly blocked tasks is known to be less than the pool's ForkJoinPool.getParallelism() level, the pool cannot guarantee that enough threads will be available to ensure progress or good performance.

The primary method for awaiting completion and extracting results of a task is join(), but there are several variants: The java.util.concurrent.Future.get() methods support interruptible and/or timed waits for completion and report results using Future conventions. Method invoke() is semantically equivalent to fork(); join() but always attempts to begin execution in the current thread. The "quiet" forms of these methods do not extract results or report exceptions. These may be useful when a set of tasks are being executed, and you need to delay processing of results or exceptions until all complete. Method invokeAll (available in multiple versions) performs the most common form of parallel invocation: forking a set of tasks and joining them all.

In the most typical usages, a fork-join pair act like a call (fork) and return (join) from a parallel recursive function. As is the case with other forms of recursive calls, returns (joins) should be performed innermost-first. For example, a.fork(); b.fork(); b.join(); a.join(); is likely to be substantially more efficient than joining a before b.

The execution status of tasks may be queried at several levels of detail: isDone() is true if a task completed in any way (including the case where a task was cancelled without executing); isCompletedNormally() is true if a task completed without cancellation or encountering an exception; isCancelled() is true if the task was cancelled (in which case getException() returns a java.util.concurrent.CancellationException); and isCompletedAbnormally() is true if a task was either cancelled or encountered an exception, in which case getException() will return either the encountered exception or java.util.concurrent.CancellationException.

The ForkJoinTask class is not usually directly subclassed. Instead, you subclass one of the abstract classes that support a particular style of fork/join processing, typically RecursiveAction for most computations that do not return results, RecursiveTask for those that do, and CountedCompleter for those in which completed actions trigger other actions. Normally, a concrete ForkJoinTask subclass declares fields comprising its parameters, established in a constructor, and then defines a compute method that somehow uses the control methods supplied by this base class.

Method join() and its variants are appropriate for use only when completion dependencies are acyclic; that is, the parallel computation can be described as a directed acyclic graph (DAG). Otherwise, executions may encounter a form of deadlock as tasks cyclically wait for each other. However, this framework supports other methods and techniques (for example the use of java.util.concurrent.Phaser, helpQuiesce(), and complete(java.lang.Object)) that may be of use in constructing custom subclasses for problems that are not statically structured as DAGs. To support such usages a ForkJoinTask may be atomically tagged with a short value using setForkJoinTaskTag(short) or compareAndSetForkJoinTaskTag(short,short) and checked using getForkJoinTaskTag(). The ForkJoinTask implementation does not use these protected methods or tags for any purpose, but they may be of use in the construction of specialized subclasses. For example, parallel graph traversals can use the supplied methods to avoid revisiting nodes/tasks that have already been processed. (Method names for tagging are bulky in part to encourage definition of methods that reflect their usage patterns.)

Most base support methods are final, to prevent overriding of implementations that are intrinsically tied to the underlying lightweight task scheduling framework. Developers creating new basic styles of fork/join processing should minimally implement protected methods exec(), setRawResult(java.lang.Object), and getRawResult(), while also introducing an abstract computational method that can be implemented in its subclasses, possibly relying on other protected methods provided by this class.

ForkJoinTasks should perform relatively small amounts of computation. Large tasks should be split into smaller subtasks, usually via recursive decomposition. As a very rough rule of thumb, a task should perform more than 100 and less than 10000 basic computational steps, and should avoid indefinite looping. If tasks are too big, then parallelism cannot improve throughput. If too small, then memory and internal task maintenance overhead may overwhelm processing.

This class provides adapt methods for java.lang.Runnable and java.util.concurrent.Callable, that may be of use when mixing execution of ForkJoinTasks with other kinds of tasks. When all tasks are of this form, consider using a pool constructed in asyncMode.

ForkJoinTasks are Serializable, which enables them to be used in extensions such as remote execution frameworks. It is sensible to serialize tasks only before or after, but not during, execution. Serialization is not relied on during execution itself.

Author(s):
Doug Lea
Since:
1.7
185
 
186
 @SuppressWarnings("restriction")
187
 public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
188
 
189
     /*
190
      * See the internal documentation of class ForkJoinPool for a
191
      * general implementation overview.  ForkJoinTasks are mainly
192
      * responsible for maintaining their "status" field amidst relays
193
      * to methods in ForkJoinWorkerThread and ForkJoinPool.
194
      *
195
      * The methods of this class are more-or-less layered into
196
      * (1) basic status maintenance
197
      * (2) execution and awaiting completion
198
      * (3) user-level methods that additionally report results.
199
      * This is sometimes hard to see because this file orders exported
200
      * methods in a way that flows well in javadocs.
201
      */
202
 
203
     /*
204
      * The status field holds run control status bits packed into a
205
      * single int to minimize footprint and to ensure atomicity (via
206
      * CAS).  Status is initially zero, and takes on nonnegative
207
      * values until completed, upon which status (anded with
208
      * DONE_MASK) holds value NORMAL, CANCELLED, or EXCEPTIONAL. Tasks
209
      * undergoing blocking waits by other threads have the SIGNAL bit
210
      * set.  Completion of a stolen task with SIGNAL set awakens any
211
      * waiters via notifyAll. Even though suboptimal for some
212
      * purposes, we use basic builtin wait/notify to take advantage of
213
      * "monitor inflation" in JVMs that we would otherwise need to
214
      * emulate to avoid adding further per-task bookkeeping overhead.
215
      * We want these monitors to be "fat", i.e., not use biasing or
216
      * thin-lock techniques, so use some odd coding idioms that tend
217
      * to avoid them, mainly by arranging that every synchronized
218
      * block performs a wait, notifyAll or both.
219
      *
220
      * These control bits occupy only (some of) the upper half (16
221
      * bits) of status field. The lower bits are used for user-defined
222
      * tags.
223
      */

   
The run status of this task
225
 
226
    volatile int status; // accessed directly by pool and workers
227
    static final int DONE_MASK   = 0xf0000000;  // mask out non-completion bits
228
    static final int NORMAL      = 0xf0000000;  // must be negative
229
    static final int CANCELLED   = 0xc0000000;  // must be < NORMAL
230
    static final int EXCEPTIONAL = 0x80000000;  // must be < CANCELLED
231
    static final int SIGNAL      = 0x00010000;  // must be >= 1 << 16
232
    static final int SMASK       = 0x0000ffff;  // short bits for tags
233
 
   
Marks completion and wakes up threads waiting to join this task.

Parameters:
completion one of NORMAL, CANCELLED, EXCEPTIONAL
Returns:
completion status on exit
240
 
241
    private int setCompletion(int completion) {
242
       for (int s;;) {
243
          if ((s = ) < 0)
244
             return s;
245
          if (.compareAndSwapInt(this, , s, s | completion)) {
246
             if ((s >>> 16) != 0)
247
                synchronized (this) { notifyAll(); }
248
             return completion;
249
          }
250
       }
251
    }

   
Primary execution method for stolen tasks. Unless done, calls exec and records status if completed, but doesn't wait for completion otherwise.

Returns:
status on exit from this method
259
 
260
    final int doExec() {
261
       int s; boolean completed;
262
       if ((s = ) >= 0) {
263
          try {
264
             completed = exec();
265
          } catch (Throwable rex) {
266
             return setExceptionalCompletion(rex);
267
          }
268
          if (completed)
269
             s = setCompletion();
270
       }
271
       return s;
272
    }

   
Tries to set SIGNAL status unless already completed. Used by ForkJoinPool. Other variants are directly incorporated into externalAwaitDone etc.

Returns:
true if successful
280
 
281
    final boolean trySetSignal() {
282
       int s = ;
283
       return s >= 0 && .compareAndSwapInt(this, , s, s | );
284
    }

   
Blocks a non-worker-thread until completion.

Returns:
status upon completion
289
 
290
    private int externalAwaitDone() {
291
       int s;
292
       ForkJoinPool.externalHelpJoin(this);
293
       boolean interrupted = false;
294
       while ((s = ) >= 0) {
295
          if (.compareAndSwapInt(this, , s, s | )) {
296
             synchronized (this) {
297
                if ( >= 0) {
298
                   try {
299
                      wait();
300
                   } catch (InterruptedException ie) {
301
                      interrupted = true;
302
                   }
303
                }
304
                else
305
                   notifyAll();
306
             }
307
          }
308
       }
309
       if (interrupted)
310
          Thread.currentThread().interrupt();
311
       return s;
312
    }

   
Blocks a non-worker-thread until completion or interruption. 
316
 
317
    private int externalInterruptibleAwaitDone() throws InterruptedException {
318
       int s;
319
       if (Thread.interrupted())
320
          throw new InterruptedException();
321
       ForkJoinPool.externalHelpJoin(this);
322
       while ((s = ) >= 0) {
323
          if (.compareAndSwapInt(this, , s, s | )) {
324
             synchronized (this) {
325
                if ( >= 0)
326
                   wait();
327
                else
328
                   notifyAll();
329
             }
330
          }
331
       }
332
       return s;
333
    }


   
Implementation for join, get, quietlyJoin. Directly handles only cases of already-completed, external wait, and unfork+exec. Others are relayed to ForkJoinPool.awaitJoin.

Returns:
status upon completion
342
 
343
    private int doJoin() {
344
       int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w;
345
       return (s = ) < 0 ? s :
346
             ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
347
                   (w = (wt = (ForkJoinWorkerThread)t).).
348
                         tryUnpush(this) && (s = doExec()) < 0 ? s :
349
                         wt.pool.awaitJoin(w, this) :
350
                   externalAwaitDone();
351
    }

   
Implementation for invoke, quietlyInvoke.

Returns:
status upon completion
357
 
358
    private int doInvoke() {
359
       int s; Thread t; ForkJoinWorkerThread wt;
360
       return (s = doExec()) < 0 ? s :
361
             ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
362
                   (wt = (ForkJoinWorkerThread)t)..awaitJoin(wt.workQueue, this) :
363
                   externalAwaitDone();
364
    }
365
 
366
    // Exception table support
367
 
   
Table of exceptions thrown by tasks, to enable reporting by callers. Because exceptions are rare, we don't directly keep them with task objects, but instead use a weak ref table. Note that cancellation exceptions don't appear in the table, but are instead recorded as status values. Note: These statics are initialized below in static block. 
376
 
377
    private static final ExceptionNode[] exceptionTable;
378
    private static final ReentrantLock exceptionTableLock;
379
    private static final ReferenceQueue<Object> exceptionTableRefQueue;

   
Fixed capacity for exceptionTable. 
383
 
384
    private static final int EXCEPTION_MAP_CAPACITY = 32;

   
Key-value nodes for exception table. The chained hash table uses identity comparisons, full locking, and weak references for keys. The table has a fixed capacity because it only maintains task exceptions long enough for joiners to access them, so should never become very large for sustained periods. However, since we do not know when the last joiner completes, we must use weak references and expunge them. We do so on each operation (hence full locking). Also, some thread in any ForkJoinPool will call helpExpungeStaleExceptions when its pool becomes isQuiescent. 
397
 
398
    static final class ExceptionNode extends WeakReference<ForkJoinTask<?>> {
399
       final Throwable ex;
400
       ExceptionNode next;
401
       final long thrower;  // use id not ref to avoid weak cycles
402
       ExceptionNode(ForkJoinTask<?> task, Throwable ex, ExceptionNode next) {
403
          super(task, );
404
          this. = ex;
405
          this. = next;
406
          this. = Thread.currentThread().getId();
407
       }
408
    }

   
Records exception and sets status.

Returns:
status on exit
414
 
415
    final int recordExceptionalCompletion(Throwable ex) {
416
       int s;
417
       if ((s = ) >= 0) {
418
          int h = System.identityHashCode(this);
419
          final ReentrantLock lock = ;
420
          lock.lock();
421
          try {
422
             expungeStaleExceptions();
423
             ExceptionNode[] t = ;
424
             int i = h & (t.length - 1);
425
             for (ExceptionNode e = t[i]; ; e = e.next) {
426
                if (e == null) {
427
                   t[i] = new ExceptionNode(this, ex, t[i]);
428
                   break;
429
                }
430
                if (e.get() == this) // already present
431
                   break;
432
             }
433
          } finally {
434
             lock.unlock();
435
          }
436
          s = setCompletion();
437
       }
438
       return s;
439
    }

   
Records exception and possibly propagates.

Returns:
status on exit
445
 
446
    private int setExceptionalCompletion(Throwable ex) {
447
       int s = recordExceptionalCompletion(ex);
448
       if ((s & ) == )
449
          internalPropagateException(ex);
450
       return s;
451
    }

   
Hook for exception propagation support for tasks with completers. 
455
 
457
    }

   
Cancels, ignoring any exceptions thrown by cancel. Used during worker and pool shutdown. Cancel is spec'ed not to throw any exceptions, but if it does anyway, we have no recourse during shutdown, so guard against this case. 
464
 
465
    static final void cancelIgnoringExceptions(ForkJoinTask<?> t) {
466
       if (t != null && t.status >= 0) {
467
          try {
468
             t.cancel(false);
469
          } catch (Throwable ignore) {
470
          }
471
       }
472
    }

   
Removes exception node and clears status. 
476
 
477
    private void clearExceptionalCompletion() {
478
       int h = System.identityHashCode(this);
479
       final ReentrantLock lock = ;
480
       lock.lock();
481
       try {
482
          ExceptionNode[] t = ;
483
          int i = h & (t.length - 1);
484
          ExceptionNode e = t[i];
485
          ExceptionNode pred = null;
486
          while (e != null) {
487
             ExceptionNode next = e.next;
488
             if (e.get() == this) {
489
                if (pred == null)
490
                   t[i] = next;
491
                else
492
                   pred.next = next;
493
                break;
494
             }
495
             pred = e;
496
             e = next;
497
          }
498
          expungeStaleExceptions();
499
           = 0;
500
       } finally {
501
          lock.unlock();
502
       }
503
    }

   
Returns a rethrowable exception for the given task, if available. To provide accurate stack traces, if the exception was not thrown by the current thread, we try to create a new exception of the same type as the one thrown, but with the recorded exception as its cause. If there is no such constructor, we instead try to use a no-arg constructor, followed by initCause, to the same effect. If none of these apply, or any fail due to other exceptions, we return the recorded exception, which is still correct, although it may contain a misleading stack trace.

Returns:
the exception, or null if none
518
 
519
    private Throwable getThrowableException() {
520
       if (( & ) != )
521
          return null;
522
       int h = System.identityHashCode(this);
523
       ExceptionNode e;
524
       final ReentrantLock lock = ;
525
       lock.lock();
526
       try {
527
          expungeStaleExceptions();
528
          ExceptionNode[] t = ;
529
          e = t[h & (t.length - 1)];
530
          while (e != null && e.get() != this)
531
             e = e.next;
532
       } finally {
533
          lock.unlock();
534
       }
535
       Throwable ex;
536
       if (e == null || (ex = e.ex) == null)
537
          return null;
538
       if (false && e.thrower != Thread.currentThread().getId()) {
539
          Class<? extends Throwable> ec = ex.getClass();
540
          try {
541
             Constructor<?> noArgCtor = null;
542
             Constructor<?>[] cs = ec.getConstructors();// public ctors only
543
             for (int i = 0; i < cs.length; ++i) {
544
                Constructor<?> c = cs[i];
545
                Class<?>[] ps = c.getParameterTypes();
546
                if (ps.length == 0)
547
                   noArgCtor = c;
548
                else if (ps.length == 1 && ps[0] == Throwable.class)
549
                   return (Throwable)(c.newInstance(ex));
550
             }
551
             if (noArgCtor != null) {
552
                Throwable wx = (Throwable)(noArgCtor.newInstance());
553
                wx.initCause(ex);
554
                return wx;
555
             }
556
          } catch (Exception ignore) {
557
          }
558
       }
559
       return ex;
560
    }

   
Poll stale refs and remove them. Call only while holding lock. 
564
 
565
    private static void expungeStaleExceptions() {
566
       for (Object x; (x = .poll()) != null;) {
567
          if (x instanceof ExceptionNode) {
568
             ForkJoinTask<?> key = ((ExceptionNode)x).get();
569
             ExceptionNode[] t = ;
570
             int i = System.identityHashCode(key) & (t.length - 1);
571
             ExceptionNode e = t[i];
572
             ExceptionNode pred = null;
573
             while (e != null) {
574
                ExceptionNode next = e.next;
575
                if (e == x) {
576
                   if (pred == null)
577
                      t[i] = next;
578
                   else
579
                      pred.next = next;
580
                   break;
581
                }
582
                pred = e;
583
                e = next;
584
             }
585
          }
586
       }
587
    }

   
If lock is available, poll stale refs and remove them. Called from ForkJoinPool when pools become quiescent. 
592
 
593
    static final void helpExpungeStaleExceptions() {
594
       final ReentrantLock lock = ;
595
       if (lock.tryLock()) {
596
          try {
597
             expungeStaleExceptions();
598
          } finally {
599
             lock.unlock();
600
          }
601
       }
602
    }

   
A version of "sneaky throw" to relay exceptions 
606
 
607
    static void rethrow(final Throwable ex) {
608
       if (ex != null) {
609
          if (ex instanceof Error)
610
             throw (Error)ex;
611
          if (ex instanceof RuntimeException)
612
             throw (RuntimeException)ex;
613
          ForkJoinTask.<RuntimeException>uncheckedThrow(ex);
614
       }
615
    }

   
The sneaky part of sneaky throw, relying on generics limitations to evade compiler complaints about rethrowing unchecked exceptions 
621
 
622
    @SuppressWarnings("unchecked") static <T extends Throwable>
623
    void uncheckedThrow(Throwable t) throws T {
624
       if (t != null)
625
          throw (T)t; // rely on vacuous cast
626
    }

   
Throws exception, if any, associated with the given status. 
630
 
631
    private void reportException(int s) {
632
       if (s == )
633
          throw new CancellationException();
634
       if (s == )
635
          rethrow(getThrowableException());
636
    }
637
 
638
    // public methods
639
 
   
Arranges to asynchronously execute this task in the pool the current task is running in, if applicable, or using the ForkJoinPool.commonPool() if not inForkJoinPool(). While it is not necessarily enforced, it is a usage error to fork a task more than once unless it has completed and been reinitialized. Subsequent modifications to the state of this task or any data it operates on are not necessarily consistently observable by any thread other than the one executing it unless preceded by a call to join() or related methods, or a call to isDone() returning true.

Returns:
this, to simplify usage
654
 
655
    public final ForkJoinTask<V> fork() {
656
       Thread t;
657
       if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
658
          ((ForkJoinWorkerThread)t)..push(this);
659
       else
660
          ..externalPush(this);
661
       return this;
662
    }

   
Returns the result of the computation when it is done. This method differs from get() in that abnormal completion results in RuntimeException or Error, not ExecutionException, and that interrupts of the calling thread do not cause the method to abruptly return by throwing InterruptedException.

Returns:
the computed result
674
 
675
    public final V join() {
676
       int s;
677
       if ((s = doJoin() & ) != )
678
          reportException(s);
679
       return getRawResult();
680
    }

   
Commences performing this task, awaits its completion if necessary, and returns its result, or throws an (unchecked) RuntimeException or Error if the underlying computation did so.

Returns:
the computed result
689
 
690
    public final V invoke() {
691
       int s;
692
       if ((s = doInvoke() & ) != )
693
          reportException(s);
694
       return getRawResult();
695
    }

   
Forks the given tasks, returning when isDone holds for each task or an (unchecked) exception is encountered, in which case the exception is rethrown. If more than one task encounters an exception, then this method throws any one of these exceptions. If any task encounters an exception, the other may be cancelled. However, the execution status of individual tasks is not guaranteed upon exceptional return. The status of each task may be obtained using getException() and related methods to check if they have been cancelled, completed normally or exceptionally, or left unprocessed.

Parameters:
t1 the first task
t2 the second task
Throws:
java.lang.NullPointerException if any task is null
713
 
714
    public static void invokeAll(ForkJoinTask<?> t1, ForkJoinTask<?> t2) {
715
       int s1, s2;
716
       t2.fork();
717
       if ((s1 = t1.doInvoke() & ) != )
718
          t1.reportException(s1);
719
       if ((s2 = t2.doJoin() & ) != )
720
          t2.reportException(s2);
721
    }

   
Forks the given tasks, returning when isDone holds for each task or an (unchecked) exception is encountered, in which case the exception is rethrown. If more than one task encounters an exception, then this method throws any one of these exceptions. If any task encounters an exception, others may be cancelled. However, the execution status of individual tasks is not guaranteed upon exceptional return. The status of each task may be obtained using getException() and related methods to check if they have been cancelled, completed normally or exceptionally, or left unprocessed.

Parameters:
tasks the tasks
Throws:
java.lang.NullPointerException if any task is null
737
 
738
    public static void invokeAll(ForkJoinTask<?>... tasks) {
739
       Throwable ex = null;
740
       int last = tasks.length - 1;
741
       for (int i = last; i >= 0; --i) {
742
          ForkJoinTask<?> t = tasks[i];
743
          if (t == null) {
744
             if (ex == null)
745
                ex = new NullPointerException();
746
          }
747
          else if (i != 0)
748
             t.fork();
749
          else if (t.doInvoke() <  && ex == null)
750
             ex = t.getException();
751
       }
752
       for (int i = 1; i <= last; ++i) {
753
          ForkJoinTask<?> t = tasks[i];
754
          if (t != null) {
755
             if (ex != null)
756
                t.cancel(false);
757
             else if (t.doJoin() < )
758
                ex = t.getException();
759
          }
760
       }
761
       if (ex != null)
762
          rethrow(ex);
763
    }

   
Forks all tasks in the specified collection, returning when isDone holds for each task or an (unchecked) exception is encountered, in which case the exception is rethrown. If more than one task encounters an exception, then this method throws any one of these exceptions. If any task encounters an exception, others may be cancelled. However, the execution status of individual tasks is not guaranteed upon exceptional return. The status of each task may be obtained using getException() and related methods to check if they have been cancelled, completed normally or exceptionally, or left unprocessed.

Parameters:
tasks the collection of tasks
Returns:
the tasks argument, to simplify usage
Throws:
java.lang.NullPointerException if tasks or any element are null
781
 
782
    public static <T extends ForkJoinTask<?>> Collection<T> invokeAll(Collection<T> tasks) {
783
       if (!(tasks instanceof RandomAccess) || !(tasks instanceof List<?>)) {
784
          invokeAll(tasks.toArray(new ForkJoinTask<?>[tasks.size()]));
785
          return tasks;
786
       }
787
       @SuppressWarnings("unchecked")
788
       List<? extends ForkJoinTask<?>> ts =
789
             (List<? extends ForkJoinTask<?>>) tasks;
790
       Throwable ex = null;
791
       int last = ts.size() - 1;
792
       for (int i = last; i >= 0; --i) {
793
          ForkJoinTask<?> t = ts.get(i);
794
          if (t == null) {
795
             if (ex == null)
796
                ex = new NullPointerException();
797
          }
798
          else if (i != 0)
799
             t.fork();
800
          else if (t.doInvoke() <  && ex == null)
801
             ex = t.getException();
802
       }
803
       for (int i = 1; i <= last; ++i) {
804
          ForkJoinTask<?> t = ts.get(i);
805
          if (t != null) {
806
             if (ex != null)
807
                t.cancel(false);
808
             else if (t.doJoin() < )
809
                ex = t.getException();
810
          }
811
       }
812
       if (ex != null)
813
          rethrow(ex);
814
       return tasks;
815
    }

   
Attempts to cancel execution of this task. This attempt will fail if the task has already completed or could not be cancelled for some other reason. If successful, and this task has not started when cancel is called, execution of this task is suppressed. After this method returns successfully, unless there is an intervening call to reinitialize(), subsequent calls to isCancelled(), isDone(), and cancel will return true and calls to join() and related methods will result in CancellationException.

This method may be overridden in subclasses, but if so, must still ensure that these properties hold. In particular, the cancel method itself must not throw exceptions.

This method is designed to be invoked by other tasks. To terminate the current task, you can just return or throw an unchecked exception from its computation method, or invoke completeExceptionally(java.lang.Throwable).

Parameters:
mayInterruptIfRunning this value has no effect in the default implementation because interrupts are not used to control cancellation.
Returns:
true if this task is now cancelled
843
 
844
    public boolean cancel(boolean mayInterruptIfRunning) {
845
       return (setCompletion() & ) == ;
846
    }
847
 
848
    public final boolean isDone() {
849
       return  < 0;
850
    }
851
 
852
    public final boolean isCancelled() {
853
       return ( & ) == ;
854
    }

   
Returns true if this task threw an exception or was cancelled.

Returns:
true if this task threw an exception or was cancelled
860
 
861
    public final boolean isCompletedAbnormally() {
862
       return  < ;
863
    }

   
Returns true if this task completed without throwing an exception and was not cancelled.

Returns:
true if this task completed without throwing an exception and was not cancelled
871
 
872
    public final boolean isCompletedNormally() {
873
       return ( & ) == ;
874
    }

   
Returns the exception thrown by the base computation, or a CancellationException if cancelled, or null if none or if the method has not yet completed.

Returns:
the exception, or null if none
882
 
883
    public final Throwable getException() {
884
       int s =  & ;
885
       return ((s >= )    ? null :
886
                     (s == ) ? new CancellationException() :
887
                           getThrowableException());
888
    }

   
Completes this task abnormally, and if not already aborted or cancelled, causes it to throw the given exception upon join and related operations. This method may be used to induce exceptions in asynchronous tasks, or to force completion of tasks that would not otherwise complete. Its use in other situations is discouraged. This method is overridable, but overridden versions must invoke super implementation to maintain guarantees.

Parameters:
ex the exception to throw. If this exception is not a RuntimeException or Error, the actual exception thrown will be a RuntimeException with cause ex.
903
 
904
    public void completeExceptionally(Throwable ex) {
905
       setExceptionalCompletion((ex instanceof RuntimeException) ||
906
             (ex instanceof Error) ? ex :
907
             new RuntimeException(ex));
908
    }

   
Completes this task, and if not already aborted or cancelled, returning the given value as the result of subsequent invocations of join and related operations. This method may be used to provide results for asynchronous tasks, or to provide alternative handling for tasks that would not otherwise complete normally. Its use in other situations is discouraged. This method is overridable, but overridden versions must invoke super implementation to maintain guarantees.

Parameters:
value the result value for this task
922
 
923
    public void complete(V value) {
924
       try {
925
          setRawResult(value);
926
       } catch (Throwable rex) {
927
          setExceptionalCompletion(rex);
928
          return;
929
       }
930
       setCompletion();
931
    }

   
Completes this task normally without setting a value. The most recent value established by setRawResult(java.lang.Object) (or null by default) will be returned as the result of subsequent invocations of join and related operations.

Since:
1.8
940
 
941
    public final void quietlyComplete() {
942
       setCompletion();
943
    }

   
Waits if necessary for the computation to complete, and then retrieves its result.

Returns:
the computed result
Throws:
java.util.concurrent.CancellationException if the computation was cancelled
java.util.concurrent.ExecutionException if the computation threw an exception
java.lang.InterruptedException if the current thread is not a member of a ForkJoinPool and was interrupted while waiting
955
 
956
    public final V get() throws InterruptedException, ExecutionException {
957
       int s = (Thread.currentThread() instanceof ForkJoinWorkerThread) ?
958
             doJoin() : externalInterruptibleAwaitDone();
959
       Throwable ex;
960
       if ((s &= ) == )
961
          throw new CancellationException();
962
       if (s ==  && (ex = getThrowableException()) != null)
963
          throw new ExecutionException(ex);
964
       return getRawResult();
965
    }

   
Waits if necessary for at most the given time for the computation to complete, and then retrieves its result, if available.

Parameters:
timeout the maximum time to wait
unit the time unit of the timeout argument
Returns:
the computed result
Throws:
java.util.concurrent.CancellationException if the computation was cancelled
java.util.concurrent.ExecutionException if the computation threw an exception
java.lang.InterruptedException if the current thread is not a member of a ForkJoinPool and was interrupted while waiting
java.util.concurrent.TimeoutException if the wait timed out
980
 
981
    public final V get(long timeout, TimeUnit unit)
983
       if (Thread.interrupted())
984
          throw new InterruptedException();
985
       // Messy in part because we measure in nanosecs, but wait in millisecs
986
       int s; long ms;
987
       long ns = unit.toNanos(timeout);
988
       if ((s = ) >= 0 && ns > 0L) {
989
          long deadline = System.nanoTime() + ns;
990
          ForkJoinPool p = null;
991
          ForkJoinPool.WorkQueue w = null;
992
          Thread t = Thread.currentThread();
993
          if (t instanceof ForkJoinWorkerThread) {
994
             ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;
995
             p = wt.pool;
996
             w = wt.workQueue;
997
             p.helpJoinOnce(w, this); // no retries on failure
998
          }
999
          else
1000
            ForkJoinPool.externalHelpJoin(this);
1001
         boolean canBlock = false;
1002
         boolean interrupted = false;
1003
         try {
1004
            while ((s = ) >= 0) {
1005
               if (w != null && w.qlock < 0)
1006
                  cancelIgnoringExceptions(this);
1007
               else if (!canBlock) {
1008
                  if (p == null || p.tryCompensate())
1009
                     canBlock = true;
1010
               }
1011
               else {
1012
                  if ((ms = ..toMillis(ns)) > 0L &&
1013
                        .compareAndSwapInt(this, , s, s | )) {
1014
                     synchronized (this) {
1015
                        if ( >= 0) {
1016
                           try {
1017
                              wait(ms);
1018
                           } catch (InterruptedException ie) {
1019
                              if (p == null)
1020
                                 interrupted = true;
1021
                           }
1022
                        }
1023
                        else
1024
                           notifyAll();
1025
                     }
1026
                  }
1027
                  if ((s = ) < 0 || interrupted ||
1028
                        (ns = deadline - System.nanoTime()) <= 0L)
1029
                     break;
1030
               }
1031
            }
1032
         } finally {
1033
            if (p != null && canBlock)
1034
               p.incrementActiveCount();
1035
         }
1036
         if (interrupted)
1037
            throw new InterruptedException();
1038
      }
1039
      if ((s &= ) != ) {
1040
         Throwable ex;
1041
         if (s == )
1042
            throw new CancellationException();
1043
         if (s != )
1044
            throw new TimeoutException();
1045
         if ((ex = getThrowableException()) != null)
1046
            throw new ExecutionException(ex);
1047
      }
1048
      return getRawResult();
1049
   }

   
Joins this task, without returning its result or throwing its exception. This method may be useful when processing collections of tasks when some have been cancelled or otherwise known to have aborted. 
1057
   public final void quietlyJoin() {
1058
      doJoin();
1059
   }

   
Commences performing this task and awaits its completion if necessary, without returning its result or throwing its exception. 
1066
   public final void quietlyInvoke() {
1067
      doInvoke();
1068
   }

   
Possibly executes tasks until the pool hosting the current task is quiescent. This method may be of use in designs in which many tasks are forked, but none are explicitly joined, instead executing them until all are processed. 
1077
   public static void helpQuiesce() {
1078
      Thread t;
1079
      if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) {
1080
         ForkJoinWorkerThread wt = (ForkJoinWorkerThread)t;
1081
         wt.pool.helpQuiescePool(wt.workQueue);
1082
      }
1083
      else
1084
         ForkJoinPool.quiesceCommonPool();
1085
   }

   
Resets the internal bookkeeping state of this task, allowing a subsequent fork. This method allows repeated reuse of this task, but only if reuse occurs when this task has either never been forked, or has been forked, then completed and all outstanding joins of this task have also completed. Effects under any other usage conditions are not guaranteed. This method may be useful when executing pre-constructed trees of subtasks in loops.

Upon completion of this method, isDone() reports false, and getException() reports null. However, the value returned by getRawResult is unaffected. To clear this value, you can invoke setRawResult(null). 

1103
   public void reinitialize() {
1104
      if (( & ) == )
1105
         clearExceptionalCompletion();
1106
      else
1107
          = 0;
1108
   }

   
Returns the pool hosting the current task execution, or null if this task is executing outside of any ForkJoinPool.

Returns:
the pool, or null if none
See also:
inForkJoinPool()
1117
   public static ForkJoinPool getPool() {
1118
      Thread t = Thread.currentThread();
1119
      return (t instanceof ForkJoinWorkerThread) ?
1120
            ((ForkJoinWorkerThread) t). : null;
1121
   }

   
Returns true if the current thread is a ForkJoinWorkerThread executing as a ForkJoinPool computation.

Returns:
true if the current thread is a ForkJoinWorkerThread executing as a ForkJoinPool computation, or false otherwise
1131
   public static boolean inForkJoinPool() {
1132
      return Thread.currentThread() instanceof ForkJoinWorkerThread;
1133
   }

   
Tries to unschedule this task for execution. This method will typically (but is not guaranteed to) succeed if this task is the most recently forked task by the current thread, and has not commenced executing in another thread. This method may be useful when arranging alternative local processing of tasks that could have been, but were not, stolen.

Returns:
true if unforked
1145
   public boolean tryUnfork() {
1146
      Thread t;
1147
      return (((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1148
                    ((ForkJoinWorkerThread)t)..tryUnpush(this) :
1149
                    ForkJoinPool.tryExternalUnpush(this));
1150
   }

   
Returns an estimate of the number of tasks that have been forked by the current worker thread but not yet executed. This value may be useful for heuristic decisions about whether to fork other tasks.

Returns:
the number of tasks
1160
   public static int getQueuedTaskCount() {
1161
      Thread t; ForkJoinPool.WorkQueue q;
1162
      if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
1163
         q = ((ForkJoinWorkerThread)t).;
1164
      else
1165
         q = ForkJoinPool.commonSubmitterQueue();
1166
      return (q == null) ? 0 : q.queueSize();
1167
   }

   
Returns an estimate of how many more locally queued tasks are held by the current worker thread than there are other worker threads that might steal them, or zero if this thread is not operating in a ForkJoinPool. This value may be useful for heuristic decisions about whether to fork other tasks. In many usages of ForkJoinTasks, at steady state, each worker should aim to maintain a small constant surplus (for example, 3) of tasks, and to process computations locally if this threshold is exceeded.

Returns:
the surplus number of tasks, which may be negative
1182
   public static int getSurplusQueuedTaskCount() {
1183
      return ForkJoinPool.getSurplusQueuedTaskCount();
1184
   }
1186
   // Extension methods
1187

   
Returns the result that would be returned by join(), even if this task completed abnormally, or null if this task is not known to have been completed. This method is designed to aid debugging, as well as to support extensions. Its use in any other context is discouraged.

Returns:
the result, or null if not completed
1197
   public abstract V getRawResult();

   
Forces the given value to be returned as a result. This method is designed to support extensions, and should not in general be called otherwise.

Parameters:
value the value
1206
   protected abstract void setRawResult(V value);

   
Immediately performs the base action of this task and returns true if, upon return from this method, this task is guaranteed to have completed normally. This method may return false otherwise, to indicate that this task is not necessarily complete (or is not known to be complete), for example in asynchronous actions that require explicit invocations of completion methods. This method may also throw an (unchecked) exception to indicate abnormal exit. This method is designed to support extensions, and should not in general be called otherwise.

Returns:
true if this task is known to have completed normally
1222
   protected abstract boolean exec();

   
Returns, but does not unschedule or execute, a task queued by the current thread but not yet executed, if one is immediately available. There is no guarantee that this task will actually be polled or executed next. Conversely, this method may return null even if a task exists but cannot be accessed without contention with other threads. This method is designed primarily to support extensions, and is unlikely to be useful otherwise.

Returns:
the next task, or null if none are available
1236
   protected static ForkJoinTask<?> peekNextLocalTask() {
1237
      Thread t; ForkJoinPool.WorkQueue q;
1238
      if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
1239
         q = ((ForkJoinWorkerThread)t).;
1240
      else
1241
         q = ForkJoinPool.commonSubmitterQueue();
1242
      return (q == null) ? null : q.peek();
1243
   }

   
Unschedules and returns, without executing, the next task queued by the current thread but not yet executed, if the current thread is operating in a ForkJoinPool. This method is designed primarily to support extensions, and is unlikely to be useful otherwise.

Returns:
the next task, or null if none are available
1254
   protected static ForkJoinTask<?> pollNextLocalTask() {
1255
      Thread t;
1256
      return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1257
            ((ForkJoinWorkerThread)t)..nextLocalTask() :
1258
            null;
1259
   }

   
If the current thread is operating in a ForkJoinPool, unschedules and returns, without executing, the next task queued by the current thread but not yet executed, if one is available, or if not available, a task that was forked by some other thread, if available. Availability may be transient, so a null result does not necessarily imply quiescence of the pool this task is operating in. This method is designed primarily to support extensions, and is unlikely to be useful otherwise.

Returns:
a task, or null if none are available
1274
   protected static ForkJoinTask<?> pollTask() {
1275
      Thread t; ForkJoinWorkerThread wt;
1276
      return ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
1277
            (wt = (ForkJoinWorkerThread)t)..nextTaskFor(wt.workQueue) :
1278
            null;
1279
   }
1281
   // tag operations
1282

   
Returns the tag for this task.

Returns:
the tag for this task
Since:
1.8
1289
   public final short getForkJoinTaskTag() {
1290
      return (short);
1291
   }

   
Atomically sets the tag value for this task.

Parameters:
tag the tag value
Returns:
the previous value of the tag
Since:
1.8
1300
   public final short setForkJoinTaskTag(short tag) {
1301
      for (int s;;) {
1302
         if (.compareAndSwapInt(this, , s = ,
1303
               (s & ~) | (tag & )))
1304
            return (short)s;
1305
      }
1306
   }

   
Atomically conditionally sets the tag value for this task. Among other applications, tags can be used as visit markers in tasks operating on graphs, as in methods that check: if (task.compareAndSetForkJoinTaskTag((short)0, (short)1)) before processing, otherwise exiting because the node has already been visited.

Parameters:
e the expected tag value
tag the new tag value
Returns:
true if successful; i.e., the current value was equal to e and is now tag.
Since:
1.8
1322
   public final boolean compareAndSetForkJoinTaskTag(short e, short tag) {
1323
      for (int s;;) {
1324
         if ((short)(s = ) != e)
1325
            return false;
1326
         if (.compareAndSwapInt(this, , s,
1327
               (s & ~) | (tag & )))
1328
            return true;
1329
      }
1330
   }

   
Adaptor for Runnables. This implements RunnableFuture to be compliant with AbstractExecutorService constraints when used in ForkJoinPool. 
1337
   static final class AdaptedRunnable<T> extends ForkJoinTask<T>
1338
         implements RunnableFuture<T> {
1339
      final Runnable runnable;
1340
      T result;
1341
      AdaptedRunnable(Runnable runnable, T result) {
1342
         if (runnable == null) throw new NullPointerException();
1343
         this. = runnable;
1344
         this. = result; // OK to set this even before completion
1345
      }
1346
      public final T getRawResult() { return ; }
1347
      public final void setRawResult(T v) {  = v; }
1348
      public final boolean exec() { .run(); return true; }
1349
      public final void run() { invoke(); }
1350
      private static final long serialVersionUID = 5232453952276885070L;
1351
   }

   
Adaptor for Runnables without results 
1356
   static final class AdaptedRunnableAction extends ForkJoinTask<Void>
1357
         implements RunnableFuture<Void> {
1358
      final Runnable runnable;
1359
      AdaptedRunnableAction(Runnable runnable) {
1360
         if (runnable == null) throw new NullPointerException();
1361
         this. = runnable;
1362
      }
1363
      public final Void getRawResult() { return null; }
1364
      public final void setRawResult(Void v) { }
1365
      public final boolean exec() { .run(); return true; }
1366
      public final void run() { invoke(); }
1367
      private static final long serialVersionUID = 5232453952276885070L;
1368
   }

   
Adaptor for Callables 
1373
   static final class AdaptedCallable<T> extends ForkJoinTask<T>
1374
         implements RunnableFuture<T> {
1375
      final Callable<? extends T> callable;
1376
      T result;
1377
      AdaptedCallable(Callable<? extends T> callable) {
1378
         if (callable == null) throw new NullPointerException();
1379
         this. = callable;
1380
      }
1381
      public final T getRawResult() { return ; }
1382
      public final void setRawResult(T v) {  = v; }
1383
      public final boolean exec() {
1384
         try {
1385
             = .call();
1386
            return true;
1387
         } catch (Error err) {
1388
            throw err;
1389
         } catch (RuntimeException rex) {
1390
            throw rex;
1391
         } catch (Exception ex) {
1392
            throw new RuntimeException(ex);
1393
         }
1394
      }
1395
      public final void run() { invoke(); }
1396
      private static final long serialVersionUID = 2838392045355241008L;
1397
   }

   
Returns a new ForkJoinTask that performs the run method of the given Runnable as its action, and returns a null result upon join().

Parameters:
runnable the runnable action
Returns:
the task
1407
   public static ForkJoinTask<?> adapt(Runnable runnable) {
1408
      return new AdaptedRunnableAction(runnable);
1409
   }

   
Returns a new ForkJoinTask that performs the run method of the given Runnable as its action, and returns the given result upon join().

Parameters:
runnable the runnable action
result the result upon completion
Returns:
the task
1420
   public static <T> ForkJoinTask<T> adapt(Runnable runnable, T result) {
1421
      return new AdaptedRunnable<T>(runnable, result);
1422
   }

   
Returns a new ForkJoinTask that performs the call method of the given Callable as its action, and returns its result upon join(), translating any checked exceptions encountered into RuntimeException.

Parameters:
callable the callable action
Returns:
the task
1433
   public static <T> ForkJoinTask<T> adapt(Callable<? extends T> callable) {
1434
      return new AdaptedCallable<T>(callable);
1435
   }
1437
   // Serialization support
1439
   private static final long serialVersionUID = -7721805057305804111L;

   
Saves this task to a stream (that is, serializes it).

SerialData:
the current run status and the exception thrown during execution, or null if none
1447
   private void writeObject(java.io.ObjectOutputStream s)
1448
         throws java.io.IOException {
1449
      s.defaultWriteObject();
1450
      s.writeObject(getException());
1451
   }

   
Reconstitutes this task from a stream (that is, deserializes it). 
1456
   private void readObject(java.io.ObjectInputStream s)
1457
         throws java.io.IOException, ClassNotFoundException {
1458
      s.defaultReadObject();
1459
      Object ex = s.readObject();
1460
      if (ex != null)
1461
         setExceptionalCompletion((Throwable)ex);
1462
   }
1464
   // Unsafe mechanics
1465
   @SuppressWarnings("internal")
1466
   private static final sun.misc.Unsafe U;
1467
   private static final long STATUS;
1469
   static {
1470
       = new ReentrantLock();
1473
      try {
1474
          = getUnsafe();
1475
         Class<?> k = ForkJoinTask.class;
1476
          = .objectFieldOffset
1477
               (k.getDeclaredField("status"));
1478
      } catch (Exception e) {
1479
         throw new Error(e);
1480
      }
1481
   }

   
Returns a sun.misc.Unsafe. Suitable for use in a 3rd party package. Replace with a simple call to Unsafe.getUnsafe when integrating into a jdk.

Returns:
a sun.misc.Unsafe
1490
   private static sun.misc.Unsafe getUnsafe() {
1491
      try {
1492
         return sun.misc.Unsafe.getUnsafe();
1493
      } catch (SecurityException tryReflectionInstead) {}
1494
      try {
1495
         return java.security.AccessController.doPrivileged
1496
               (new java.security.PrivilegedExceptionAction<sun.misc.Unsafe>() {
1497
                  public sun.misc.Unsafe run() throws Exception {
1498
                     Class<sun.misc.Unsafe> k = sun.misc.Unsafe.class;
1499
                     for (java.lang.reflect.Field f : k.getDeclaredFields()) {
1500
                        f.setAccessible(true);
1501
                        Object x = f.get(null);
1502
                        if (k.isInstance(x))
1503
                           return k.cast(x);
1504
                     }
1505
                     throw new NoSuchFieldError("the Unsafe");
1506
                  }});
1507
      } catch (java.security.PrivilegedActionException e) {
1508
         throw new RuntimeException("Could not initialize intrinsics",
1509
               e.getCause());
1510
      }
1511
   }