summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/actors/scala/actors/Actor.scala4
-rw-r--r--src/actors/scala/actors/ForkJoinScheduler.scala94
-rw-r--r--src/actors/scala/actors/Scheduler.scala3
-rw-r--r--src/actors/scala/actors/SimpleExecutorScheduler.scala7
-rw-r--r--src/actors/scala/actors/TerminationMonitor.scala5
-rw-r--r--src/actors/scala/actors/forkjoin/ForkJoinPool.java1785
-rw-r--r--src/actors/scala/actors/forkjoin/ForkJoinTask.java1053
-rw-r--r--src/actors/scala/actors/forkjoin/ForkJoinWorkerThread.java736
-rw-r--r--src/actors/scala/actors/forkjoin/LinkedTransferQueue.java840
-rw-r--r--src/actors/scala/actors/forkjoin/Phaser.java960
-rw-r--r--src/actors/scala/actors/forkjoin/RecursiveAction.java151
-rw-r--r--src/actors/scala/actors/forkjoin/RecursiveTask.java71
-rw-r--r--src/actors/scala/actors/forkjoin/ThreadLocalRandom.java186
-rw-r--r--src/actors/scala/actors/forkjoin/TransferQueue.java118
-rw-r--r--src/actors/scala/actors/forkjoin/package-info.java29
15 files changed, 6032 insertions, 10 deletions
diff --git a/src/actors/scala/actors/Actor.scala b/src/actors/scala/actors/Actor.scala
index dfbd46a1a7..91fa746c64 100644
--- a/src/actors/scala/actors/Actor.scala
+++ b/src/actors/scala/actors/Actor.scala
@@ -410,8 +410,8 @@ trait Actor extends Reactor with AbstractActor {
}
if (isSuspended) {
synchronized {
- senders = item._2 :: senders
received = Some(item._1)
+ senders = item._2 :: senders
resumeActor()
}
} else {
@@ -597,12 +597,12 @@ trait Actor extends Reactor with AbstractActor {
receiveTimeout
} else {
waitingFor = f.isDefinedAt
+ continuation = f
val thisActor = this
onTimeout = Some(new TimerTask {
def run() { thisActor.send(TIMEOUT, thisActor) }
})
Actor.timer.schedule(onTimeout.get, msec)
- continuation = f
done = true
() => {}
}
diff --git a/src/actors/scala/actors/ForkJoinScheduler.scala b/src/actors/scala/actors/ForkJoinScheduler.scala
new file mode 100644
index 0000000000..850706eb72
--- /dev/null
+++ b/src/actors/scala/actors/ForkJoinScheduler.scala
@@ -0,0 +1,94 @@
+package scala.actors
+
+import java.lang.Thread.State
+import forkjoin._
+
+class ForkJoinScheduler extends Thread with IScheduler with TerminationMonitor {
+
+ private val pool = {
+ val p = new ForkJoinPool()
+ Debug.info(this+": parallelism "+p.getParallelism())
+ Debug.info(this+": max pool size "+p.getMaximumPoolSize())
+ p
+ }
+
+ private var terminating = false
+
+ private val CHECK_FREQ = 50
+
+ private def allWorkersWaiting: Boolean =
+ pool.workers.forall(t => {
+ if (t == null)
+ true
+ else {
+ val s = t.getState()
+ s == State.WAITING || s == State.TIMED_WAITING
+ }
+ })
+
+ override def run() {
+ try {
+ while (true) {
+ this.synchronized {
+ try {
+ wait(CHECK_FREQ)
+ } catch {
+ case _: InterruptedException =>
+ }
+ if (terminating)
+ throw new QuitException
+
+ if (allTerminated) {
+ //Debug.info(this+": all actors terminated")
+ throw new QuitException
+ }
+
+ if (!pool.isQuiescent && allWorkersWaiting) {
+ //Debug.info(this+": all workers blocked")
+ val par = pool.getParallelism()
+ //Debug.info(this+": parallelism "+par)
+ //Debug.info(this+": max pool size "+pool.getMaximumPoolSize())
+ if (par < pool.getMaximumPoolSize()) {
+ pool.setParallelism(par + 1)
+ }
+ }
+ }
+ }
+ } catch {
+ case _: QuitException =>
+ Debug.info(this+": initiating shutdown...")
+ pool.shutdown()
+ // allow thread to exit
+ }
+ }
+
+ def execute(task: Runnable) {
+ val recAction = new RecursiveAction {
+ def compute() = task.run()
+ }
+ val thread = Thread.currentThread()
+ if (thread.isInstanceOf[ForkJoinWorkerThread])
+ recAction.fork()
+ else
+ pool.execute(task)
+ }
+
+ /** Submits a closure for execution.
+ *
+ * @param fun the closure to be executed
+ */
+ def execute(fun: => Unit): Unit =
+ execute(new Runnable {
+ def run() { fun }
+ })
+
+ /** Shuts down the scheduler.
+ */
+ def shutdown(): Unit = synchronized {
+ terminating = true
+ }
+
+ def isActive =
+ !pool.isShutdown()
+
+}
diff --git a/src/actors/scala/actors/Scheduler.scala b/src/actors/scala/actors/Scheduler.scala
index a7935edabf..5f5f1705e7 100644
--- a/src/actors/scala/actors/Scheduler.scala
+++ b/src/actors/scala/actors/Scheduler.scala
@@ -25,6 +25,7 @@ object Scheduler extends DelegatingScheduler {
Debug.info("initializing "+this+"...")
def makeNewScheduler: IScheduler = {
+ /*
val workQueue = new LinkedBlockingQueue[Runnable](100000)
val threadPool = new ThreadPoolExecutor(ThreadPoolConfig.corePoolSize,
ThreadPoolConfig.maxPoolSize,
@@ -32,6 +33,8 @@ object Scheduler extends DelegatingScheduler {
TimeUnit.MILLISECONDS,
workQueue)
val s = new SimpleExecutorScheduler(threadPool, true)
+ */
+ val s = new ForkJoinScheduler
s.start()
s
}
diff --git a/src/actors/scala/actors/SimpleExecutorScheduler.scala b/src/actors/scala/actors/SimpleExecutorScheduler.scala
index 1499aeb7be..ea9a3ae629 100644
--- a/src/actors/scala/actors/SimpleExecutorScheduler.scala
+++ b/src/actors/scala/actors/SimpleExecutorScheduler.scala
@@ -31,13 +31,6 @@ import java.util.concurrent.{ExecutorService, RejectedExecutionException}
class SimpleExecutorScheduler(protected var executor: ExecutorService,
protected var terminate: Boolean) extends TerminationService(terminate) {
- /* Maintains per actor one closure that is executed
- * when the actor terminates.
- */
- protected val termHandlers = new HashMap[Reactor, () => Unit]
-
- private var pendingReactions = 0
-
/* This constructor (and the var above) is currently only used to work
* around a bug in scaladoc, which cannot deal with early initializers
* (to be used in subclasses such as DefaultExecutorScheduler) properly.
diff --git a/src/actors/scala/actors/TerminationMonitor.scala b/src/actors/scala/actors/TerminationMonitor.scala
index ba76d19145..ce0adc2e59 100644
--- a/src/actors/scala/actors/TerminationMonitor.scala
+++ b/src/actors/scala/actors/TerminationMonitor.scala
@@ -16,10 +16,13 @@ trait TerminationMonitor extends IScheduler {
private var pendingReactions = 0
private val termHandlers = new HashMap[Reactor, () => Unit]
+ private var started = false
/** newActor is invoked whenever a new actor is started. */
def newActor(a: Reactor) = synchronized {
pendingReactions += 1
+ if (!started)
+ started = true
}
/** Registers a closure to be executed when the specified
@@ -53,7 +56,7 @@ trait TerminationMonitor extends IScheduler {
}
protected def allTerminated: Boolean = synchronized {
- pendingReactions <= 0
+ started && pendingReactions <= 0
}
}
diff --git a/src/actors/scala/actors/forkjoin/ForkJoinPool.java b/src/actors/scala/actors/forkjoin/ForkJoinPool.java
new file mode 100644
index 0000000000..765e2e1284
--- /dev/null
+++ b/src/actors/scala/actors/forkjoin/ForkJoinPool.java
@@ -0,0 +1,1785 @@
+/*
+ * Written by Doug Lea with assistance from members of JCP JSR-166
+ * Expert Group and released to the public domain, as explained at
+ * http://creativecommons.org/licenses/publicdomain
+ */
+
+package scala.actors.forkjoin;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.locks.*;
+import java.util.concurrent.atomic.*;
+import sun.misc.Unsafe;
+import java.lang.reflect.*;
+
+
+interface RunnableFuture<T> extends Runnable {
+ //TR placeholder for java.util.concurrent.RunnableFuture
+}
+
+class Arrays {
+ //TR placeholder for java.util.Arrays.copyOf
+ public static <T> T[] copyOf(T[] original, int newLength) {
+ T[] copy = (T[])new Object[newLength];
+ System.arraycopy(original, 0, copy, 0, Math.min(newLength, original.length));
+ return copy;
+ }
+
+}
+
+
+/**
+ * An {@link ExecutorService} for running {@link ForkJoinTask}s. A
+ * ForkJoinPool provides the entry point for submissions from
+ * non-ForkJoinTasks, as well as management and monitoring operations.
+ * Normally a single ForkJoinPool is used for a large number of
+ * submitted tasks. Otherwise, use would not usually outweigh the
+ * construction and bookkeeping overhead of creating a large set of
+ * threads.
+ *
+ * <p>ForkJoinPools differ from other kinds of Executors mainly in
+ * that they provide <em>work-stealing</em>: all threads in the pool
+ * attempt to find and execute subtasks created by other active tasks
+ * (eventually blocking if none exist). This makes them efficient when
+ * most tasks spawn other subtasks (as do most ForkJoinTasks), as well
+ * as the mixed execution of some plain Runnable- or Callable- based
+ * activities along with ForkJoinTasks. Otherwise, other
+ * ExecutorService implementations are typically more appropriate
+ * choices.
+ *
+ * <p>A ForkJoinPool may be constructed with a given parallelism level
+ * (target pool size), which it attempts to maintain by dynamically
+ * adding, suspending, or resuming threads, even if some tasks are
+ * waiting to join others. However, no such adjustments are performed
+ * in the face of blocked IO or other unmanaged synchronization. The
+ * nested <code>ManagedBlocker</code> interface enables extension of
+ * the kinds of synchronization accommodated. The target parallelism
+ * level may also be changed dynamically (<code>setParallelism</code>)
+ * and dynamically thread construction can be limited using methods
+ * <code>setMaximumPoolSize</code> and/or
+ * <code>setMaintainsParallelism</code>.
+ *
+ * <p>In addition to execution and lifecycle control methods, this
+ * class provides status check methods (for example
+ * <code>getStealCount</code>) that are intended to aid in developing,
+ * tuning, and monitoring fork/join applications. Also, method
+ * <code>toString</code> returns indications of pool state in a
+ * convenient form for informal monitoring.
+ *
+ * <p><b>Implementation notes</b>: This implementation restricts the
+ * maximum number of running threads to 32767. Attempts to create
+ * pools with greater than the maximum result in
+ * IllegalArgumentExceptions.
+ */
+public class ForkJoinPool /*extends AbstractExecutorService*/ {
+
+ /*
+ * See the extended comments interspersed below for design,
+ * rationale, and walkthroughs.
+ */
+
+ /** Mask for packing and unpacking shorts */
+ private static final int shortMask = 0xffff;
+
+ /** Max pool size -- must be a power of two minus 1 */
+ private static final int MAX_THREADS = 0x7FFF;
+
+ /**
+ * Factory for creating new ForkJoinWorkerThreads. A
+ * ForkJoinWorkerThreadFactory must be defined and used for
+ * ForkJoinWorkerThread subclasses that extend base functionality
+ * or initialize threads with different contexts.
+ */
+ public static interface ForkJoinWorkerThreadFactory {
+ /**
+ * Returns a new worker thread operating in the given pool.
+ *
+ * @param pool the pool this thread works in
+ * @throws NullPointerException if pool is null;
+ */
+ public ForkJoinWorkerThread newThread(ForkJoinPool pool);
+ }
+
+ /**
+ * Default ForkJoinWorkerThreadFactory implementation, creates a
+ * new ForkJoinWorkerThread.
+ */
+ static class DefaultForkJoinWorkerThreadFactory
+ implements ForkJoinWorkerThreadFactory {
+ public ForkJoinWorkerThread newThread(ForkJoinPool pool) {
+ try {
+ return new ForkJoinWorkerThread(pool);
+ } catch (OutOfMemoryError oom) {
+ return null;
+ }
+ }
+ }
+
+ /**
+ * Creates a new ForkJoinWorkerThread. This factory is used unless
+ * overridden in ForkJoinPool constructors.
+ */
+ public static final ForkJoinWorkerThreadFactory
+ defaultForkJoinWorkerThreadFactory =
+ new DefaultForkJoinWorkerThreadFactory();
+
+ /**
+ * Permission required for callers of methods that may start or
+ * kill threads.
+ */
+ private static final RuntimePermission modifyThreadPermission =
+ new RuntimePermission("modifyThread");
+
+ /**
+ * If there is a security manager, makes sure caller has
+ * permission to modify threads.
+ */
+ private static void checkPermission() {
+ SecurityManager security = System.getSecurityManager();
+ if (security != null)
+ security.checkPermission(modifyThreadPermission);
+ }
+
+ /**
+ * Generator for assigning sequence numbers as pool names.
+ */
+ private static final AtomicInteger poolNumberGenerator =
+ new AtomicInteger();
+
+ /**
+ * Array holding all worker threads in the pool. Array size must
+ * be a power of two. Updates and replacements are protected by
+ * workerLock, but it is always kept in a consistent enough state
+ * to be randomly accessed without locking by workers performing
+ * work-stealing.
+ */
+ public volatile ForkJoinWorkerThread[] workers;
+
+ /**
+ * Lock protecting access to workers.
+ */
+ private final ReentrantLock workerLock;
+
+ /**
+ * Condition for awaitTermination.
+ */
+ private final Condition termination;
+
+ /**
+ * The uncaught exception handler used when any worker
+ * abrupty terminates
+ */
+ private Thread.UncaughtExceptionHandler ueh;
+
+ /**
+ * Creation factory for worker threads.
+ */
+ private final ForkJoinWorkerThreadFactory factory;
+
+ /**
+ * Head of stack of threads that were created to maintain
+ * parallelism when other threads blocked, but have since
+ * suspended when the parallelism level rose.
+ */
+ private volatile WaitQueueNode spareStack;
+
+ /**
+ * Sum of per-thread steal counts, updated only when threads are
+ * idle or terminating.
+ */
+ private final AtomicLong stealCount;
+
+ /**
+ * Queue for external submissions.
+ */
+ private final LinkedTransferQueue<ForkJoinTask<?>> submissionQueue;
+
+ /**
+ * Head of Treiber stack for barrier sync. See below for explanation
+ */
+ private volatile WaitQueueNode syncStack;
+
+ /**
+ * The count for event barrier
+ */
+ private volatile long eventCount;
+
+ /**
+ * Pool number, just for assigning useful names to worker threads
+ */
+ private final int poolNumber;
+
+ /**
+ * The maximum allowed pool size
+ */
+ private volatile int maxPoolSize;
+
+ /**
+ * The desired parallelism level, updated only under workerLock.
+ */
+ private volatile int parallelism;
+
+ /**
+ * Holds number of total (i.e., created and not yet terminated)
+ * and running (i.e., not blocked on joins or other managed sync)
+ * threads, packed into one int to ensure consistent snapshot when
+ * making decisions about creating and suspending spare
+ * threads. Updated only by CAS. Note: CASes in
+ * updateRunningCount and preJoin running active count is in low
+ * word, so need to be modified if this changes
+ */
+ private volatile int workerCounts;
+
+ private static int totalCountOf(int s) { return s >>> 16; }
+ private static int runningCountOf(int s) { return s & shortMask; }
+ private static int workerCountsFor(int t, int r) { return (t << 16) + r; }
+
+ /**
+ * Add delta (which may be negative) to running count. This must
+ * be called before (with negative arg) and after (with positive)
+ * any managed synchronization (i.e., mainly, joins)
+ * @param delta the number to add
+ */
+ final void updateRunningCount(int delta) {
+ int s;
+ do;while (!casWorkerCounts(s = workerCounts, s + delta));
+ }
+
+ /**
+ * Add delta (which may be negative) to both total and running
+ * count. This must be called upon creation and termination of
+ * worker threads.
+ * @param delta the number to add
+ */
+ private void updateWorkerCount(int delta) {
+ int d = delta + (delta << 16); // add to both lo and hi parts
+ int s;
+ do;while (!casWorkerCounts(s = workerCounts, s + d));
+ }
+
+ /**
+ * Lifecycle control. High word contains runState, low word
+ * contains the number of workers that are (probably) executing
+ * tasks. This value is atomically incremented before a worker
+ * gets a task to run, and decremented when worker has no tasks
+ * and cannot find any. These two fields are bundled together to
+ * support correct termination triggering. Note: activeCount
+ * CAS'es cheat by assuming active count is in low word, so need
+ * to be modified if this changes
+ */
+ private volatile int runControl;
+
+ // RunState values. Order among values matters
+ private static final int RUNNING = 0;
+ private static final int SHUTDOWN = 1;
+ private static final int TERMINATING = 2;
+ private static final int TERMINATED = 3;
+
+ private static int runStateOf(int c) { return c >>> 16; }
+ private static int activeCountOf(int c) { return c & shortMask; }
+ private static int runControlFor(int r, int a) { return (r << 16) + a; }
+
+ /**
+ * Try incrementing active count; fail on contention. Called by
+ * workers before/during executing tasks.
+ * @return true on success;
+ */
+ final boolean tryIncrementActiveCount() {
+ int c = runControl;
+ return casRunControl(c, c+1);
+ }
+
+ /**
+ * Try decrementing active count; fail on contention.
+ * Possibly trigger termination on success
+ * Called by workers when they can't find tasks.
+ * @return true on success
+ */
+ final boolean tryDecrementActiveCount() {
+ int c = runControl;
+ int nextc = c - 1;
+ if (!casRunControl(c, nextc))
+ return false;
+ if (canTerminateOnShutdown(nextc))
+ terminateOnShutdown();
+ return true;
+ }
+
+ /**
+ * Return true if argument represents zero active count and
+ * nonzero runstate, which is the triggering condition for
+ * terminating on shutdown.
+ */
+ private static boolean canTerminateOnShutdown(int c) {
+ return ((c & -c) >>> 16) != 0; // i.e. least bit is nonzero runState bit
+ }
+
+ /**
+ * Transition run state to at least the given state. Return true
+ * if not already at least given state.
+ */
+ private boolean transitionRunStateTo(int state) {
+ for (;;) {
+ int c = runControl;
+ if (runStateOf(c) >= state)
+ return false;
+ if (casRunControl(c, runControlFor(state, activeCountOf(c))))
+ return true;
+ }
+ }
+
+ /**
+ * Controls whether to add spares to maintain parallelism
+ */
+ private volatile boolean maintainsParallelism;
+
+ // Constructors
+
+ /**
+ * Creates a ForkJoinPool with a pool size equal to the number of
+ * processors available on the system and using the default
+ * ForkJoinWorkerThreadFactory,
+ * @throws SecurityException if a security manager exists and
+ * the caller is not permitted to modify threads
+ * because it does not hold {@link
+ * java.lang.RuntimePermission}<code>("modifyThread")</code>,
+ */
+ public ForkJoinPool() {
+ this(Runtime.getRuntime().availableProcessors(),
+ defaultForkJoinWorkerThreadFactory);
+ }
+
+ /**
+ * Creates a ForkJoinPool with the indicated parellelism level
+ * threads, and using the default ForkJoinWorkerThreadFactory,
+ * @param parallelism the number of worker threads
+ * @throws IllegalArgumentException if parallelism less than or
+ * equal to zero
+ * @throws SecurityException if a security manager exists and
+ * the caller is not permitted to modify threads
+ * because it does not hold {@link
+ * java.lang.RuntimePermission}<code>("modifyThread")</code>,
+ */
+ public ForkJoinPool(int parallelism) {
+ this(parallelism, defaultForkJoinWorkerThreadFactory);
+ }
+
+ /**
+ * Creates a ForkJoinPool with parallelism equal to the number of
+ * processors available on the system and using the given
+ * ForkJoinWorkerThreadFactory,
+ * @param factory the factory for creating new threads
+ * @throws NullPointerException if factory is null
+ * @throws SecurityException if a security manager exists and
+ * the caller is not permitted to modify threads
+ * because it does not hold {@link
+ * java.lang.RuntimePermission}<code>("modifyThread")</code>,
+ */
+ public ForkJoinPool(ForkJoinWorkerThreadFactory factory) {
+ this(Runtime.getRuntime().availableProcessors(), factory);
+ }
+
+ /**
+ * Creates a ForkJoinPool with the given parallelism and factory.
+ *
+ * @param parallelism the targeted number of worker threads
+ * @param factory the factory for creating new threads
+ * @throws IllegalArgumentException if parallelism less than or
+ * equal to zero, or greater than implementation limit.
+ * @throws NullPointerException if factory is null
+ * @throws SecurityException if a security manager exists and
+ * the caller is not permitted to modify threads
+ * because it does not hold {@link
+ * java.lang.RuntimePermission}<code>("modifyThread")</code>,
+ */
+ public ForkJoinPool(int parallelism, ForkJoinWorkerThreadFactory factory) {
+ if (parallelism <= 0 || parallelism > MAX_THREADS)
+ throw new IllegalArgumentException();
+ if (factory == null)
+ throw new NullPointerException();
+ checkPermission();
+ this.factory = factory;
+ this.parallelism = parallelism;
+ this.maxPoolSize = MAX_THREADS;
+ this.maintainsParallelism = true;
+ this.poolNumber = poolNumberGenerator.incrementAndGet();
+ this.workerLock = new ReentrantLock();
+ this.termination = workerLock.newCondition();
+ this.stealCount = new AtomicLong();
+ this.submissionQueue = new LinkedTransferQueue<ForkJoinTask<?>>();
+ createAndStartInitialWorkers(parallelism);
+ }
+
+ /**
+ * Create new worker using factory.
+ * @param index the index to assign worker
+ * @return new worker, or null of factory failed
+ */
+ private ForkJoinWorkerThread createWorker(int index) {
+ Thread.UncaughtExceptionHandler h = ueh;
+ ForkJoinWorkerThread w = factory.newThread(this);
+ if (w != null) {
+ w.poolIndex = index;
+ w.setDaemon(true);
+ w.setName("ForkJoinPool-" + poolNumber + "-worker-" + index);
+ if (h != null)
+ w.setUncaughtExceptionHandler(h);
+ }
+ return w;
+ }
+
+ /**
+ * Return a good size for worker array given pool size.
+ * Currently requires size to be a power of two.
+ */
+ private static int arraySizeFor(int ps) {
+ return ps <= 1? 1 : (1 << (32 - Integer.numberOfLeadingZeros(ps-1)));
+ }
+
+ public static ForkJoinWorkerThread[] copyOfWorkers(ForkJoinWorkerThread[] original, int newLength) {
+ ForkJoinWorkerThread[] copy = new ForkJoinWorkerThread[newLength];
+ System.arraycopy(original, 0, copy, 0, Math.min(newLength, original.length));
+ return copy;
+ }
+
+ /**
+ * Create or resize array if necessary to hold newLength
+ * @return the array
+ */
+ private ForkJoinWorkerThread[] ensureWorkerArrayCapacity(int newLength) {
+ ForkJoinWorkerThread[] ws = workers;
+ if (ws == null)
+ return workers = new ForkJoinWorkerThread[arraySizeFor(newLength)];
+ else if (newLength > ws.length)
+ return workers = copyOfWorkers(ws, arraySizeFor(newLength));
+ else
+ return ws;
+ }
+
+ /**
+ * Try to shrink workers into smaller array after one or more terminate
+ */
+ private void tryShrinkWorkerArray() {
+ ForkJoinWorkerThread[] ws = workers;
+ int len = ws.length;
+ int last = len - 1;
+ while (last >= 0 && ws[last] == null)
+ --last;
+ int newLength = arraySizeFor(last+1);
+ if (newLength < len)
+ workers = Arrays.copyOf(ws, newLength);
+ }
+
+ /**
+ * Initial worker array and worker creation and startup. (This
+ * must be done under lock to avoid interference by some of the
+ * newly started threads while creating others.)
+ */
+ private void createAndStartInitialWorkers(int ps) {
+ final ReentrantLock lock = this.workerLock;
+ lock.lock();
+ try {
+ ForkJoinWorkerThread[] ws = ensureWorkerArrayCapacity(ps);
+ for (int i = 0; i < ps; ++i) {
+ ForkJoinWorkerThread w = createWorker(i);
+ if (w != null) {
+ ws[i] = w;
+ w.start();
+ updateWorkerCount(1);
+ }
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Worker creation and startup for threads added via setParallelism.
+ */
+ private void createAndStartAddedWorkers() {
+ resumeAllSpares(); // Allow spares to convert to nonspare
+ int ps = parallelism;
+ ForkJoinWorkerThread[] ws = ensureWorkerArrayCapacity(ps);
+ int len = ws.length;
+ // Sweep through slots, to keep lowest indices most populated
+ int k = 0;
+ while (k < len) {
+ if (ws[k] != null) {
+ ++k;
+ continue;
+ }
+ int s = workerCounts;
+ int tc = totalCountOf(s);
+ int rc = runningCountOf(s);
+ if (rc >= ps || tc >= ps)
+ break;
+ if (casWorkerCounts (s, workerCountsFor(tc+1, rc+1))) {
+ ForkJoinWorkerThread w = createWorker(k);
+ if (w != null) {
+ ws[k++] = w;
+ w.start();
+ }
+ else {
+ updateWorkerCount(-1); // back out on failed creation
+ break;
+ }
+ }
+ }
+ }
+
+ // Execution methods
+
+ /**
+ * Common code for execute, invoke and submit
+ */
+ private <T> void doSubmit(ForkJoinTask<T> task) {
+ if (isShutdown())
+ throw new RejectedExecutionException();
+ submissionQueue.offer(task);
+ signalIdleWorkers();
+ }
+
+ /**
+ * Performs the given task; returning its result upon completion
+ * @param task the task
+ * @return the task's result
+ * @throws NullPointerException if task is null
+ * @throws RejectedExecutionException if pool is shut down
+ */
+ public <T> T invoke(ForkJoinTask<T> task) {
+ doSubmit(task);
+ return task.join();
+ }
+
+ /**
+ * Arranges for (asynchronous) execution of the given task.
+ * @param task the task
+ * @throws NullPointerException if task is null
+ * @throws RejectedExecutionException if pool is shut down
+ */
+ public <T> void execute(ForkJoinTask<T> task) {
+ doSubmit(task);
+ }
+
+ // AbstractExecutorService methods
+
+ public void execute(Runnable task) {
+ doSubmit(new AdaptedRunnable<Void>(task, null));
+ }
+
+ public <T> ForkJoinTask<T> submit(Callable<T> task) {
+ ForkJoinTask<T> job = new AdaptedCallable<T>(task);
+ doSubmit(job);
+ return job;
+ }
+
+ public <T> ForkJoinTask<T> submit(Runnable task, T result) {
+ ForkJoinTask<T> job = new AdaptedRunnable<T>(task, result);
+ doSubmit(job);
+ return job;
+ }
+
+ public ForkJoinTask<?> submit(Runnable task) {
+ ForkJoinTask<Void> job = new AdaptedRunnable<Void>(task, null);
+ doSubmit(job);
+ return job;
+ }
+
+ /**
+ * Adaptor for Runnables. This implements RunnableFuture
+ * to be compliant with AbstractExecutorService constraints
+ */
+ static final class AdaptedRunnable<T> extends ForkJoinTask<T>
+ implements RunnableFuture<T> {
+ final Runnable runnable;
+ final T resultOnCompletion;
+ T result;
+ AdaptedRunnable(Runnable runnable, T result) {
+ if (runnable == null) throw new NullPointerException();
+ this.runnable = runnable;
+ this.resultOnCompletion = result;
+ }
+ public T getRawResult() { return result; }
+ public void setRawResult(T v) { result = v; }
+ public boolean exec() {
+ runnable.run();
+ result = resultOnCompletion;
+ return true;
+ }
+ public void run() { invoke(); }
+ }
+
+ /**
+ * Adaptor for Callables
+ */
+ static final class AdaptedCallable<T> extends ForkJoinTask<T>
+ implements RunnableFuture<T> {
+ final Callable<T> callable;
+ T result;
+ AdaptedCallable(Callable<T> callable) {
+ if (callable == null) throw new NullPointerException();
+ this.callable = callable;
+ }
+ public T getRawResult() { return result; }
+ public void setRawResult(T v) { result = v; }
+ public boolean exec() {
+ try {
+ result = callable.call();
+ return true;
+ } catch (Error err) {
+ throw err;
+ } catch (RuntimeException rex) {
+ throw rex;
+ } catch (Exception ex) {
+ throw new RuntimeException(ex);
+ }
+ }
+ public void run() { invoke(); }
+ }
+
+ public <T> List<Future<T>> invokeAll(Collection<Callable<T>> tasks) {
+ ArrayList<ForkJoinTask<T>> ts =
+ new ArrayList<ForkJoinTask<T>>(tasks.size());
+ for (Callable<T> c : tasks)
+ ts.add(new AdaptedCallable<T>(c));
+ invoke(new InvokeAll<T>(ts));
+ return (List<Future<T>>)(List)ts;
+ }
+
+ static final class InvokeAll<T> extends RecursiveAction {
+ final ArrayList<ForkJoinTask<T>> tasks;
+ InvokeAll(ArrayList<ForkJoinTask<T>> tasks) { this.tasks = tasks; }
+ public void compute() {
+ try { invokeAll(tasks); } catch(Exception ignore) {}
+ }
+ }
+
+ // Configuration and status settings and queries
+
+ /**
+ * Returns the factory used for constructing new workers
+ *
+ * @return the factory used for constructing new workers
+ */
+ public ForkJoinWorkerThreadFactory getFactory() {
+ return factory;
+ }
+
+ /**
+ * Returns the handler for internal worker threads that terminate
+ * due to unrecoverable errors encountered while executing tasks.
+ * @return the handler, or null if none
+ */
+ public Thread.UncaughtExceptionHandler getUncaughtExceptionHandler() {
+ Thread.UncaughtExceptionHandler h;
+ final ReentrantLock lock = this.workerLock;
+ lock.lock();
+ try {
+ h = ueh;
+ } finally {
+ lock.unlock();
+ }
+ return h;
+ }
+
+ /**
+ * Sets the handler for internal worker threads that terminate due
+ * to unrecoverable errors encountered while executing tasks.
+ * Unless set, the current default or ThreadGroup handler is used
+ * as handler.
+ *
+ * @param h the new handler
+ * @return the old handler, or null if none
+ * @throws SecurityException if a security manager exists and
+ * the caller is not permitted to modify threads
+ * because it does not hold {@link
+ * java.lang.RuntimePermission}<code>("modifyThread")</code>,
+ */
+ public Thread.UncaughtExceptionHandler
+ setUncaughtExceptionHandler(Thread.UncaughtExceptionHandler h) {
+ checkPermission();
+ Thread.UncaughtExceptionHandler old = null;
+ final ReentrantLock lock = this.workerLock;
+ lock.lock();
+ try {
+ old = ueh;
+ ueh = h;
+ ForkJoinWorkerThread[] ws = workers;
+ for (int i = 0; i < ws.length; ++i) {
+ ForkJoinWorkerThread w = ws[i];
+ if (w != null)
+ w.setUncaughtExceptionHandler(h);
+ }
+ } finally {
+ lock.unlock();
+ }
+ return old;
+ }
+
+
+ /**
+ * Sets the target paralleism level of this pool.
+ * @param parallelism the target parallelism
+ * @throws IllegalArgumentException if parallelism less than or
+ * equal to zero or greater than maximum size bounds.
+ * @throws SecurityException if a security manager exists and
+ * the caller is not permitted to modify threads
+ * because it does not hold {@link
+ * java.lang.RuntimePermission}<code>("modifyThread")</code>,
+ */
+ public void setParallelism(int parallelism) {
+ checkPermission();
+ if (parallelism <= 0 || parallelism > maxPoolSize)
+ throw new IllegalArgumentException();
+ final ReentrantLock lock = this.workerLock;
+ lock.lock();
+ try {
+ if (!isTerminating()) {
+ int p = this.parallelism;
+ this.parallelism = parallelism;
+ if (parallelism > p)
+ createAndStartAddedWorkers();
+ else
+ trimSpares();
+ }
+ } finally {
+ lock.unlock();
+ }
+ signalIdleWorkers();
+ }
+
+ /**
+ * Returns the targeted number of worker threads in this pool.
+ *
+ * @return the targeted number of worker threads in this pool
+ */
+ public int getParallelism() {
+ return parallelism;
+ }
+
+ /**
+ * Returns the number of worker threads that have started but not
+ * yet terminated. This result returned by this method may differ
+ * from <code>getParallelism</code> when threads are created to
+ * maintain parallelism when others are cooperatively blocked.
+ *
+ * @return the number of worker threads
+ */
+ public int getPoolSize() {
+ return totalCountOf(workerCounts);
+ }
+
+ /**
+ * Returns the maximum number of threads allowed to exist in the
+ * pool, even if there are insufficient unblocked running threads.
+ * @return the maximum
+ */
+ public int getMaximumPoolSize() {
+ return maxPoolSize;
+ }
+
+ /**
+ * Sets the maximum number of threads allowed to exist in the
+ * pool, even if there are insufficient unblocked running threads.
+ * Setting this value has no effect on current pool size. It
+ * controls construction of new threads.
+ * @throws IllegalArgumentException if negative or greater then
+ * internal implementation limit.
+ */
+ public void setMaximumPoolSize(int newMax) {
+ if (newMax < 0 || newMax > MAX_THREADS)
+ throw new IllegalArgumentException();
+ maxPoolSize = newMax;
+ }
+
+
+ /**
+ * Returns true if this pool dynamically maintains its target
+ * parallelism level. If false, new threads are added only to
+ * avoid possible starvation.
+ * This setting is by default true;
+ * @return true if maintains parallelism
+ */
+ public boolean getMaintainsParallelism() {
+ return maintainsParallelism;
+ }
+
+ /**
+ * Sets whether this pool dynamically maintains its target
+ * parallelism level. If false, new threads are added only to
+ * avoid possible starvation.
+ * @param enable true to maintains parallelism
+ */
+ public void setMaintainsParallelism(boolean enable) {
+ maintainsParallelism = enable;
+ }
+
+ /**
+ * Returns an estimate of the number of worker threads that are
+ * not blocked waiting to join tasks or for other managed
+ * synchronization.
+ *
+ * @return the number of worker threads
+ */
+ public int getRunningThreadCount() {
+ return runningCountOf(workerCounts);
+ }
+
+ /**
+ * Returns an estimate of the number of threads that are currently
+ * stealing or executing tasks. This method may overestimate the
+ * number of active threads.
+ * @return the number of active threads.
+ */
+ public int getActiveThreadCount() {
+ return activeCountOf(runControl);
+ }
+
+ /**
+ * Returns an estimate of the number of threads that are currently
+ * idle waiting for tasks. This method may underestimate the
+ * number of idle threads.
+ * @return the number of idle threads.
+ */
+ final int getIdleThreadCount() {
+ int c = runningCountOf(workerCounts) - activeCountOf(runControl);
+ return (c <= 0)? 0 : c;
+ }
+
+ /**
+ * Returns true if all worker threads are currently idle. An idle
+ * worker is one that cannot obtain a task to execute because none
+ * are available to steal from other threads, and there are no
+ * pending submissions to the pool. This method is conservative:
+ * It might not return true immediately upon idleness of all
+ * threads, but will eventually become true if threads remain
+ * inactive.
+ * @return true if all threads are currently idle
+ */
+ public boolean isQuiescent() {
+ return activeCountOf(runControl) == 0;
+ }
+
+ /**
+ * Returns an estimate of the total number of tasks stolen from
+ * one thread's work queue by another. The reported value
+ * underestimates the actual total number of steals when the pool
+ * is not quiescent. This value may be useful for monitoring and
+ * tuning fork/join programs: In general, steal counts should be
+ * high enough to keep threads busy, but low enough to avoid
+ * overhead and contention across threads.
+ * @return the number of steals.
+ */
+ public long getStealCount() {
+ return stealCount.get();
+ }
+
+ /**
+ * Accumulate steal count from a worker. Call only
+ * when worker known to be idle.
+ */
+ private void updateStealCount(ForkJoinWorkerThread w) {
+ int sc = w.getAndClearStealCount();
+ if (sc != 0)
+ stealCount.addAndGet(sc);
+ }
+
+ /**
+ * Returns an estimate of the total number of tasks currently held
+ * in queues by worker threads (but not including tasks submitted
+ * to the pool that have not begun executing). This value is only
+ * an approximation, obtained by iterating across all threads in
+ * the pool. This method may be useful for tuning task
+ * granularities.
+ * @return the number of queued tasks.
+ */
+ public long getQueuedTaskCount() {
+ long count = 0;
+ ForkJoinWorkerThread[] ws = workers;
+ for (int i = 0; i < ws.length; ++i) {
+ ForkJoinWorkerThread t = ws[i];
+ if (t != null)
+ count += t.getQueueSize();
+ }
+ return count;
+ }
+
+ /**
+ * Returns an estimate of the number tasks submitted to this pool
+ * that have not yet begun executing. This method takes time
+ * proportional to the number of submissions.
+ * @return the number of queued submissions.
+ */
+ public int getQueuedSubmissionCount() {
+ return submissionQueue.size();
+ }
+
+ /**
+ * Returns true if there are any tasks submitted to this pool
+ * that have not yet begun executing.
+ * @return <code>true</code> if there are any queued submissions.
+ */
+ public boolean hasQueuedSubmissions() {
+ return !submissionQueue.isEmpty();
+ }
+
+ /**
+ * Removes and returns the next unexecuted submission if one is
+ * available. This method may be useful in extensions to this
+ * class that re-assign work in systems with multiple pools.
+ * @return the next submission, or null if none
+ */
+ protected ForkJoinTask<?> pollSubmission() {
+ return submissionQueue.poll();
+ }
+
+ /**
+ * Returns a string identifying this pool, as well as its state,
+ * including indications of run state, parallelism level, and
+ * worker and task counts.
+ *
+ * @return a string identifying this pool, as well as its state
+ */
+ public String toString() {
+ int ps = parallelism;
+ int wc = workerCounts;
+ int rc = runControl;
+ long st = getStealCount();
+ long qt = getQueuedTaskCount();
+ long qs = getQueuedSubmissionCount();
+ return super.toString() +
+ "[" + runStateToString(runStateOf(rc)) +
+ ", parallelism = " + ps +
+ ", size = " + totalCountOf(wc) +
+ ", active = " + activeCountOf(rc) +
+ ", running = " + runningCountOf(wc) +
+ ", steals = " + st +
+ ", tasks = " + qt +
+ ", submissions = " + qs +
+ "]";
+ }
+
+ private static String runStateToString(int rs) {
+ switch(rs) {
+ case RUNNING: return "Running";
+ case SHUTDOWN: return "Shutting down";
+ case TERMINATING: return "Terminating";
+ case TERMINATED: return "Terminated";
+ default: throw new Error("Unknown run state");
+ }
+ }
+
+ // lifecycle control
+
+ /**
+ * Initiates an orderly shutdown in which previously submitted
+ * tasks are executed, but no new tasks will be accepted.
+ * Invocation has no additional effect if already shut down.
+ * Tasks that are in the process of being submitted concurrently
+ * during the course of this method may or may not be rejected.
+ * @throws SecurityException if a security manager exists and
+ * the caller is not permitted to modify threads
+ * because it does not hold {@link
+ * java.lang.RuntimePermission}<code>("modifyThread")</code>,
+ */
+ public void shutdown() {
+ checkPermission();
+ transitionRunStateTo(SHUTDOWN);
+ if (canTerminateOnShutdown(runControl))
+ terminateOnShutdown();
+ }
+
+ /**
+ * Attempts to stop all actively executing tasks, and cancels all
+ * waiting tasks. Tasks that are in the process of being
+ * submitted or executed concurrently during the course of this
+ * method may or may not be rejected. Unlike some other executors,
+ * this method cancels rather than collects non-executed tasks,
+ * so always returns an empty list.
+ * @return an empty list
+ * @throws SecurityException if a security manager exists and
+ * the caller is not permitted to modify threads
+ * because it does not hold {@link
+ * java.lang.RuntimePermission}<code>("modifyThread")</code>,
+ */
+ public List<Runnable> shutdownNow() {
+ checkPermission();
+ terminate();
+ return Collections.emptyList();
+ }
+
+ /**
+ * Returns <code>true</code> if all tasks have completed following shut down.
+ *
+ * @return <code>true</code> if all tasks have completed following shut down
+ */
+ public boolean isTerminated() {
+ return runStateOf(runControl) == TERMINATED;
+ }
+
+ /**
+ * Returns <code>true</code> if the process of termination has
+ * commenced but possibly not yet completed.
+ *
+ * @return <code>true</code> if terminating
+ */
+ public boolean isTerminating() {
+ return runStateOf(runControl) >= TERMINATING;
+ }
+
+ /**
+ * Returns <code>true</code> if this pool has been shut down.
+ *
+ * @return <code>true</code> if this pool has been shut down
+ */
+ public boolean isShutdown() {
+ return runStateOf(runControl) >= SHUTDOWN;
+ }
+
+ /**
+ * Blocks until all tasks have completed execution after a shutdown
+ * request, or the timeout occurs, or the current thread is
+ * interrupted, whichever happens first.
+ *
+ * @param timeout the maximum time to wait
+ * @param unit the time unit of the timeout argument
+ * @return <code>true</code> if this executor terminated and
+ * <code>false</code> if the timeout elapsed before termination
+ * @throws InterruptedException if interrupted while waiting
+ */
+ public boolean awaitTermination(long timeout, TimeUnit unit)
+ throws InterruptedException {
+ long nanos = unit.toNanos(timeout);
+ final ReentrantLock lock = this.workerLock;
+ lock.lock();
+ try {
+ for (;;) {
+ if (isTerminated())
+ return true;
+ if (nanos <= 0)
+ return false;
+ nanos = termination.awaitNanos(nanos);
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ // Shutdown and termination support
+
+ /**
+ * Callback from terminating worker. Null out the corresponding
+ * workers slot, and if terminating, try to terminate, else try to
+ * shrink workers array.
+ * @param w the worker
+ */
+ final void workerTerminated(ForkJoinWorkerThread w) {
+ updateStealCount(w);
+ updateWorkerCount(-1);
+ final ReentrantLock lock = this.workerLock;
+ lock.lock();
+ try {
+ ForkJoinWorkerThread[] ws = workers;
+ int idx = w.poolIndex;
+ if (idx >= 0 && idx < ws.length && ws[idx] == w)
+ ws[idx] = null;
+ if (totalCountOf(workerCounts) == 0) {
+ terminate(); // no-op if already terminating
+ transitionRunStateTo(TERMINATED);
+ termination.signalAll();
+ }
+ else if (!isTerminating()) {
+ tryShrinkWorkerArray();
+ tryResumeSpare(true); // allow replacement
+ }
+ } finally {
+ lock.unlock();
+ }
+ signalIdleWorkers();
+ }
+
+ /**
+ * Initiate termination.
+ */
+ private void terminate() {
+ if (transitionRunStateTo(TERMINATING)) {
+ stopAllWorkers();
+ resumeAllSpares();
+ signalIdleWorkers();
+ cancelQueuedSubmissions();
+ cancelQueuedWorkerTasks();
+ interruptUnterminatedWorkers();
+ signalIdleWorkers(); // resignal after interrupt
+ }
+ }
+
+ /**
+ * Possibly terminate when on shutdown state
+ */
+ private void terminateOnShutdown() {
+ if (!hasQueuedSubmissions() && canTerminateOnShutdown(runControl))
+ terminate();
+ }
+
+ /**
+ * Clear out and cancel submissions
+ */
+ private void cancelQueuedSubmissions() {
+ ForkJoinTask<?> task;
+ while ((task = pollSubmission()) != null)
+ task.cancel(false);
+ }
+
+ /**
+ * Clean out worker queues.
+ */
+ private void cancelQueuedWorkerTasks() {
+ final ReentrantLock lock = this.workerLock;
+ lock.lock();
+ try {
+ ForkJoinWorkerThread[] ws = workers;
+ for (int i = 0; i < ws.length; ++i) {
+ ForkJoinWorkerThread t = ws[i];
+ if (t != null)
+ t.cancelTasks();
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Set each worker's status to terminating. Requires lock to avoid
+ * conflicts with add/remove
+ */
+ private void stopAllWorkers() {
+ final ReentrantLock lock = this.workerLock;
+ lock.lock();
+ try {
+ ForkJoinWorkerThread[] ws = workers;
+ for (int i = 0; i < ws.length; ++i) {
+ ForkJoinWorkerThread t = ws[i];
+ if (t != null)
+ t.shutdownNow();
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Interrupt all unterminated workers. This is not required for
+ * sake of internal control, but may help unstick user code during
+ * shutdown.
+ */
+ private void interruptUnterminatedWorkers() {
+ final ReentrantLock lock = this.workerLock;
+ lock.lock();
+ try {
+ ForkJoinWorkerThread[] ws = workers;
+ for (int i = 0; i < ws.length; ++i) {
+ ForkJoinWorkerThread t = ws[i];
+ if (t != null && !t.isTerminated()) {
+ try {
+ t.interrupt();
+ } catch (SecurityException ignore) {
+ }
+ }
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
+
+ /*
+ * Nodes for event barrier to manage idle threads. Queue nodes
+ * are basic Treiber stack nodes, also used for spare stack.
+ *
+ * The event barrier has an event count and a wait queue (actually
+ * a Treiber stack). Workers are enabled to look for work when
+ * the eventCount is incremented. If they fail to find work, they
+ * may wait for next count. Upon release, threads help others wake
+ * up.
+ *
+ * Synchronization events occur only in enough contexts to
+ * maintain overall liveness:
+ *
+ * - Submission of a new task to the pool
+ * - Resizes or other changes to the workers array
+ * - pool termination
+ * - A worker pushing a task on an empty queue
+ *
+ * The case of pushing a task occurs often enough, and is heavy
+ * enough compared to simple stack pushes, to require special
+ * handling: Method signalWork returns without advancing count if
+ * the queue appears to be empty. This would ordinarily result in
+ * races causing some queued waiters not to be woken up. To avoid
+ * this, the first worker enqueued in method sync (see
+ * syncIsReleasable) rescans for tasks after being enqueued, and
+ * helps signal if any are found. This works well because the
+ * worker has nothing better to do, and so might as well help
+ * alleviate the overhead and contention on the threads actually
+ * doing work. Also, since event counts increments on task
+ * availability exist to maintain liveness (rather than to force
+ * refreshes etc), it is OK for callers to exit early if
+ * contending with another signaller.
+ */
+ static final class WaitQueueNode {
+ WaitQueueNode next; // only written before enqueued
+ volatile ForkJoinWorkerThread thread; // nulled to cancel wait
+ final long count; // unused for spare stack
+
+ WaitQueueNode(long c, ForkJoinWorkerThread w) {
+ count = c;
+ thread = w;
+ }
+
+ /**
+ * Wake up waiter, returning false if known to already
+ */
+ boolean signal() {
+ ForkJoinWorkerThread t = thread;
+ if (t == null)
+ return false;
+ thread = null;
+ LockSupport.unpark(t);
+ return true;
+ }
+
+ /**
+ * Await release on sync
+ */
+ void awaitSyncRelease(ForkJoinPool p) {
+ while (thread != null && !p.syncIsReleasable(this))
+ LockSupport.park();//TR park(this);
+ }
+
+ /**
+ * Await resumption as spare
+ */
+ void awaitSpareRelease() {
+ while (thread != null) {
+ if (!Thread.interrupted())
+ LockSupport.park();//TR park(this);
+ }
+ }
+ }
+
+ /**
+ * Ensures that no thread is waiting for count to advance from the
+ * current value of eventCount read on entry to this method, by
+ * releasing waiting threads if necessary.
+ * @return the count
+ */
+ final long ensureSync() {
+ long c = eventCount;
+ WaitQueueNode q;
+ while ((q = syncStack) != null && q.count < c) {
+ if (casBarrierStack(q, null)) {
+ do {
+ q.signal();
+ } while ((q = q.next) != null);
+ break;
+ }
+ }
+ return c;
+ }
+
+ /**
+ * Increments event count and releases waiting threads.
+ */
+ private void signalIdleWorkers() {
+ long c;
+ do;while (!casEventCount(c = eventCount, c+1));
+ ensureSync();
+ }
+
+ /**
+ * Signal threads waiting to poll a task. Because method sync
+ * rechecks availability, it is OK to only proceed if queue
+ * appears to be non-empty, and OK to skip under contention to
+ * increment count (since some other thread succeeded).
+ */
+ final void signalWork() {
+ long c;
+ WaitQueueNode q;
+ if (syncStack != null &&
+ casEventCount(c = eventCount, c+1) &&
+ (((q = syncStack) != null && q.count <= c) &&
+ (!casBarrierStack(q, q.next) || !q.signal())))
+ ensureSync();
+ }
+
+ /**
+ * Waits until event count advances from last value held by
+ * caller, or if excess threads, caller is resumed as spare, or
+ * caller or pool is terminating. Updates caller's event on exit.
+ * @param w the calling worker thread
+ */
+ final void sync(ForkJoinWorkerThread w) {
+ updateStealCount(w); // Transfer w's count while it is idle
+
+ while (!w.isShutdown() && !isTerminating() && !suspendIfSpare(w)) {
+ long prev = w.lastEventCount;
+ WaitQueueNode node = null;
+ WaitQueueNode h;
+ while (eventCount == prev &&
+ ((h = syncStack) == null || h.count == prev)) {
+ if (node == null)
+ node = new WaitQueueNode(prev, w);
+ if (casBarrierStack(node.next = h, node)) {
+ node.awaitSyncRelease(this);
+ break;
+ }
+ }
+ long ec = ensureSync();
+ if (ec != prev) {
+ w.lastEventCount = ec;
+ break;
+ }
+ }
+ }
+
+ /**
+ * Returns true if worker waiting on sync can proceed:
+ * - on signal (thread == null)
+ * - on event count advance (winning race to notify vs signaller)
+ * - on Interrupt
+ * - if the first queued node, we find work available
+ * If node was not signalled and event count not advanced on exit,
+ * then we also help advance event count.
+ * @return true if node can be released
+ */
+ final boolean syncIsReleasable(WaitQueueNode node) {
+ long prev = node.count;
+ if (!Thread.interrupted() && node.thread != null &&
+ (node.next != null ||
+ !ForkJoinWorkerThread.hasQueuedTasks(workers)) &&
+ eventCount == prev)
+ return false;
+ if (node.thread != null) {
+ node.thread = null;
+ long ec = eventCount;
+ if (prev <= ec) // help signal
+ casEventCount(ec, ec+1);
+ }
+ return true;
+ }
+
+ /**
+ * Returns true if a new sync event occurred since last call to
+ * sync or this method, if so, updating caller's count.
+ */
+ final boolean hasNewSyncEvent(ForkJoinWorkerThread w) {
+ long lc = w.lastEventCount;
+ long ec = ensureSync();
+ if (ec == lc)
+ return false;
+ w.lastEventCount = ec;
+ return true;
+ }
+
+ // Parallelism maintenance
+
+ /**
+ * Decrement running count; if too low, add spare.
+ *
+ * Conceptually, all we need to do here is add or resume a
+ * spare thread when one is about to block (and remove or
+ * suspend it later when unblocked -- see suspendIfSpare).
+ * However, implementing this idea requires coping with
+ * several problems: We have imperfect information about the
+ * states of threads. Some count updates can and usually do
+ * lag run state changes, despite arrangements to keep them
+ * accurate (for example, when possible, updating counts
+ * before signalling or resuming), especially when running on
+ * dynamic JVMs that don't optimize the infrequent paths that
+ * update counts. Generating too many threads can make these
+ * problems become worse, because excess threads are more
+ * likely to be context-switched with others, slowing them all
+ * down, especially if there is no work available, so all are
+ * busy scanning or idling. Also, excess spare threads can
+ * only be suspended or removed when they are idle, not
+ * immediately when they aren't needed. So adding threads will
+ * raise parallelism level for longer than necessary. Also,
+ * FJ applications often enounter highly transient peaks when
+ * many threads are blocked joining, but for less time than it
+ * takes to create or resume spares.
+ *
+ * @param joinMe if non-null, return early if done
+ * @param maintainParallelism if true, try to stay within
+ * target counts, else create only to avoid starvation
+ * @return true if joinMe known to be done
+ */
+ final boolean preJoin(ForkJoinTask<?> joinMe, boolean maintainParallelism) {
+ maintainParallelism &= maintainsParallelism; // overrride
+ boolean dec = false; // true when running count decremented
+ while (spareStack == null || !tryResumeSpare(dec)) {
+ int counts = workerCounts;
+ if (dec || (dec = casWorkerCounts(counts, --counts))) { // CAS cheat
+ if (!needSpare(counts, maintainParallelism))
+ break;
+ if (joinMe.status < 0)
+ return true;
+ if (tryAddSpare(counts))
+ break;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Same idea as preJoin
+ */
+ final boolean preBlock(ManagedBlocker blocker, boolean maintainParallelism){
+ maintainParallelism &= maintainsParallelism;
+ boolean dec = false;
+ while (spareStack == null || !tryResumeSpare(dec)) {
+ int counts = workerCounts;
+ if (dec || (dec = casWorkerCounts(counts, --counts))) {
+ if (!needSpare(counts, maintainParallelism))
+ break;
+ if (blocker.isReleasable())
+ return true;
+ if (tryAddSpare(counts))
+ break;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Returns true if a spare thread appears to be needed. If
+ * maintaining parallelism, returns true when the deficit in
+ * running threads is more than the surplus of total threads, and
+ * there is apparently some work to do. This self-limiting rule
+ * means that the more threads that have already been added, the
+ * less parallelism we will tolerate before adding another.
+ * @param counts current worker counts
+ * @param maintainParallelism try to maintain parallelism
+ */
+ private boolean needSpare(int counts, boolean maintainParallelism) {
+ int ps = parallelism;
+ int rc = runningCountOf(counts);
+ int tc = totalCountOf(counts);
+ int runningDeficit = ps - rc;
+ int totalSurplus = tc - ps;
+ return (tc < maxPoolSize &&
+ (rc == 0 || totalSurplus < 0 ||
+ (maintainParallelism &&
+ runningDeficit > totalSurplus &&
+ ForkJoinWorkerThread.hasQueuedTasks(workers))));
+ }
+
+ /**
+ * Add a spare worker if lock available and no more than the
+ * expected numbers of threads exist
+ * @return true if successful
+ */
+ private boolean tryAddSpare(int expectedCounts) {
+ final ReentrantLock lock = this.workerLock;
+ int expectedRunning = runningCountOf(expectedCounts);
+ int expectedTotal = totalCountOf(expectedCounts);
+ boolean success = false;
+ boolean locked = false;
+ // confirm counts while locking; CAS after obtaining lock
+ try {
+ for (;;) {
+ int s = workerCounts;
+ int tc = totalCountOf(s);
+ int rc = runningCountOf(s);
+ if (rc > expectedRunning || tc > expectedTotal)
+ break;
+ if (!locked && !(locked = lock.tryLock()))
+ break;
+ if (casWorkerCounts(s, workerCountsFor(tc+1, rc+1))) {
+ createAndStartSpare(tc);
+ success = true;
+ break;
+ }
+ }
+ } finally {
+ if (locked)
+ lock.unlock();
+ }
+ return success;
+ }
+
+ /**
+ * Add the kth spare worker. On entry, pool coounts are already
+ * adjusted to reflect addition.
+ */
+ private void createAndStartSpare(int k) {
+ ForkJoinWorkerThread w = null;
+ ForkJoinWorkerThread[] ws = ensureWorkerArrayCapacity(k + 1);
+ int len = ws.length;
+ // Probably, we can place at slot k. If not, find empty slot
+ if (k < len && ws[k] != null) {
+ for (k = 0; k < len && ws[k] != null; ++k)
+ ;
+ }
+ if (k < len && !isTerminating() && (w = createWorker(k)) != null) {
+ ws[k] = w;
+ w.start();
+ }
+ else
+ updateWorkerCount(-1); // adjust on failure
+ signalIdleWorkers();
+ }
+
+ /**
+ * Suspend calling thread w if there are excess threads. Called
+ * only from sync. Spares are enqueued in a Treiber stack
+ * using the same WaitQueueNodes as barriers. They are resumed
+ * mainly in preJoin, but are also woken on pool events that
+ * require all threads to check run state.
+ * @param w the caller
+ */
+ private boolean suspendIfSpare(ForkJoinWorkerThread w) {
+ WaitQueueNode node = null;
+ int s;
+ while (parallelism < runningCountOf(s = workerCounts)) {
+ if (node == null)
+ node = new WaitQueueNode(0, w);
+ if (casWorkerCounts(s, s-1)) { // representation-dependent
+ // push onto stack
+ do;while (!casSpareStack(node.next = spareStack, node));
+ // block until released by resumeSpare
+ node.awaitSpareRelease();
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Try to pop and resume a spare thread.
+ * @param updateCount if true, increment running count on success
+ * @return true if successful
+ */
+ private boolean tryResumeSpare(boolean updateCount) {
+ WaitQueueNode q;
+ while ((q = spareStack) != null) {
+ if (casSpareStack(q, q.next)) {
+ if (updateCount)
+ updateRunningCount(1);
+ q.signal();
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Pop and resume all spare threads. Same idea as ensureSync.
+ * @return true if any spares released
+ */
+ private boolean resumeAllSpares() {
+ WaitQueueNode q;
+ while ( (q = spareStack) != null) {
+ if (casSpareStack(q, null)) {
+ do {
+ updateRunningCount(1);
+ q.signal();
+ } while ((q = q.next) != null);
+ return true;
+ }
+ }
+ return false;
+ }
+
+ /**
+ * Pop and shutdown excessive spare threads. Call only while
+ * holding lock. This is not guaranteed to eliminate all excess
+ * threads, only those suspended as spares, which are the ones
+ * unlikely to be needed in the future.
+ */
+ private void trimSpares() {
+ int surplus = totalCountOf(workerCounts) - parallelism;
+ WaitQueueNode q;
+ while (surplus > 0 && (q = spareStack) != null) {
+ if (casSpareStack(q, null)) {
+ do {
+ updateRunningCount(1);
+ ForkJoinWorkerThread w = q.thread;
+ if (w != null && surplus > 0 &&
+ runningCountOf(workerCounts) > 0 && w.shutdown())
+ --surplus;
+ q.signal();
+ } while ((q = q.next) != null);
+ }
+ }
+ }
+
+ /**
+ * Interface for extending managed parallelism for tasks running
+ * in ForkJoinPools. A ManagedBlocker provides two methods.
+ * Method <code>isReleasable</code> must return true if blocking is not
+ * necessary. Method <code>block</code> blocks the current thread
+ * if necessary (perhaps internally invoking isReleasable before
+ * actually blocking.).
+ * <p>For example, here is a ManagedBlocker based on a
+ * ReentrantLock:
+ * <pre>
+ * class ManagedLocker implements ManagedBlocker {
+ * final ReentrantLock lock;
+ * boolean hasLock = false;
+ * ManagedLocker(ReentrantLock lock) { this.lock = lock; }
+ * public boolean block() {
+ * if (!hasLock)
+ * lock.lock();
+ * return true;
+ * }
+ * public boolean isReleasable() {
+ * return hasLock || (hasLock = lock.tryLock());
+ * }
+ * }
+ * </pre>
+ */
+ public static interface ManagedBlocker {
+ /**
+ * Possibly blocks the current thread, for example waiting for
+ * a lock or condition.
+ * @return true if no additional blocking is necessary (i.e.,
+ * if isReleasable would return true).
+ * @throws InterruptedException if interrupted while waiting
+ * (the method is not required to do so, but is allowe to).
+ */
+ boolean block() throws InterruptedException;
+
+ /**
+ * Returns true if blocking is unnecessary.
+ */
+ boolean isReleasable();
+ }
+
+ /**
+ * Blocks in accord with the given blocker. If the current thread
+ * is a ForkJoinWorkerThread, this method possibly arranges for a
+ * spare thread to be activated if necessary to ensure parallelism
+ * while the current thread is blocked. If
+ * <code>maintainParallelism</code> is true and the pool supports
+ * it ({@link #getMaintainsParallelism}), this method attempts to
+ * maintain the pool's nominal parallelism. Otherwise if activates
+ * a thread only if necessary to avoid complete starvation. This
+ * option may be preferable when blockages use timeouts, or are
+ * almost always brief.
+ *
+ * <p> If the caller is not a ForkJoinTask, this method is behaviorally
+ * equivalent to
+ * <pre>
+ * while (!blocker.isReleasable())
+ * if (blocker.block())
+ * return;
+ * </pre>
+ * If the caller is a ForkJoinTask, then the pool may first
+ * be expanded to ensure parallelism, and later adjusted.
+ *
+ * @param blocker the blocker
+ * @param maintainParallelism if true and supported by this pool,
+ * attempt to maintain the pool's nominal parallelism; otherwise
+ * activate a thread only if necessary to avoid complete
+ * starvation.
+ * @throws InterruptedException if blocker.block did so.
+ */
+ public static void managedBlock(ManagedBlocker blocker,
+ boolean maintainParallelism)
+ throws InterruptedException {
+ Thread t = Thread.currentThread();
+ ForkJoinPool pool = (t instanceof ForkJoinWorkerThread?
+ ((ForkJoinWorkerThread)t).pool : null);
+ if (!blocker.isReleasable()) {
+ try {
+ if (pool == null ||
+ !pool.preBlock(blocker, maintainParallelism))
+ awaitBlocker(blocker);
+ } finally {
+ if (pool != null)
+ pool.updateRunningCount(1);
+ }
+ }
+ }
+
+ private static void awaitBlocker(ManagedBlocker blocker)
+ throws InterruptedException {
+ do;while (!blocker.isReleasable() && !blocker.block());
+ }
+
+ // AbstractExecutorService overrides
+
+ protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
+ return new AdaptedRunnable(runnable, value);
+ }
+
+ protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
+ return new AdaptedCallable(callable);
+ }
+
+
+ // Temporary Unsafe mechanics for preliminary release
+ private static Unsafe getUnsafe() throws Throwable {
+ try {
+ return Unsafe.getUnsafe();
+ } catch (SecurityException se) {
+ try {
+ return java.security.AccessController.doPrivileged
+ (new java.security.PrivilegedExceptionAction<Unsafe>() {
+ public Unsafe run() throws Exception {
+ return getUnsafePrivileged();
+ }});
+ } catch (java.security.PrivilegedActionException e) {
+ throw e.getCause();
+ }
+ }
+ }
+
+ private static Unsafe getUnsafePrivileged()
+ throws NoSuchFieldException, IllegalAccessException {
+ Field f = Unsafe.class.getDeclaredField("theUnsafe");
+ f.setAccessible(true);
+ return (Unsafe) f.get(null);
+ }
+
+ private static long fieldOffset(String fieldName)
+ throws NoSuchFieldException {
+ return _unsafe.objectFieldOffset
+ (ForkJoinPool.class.getDeclaredField(fieldName));
+ }
+
+ static final Unsafe _unsafe;
+ static final long eventCountOffset;
+ static final long workerCountsOffset;
+ static final long runControlOffset;
+ static final long syncStackOffset;
+ static final long spareStackOffset;
+
+ static {
+ try {
+ _unsafe = getUnsafe();
+ eventCountOffset = fieldOffset("eventCount");
+ workerCountsOffset = fieldOffset("workerCounts");
+ runControlOffset = fieldOffset("runControl");
+ syncStackOffset = fieldOffset("syncStack");
+ spareStackOffset = fieldOffset("spareStack");
+ } catch (Throwable e) {
+ throw new RuntimeException("Could not initialize intrinsics", e);
+ }
+ }
+
+ private boolean casEventCount(long cmp, long val) {
+ return _unsafe.compareAndSwapLong(this, eventCountOffset, cmp, val);
+ }
+ private boolean casWorkerCounts(int cmp, int val) {
+ return _unsafe.compareAndSwapInt(this, workerCountsOffset, cmp, val);
+ }
+ private boolean casRunControl(int cmp, int val) {
+ return _unsafe.compareAndSwapInt(this, runControlOffset, cmp, val);
+ }
+ private boolean casSpareStack(WaitQueueNode cmp, WaitQueueNode val) {
+ return _unsafe.compareAndSwapObject(this, spareStackOffset, cmp, val);
+ }
+ private boolean casBarrierStack(WaitQueueNode cmp, WaitQueueNode val) {
+ return _unsafe.compareAndSwapObject(this, syncStackOffset, cmp, val);
+ }
+}
diff --git a/src/actors/scala/actors/forkjoin/ForkJoinTask.java b/src/actors/scala/actors/forkjoin/ForkJoinTask.java
new file mode 100644
index 0000000000..5e40940ff3
--- /dev/null
+++ b/src/actors/scala/actors/forkjoin/ForkJoinTask.java
@@ -0,0 +1,1053 @@
+/*
+ * Written by Doug Lea with assistance from members of JCP JSR-166
+ * Expert Group and released to the public domain, as explained at
+ * http://creativecommons.org/licenses/publicdomain
+ */
+
+package scala.actors.forkjoin;
+import java.io.Serializable;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
+import sun.misc.Unsafe;
+import java.lang.reflect.*;
+
+/**
+ * Abstract base class for tasks that run within a {@link
+ * 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.
+ *
+ * <p> A "main" ForkJoinTask begins execution when submitted to a
+ * {@link ForkJoinPool}. Once started, it will usually in turn start
+ * other subtasks. As indicated by the name of this class, many
+ * programs using ForkJoinTasks employ only methods <code>fork</code>
+ * and <code>join</code>, or derivatives such as
+ * <code>invokeAll</code>. 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.
+ *
+ * <p>A ForkJoinTask is a lightweight form of {@link Future}. The
+ * efficiency of ForkJoinTasks stems from a set of restrictions (that
+ * are only partially statically enforceable) reflecting their
+ * intended use as computational tasks calculating pure functions or
+ * operating on purely isolated objects. The primary coordination
+ * mechanisms are {@link #fork}, that arranges asynchronous execution,
+ * and {@link #join}, that doesn't proceed until the task's result has
+ * been computed. Computations should avoid <code>synchronized</code>
+ * 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. Tasks should also not perform blocking IO,
+ * and should ideally access variables that are completely independent
+ * of those accessed by other running tasks. Minor breaches of these
+ * restrictions, for example using shared output streams, may be
+ * tolerable in practice, but frequent use may result in poor
+ * performance, and the potential to indefinitely stall if the number
+ * of threads not waiting for IO or other external synchronization
+ * becomes exhausted. This usage restriction is in part enforced by
+ * not permitting checked exceptions such as <code>IOExceptions</code>
+ * to be thrown. However, computations may still encounter unchecked
+ * exceptions, that are rethrown to callers attempting join
+ * them. These exceptions may additionally include
+ * RejectedExecutionExceptions stemming from internal resource
+ * exhaustion such as failure to allocate internal task queues.
+ *
+ * <p>The primary method for awaiting completion and extracting
+ * results of a task is {@link #join}, but there are several variants:
+ * The {@link Future#get} methods support interruptible and/or timed
+ * waits for completion and report results using <code>Future</code>
+ * conventions. Method {@link #helpJoin} enables callers to actively
+ * execute other tasks while awaiting joins, which is sometimes more
+ * efficient but only applies when all subtasks are known to be
+ * strictly tree-structured. Method {@link #invoke} is semantically
+ * equivalent to <code>fork(); join()</code> but always attempts to
+ * begin execution in the current thread. The "<em>quiet</em>" 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 <code>invokeAll</code> (available in multiple versions)
+ * performs the most common form of parallel invocation: forking a set
+ * of tasks and joining them all.
+ *
+ * <p> 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. Normally, a concrete
+ * ForkJoinTask subclass declares fields comprising its parameters,
+ * established in a constructor, and then defines a <code>compute</code>
+ * method that somehow uses the control methods supplied by this base
+ * class. While these methods have <code>public</code> access (to allow
+ * instances of different task subclasses to call each others
+ * methods), some of them may only be called from within other
+ * ForkJoinTasks. Attempts to invoke them in other contexts result in
+ * exceptions or errors possibly including ClassCastException.
+ *
+ * <p>Most base support methods are <code>final</code> because their
+ * implementations are intrinsically tied to the underlying
+ * lightweight task scheduling framework, and so cannot be overridden.
+ * Developers creating new basic styles of fork/join processing should
+ * minimally implement <code>protected</code> methods
+ * <code>exec</code>, <code>setRawResult</code>, and
+ * <code>getRawResult</code>, while also introducing an abstract
+ * computational method that can be implemented in its subclasses,
+ * possibly relying on other <code>protected</code> methods provided
+ * by this class.
+ *
+ * <p>ForkJoinTasks should perform relatively small amounts of
+ * computations, othewise splitting into smaller tasks. As a very
+ * rough rule of thumb, a task should perform more than 100 and less
+ * than 10000 basic computational steps. If tasks are too big, then
+ * parellelism cannot improve throughput. If too small, then memory
+ * and internal task maintenance overhead may overwhelm processing.
+ *
+ * <p>ForkJoinTasks are <code>Serializable</code>, which enables them
+ * to be used in extensions such as remote execution frameworks. It is
+ * in general sensible to serialize tasks only before or after, but
+ * not during execution. Serialization is not relied on during
+ * execution itself.
+ */
+public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
+
+ /**
+ * Run control status bits packed into a single int to minimize
+ * footprint and to ensure atomicity (via CAS). Status is
+ * initially zero, and takes on nonnegative values until
+ * completed, upon which status holds COMPLETED. CANCELLED, or
+ * EXCEPTIONAL, which use the top 3 bits. Tasks undergoing
+ * blocking waits by other threads have SIGNAL_MASK bits set --
+ * bit 15 for external (nonFJ) waits, and the rest a count of
+ * waiting FJ threads. (This representation relies on
+ * ForkJoinPool max thread limits). Completion of a stolen task
+ * with SIGNAL_MASK bits set awakens waiter via notifyAll. Even
+ * though suboptimal for some purposes, we use basic builtin
+ * wait/notify to take advantage of "monitor inflation" in JVMs
+ * that we would otherwise need to emulate to avoid adding further
+ * per-task bookkeeping overhead. Note that bits 16-28 are
+ * currently unused. Also value 0x80000000 is available as spare
+ * completion value.
+ */
+ volatile int status; // accessed directy by pool and workers
+
+ static final int COMPLETION_MASK = 0xe0000000;
+ static final int NORMAL = 0xe0000000; // == mask
+ static final int CANCELLED = 0xc0000000;
+ static final int EXCEPTIONAL = 0xa0000000;
+ static final int SIGNAL_MASK = 0x0000ffff;
+ static final int INTERNAL_SIGNAL_MASK = 0x00007fff;
+ static final int EXTERNAL_SIGNAL = 0x00008000; // top bit of low word
+
+ /**
+ * 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 us a weak ref table. Note
+ * that cancellation exceptions don't appear in the table, but are
+ * instead recorded as status values.
+ * Todo: Use ConcurrentReferenceHashMap
+ */
+ static final Map<ForkJoinTask<?>, Throwable> exceptionMap =
+ Collections.synchronizedMap
+ (new WeakHashMap<ForkJoinTask<?>, Throwable>());
+
+ // within-package utilities
+
+ /**
+ * Get current worker thread, or null if not a worker thread
+ */
+ static ForkJoinWorkerThread getWorker() {
+ Thread t = Thread.currentThread();
+ return ((t instanceof ForkJoinWorkerThread)?
+ (ForkJoinWorkerThread)t : null);
+ }
+
+ final boolean casStatus(int cmp, int val) {
+ return _unsafe.compareAndSwapInt(this, statusOffset, cmp, val);
+ }
+
+ /**
+ * Workaround for not being able to rethrow unchecked exceptions.
+ */
+ static void rethrowException(Throwable ex) {
+ if (ex != null)
+ _unsafe.throwException(ex);
+ }
+
+ // Setting completion status
+
+ /**
+ * Mark completion and wake up threads waiting to join this task.
+ * @param completion one of NORMAL, CANCELLED, EXCEPTIONAL
+ */
+ final void setCompletion(int completion) {
+ ForkJoinPool pool = getPool();
+ if (pool != null) {
+ int s; // Clear signal bits while setting completion status
+ do;while ((s = status) >= 0 && !casStatus(s, completion));
+
+ if ((s & SIGNAL_MASK) != 0) {
+ if ((s &= INTERNAL_SIGNAL_MASK) != 0)
+ pool.updateRunningCount(s);
+ synchronized(this) { notifyAll(); }
+ }
+ }
+ else
+ externallySetCompletion(completion);
+ }
+
+ /**
+ * Version of setCompletion for non-FJ threads. Leaves signal
+ * bits for unblocked threads to adjust, and always notifies.
+ */
+ private void externallySetCompletion(int completion) {
+ int s;
+ do;while ((s = status) >= 0 &&
+ !casStatus(s, (s & SIGNAL_MASK) | completion));
+ synchronized(this) { notifyAll(); }
+ }
+
+ /**
+ * Sets status to indicate normal completion
+ */
+ final void setNormalCompletion() {
+ // Try typical fast case -- single CAS, no signal, not already done.
+ // Manually expand casStatus to improve chances of inlining it
+ if (!_unsafe.compareAndSwapInt(this, statusOffset, 0, NORMAL))
+ setCompletion(NORMAL);
+ }
+
+ // internal waiting and notification
+
+ /**
+ * Performs the actual monitor wait for awaitDone
+ */
+ private void doAwaitDone() {
+ // Minimize lock bias and in/de-flation effects by maximizing
+ // chances of waiting inside sync
+ try {
+ while (status >= 0)
+ synchronized(this) { if (status >= 0) wait(); }
+ } catch (InterruptedException ie) {
+ onInterruptedWait();
+ }
+ }
+
+ /**
+ * Performs the actual monitor wait for awaitDone
+ */
+ private void doAwaitDone(long startTime, long nanos) {
+ synchronized(this) {
+ try {
+ while (status >= 0) {
+ long nt = nanos - System.nanoTime() - startTime;
+ if (nt <= 0)
+ break;
+ wait(nt / 1000000, (int)(nt % 1000000));
+ }
+ } catch (InterruptedException ie) {
+ onInterruptedWait();
+ }
+ }
+ }
+
+ // Awaiting completion
+
+ /**
+ * Sets status to indicate there is joiner, then waits for join,
+ * surrounded with pool notifications.
+ * @return status upon exit
+ */
+ private int awaitDone(ForkJoinWorkerThread w, boolean maintainParallelism) {
+ ForkJoinPool pool = w == null? null : w.pool;
+ int s;
+ while ((s = status) >= 0) {
+ if (casStatus(s, pool == null? s|EXTERNAL_SIGNAL : s+1)) {
+ if (pool == null || !pool.preJoin(this, maintainParallelism))
+ doAwaitDone();
+ if (((s = status) & INTERNAL_SIGNAL_MASK) != 0)
+ adjustPoolCountsOnUnblock(pool);
+ break;
+ }
+ }
+ return s;
+ }
+
+ /**
+ * Timed version of awaitDone
+ * @return status upon exit
+ */
+ private int awaitDone(ForkJoinWorkerThread w, long nanos) {
+ ForkJoinPool pool = w == null? null : w.pool;
+ int s;
+ while ((s = status) >= 0) {
+ if (casStatus(s, pool == null? s|EXTERNAL_SIGNAL : s+1)) {
+ long startTime = System.nanoTime();
+ if (pool == null || !pool.preJoin(this, false))
+ doAwaitDone(startTime, nanos);
+ if ((s = status) >= 0) {
+ adjustPoolCountsOnCancelledWait(pool);
+ s = status;
+ }
+ if (s < 0 && (s & INTERNAL_SIGNAL_MASK) != 0)
+ adjustPoolCountsOnUnblock(pool);
+ break;
+ }
+ }
+ return s;
+ }
+
+ /**
+ * Notify pool that thread is unblocked. Called by signalled
+ * threads when woken by non-FJ threads (which is atypical).
+ */
+ private void adjustPoolCountsOnUnblock(ForkJoinPool pool) {
+ int s;
+ do;while ((s = status) < 0 && !casStatus(s, s & COMPLETION_MASK));
+ if (pool != null && (s &= INTERNAL_SIGNAL_MASK) != 0)
+ pool.updateRunningCount(s);
+ }
+
+ /**
+ * Notify pool to adjust counts on cancelled or timed out wait
+ */
+ private void adjustPoolCountsOnCancelledWait(ForkJoinPool pool) {
+ if (pool != null) {
+ int s;
+ while ((s = status) >= 0 && (s & INTERNAL_SIGNAL_MASK) != 0) {
+ if (casStatus(s, s - 1)) {
+ pool.updateRunningCount(1);
+ break;
+ }
+ }
+ }
+ }
+
+ /**
+ * Handle interruptions during waits.
+ */
+ private void onInterruptedWait() {
+ ForkJoinWorkerThread w = getWorker();
+ if (w == null)
+ Thread.currentThread().interrupt(); // re-interrupt
+ else if (w.isTerminating())
+ cancelIgnoringExceptions();
+ // else if FJworker, ignore interrupt
+ }
+
+ // Recording and reporting exceptions
+
+ private void setDoneExceptionally(Throwable rex) {
+ exceptionMap.put(this, rex);
+ setCompletion(EXCEPTIONAL);
+ }
+
+ /**
+ * Throws the exception associated with status s;
+ * @throws the exception
+ */
+ private void reportException(int s) {
+ if ((s &= COMPLETION_MASK) < NORMAL) {
+ if (s == CANCELLED)
+ throw new CancellationException();
+ else
+ rethrowException(exceptionMap.get(this));
+ }
+ }
+
+ /**
+ * Returns result or throws exception using j.u.c.Future conventions
+ * Only call when isDone known to be true.
+ */
+ private V reportFutureResult()
+ throws ExecutionException, InterruptedException {
+ int s = status & COMPLETION_MASK;
+ if (s < NORMAL) {
+ Throwable ex;
+ if (s == CANCELLED)
+ throw new CancellationException();
+ if (s == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null)
+ throw new ExecutionException(ex);
+ if (Thread.interrupted())
+ throw new InterruptedException();
+ }
+ return getRawResult();
+ }
+
+ /**
+ * Returns result or throws exception using j.u.c.Future conventions
+ * with timeouts
+ */
+ private V reportTimedFutureResult()
+ throws InterruptedException, ExecutionException, TimeoutException {
+ Throwable ex;
+ int s = status & COMPLETION_MASK;
+ if (s == NORMAL)
+ return getRawResult();
+ if (s == CANCELLED)
+ throw new CancellationException();
+ if (s == EXCEPTIONAL && (ex = exceptionMap.get(this)) != null)
+ throw new ExecutionException(ex);
+ if (Thread.interrupted())
+ throw new InterruptedException();
+ throw new TimeoutException();
+ }
+
+ // internal execution methods
+
+ /**
+ * Calls exec, recording completion, and rethrowing exception if
+ * encountered. Caller should normally check status before calling
+ * @return true if completed normally
+ */
+ private boolean tryExec() {
+ try { // try block must contain only call to exec
+ if (!exec())
+ return false;
+ } catch (Throwable rex) {
+ setDoneExceptionally(rex);
+ rethrowException(rex);
+ return false; // not reached
+ }
+ setNormalCompletion();
+ return true;
+ }
+
+ /**
+ * Main execution method used by worker threads. Invokes
+ * base computation unless already complete
+ */
+ final void quietlyExec() {
+ if (status >= 0) {
+ try {
+ if (!exec())
+ return;
+ } catch(Throwable rex) {
+ setDoneExceptionally(rex);
+ return;
+ }
+ setNormalCompletion();
+ }
+ }
+
+ /**
+ * Calls exec, recording but not rethrowing exception
+ * Caller should normally check status before calling
+ * @return true if completed normally
+ */
+ private boolean tryQuietlyInvoke() {
+ try {
+ if (!exec())
+ return false;
+ } catch (Throwable rex) {
+ setDoneExceptionally(rex);
+ return false;
+ }
+ setNormalCompletion();
+ return true;
+ }
+
+ /**
+ * Cancel, ignoring any exceptions it throws
+ */
+ final void cancelIgnoringExceptions() {
+ try {
+ cancel(false);
+ } catch(Throwable ignore) {
+ }
+ }
+
+ /**
+ * Main implementation of helpJoin
+ */
+ private int busyJoin(ForkJoinWorkerThread w) {
+ int s;
+ ForkJoinTask<?> t;
+ while ((s = status) >= 0 && (t = w.scanWhileJoining(this)) != null)
+ t.quietlyExec();
+ return (s >= 0)? awaitDone(w, false) : s; // block if no work
+ }
+
+ // public methods
+
+ /**
+ * Arranges to asynchronously execute this task. 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. This
+ * method may be invoked only from within ForkJoinTask
+ * computations. Attempts to invoke in other contexts result in
+ * exceptions or errors possibly including ClassCastException.
+ */
+ public final void fork() {
+ ((ForkJoinWorkerThread)(Thread.currentThread())).pushTask(this);
+ }
+
+ /**
+ * Returns the result of the computation when it is ready.
+ * This method differs from <code>get</code> in that abnormal
+ * completion results in RuntimeExceptions or Errors, not
+ * ExecutionExceptions.
+ *
+ * @return the computed result
+ */
+ public final V join() {
+ ForkJoinWorkerThread w = getWorker();
+ if (w == null || status < 0 || !w.unpushTask(this) || !tryExec())
+ reportException(awaitDone(w, true));
+ return getRawResult();
+ }
+
+ /**
+ * Commences performing this task, awaits its completion if
+ * necessary, and return its result.
+ * @throws Throwable (a RuntimeException, Error, or unchecked
+ * exception) if the underlying computation did so.
+ * @return the computed result
+ */
+ public final V invoke() {
+ if (status >= 0 && tryExec())
+ return getRawResult();
+ else
+ return join();
+ }
+
+ /**
+ * Forks both tasks, returning when <code>isDone</code> holds for
+ * both of them or an exception is encountered. This method may be
+ * invoked only from within ForkJoinTask computations. Attempts to
+ * invoke in other contexts result in exceptions or errors
+ * possibly including ClassCastException.
+ * @param t1 one task
+ * @param t2 the other task
+ * @throws NullPointerException if t1 or t2 are null
+ * @throws RuntimeException or Error if either task did so.
+ */
+ public static void invokeAll(ForkJoinTask<?>t1, ForkJoinTask<?> t2) {
+ t2.fork();
+ t1.invoke();
+ t2.join();
+ }
+
+ /**
+ * Forks the given tasks, returning when <code>isDone</code> holds
+ * for all of them. If any task encounters an exception, others
+ * may be cancelled. This method may be invoked only from within
+ * ForkJoinTask computations. Attempts to invoke in other contexts
+ * result in exceptions or errors possibly including ClassCastException.
+ * @param tasks the array of tasks
+ * @throws NullPointerException if tasks or any element are null.
+ * @throws RuntimeException or Error if any task did so.
+ */
+ public static void invokeAll(ForkJoinTask<?>... tasks) {
+ Throwable ex = null;
+ int last = tasks.length - 1;
+ for (int i = last; i >= 0; --i) {
+ ForkJoinTask<?> t = tasks[i];
+ if (t == null) {
+ if (ex == null)
+ ex = new NullPointerException();
+ }
+ else if (i != 0)
+ t.fork();
+ else {
+ t.quietlyInvoke();
+ if (ex == null)
+ ex = t.getException();
+ }
+ }
+ for (int i = 1; i <= last; ++i) {
+ ForkJoinTask<?> t = tasks[i];
+ if (t != null) {
+ if (ex != null)
+ t.cancel(false);
+ else {
+ t.quietlyJoin();
+ if (ex == null)
+ ex = t.getException();
+ }
+ }
+ }
+ if (ex != null)
+ rethrowException(ex);
+ }
+
+ /**
+ * Forks all tasks in the collection, returning when
+ * <code>isDone</code> holds for all of them. If any task
+ * encounters an exception, others may be cancelled. This method
+ * may be invoked only from within ForkJoinTask
+ * computations. Attempts to invoke in other contexts resul!t in
+ * exceptions or errors possibly including ClassCastException.
+ * @param tasks the collection of tasks
+ * @throws NullPointerException if tasks or any element are null.
+ * @throws RuntimeException or Error if any task did so.
+ */
+ public static void invokeAll(Collection<? extends ForkJoinTask<?>> tasks) {
+ if (!(tasks instanceof List)) {
+ invokeAll(tasks.toArray(new ForkJoinTask[tasks.size()]));
+ return;
+ }
+ List<? extends ForkJoinTask<?>> ts =
+ (List<? extends ForkJoinTask<?>>)tasks;
+ Throwable ex = null;
+ int last = ts.size() - 1;
+ for (int i = last; i >= 0; --i) {
+ ForkJoinTask<?> t = ts.get(i);
+ if (t == null) {
+ if (ex == null)
+ ex = new NullPointerException();
+ }
+ else if (i != 0)
+ t.fork();
+ else {
+ t.quietlyInvoke();
+ if (ex == null)
+ ex = t.getException();
+ }
+ }
+ for (int i = 1; i <= last; ++i) {
+ ForkJoinTask<?> t = ts.get(i);
+ if (t != null) {
+ if (ex != null)
+ t.cancel(false);
+ else {
+ t.quietlyJoin();
+ if (ex == null)
+ ex = t.getException();
+ }
+ }
+ }
+ if (ex != null)
+ rethrowException(ex);
+ }
+
+ /**
+ * Returns true if the computation performed by this task has
+ * completed (or has been cancelled).
+ * @return true if this computation has completed
+ */
+ public final boolean isDone() {
+ return status < 0;
+ }
+
+ /**
+ * Returns true if this task was cancelled.
+ * @return true if this task was cancelled
+ */
+ public final boolean isCancelled() {
+ return (status & COMPLETION_MASK) == CANCELLED;
+ }
+
+ /**
+ * Asserts that the results of this task's computation will not be
+ * used. If a cancellation occurs before atempting to execute this
+ * task, then execution will be suppressed, <code>isCancelled</code>
+ * will report true, and <code>join</code> will result in a
+ * <code>CancellationException</code> being thrown. Otherwise, when
+ * cancellation races with completion, there are no guarantees
+ * about whether <code>isCancelled</code> will report true, whether
+ * <code>join</code> will return normally or via an exception, or
+ * whether these behaviors will remain consistent upon repeated
+ * invocation.
+ *
+ * <p>This method may be overridden in subclasses, but if so, must
+ * still ensure that these minimal properties hold. In particular,
+ * the cancel method itself must not throw exceptions.
+ *
+ * <p> This method is designed to be invoked by <em>other</em>
+ * tasks. To terminate the current task, you can just return or
+ * throw an unchecked exception from its computation method, or
+ * invoke <code>completeExceptionally</code>.
+ *
+ * @param mayInterruptIfRunning this value is ignored in the
+ * default implementation because tasks are not in general
+ * cancelled via interruption.
+ *
+ * @return true if this task is now cancelled
+ */
+ public boolean cancel(boolean mayInterruptIfRunning) {
+ setCompletion(CANCELLED);
+ return (status & COMPLETION_MASK) == CANCELLED;
+ }
+
+ /**
+ * Returns true if this task threw an exception or was cancelled
+ * @return true if this task threw an exception or was cancelled
+ */
+ public final boolean isCompletedAbnormally() {
+ return (status & COMPLETION_MASK) < NORMAL;
+ }
+
+ /**
+ * 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.
+ * @return the exception, or null if none
+ */
+ public final Throwable getException() {
+ int s = status & COMPLETION_MASK;
+ if (s >= NORMAL)
+ return null;
+ if (s == CANCELLED)
+ return new CancellationException();
+ return exceptionMap.get(this);
+ }
+
+ /**
+ * Completes this task abnormally, and if not already aborted or
+ * cancelled, causes it to throw the given exception upon
+ * <code>join</code> 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 likely to be wrong. This method is
+ * overridable, but overridden versions must invoke <code>super</code>
+ * implementation to maintain guarantees.
+ *
+ * @param 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.
+ */
+ public void completeExceptionally(Throwable ex) {
+ setDoneExceptionally((ex instanceof RuntimeException) ||
+ (ex instanceof Error)? ex :
+ new RuntimeException(ex));
+ }
+
+ /**
+ * Completes this task, and if not already aborted or cancelled,
+ * returning a <code>null</code> result upon <code>join</code> 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 likely to be wrong. This method is
+ * overridable, but overridden versions must invoke <code>super</code>
+ * implementation to maintain guarantees.
+ *
+ * @param value the result value for this task.
+ */
+ public void complete(V value) {
+ try {
+ setRawResult(value);
+ } catch(Throwable rex) {
+ setDoneExceptionally(rex);
+ return;
+ }
+ setNormalCompletion();
+ }
+
+ public final V get() throws InterruptedException, ExecutionException {
+ ForkJoinWorkerThread w = getWorker();
+ if (w == null || status < 0 || !w.unpushTask(this) || !tryQuietlyInvoke())
+ awaitDone(w, true);
+ return reportFutureResult();
+ }
+
+ public final V get(long timeout, TimeUnit unit)
+ throws InterruptedException, ExecutionException, TimeoutException {
+ ForkJoinWorkerThread w = getWorker();
+ if (w == null || status < 0 || !w.unpushTask(this) || !tryQuietlyInvoke())
+ awaitDone(w, unit.toNanos(timeout));
+ return reportTimedFutureResult();
+ }
+
+ /**
+ * Possibly executes other tasks until this task is ready, then
+ * returns the result of the computation. This method may be more
+ * efficient than <code>join</code>, but is only applicable when
+ * there are no potemtial dependencies between continuation of the
+ * current task and that of any other task that might be executed
+ * while helping. (This usually holds for pure divide-and-conquer
+ * tasks). This method may be invoked only from within
+ * ForkJoinTask computations. Attempts to invoke in other contexts
+ * resul!t in exceptions or errors possibly including ClassCastException.
+ * @return the computed result
+ */
+ public final V helpJoin() {
+ ForkJoinWorkerThread w = (ForkJoinWorkerThread)(Thread.currentThread());
+ if (status < 0 || !w.unpushTask(this) || !tryExec())
+ reportException(busyJoin(w));
+ return getRawResult();
+ }
+
+ /**
+ * Possibly executes other tasks until this task is ready. This
+ * method may be invoked only from within ForkJoinTask
+ * computations. Attempts to invoke in other contexts resul!t in
+ * exceptions or errors possibly including ClassCastException.
+ */
+ public final void quietlyHelpJoin() {
+ if (status >= 0) {
+ ForkJoinWorkerThread w =
+ (ForkJoinWorkerThread)(Thread.currentThread());
+ if (!w.unpushTask(this) || !tryQuietlyInvoke())
+ busyJoin(w);
+ }
+ }
+
+ /**
+ * Joins this task, without returning its result or throwing an
+ * exception. This method may be useful when processing
+ * collections of tasks when some have been cancelled or otherwise
+ * known to have aborted.
+ */
+ public final void quietlyJoin() {
+ if (status >= 0) {
+ ForkJoinWorkerThread w = getWorker();
+ if (w == null || !w.unpushTask(this) || !tryQuietlyInvoke())
+ awaitDone(w, true);
+ }
+ }
+
+ /**
+ * Commences performing this task and awaits its completion if
+ * necessary, without returning its result or throwing an
+ * exception. This method may be useful when processing
+ * collections of tasks when some have been cancelled or otherwise
+ * known to have aborted.
+ */
+ public final void quietlyInvoke() {
+ if (status >= 0 && !tryQuietlyInvoke())
+ quietlyJoin();
+ }
+
+ /**
+ * Possibly executes tasks until the pool hosting the current task
+ * {@link ForkJoinPool#isQuiescent}. 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.
+ */
+ public static void helpQuiesce() {
+ ((ForkJoinWorkerThread)(Thread.currentThread())).
+ helpQuiescePool();
+ }
+
+ /**
+ * Resets the internal bookkeeping state of this task, allowing a
+ * subsequent <code>fork</code>. 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, and are
+ * almost surely wrong. This method may be useful when executing
+ * pre-constructed trees of subtasks in loops.
+ */
+ public void reinitialize() {
+ if ((status & COMPLETION_MASK) == EXCEPTIONAL)
+ exceptionMap.remove(this);
+ status = 0;
+ }
+
+ /**
+ * Returns the pool hosting the current task execution, or null
+ * if this task is executing outside of any pool.
+ * @return the pool, or null if none.
+ */
+ public static ForkJoinPool getPool() {
+ Thread t = Thread.currentThread();
+ return ((t instanceof ForkJoinWorkerThread)?
+ ((ForkJoinWorkerThread)t).pool : null);
+ }
+
+ /**
+ * Tries to unschedule this task for execution. This method will
+ * typically 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. This method may be invoked only from within
+ * ForkJoinTask computations. Attempts to invoke in other contexts
+ * result in exceptions or errors possibly including ClassCastException.
+ * @return true if unforked
+ */
+ public boolean tryUnfork() {
+ return ((ForkJoinWorkerThread)(Thread.currentThread())).unpushTask(this);
+ }
+
+ /**
+ * 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.
+ * @return the number of tasks
+ */
+ public static int getQueuedTaskCount() {
+ return ((ForkJoinWorkerThread)(Thread.currentThread())).
+ getQueueSize();
+ }
+
+ /**
+ * Returns a 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. 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.
+ * @return the surplus number of tasks, which may be negative
+ */
+ public static int getSurplusQueuedTaskCount() {
+ return ((ForkJoinWorkerThread)(Thread.currentThread()))
+ .getEstimatedSurplusTaskCount();
+ }
+
+ // Extension methods
+
+ /**
+ * Returns the result that would be returned by <code>join</code>,
+ * 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.
+ *
+ * @return the result, or null if not completed.
+ */
+ 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.
+ *
+ * @param value the value
+ */
+ protected abstract void setRawResult(V value);
+
+ /**
+ * Immediately performs the base action of this task. This method
+ * is designed to support extensions, and should not in general be
+ * called otherwise. The return value controls whether this task
+ * is considered to be done normally. It may return false in
+ * asynchronous actions that require explicit invocations of
+ * <code>complete</code> to become joinable. It may throw exceptions
+ * to indicate abnormal exit.
+ * @return true if completed normally
+ * @throws Error or RuntimeException if encountered during computation
+ */
+ protected abstract boolean exec();
+
+ /**
+ * Returns, but does not unschedule or execute, the task most
+ * recently forked by the current thread but not yet executed, if
+ * one is available. There is no guarantee that this task will
+ * actually be polled or executed next.
+ * This method is designed primarily to support extensions,
+ * and is unlikely to be useful otherwise.
+ * This method may be invoked only from within
+ * ForkJoinTask computations. Attempts to invoke in other contexts
+ * result in exceptions or errors possibly including ClassCastException.
+ *
+ * @return the next task, or null if none are available
+ */
+ protected static ForkJoinTask<?> peekNextLocalTask() {
+ return ((ForkJoinWorkerThread)(Thread.currentThread())).peekTask();
+ }
+
+ /**
+ * Unschedules and returns, without executing, the task most
+ * recently forked by the current thread but not yet executed.
+ * This method is designed primarily to support extensions,
+ * and is unlikely to be useful otherwise.
+ * This method may be invoked only from within
+ * ForkJoinTask computations. Attempts to invoke in other contexts
+ * result in exceptions or errors possibly including ClassCastException.
+ *
+ * @return the next task, or null if none are available
+ */
+ protected static ForkJoinTask<?> pollNextLocalTask() {
+ return ((ForkJoinWorkerThread)(Thread.currentThread())).popTask();
+ }
+
+ /**
+ * Unschedules and returns, without executing, the task most
+ * recently forked 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 <code>null</code> result does not necessarily
+ * imply quiecence of the pool this task is operating in.
+ * This method is designed primarily to support extensions,
+ * and is unlikely to be useful otherwise.
+ * This method may be invoked only from within
+ * ForkJoinTask computations. Attempts to invoke in other contexts
+ * result in exceptions or errors possibly including ClassCastException.
+ *
+ * @return a task, or null if none are available
+ */
+ protected static ForkJoinTask<?> pollTask() {
+ return ((ForkJoinWorkerThread)(Thread.currentThread())).
+ pollTask();
+ }
+
+ // Serialization support
+
+ private static final long serialVersionUID = -7721805057305804111L;
+
+ /**
+ * Save the state to a stream.
+ *
+ * @serialData the current run status and the exception thrown
+ * during execution, or null if none.
+ * @param s the stream
+ */
+ private void writeObject(java.io.ObjectOutputStream s)
+ throws java.io.IOException {
+ s.defaultWriteObject();
+ s.writeObject(getException());
+ }
+
+ /**
+ * Reconstitute the instance from a stream.
+ * @param s the stream
+ */
+ private void readObject(java.io.ObjectInputStream s)
+ throws java.io.IOException, ClassNotFoundException {
+ s.defaultReadObject();
+ status &= ~INTERNAL_SIGNAL_MASK; // clear internal signal counts
+ status |= EXTERNAL_SIGNAL; // conservatively set external signal
+ Object ex = s.readObject();
+ if (ex != null)
+ setDoneExceptionally((Throwable)ex);
+ }
+
+ // Temporary Unsafe mechanics for preliminary release
+ private static Unsafe getUnsafe() throws Throwable {
+ try {
+ return Unsafe.getUnsafe();
+ } catch (SecurityException se) {
+ try {
+ return java.security.AccessController.doPrivileged
+ (new java.security.PrivilegedExceptionAction<Unsafe>() {
+ public Unsafe run() throws Exception {
+ return getUnsafePrivileged();
+ }});
+ } catch (java.security.PrivilegedActionException e) {
+ throw e.getCause();
+ }
+ }
+ }
+
+ private static Unsafe getUnsafePrivileged()
+ throws NoSuchFieldException, IllegalAccessException {
+ Field f = Unsafe.class.getDeclaredField("theUnsafe");
+ f.setAccessible(true);
+ return (Unsafe) f.get(null);
+ }
+
+ private static long fieldOffset(String fieldName)
+ throws NoSuchFieldException {
+ return _unsafe.objectFieldOffset
+ (ForkJoinTask.class.getDeclaredField(fieldName));
+ }
+
+ static final Unsafe _unsafe;
+ static final long statusOffset;
+
+ static {
+ try {
+ _unsafe = getUnsafe();
+ statusOffset = fieldOffset("status");
+ } catch (Throwable e) {
+ throw new RuntimeException("Could not initialize intrinsics", e);
+ }
+ }
+
+}
diff --git a/src/actors/scala/actors/forkjoin/ForkJoinWorkerThread.java b/src/actors/scala/actors/forkjoin/ForkJoinWorkerThread.java
new file mode 100644
index 0000000000..ad22a94afa
--- /dev/null
+++ b/src/actors/scala/actors/forkjoin/ForkJoinWorkerThread.java
@@ -0,0 +1,736 @@
+/*
+ * Written by Doug Lea with assistance from members of JCP JSR-166
+ * Expert Group and released to the public domain, as explained at
+ * http://creativecommons.org/licenses/publicdomain
+ */
+
+package scala.actors.forkjoin;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
+import java.util.concurrent.locks.*;
+import sun.misc.Unsafe;
+import java.lang.reflect.*;
+
+/**
+ * A thread managed by a {@link ForkJoinPool}. This class is
+ * subclassable solely for the sake of adding functionality -- there
+ * are no overridable methods dealing with scheduling or
+ * execution. However, you can override initialization and termination
+ * cleanup methods surrounding the main task processing loop. If you
+ * do create such a subclass, you will also need to supply a custom
+ * ForkJoinWorkerThreadFactory to use it in a ForkJoinPool.
+ *
+ */
+public class ForkJoinWorkerThread extends Thread {
+ /*
+ * Algorithm overview:
+ *
+ * 1. Work-Stealing: Work-stealing queues are special forms of
+ * Deques that support only three of the four possible
+ * end-operations -- push, pop, and deq (aka steal), and only do
+ * so under the constraints that push and pop are called only from
+ * the owning thread, while deq may be called from other threads.
+ * (If you are unfamiliar with them, you probably want to read
+ * Herlihy and Shavit's book "The Art of Multiprocessor
+ * programming", chapter 16 describing these in more detail before
+ * proceeding.) The main work-stealing queue design is roughly
+ * similar to "Dynamic Circular Work-Stealing Deque" by David
+ * Chase and Yossi Lev, SPAA 2005
+ * (http://research.sun.com/scalable/pubs/index.html). The main
+ * difference ultimately stems from gc requirements that we null
+ * out taken slots as soon as we can, to maintain as small a
+ * footprint as possible even in programs generating huge numbers
+ * of tasks. To accomplish this, we shift the CAS arbitrating pop
+ * vs deq (steal) from being on the indices ("base" and "sp") to
+ * the slots themselves (mainly via method "casSlotNull()"). So,
+ * both a successful pop and deq mainly entail CAS'ing a nonnull
+ * slot to null. Because we rely on CASes of references, we do
+ * not need tag bits on base or sp. They are simple ints as used
+ * in any circular array-based queue (see for example ArrayDeque).
+ * Updates to the indices must still be ordered in a way that
+ * guarantees that (sp - base) > 0 means the queue is empty, but
+ * otherwise may err on the side of possibly making the queue
+ * appear nonempty when a push, pop, or deq have not fully
+ * committed. Note that this means that the deq operation,
+ * considered individually, is not wait-free. One thief cannot
+ * successfully continue until another in-progress one (or, if
+ * previously empty, a push) completes. However, in the
+ * aggregate, we ensure at least probablistic non-blockingness. If
+ * an attempted steal fails, a thief always chooses a different
+ * random victim target to try next. So, in order for one thief to
+ * progress, it suffices for any in-progress deq or new push on
+ * any empty queue to complete. One reason this works well here is
+ * that apparently-nonempty often means soon-to-be-stealable,
+ * which gives threads a chance to activate if necessary before
+ * stealing (see below).
+ *
+ * Efficient implementation of this approach currently relies on
+ * an uncomfortable amount of "Unsafe" mechanics. To maintain
+ * correct orderings, reads and writes of variable base require
+ * volatile ordering. Variable sp does not require volatile write
+ * but needs cheaper store-ordering on writes. Because they are
+ * protected by volatile base reads, reads of the queue array and
+ * its slots do not need volatile load semantics, but writes (in
+ * push) require store order and CASes (in pop and deq) require
+ * (volatile) CAS semantics. Since these combinations aren't
+ * supported using ordinary volatiles, the only way to accomplish
+ * these effciently is to use direct Unsafe calls. (Using external
+ * AtomicIntegers and AtomicReferenceArrays for the indices and
+ * array is significantly slower because of memory locality and
+ * indirection effects.) Further, performance on most platforms is
+ * very sensitive to placement and sizing of the (resizable) queue
+ * array. Even though these queues don't usually become all that
+ * big, the initial size must be large enough to counteract cache
+ * contention effects across multiple queues (especially in the
+ * presence of GC cardmarking). Also, to improve thread-locality,
+ * queues are currently initialized immediately after the thread
+ * gets the initial signal to start processing tasks. However,
+ * all queue-related methods except pushTask are written in a way
+ * that allows them to instead be lazily allocated and/or disposed
+ * of when empty. All together, these low-level implementation
+ * choices produce as much as a factor of 4 performance
+ * improvement compared to naive implementations, and enable the
+ * processing of billions of tasks per second, sometimes at the
+ * expense of ugliness.
+ *
+ * 2. Run control: The primary run control is based on a global
+ * counter (activeCount) held by the pool. It uses an algorithm
+ * similar to that in Herlihy and Shavit section 17.6 to cause
+ * threads to eventually block when all threads declare they are
+ * inactive. (See variable "scans".) For this to work, threads
+ * must be declared active when executing tasks, and before
+ * stealing a task. They must be inactive before blocking on the
+ * Pool Barrier (awaiting a new submission or other Pool
+ * event). In between, there is some free play which we take
+ * advantage of to avoid contention and rapid flickering of the
+ * global activeCount: If inactive, we activate only if a victim
+ * queue appears to be nonempty (see above). Similarly, a thread
+ * tries to inactivate only after a full scan of other threads.
+ * The net effect is that contention on activeCount is rarely a
+ * measurable performance issue. (There are also a few other cases
+ * where we scan for work rather than retry/block upon
+ * contention.)
+ *
+ * 3. Selection control. We maintain policy of always choosing to
+ * run local tasks rather than stealing, and always trying to
+ * steal tasks before trying to run a new submission. All steals
+ * are currently performed in randomly-chosen deq-order. It may be
+ * worthwhile to bias these with locality / anti-locality
+ * information, but doing this well probably requires more
+ * lower-level information from JVMs than currently provided.
+ */
+
+ /**
+ * Capacity of work-stealing queue array upon initialization.
+ * Must be a power of two. Initial size must be at least 2, but is
+ * padded to minimize cache effects.
+ */
+ private static final int INITIAL_QUEUE_CAPACITY = 1 << 13;
+
+ /**
+ * Maximum work-stealing queue array size. Must be less than or
+ * equal to 1 << 28 to ensure lack of index wraparound. (This
+ * is less than usual bounds, because we need leftshift by 3
+ * to be in int range).
+ */
+ private static final int MAXIMUM_QUEUE_CAPACITY = 1 << 28;
+
+ /**
+ * The pool this thread works in. Accessed directly by ForkJoinTask
+ */
+ final ForkJoinPool pool;
+
+ /**
+ * The work-stealing queue array. Size must be a power of two.
+ * Initialized when thread starts, to improve memory locality.
+ */
+ private ForkJoinTask<?>[] queue;
+
+ /**
+ * Index (mod queue.length) of next queue slot to push to or pop
+ * from. It is written only by owner thread, via ordered store.
+ * Both sp and base are allowed to wrap around on overflow, but
+ * (sp - base) still estimates size.
+ */
+ private volatile int sp;
+
+ /**
+ * Index (mod queue.length) of least valid queue slot, which is
+ * always the next position to steal from if nonempty.
+ */
+ private volatile int base;
+
+ /**
+ * Activity status. When true, this worker is considered active.
+ * Must be false upon construction. It must be true when executing
+ * tasks, and BEFORE stealing a task. It must be false before
+ * calling pool.sync
+ */
+ private boolean active;
+
+ /**
+ * Run state of this worker. Supports simple versions of the usual
+ * shutdown/shutdownNow control.
+ */
+ private volatile int runState;
+
+ /**
+ * Seed for random number generator for choosing steal victims.
+ * Uses Marsaglia xorshift. Must be nonzero upon initialization.
+ */
+ private int seed;
+
+ /**
+ * Number of steals, transferred to pool when idle
+ */
+ private int stealCount;
+
+ /**
+ * Index of this worker in pool array. Set once by pool before
+ * running, and accessed directly by pool during cleanup etc
+ */
+ int poolIndex;
+
+ /**
+ * The last barrier event waited for. Accessed in pool callback
+ * methods, but only by current thread.
+ */
+ long lastEventCount;
+
+ /**
+ * Creates a ForkJoinWorkerThread operating in the given pool.
+ * @param pool the pool this thread works in
+ * @throws NullPointerException if pool is null
+ */
+ protected ForkJoinWorkerThread(ForkJoinPool pool) {
+ if (pool == null) throw new NullPointerException();
+ this.pool = pool;
+ // Note: poolIndex is set by pool during construction
+ // Remaining initialization is deferred to onStart
+ }
+
+ // Public access methods
+
+ /**
+ * Returns the pool hosting this thread
+ * @return the pool
+ */
+ public ForkJoinPool getPool() {
+ return pool;
+ }
+
+ /**
+ * Returns the index number of this thread in its pool. The
+ * returned value ranges from zero to the maximum number of
+ * threads (minus one) that have ever been created in the pool.
+ * This method may be useful for applications that track status or
+ * collect results per-worker rather than per-task.
+ * @return the index number.
+ */
+ public int getPoolIndex() {
+ return poolIndex;
+ }
+
+
+ // Runstate management
+
+ // Runstate values. Order matters
+ private static final int RUNNING = 0;
+ private static final int SHUTDOWN = 1;
+ private static final int TERMINATING = 2;
+ private static final int TERMINATED = 3;
+
+ final boolean isShutdown() { return runState >= SHUTDOWN; }
+ final boolean isTerminating() { return runState >= TERMINATING; }
+ final boolean isTerminated() { return runState == TERMINATED; }
+ final boolean shutdown() { return transitionRunStateTo(SHUTDOWN); }
+ final boolean shutdownNow() { return transitionRunStateTo(TERMINATING); }
+
+ /**
+ * Transition to at least the given state. Return true if not
+ * already at least given state.
+ */
+ private boolean transitionRunStateTo(int state) {
+ for (;;) {
+ int s = runState;
+ if (s >= state)
+ return false;
+ if (_unsafe.compareAndSwapInt(this, runStateOffset, s, state))
+ return true;
+ }
+ }
+
+ /**
+ * Try to set status to active; fail on contention
+ */
+ private boolean tryActivate() {
+ if (!active) {
+ if (!pool.tryIncrementActiveCount())
+ return false;
+ active = true;
+ }
+ return true;
+ }
+
+ /**
+ * Try to set status to active; fail on contention
+ */
+ private boolean tryInactivate() {
+ if (active) {
+ if (!pool.tryDecrementActiveCount())
+ return false;
+ active = false;
+ }
+ return true;
+ }
+
+ /**
+ * Computes next value for random victim probe. Scans don't
+ * require a very high quality generator, but also not a crummy
+ * one. Marsaglia xor-shift is cheap and works well.
+ */
+ private static int xorShift(int r) {
+ r ^= r << 1;
+ r ^= r >>> 3;
+ r ^= r << 10;
+ return r;
+ }
+
+ // Lifecycle methods
+
+ /**
+ * This method is required to be public, but should never be
+ * called explicitly. It performs the main run loop to execute
+ * ForkJoinTasks.
+ */
+ public void run() {
+ Throwable exception = null;
+ try {
+ onStart();
+ pool.sync(this); // await first pool event
+ mainLoop();
+ } catch (Throwable ex) {
+ exception = ex;
+ } finally {
+ onTermination(exception);
+ }
+ }
+
+ /**
+ * Execute tasks until shut down.
+ */
+ private void mainLoop() {
+ while (!isShutdown()) {
+ ForkJoinTask<?> t = pollTask();
+ if (t != null || (t = pollSubmission()) != null)
+ t.quietlyExec();
+ else if (tryInactivate())
+ pool.sync(this);
+ }
+ }
+
+ /**
+ * Initializes internal state after construction but before
+ * processing any tasks. If you override this method, you must
+ * invoke super.onStart() at the beginning of the method.
+ * Initialization requires care: Most fields must have legal
+ * default values, to ensure that attempted accesses from other
+ * threads work correctly even before this thread starts
+ * processing tasks.
+ */
+ protected void onStart() {
+ // Allocate while starting to improve chances of thread-local
+ // isolation
+ queue = new ForkJoinTask<?>[INITIAL_QUEUE_CAPACITY];
+ // Initial value of seed need not be especially random but
+ // should differ across workers and must be nonzero
+ int p = poolIndex + 1;
+ seed = p + (p << 8) + (p << 16) + (p << 24); // spread bits
+ }
+
+ /**
+ * Perform cleanup associated with termination of this worker
+ * thread. If you override this method, you must invoke
+ * super.onTermination at the end of the overridden method.
+ *
+ * @param exception the exception causing this thread to abort due
+ * to an unrecoverable error, or null if completed normally.
+ */
+ protected void onTermination(Throwable exception) {
+ // Execute remaining local tasks unless aborting or terminating
+ while (exception == null && !pool.isTerminating() && base != sp) {
+ try {
+ ForkJoinTask<?> t = popTask();
+ if (t != null)
+ t.quietlyExec();
+ } catch(Throwable ex) {
+ exception = ex;
+ }
+ }
+ // Cancel other tasks, transition status, notify pool, and
+ // propagate exception to uncaught exception handler
+ try {
+ do;while (!tryInactivate()); // ensure inactive
+ cancelTasks();
+ runState = TERMINATED;
+ pool.workerTerminated(this);
+ } catch (Throwable ex) { // Shouldn't ever happen
+ if (exception == null) // but if so, at least rethrown
+ exception = ex;
+ } finally {
+ if (exception != null)
+ ForkJoinTask.rethrowException(exception);
+ }
+ }
+
+ // Intrinsics-based support for queue operations.
+
+ /**
+ * Add in store-order the given task at given slot of q to
+ * null. Caller must ensure q is nonnull and index is in range.
+ */
+ private static void setSlot(ForkJoinTask<?>[] q, int i,
+ ForkJoinTask<?> t){
+//TR _unsafe.putOrderedObject((Object)q, (i << qShift) + qBase, (Object)t);
+ _unsafe.putObjectVolatile((Object)q, (i << qShift) + qBase, (Object)t);
+ }
+
+ /**
+ * CAS given slot of q to null. Caller must ensure q is nonnull
+ * and index is in range.
+ */
+ private static boolean casSlotNull(ForkJoinTask<?>[] q, int i,
+ ForkJoinTask<?> t) {
+ return _unsafe.compareAndSwapObject(q, (i << qShift) + qBase, t, null);
+ }
+
+ /**
+ * Sets sp in store-order.
+ */
+ private void storeSp(int s) {
+//TR _unsafe.putOrderedInt(this, spOffset, s);
+ _unsafe.putIntVolatile(this, spOffset, s);
+ }
+
+ // Main queue methods
+
+ /**
+ * Pushes a task. Called only by current thread.
+ * @param t the task. Caller must ensure nonnull
+ */
+ final void pushTask(ForkJoinTask<?> t) {
+ ForkJoinTask<?>[] q = queue;
+ int mask = q.length - 1;
+ int s = sp;
+ setSlot(q, s & mask, t);
+ storeSp(++s);
+ if ((s -= base) == 1)
+ pool.signalWork();
+ else if (s >= mask)
+ growQueue();
+ }
+
+ /**
+ * Tries to take a task from the base of the queue, failing if
+ * either empty or contended.
+ * @return a task, or null if none or contended.
+ */
+ private ForkJoinTask<?> deqTask() {
+ ForkJoinTask<?> t;
+ ForkJoinTask<?>[] q;
+ int i;
+ int b;
+ if (sp != (b = base) &&
+ (q = queue) != null && // must read q after b
+ (t = q[i = (q.length - 1) & b]) != null &&
+ casSlotNull(q, i, t)) {
+ base = b + 1;
+ return t;
+ }
+ return null;
+ }
+
+ /**
+ * Returns a popped task, or null if empty. Ensures active status
+ * if nonnull. Called only by current thread.
+ */
+ final ForkJoinTask<?> popTask() {
+ int s = sp;
+ while (s != base) {
+ if (tryActivate()) {
+ ForkJoinTask<?>[] q = queue;
+ int mask = q.length - 1;
+ int i = (s - 1) & mask;
+ ForkJoinTask<?> t = q[i];
+ if (t == null || !casSlotNull(q, i, t))
+ break;
+ storeSp(s - 1);
+ return t;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Specialized version of popTask to pop only if
+ * topmost element is the given task. Called only
+ * by current thread while active.
+ * @param t the task. Caller must ensure nonnull
+ */
+ final boolean unpushTask(ForkJoinTask<?> t) {
+ ForkJoinTask<?>[] q = queue;
+ int mask = q.length - 1;
+ int s = sp - 1;
+ if (casSlotNull(q, s & mask, t)) {
+ storeSp(s);
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * Returns next task to pop.
+ */
+ final ForkJoinTask<?> peekTask() {
+ ForkJoinTask<?>[] q = queue;
+ return q == null? null : q[(sp - 1) & (q.length - 1)];
+ }
+
+ /**
+ * Doubles queue array size. Transfers elements by emulating
+ * steals (deqs) from old array and placing, oldest first, into
+ * new array.
+ */
+ private void growQueue() {
+ ForkJoinTask<?>[] oldQ = queue;
+ int oldSize = oldQ.length;
+ int newSize = oldSize << 1;
+ if (newSize > MAXIMUM_QUEUE_CAPACITY)
+ throw new RejectedExecutionException("Queue capacity exceeded");
+ ForkJoinTask<?>[] newQ = queue = new ForkJoinTask<?>[newSize];
+
+ int b = base;
+ int bf = b + oldSize;
+ int oldMask = oldSize - 1;
+ int newMask = newSize - 1;
+ do {
+ int oldIndex = b & oldMask;
+ ForkJoinTask<?> t = oldQ[oldIndex];
+ if (t != null && !casSlotNull(oldQ, oldIndex, t))
+ t = null;
+ setSlot(newQ, b & newMask, t);
+ } while (++b != bf);
+ pool.signalWork();
+ }
+
+ /**
+ * Tries to steal a task from another worker. Starts at a random
+ * index of workers array, and probes workers until finding one
+ * with non-empty queue or finding that all are empty. It
+ * randomly selects the first n probes. If these are empty, it
+ * resorts to a full circular traversal, which is necessary to
+ * accurately set active status by caller. Also restarts if pool
+ * events occurred since last scan, which forces refresh of
+ * workers array, in case barrier was associated with resize.
+ *
+ * This method must be both fast and quiet -- usually avoiding
+ * memory accesses that could disrupt cache sharing etc other than
+ * those needed to check for and take tasks. This accounts for,
+ * among other things, updating random seed in place without
+ * storing it until exit.
+ *
+ * @return a task, or null if none found
+ */
+ private ForkJoinTask<?> scan() {
+ ForkJoinTask<?> t = null;
+ int r = seed; // extract once to keep scan quiet
+ ForkJoinWorkerThread[] ws; // refreshed on outer loop
+ int mask; // must be power 2 minus 1 and > 0
+ outer:do {
+ if ((ws = pool.workers) != null && (mask = ws.length - 1) > 0) {
+ int idx = r;
+ int probes = ~mask; // use random index while negative
+ for (;;) {
+ r = xorShift(r); // update random seed
+ ForkJoinWorkerThread v = ws[mask & idx];
+ if (v == null || v.sp == v.base) {
+ if (probes <= mask)
+ idx = (probes++ < 0)? r : (idx + 1);
+ else
+ break;
+ }
+ else if (!tryActivate() || (t = v.deqTask()) == null)
+ continue outer; // restart on contention
+ else
+ break outer;
+ }
+ }
+ } while (pool.hasNewSyncEvent(this)); // retry on pool events
+ seed = r;
+ return t;
+ }
+
+ /**
+ * Pops or steals a task
+ * @return a task, if available
+ */
+ final ForkJoinTask<?> pollTask() {
+ ForkJoinTask<?> t = popTask();
+ if (t == null && (t = scan()) != null)
+ ++stealCount;
+ return t;
+ }
+
+ /**
+ * Returns a pool submission, if one exists, activating first.
+ * @return a submission, if available
+ */
+ private ForkJoinTask<?> pollSubmission() {
+ ForkJoinPool p = pool;
+ while (p.hasQueuedSubmissions()) {
+ ForkJoinTask<?> t;
+ if (tryActivate() && (t = p.pollSubmission()) != null)
+ return t;
+ }
+ return null;
+ }
+
+ // Methods accessed only by Pool
+
+ /**
+ * Removes and cancels all tasks in queue. Can be called from any
+ * thread.
+ */
+ final void cancelTasks() {
+ ForkJoinTask<?> t;
+ while (base != sp && (t = deqTask()) != null)
+ t.cancelIgnoringExceptions();
+ }
+
+ /**
+ * Get and clear steal count for accumulation by pool. Called
+ * only when known to be idle (in pool.sync and termination).
+ */
+ final int getAndClearStealCount() {
+ int sc = stealCount;
+ stealCount = 0;
+ return sc;
+ }
+
+ /**
+ * Returns true if at least one worker in the given array appears
+ * to have at least one queued task.
+ * @param ws array of workers
+ */
+ static boolean hasQueuedTasks(ForkJoinWorkerThread[] ws) {
+ if (ws != null) {
+ int len = ws.length;
+ for (int j = 0; j < 2; ++j) { // need two passes for clean sweep
+ for (int i = 0; i < len; ++i) {
+ ForkJoinWorkerThread w = ws[i];
+ if (w != null && w.sp != w.base)
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
+ // Support methods for ForkJoinTask
+
+ /**
+ * Returns an estimate of the number of tasks in the queue.
+ */
+ final int getQueueSize() {
+ int n = sp - base;
+ return n < 0? 0 : n; // suppress momentarily negative values
+ }
+
+ /**
+ * Returns an estimate of the number of tasks, offset by a
+ * function of number of idle workers.
+ */
+ final int getEstimatedSurplusTaskCount() {
+ // The halving approximates weighting idle vs non-idle workers
+ return (sp - base) - (pool.getIdleThreadCount() >>> 1);
+ }
+
+ /**
+ * Scan, returning early if joinMe done
+ */
+ final ForkJoinTask<?> scanWhileJoining(ForkJoinTask<?> joinMe) {
+ ForkJoinTask<?> t = pollTask();
+ if (t != null && joinMe.status < 0 && sp == base) {
+ pushTask(t); // unsteal if done and this task would be stealable
+ t = null;
+ }
+ return t;
+ }
+
+ /**
+ * Runs tasks until pool isQuiescent
+ */
+ final void helpQuiescePool() {
+ for (;;) {
+ ForkJoinTask<?> t = pollTask();
+ if (t != null)
+ t.quietlyExec();
+ else if (tryInactivate() && pool.isQuiescent())
+ break;
+ }
+ do;while (!tryActivate()); // re-activate on exit
+ }
+
+ // Temporary Unsafe mechanics for preliminary release
+ private static Unsafe getUnsafe() throws Throwable {
+ try {
+ return Unsafe.getUnsafe();
+ } catch (SecurityException se) {
+ try {
+ return java.security.AccessController.doPrivileged
+ (new java.security.PrivilegedExceptionAction<Unsafe>() {
+ public Unsafe run() throws Exception {
+ return getUnsafePrivileged();
+ }});
+ } catch (java.security.PrivilegedActionException e) {
+ throw e.getCause();
+ }
+ }
+ }
+
+ private static Unsafe getUnsafePrivileged()
+ throws NoSuchFieldException, IllegalAccessException {
+ Field f = Unsafe.class.getDeclaredField("theUnsafe");
+ f.setAccessible(true);
+ return (Unsafe) f.get(null);
+ }
+
+ private static long fieldOffset(String fieldName)
+ throws NoSuchFieldException {
+ return _unsafe.objectFieldOffset
+ (ForkJoinWorkerThread.class.getDeclaredField(fieldName));
+ }
+
+ static final Unsafe _unsafe;
+ static final long baseOffset;
+ static final long spOffset;
+ static final long runStateOffset;
+ static final long qBase;
+ static final int qShift;
+ static {
+ try {
+ _unsafe = getUnsafe();
+ baseOffset = fieldOffset("base");
+ spOffset = fieldOffset("sp");
+ runStateOffset = fieldOffset("runState");
+ qBase = _unsafe.arrayBaseOffset(ForkJoinTask[].class);
+ int s = _unsafe.arrayIndexScale(ForkJoinTask[].class);
+ if ((s & (s-1)) != 0)
+ throw new Error("data type scale not a power of two");
+ qShift = 31 - Integer.numberOfLeadingZeros(s);
+ } catch (Throwable e) {
+ throw new RuntimeException("Could not initialize intrinsics", e);
+ }
+ }
+}
diff --git a/src/actors/scala/actors/forkjoin/LinkedTransferQueue.java b/src/actors/scala/actors/forkjoin/LinkedTransferQueue.java
new file mode 100644
index 0000000000..d4a9760dfd
--- /dev/null
+++ b/src/actors/scala/actors/forkjoin/LinkedTransferQueue.java
@@ -0,0 +1,840 @@
+/*
+ * Written by Doug Lea with assistance from members of JCP JSR-166
+ * Expert Group and released to the public domain, as explained at
+ * http://creativecommons.org/licenses/publicdomain
+ */
+
+package scala.actors.forkjoin;
+import java.util.concurrent.*;
+import java.util.concurrent.locks.*;
+import java.util.concurrent.atomic.*;
+import java.util.*;
+import java.io.*;
+import sun.misc.Unsafe;
+import java.lang.reflect.*;
+
+/**
+ * An unbounded {@linkplain TransferQueue} based on linked nodes.
+ * This queue orders elements FIFO (first-in-first-out) with respect
+ * to any given producer. The <em>head</em> of the queue is that
+ * element that has been on the queue the longest time for some
+ * producer. The <em>tail</em> of the queue is that element that has
+ * been on the queue the shortest time for some producer.
+ *
+ * <p>Beware that, unlike in most collections, the {@code size}
+ * method is <em>NOT</em> a constant-time operation. Because of the
+ * asynchronous nature of these queues, determining the current number
+ * of elements requires a traversal of the elements.
+ *
+ * <p>This class and its iterator implement all of the
+ * <em>optional</em> methods of the {@link Collection} and {@link
+ * Iterator} interfaces.
+ *
+ * <p>Memory consistency effects: As with other concurrent
+ * collections, actions in a thread prior to placing an object into a
+ * {@code LinkedTransferQueue}
+ * <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a>
+ * actions subsequent to the access or removal of that element from
+ * the {@code LinkedTransferQueue} in another thread.
+ *
+ * <p>This class is a member of the
+ * <a href="{@docRoot}/../technotes/guides/collections/index.html">
+ * Java Collections Framework</a>.
+ *
+ * @since 1.7
+ * @author Doug Lea
+ * @param <E> the type of elements held in this collection
+ *
+ */
+public class LinkedTransferQueue<E> extends AbstractQueue<E>
+ implements TransferQueue<E>, java.io.Serializable {
+ private static final long serialVersionUID = -3223113410248163686L;
+
+ /*
+ * This class extends the approach used in FIFO-mode
+ * SynchronousQueues. See the internal documentation, as well as
+ * the PPoPP 2006 paper "Scalable Synchronous Queues" by Scherer,
+ * Lea & Scott
+ * (http://www.cs.rice.edu/~wns1/papers/2006-PPoPP-SQ.pdf)
+ *
+ * The main extension is to provide different Wait modes for the
+ * main "xfer" method that puts or takes items. These don't
+ * impact the basic dual-queue logic, but instead control whether
+ * or how threads block upon insertion of request or data nodes
+ * into the dual queue. It also uses slightly different
+ * conventions for tracking whether nodes are off-list or
+ * cancelled.
+ */
+
+ // Wait modes for xfer method
+ static final int NOWAIT = 0;
+ static final int TIMEOUT = 1;
+ static final int WAIT = 2;
+
+ /** The number of CPUs, for spin control */
+ static final int NCPUS = Runtime.getRuntime().availableProcessors();
+
+ /**
+ * The number of times to spin before blocking in timed waits.
+ * The value is empirically derived -- it works well across a
+ * variety of processors and OSes. Empirically, the best value
+ * seems not to vary with number of CPUs (beyond 2) so is just
+ * a constant.
+ */
+ static final int maxTimedSpins = (NCPUS < 2)? 0 : 32;
+
+ /**
+ * The number of times to spin before blocking in untimed waits.
+ * This is greater than timed value because untimed waits spin
+ * faster since they don't need to check times on each spin.
+ */
+ static final int maxUntimedSpins = maxTimedSpins * 16;
+
+ /**
+ * The number of nanoseconds for which it is faster to spin
+ * rather than to use timed park. A rough estimate suffices.
+ */
+ static final long spinForTimeoutThreshold = 1000L;
+
+ /**
+ * Node class for LinkedTransferQueue. Opportunistically
+ * subclasses from AtomicReference to represent item. Uses Object,
+ * not E, to allow setting item to "this" after use, to avoid
+ * garbage retention. Similarly, setting the next field to this is
+ * used as sentinel that node is off list.
+ */
+ static final class QNode extends AtomicReference<Object> {
+ volatile QNode next;
+ volatile Thread waiter; // to control park/unpark
+ final boolean isData;
+ QNode(Object item, boolean isData) {
+ super(item);
+ this.isData = isData;
+ }
+
+ static final AtomicReferenceFieldUpdater<QNode, QNode>
+ nextUpdater = AtomicReferenceFieldUpdater.newUpdater
+ (QNode.class, QNode.class, "next");
+
+ final boolean casNext(QNode cmp, QNode val) {
+ return nextUpdater.compareAndSet(this, cmp, val);
+ }
+
+ final void clearNext() {
+ nextUpdater.set(this,this);//TR;lazySet(this, this);
+ }
+
+ }
+
+ /**
+ * Padded version of AtomicReference used for head, tail and
+ * cleanMe, to alleviate contention across threads CASing one vs
+ * the other.
+ */
+ static final class PaddedAtomicReference<T> extends AtomicReference<T> {
+ // enough padding for 64bytes with 4byte refs
+ Object p0, p1, p2, p3, p4, p5, p6, p7, p8, p9, pa, pb, pc, pd, pe;
+ PaddedAtomicReference(T r) { super(r); }
+ }
+
+
+ /** head of the queue */
+ private transient final PaddedAtomicReference<QNode> head;
+ /** tail of the queue */
+ private transient final PaddedAtomicReference<QNode> tail;
+
+ /**
+ * Reference to a cancelled node that might not yet have been
+ * unlinked from queue because it was the last inserted node
+ * when it cancelled.
+ */
+ private transient final PaddedAtomicReference<QNode> cleanMe;
+
+ /**
+ * Tries to cas nh as new head; if successful, unlink
+ * old head's next node to avoid garbage retention.
+ */
+ private boolean advanceHead(QNode h, QNode nh) {
+ if (h == head.get() && head.compareAndSet(h, nh)) {
+ h.clearNext(); // forget old next
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * Puts or takes an item. Used for most queue operations (except
+ * poll() and tryTransfer()). See the similar code in
+ * SynchronousQueue for detailed explanation.
+ *
+ * @param e the item or if null, signifies that this is a take
+ * @param mode the wait mode: NOWAIT, TIMEOUT, WAIT
+ * @param nanos timeout in nanosecs, used only if mode is TIMEOUT
+ * @return an item, or null on failure
+ */
+ private Object xfer(Object e, int mode, long nanos) {
+ boolean isData = (e != null);
+ QNode s = null;
+ final PaddedAtomicReference<QNode> head = this.head;
+ final PaddedAtomicReference<QNode> tail = this.tail;
+
+ for (;;) {
+ QNode t = tail.get();
+ QNode h = head.get();
+
+ if (t != null && (t == h || t.isData == isData)) {
+ if (s == null)
+ s = new QNode(e, isData);
+ QNode last = t.next;
+ if (last != null) {
+ if (t == tail.get())
+ tail.compareAndSet(t, last);
+ }
+ else if (t.casNext(null, s)) {
+ tail.compareAndSet(t, s);
+ return awaitFulfill(t, s, e, mode, nanos);
+ }
+ }
+
+ else if (h != null) {
+ QNode first = h.next;
+ if (t == tail.get() && first != null &&
+ advanceHead(h, first)) {
+ Object x = first.get();
+ if (x != first && first.compareAndSet(x, e)) {
+ LockSupport.unpark(first.waiter);
+ return isData? e : x;
+ }
+ }
+ }
+ }
+ }
+
+
+ /**
+ * Version of xfer for poll() and tryTransfer, which
+ * simplifies control paths both here and in xfer.
+ */
+ private Object fulfill(Object e) {
+ boolean isData = (e != null);
+ final PaddedAtomicReference<QNode> head = this.head;
+ final PaddedAtomicReference<QNode> tail = this.tail;
+
+ for (;;) {
+ QNode t = tail.get();
+ QNode h = head.get();
+
+ if (t != null && (t == h || t.isData == isData)) {
+ QNode last = t.next;
+ if (t == tail.get()) {
+ if (last != null)
+ tail.compareAndSet(t, last);
+ else
+ return null;
+ }
+ }
+ else if (h != null) {
+ QNode first = h.next;
+ if (t == tail.get() &&
+ first != null &&
+ advanceHead(h, first)) {
+ Object x = first.get();
+ if (x != first && first.compareAndSet(x, e)) {
+ LockSupport.unpark(first.waiter);
+ return isData? e : x;
+ }
+ }
+ }
+ }
+ }
+
+ /**
+ * Spins/blocks until node s is fulfilled or caller gives up,
+ * depending on wait mode.
+ *
+ * @param pred the predecessor of waiting node
+ * @param s the waiting node
+ * @param e the comparison value for checking match
+ * @param mode mode
+ * @param nanos timeout value
+ * @return matched item, or s if cancelled
+ */
+ private Object awaitFulfill(QNode pred, QNode s, Object e,
+ int mode, long nanos) {
+ if (mode == NOWAIT)
+ return null;
+
+ long lastTime = (mode == TIMEOUT)? System.nanoTime() : 0;
+ Thread w = Thread.currentThread();
+ int spins = -1; // set to desired spin count below
+ for (;;) {
+ if (w.isInterrupted())
+ s.compareAndSet(e, s);
+ Object x = s.get();
+ if (x != e) { // Node was matched or cancelled
+ advanceHead(pred, s); // unlink if head
+ if (x == s) { // was cancelled
+ clean(pred, s);
+ return null;
+ }
+ else if (x != null) {
+ s.set(s); // avoid garbage retention
+ return x;
+ }
+ else
+ return e;
+ }
+ if (mode == TIMEOUT) {
+ long now = System.nanoTime();
+ nanos -= now - lastTime;
+ lastTime = now;
+ if (nanos <= 0) {
+ s.compareAndSet(e, s); // try to cancel
+ continue;
+ }
+ }
+ if (spins < 0) {
+ QNode h = head.get(); // only spin if at head
+ spins = ((h != null && h.next == s) ?
+ (mode == TIMEOUT?
+ maxTimedSpins : maxUntimedSpins) : 0);
+ }
+ if (spins > 0)
+ --spins;
+ else if (s.waiter == null)
+ s.waiter = w;
+ else if (mode != TIMEOUT) {
+ LockSupport.park();//TR(this);
+ s.waiter = null;
+ spins = -1;
+ }
+ else if (nanos > spinForTimeoutThreshold) {
+ LockSupport.parkNanos(nanos);//(this, nanos);
+ s.waiter = null;
+ spins = -1;
+ }
+ }
+ }
+
+ /**
+ * Returns validated tail for use in cleaning methods.
+ */
+ private QNode getValidatedTail() {
+ for (;;) {
+ QNode h = head.get();
+ QNode first = h.next;
+ if (first != null && first.next == first) { // help advance
+ advanceHead(h, first);
+ continue;
+ }
+ QNode t = tail.get();
+ QNode last = t.next;
+ if (t == tail.get()) {
+ if (last != null)
+ tail.compareAndSet(t, last); // help advance
+ else
+ return t;
+ }
+ }
+ }
+
+ /**
+ * Gets rid of cancelled node s with original predecessor pred.
+ *
+ * @param pred predecessor of cancelled node
+ * @param s the cancelled node
+ */
+ private void clean(QNode pred, QNode s) {
+ Thread w = s.waiter;
+ if (w != null) { // Wake up thread
+ s.waiter = null;
+ if (w != Thread.currentThread())
+ LockSupport.unpark(w);
+ }
+
+ if (pred == null)
+ return;
+
+ /*
+ * At any given time, exactly one node on list cannot be
+ * deleted -- the last inserted node. To accommodate this, if
+ * we cannot delete s, we save its predecessor as "cleanMe",
+ * processing the previously saved version first. At least one
+ * of node s or the node previously saved can always be
+ * processed, so this always terminates.
+ */
+ while (pred.next == s) {
+ QNode oldpred = reclean(); // First, help get rid of cleanMe
+ QNode t = getValidatedTail();
+ if (s != t) { // If not tail, try to unsplice
+ QNode sn = s.next; // s.next == s means s already off list
+ if (sn == s || pred.casNext(s, sn))
+ break;
+ }
+ else if (oldpred == pred || // Already saved
+ (oldpred == null && cleanMe.compareAndSet(null, pred)))
+ break; // Postpone cleaning
+ }
+ }
+
+ /**
+ * Tries to unsplice the cancelled node held in cleanMe that was
+ * previously uncleanable because it was at tail.
+ *
+ * @return current cleanMe node (or null)
+ */
+ private QNode reclean() {
+ /*
+ * cleanMe is, or at one time was, predecessor of cancelled
+ * node s that was the tail so could not be unspliced. If s
+ * is no longer the tail, try to unsplice if necessary and
+ * make cleanMe slot available. This differs from similar
+ * code in clean() because we must check that pred still
+ * points to a cancelled node that must be unspliced -- if
+ * not, we can (must) clear cleanMe without unsplicing.
+ * This can loop only due to contention on casNext or
+ * clearing cleanMe.
+ */
+ QNode pred;
+ while ((pred = cleanMe.get()) != null) {
+ QNode t = getValidatedTail();
+ QNode s = pred.next;
+ if (s != t) {
+ QNode sn;
+ if (s == null || s == pred || s.get() != s ||
+ (sn = s.next) == s || pred.casNext(s, sn))
+ cleanMe.compareAndSet(pred, null);
+ }
+ else // s is still tail; cannot clean
+ break;
+ }
+ return pred;
+ }
+
+ /**
+ * Creates an initially empty {@code LinkedTransferQueue}.
+ */
+ public LinkedTransferQueue() {
+ QNode dummy = new QNode(null, false);
+ head = new PaddedAtomicReference<QNode>(dummy);
+ tail = new PaddedAtomicReference<QNode>(dummy);
+ cleanMe = new PaddedAtomicReference<QNode>(null);
+ }
+
+ /**
+ * Creates a {@code LinkedTransferQueue}
+ * initially containing the elements of the given collection,
+ * added in traversal order of the collection's iterator.
+ *
+ * @param c the collection of elements to initially contain
+ * @throws NullPointerException if the specified collection or any
+ * of its elements are null
+ */
+ public LinkedTransferQueue(Collection<? extends E> c) {
+ this();
+ addAll(c);
+ }
+
+ public void put(E e) throws InterruptedException {
+ if (e == null) throw new NullPointerException();
+ if (Thread.interrupted()) throw new InterruptedException();
+ xfer(e, NOWAIT, 0);
+ }
+
+ public boolean offer(E e, long timeout, TimeUnit unit)
+ throws InterruptedException {
+ if (e == null) throw new NullPointerException();
+ if (Thread.interrupted()) throw new InterruptedException();
+ xfer(e, NOWAIT, 0);
+ return true;
+ }
+
+ public boolean offer(E e) {
+ if (e == null) throw new NullPointerException();
+ xfer(e, NOWAIT, 0);
+ return true;
+ }
+
+ public boolean add(E e) {
+ if (e == null) throw new NullPointerException();
+ xfer(e, NOWAIT, 0);
+ return true;
+ }
+
+ public void transfer(E e) throws InterruptedException {
+ if (e == null) throw new NullPointerException();
+ if (xfer(e, WAIT, 0) == null) {
+ Thread.interrupted();
+ throw new InterruptedException();
+ }
+ }
+
+ public boolean tryTransfer(E e, long timeout, TimeUnit unit)
+ throws InterruptedException {
+ if (e == null) throw new NullPointerException();
+ if (xfer(e, TIMEOUT, unit.toNanos(timeout)) != null)
+ return true;
+ if (!Thread.interrupted())
+ return false;
+ throw new InterruptedException();
+ }
+
+ public boolean tryTransfer(E e) {
+ if (e == null) throw new NullPointerException();
+ return fulfill(e) != null;
+ }
+
+ public E take() throws InterruptedException {
+ Object e = xfer(null, WAIT, 0);
+ if (e != null)
+ return (E)e;
+ Thread.interrupted();
+ throw new InterruptedException();
+ }
+
+ public E poll(long timeout, TimeUnit unit) throws InterruptedException {
+ Object e = xfer(null, TIMEOUT, unit.toNanos(timeout));
+ if (e != null || !Thread.interrupted())
+ return (E)e;
+ throw new InterruptedException();
+ }
+
+ public E poll() {
+ return (E)fulfill(null);
+ }
+
+ public int drainTo(Collection<? super E> c) {
+ if (c == null)
+ throw new NullPointerException();
+ if (c == this)
+ throw new IllegalArgumentException();
+ int n = 0;
+ E e;
+ while ( (e = poll()) != null) {
+ c.add(e);
+ ++n;
+ }
+ return n;
+ }
+
+ public int drainTo(Collection<? super E> c, int maxElements) {
+ if (c == null)
+ throw new NullPointerException();
+ if (c == this)
+ throw new IllegalArgumentException();
+ int n = 0;
+ E e;
+ while (n < maxElements && (e = poll()) != null) {
+ c.add(e);
+ ++n;
+ }
+ return n;
+ }
+
+ // Traversal-based methods
+
+ /**
+ * Returns head after performing any outstanding helping steps.
+ */
+ private QNode traversalHead() {
+ for (;;) {
+ QNode t = tail.get();
+ QNode h = head.get();
+ if (h != null && t != null) {
+ QNode last = t.next;
+ QNode first = h.next;
+ if (t == tail.get()) {
+ if (last != null)
+ tail.compareAndSet(t, last);
+ else if (first != null) {
+ Object x = first.get();
+ if (x == first)
+ advanceHead(h, first);
+ else
+ return h;
+ }
+ else
+ return h;
+ }
+ }
+ reclean();
+ }
+ }
+
+
+ public Iterator<E> iterator() {
+ return new Itr();
+ }
+
+ /**
+ * Iterators. Basic strategy is to traverse list, treating
+ * non-data (i.e., request) nodes as terminating list.
+ * Once a valid data node is found, the item is cached
+ * so that the next call to next() will return it even
+ * if subsequently removed.
+ */
+ class Itr implements Iterator<E> {
+ QNode next; // node to return next
+ QNode pnext; // predecessor of next
+ QNode snext; // successor of next
+ QNode curr; // last returned node, for remove()
+ QNode pcurr; // predecessor of curr, for remove()
+ E nextItem; // Cache of next item, once commited to in next
+
+ Itr() {
+ findNext();
+ }
+
+ /**
+ * Ensures next points to next valid node, or null if none.
+ */
+ void findNext() {
+ for (;;) {
+ QNode pred = pnext;
+ QNode q = next;
+ if (pred == null || pred == q) {
+ pred = traversalHead();
+ q = pred.next;
+ }
+ if (q == null || !q.isData) {
+ next = null;
+ return;
+ }
+ Object x = q.get();
+ QNode s = q.next;
+ if (x != null && q != x && q != s) {
+ nextItem = (E)x;
+ snext = s;
+ pnext = pred;
+ next = q;
+ return;
+ }
+ pnext = q;
+ next = s;
+ }
+ }
+
+ public boolean hasNext() {
+ return next != null;
+ }
+
+ public E next() {
+ if (next == null) throw new NoSuchElementException();
+ pcurr = pnext;
+ curr = next;
+ pnext = next;
+ next = snext;
+ E x = nextItem;
+ findNext();
+ return x;
+ }
+
+ public void remove() {
+ QNode p = curr;
+ if (p == null)
+ throw new IllegalStateException();
+ Object x = p.get();
+ if (x != null && x != p && p.compareAndSet(x, p))
+ clean(pcurr, p);
+ }
+ }
+
+ public E peek() {
+ for (;;) {
+ QNode h = traversalHead();
+ QNode p = h.next;
+ if (p == null)
+ return null;
+ Object x = p.get();
+ if (p != x) {
+ if (!p.isData)
+ return null;
+ if (x != null)
+ return (E)x;
+ }
+ }
+ }
+
+ public boolean isEmpty() {
+ for (;;) {
+ QNode h = traversalHead();
+ QNode p = h.next;
+ if (p == null)
+ return true;
+ Object x = p.get();
+ if (p != x) {
+ if (!p.isData)
+ return true;
+ if (x != null)
+ return false;
+ }
+ }
+ }
+
+ public boolean hasWaitingConsumer() {
+ for (;;) {
+ QNode h = traversalHead();
+ QNode p = h.next;
+ if (p == null)
+ return false;
+ Object x = p.get();
+ if (p != x)
+ return !p.isData;
+ }
+ }
+
+ /**
+ * Returns the number of elements in this queue. If this queue
+ * contains more than {@code Integer.MAX_VALUE} elements, returns
+ * {@code Integer.MAX_VALUE}.
+ *
+ * <p>Beware that, unlike in most collections, this method is
+ * <em>NOT</em> a constant-time operation. Because of the
+ * asynchronous nature of these queues, determining the current
+ * number of elements requires an O(n) traversal.
+ *
+ * @return the number of elements in this queue
+ */
+ public int size() {
+ int count = 0;
+ QNode h = traversalHead();
+ for (QNode p = h.next; p != null && p.isData; p = p.next) {
+ Object x = p.get();
+ if (x != null && x != p) {
+ if (++count == Integer.MAX_VALUE) // saturated
+ break;
+ }
+ }
+ return count;
+ }
+
+ public int getWaitingConsumerCount() {
+ int count = 0;
+ QNode h = traversalHead();
+ for (QNode p = h.next; p != null && !p.isData; p = p.next) {
+ if (p.get() == null) {
+ if (++count == Integer.MAX_VALUE)
+ break;
+ }
+ }
+ return count;
+ }
+
+ public int remainingCapacity() {
+ return Integer.MAX_VALUE;
+ }
+
+ public boolean remove(Object o) {
+ if (o == null)
+ return false;
+ for (;;) {
+ QNode pred = traversalHead();
+ for (;;) {
+ QNode q = pred.next;
+ if (q == null || !q.isData)
+ return false;
+ if (q == pred) // restart
+ break;
+ Object x = q.get();
+ if (x != null && x != q && o.equals(x) &&
+ q.compareAndSet(x, q)) {
+ clean(pred, q);
+ return true;
+ }
+ pred = q;
+ }
+ }
+ }
+
+ /**
+ * Save the state to a stream (that is, serialize it).
+ *
+ * @serialData All of the elements (each an {@code E}) in
+ * the proper order, followed by a null
+ * @param s the stream
+ */
+ private void writeObject(java.io.ObjectOutputStream s)
+ throws java.io.IOException {
+ s.defaultWriteObject();
+ for (E e : this)
+ s.writeObject(e);
+ // Use trailing null as sentinel
+ s.writeObject(null);
+ }
+
+ /**
+ * Reconstitute the Queue instance from a stream (that is,
+ * deserialize it).
+ * @param s the stream
+ */
+ private void readObject(java.io.ObjectInputStream s)
+ throws java.io.IOException, ClassNotFoundException {
+ s.defaultReadObject();
+ resetHeadAndTail();
+ for (;;) {
+ E item = (E)s.readObject();
+ if (item == null)
+ break;
+ else
+ offer(item);
+ }
+ }
+
+
+ // Support for resetting head/tail while deserializing
+ private void resetHeadAndTail() {
+ QNode dummy = new QNode(null, false);
+ _unsafe.putObjectVolatile(this, headOffset,
+ new PaddedAtomicReference<QNode>(dummy));
+ _unsafe.putObjectVolatile(this, tailOffset,
+ new PaddedAtomicReference<QNode>(dummy));
+ _unsafe.putObjectVolatile(this, cleanMeOffset,
+ new PaddedAtomicReference<QNode>(null));
+ }
+
+ // Temporary Unsafe mechanics for preliminary release
+ private static Unsafe getUnsafe() throws Throwable {
+ try {
+ return Unsafe.getUnsafe();
+ } catch (SecurityException se) {
+ try {
+ return java.security.AccessController.doPrivileged
+ (new java.security.PrivilegedExceptionAction<Unsafe>() {
+ public Unsafe run() throws Exception {
+ return getUnsafePrivileged();
+ }});
+ } catch (java.security.PrivilegedActionException e) {
+ throw e.getCause();
+ }
+ }
+ }
+
+ private static Unsafe getUnsafePrivileged()
+ throws NoSuchFieldException, IllegalAccessException {
+ Field f = Unsafe.class.getDeclaredField("theUnsafe");
+ f.setAccessible(true);
+ return (Unsafe) f.get(null);
+ }
+
+ private static long fieldOffset(String fieldName)
+ throws NoSuchFieldException {
+ return _unsafe.objectFieldOffset
+ (LinkedTransferQueue.class.getDeclaredField(fieldName));
+ }
+
+ private static final Unsafe _unsafe;
+ private static final long headOffset;
+ private static final long tailOffset;
+ private static final long cleanMeOffset;
+ static {
+ try {
+ _unsafe = getUnsafe();
+ headOffset = fieldOffset("head");
+ tailOffset = fieldOffset("tail");
+ cleanMeOffset = fieldOffset("cleanMe");
+ } catch (Throwable e) {
+ throw new RuntimeException("Could not initialize intrinsics", e);
+ }
+ }
+
+}
diff --git a/src/actors/scala/actors/forkjoin/Phaser.java b/src/actors/scala/actors/forkjoin/Phaser.java
new file mode 100644
index 0000000000..f87531a6b9
--- /dev/null
+++ b/src/actors/scala/actors/forkjoin/Phaser.java
@@ -0,0 +1,960 @@
+/*
+ * Written by Doug Lea with assistance from members of JCP JSR-166
+ * Expert Group and released to the public domain, as explained at
+ * http://creativecommons.org/licenses/publicdomain
+ */
+
+package scala.actors.forkjoin;
+
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.*;
+import java.util.concurrent.locks.LockSupport;
+import sun.misc.Unsafe;
+import java.lang.reflect.*;
+
+/**
+ * A reusable synchronization barrier, similar in functionality to a
+ * {@link java.util.concurrent.CyclicBarrier CyclicBarrier} and
+ * {@link java.util.concurrent.CountDownLatch CountDownLatch}
+ * but supporting more flexible usage.
+ *
+ * <ul>
+ *
+ * <li> The number of parties synchronizing on a phaser may vary over
+ * time. A task may register to be a party at any time, and may
+ * deregister upon arriving at the barrier. As is the case with most
+ * basic synchronization constructs, registration and deregistration
+ * affect only internal counts; they do not establish any further
+ * internal bookkeeping, so tasks cannot query whether they are
+ * registered. (However, you can introduce such bookkeeping by
+ * subclassing this class.)
+ *
+ * <li> Each generation has an associated phase value, starting at
+ * zero, and advancing when all parties reach the barrier (wrapping
+ * around to zero after reaching {@code Integer.MAX_VALUE}).
+ *
+ * <li> Like a CyclicBarrier, a Phaser may be repeatedly awaited.
+ * Method {@code arriveAndAwaitAdvance} has effect analogous to
+ * {@code CyclicBarrier.await}. However, Phasers separate two
+ * aspects of coordination, that may also be invoked independently:
+ *
+ * <ul>
+ *
+ * <li> Arriving at a barrier. Methods {@code arrive} and
+ * {@code arriveAndDeregister} do not block, but return
+ * the phase value current upon entry to the method.
+ *
+ * <li> Awaiting others. Method {@code awaitAdvance} requires an
+ * argument indicating the entry phase, and returns when the
+ * barrier advances to a new phase.
+ * </ul>
+ *
+ *
+ * <li> Barrier actions, performed by the task triggering a phase
+ * advance while others may be waiting, are arranged by overriding
+ * method {@code onAdvance}, that also controls termination.
+ * Overriding this method may be used to similar but more flexible
+ * effect as providing a barrier action to a CyclicBarrier.
+ *
+ * <li> Phasers may enter a <em>termination</em> state in which all
+ * actions immediately return without updating phaser state or waiting
+ * for advance, and indicating (via a negative phase value) that
+ * execution is complete. Termination is triggered by executing the
+ * overridable {@code onAdvance} method that is invoked each time the
+ * barrier is about to be tripped. When a Phaser is controlling an
+ * action with a fixed number of iterations, it is often convenient to
+ * override this method to cause termination when the current phase
+ * number reaches a threshold. Method {@code forceTermination} is also
+ * available to abruptly release waiting threads and allow them to
+ * terminate.
+ *
+ * <li> Phasers may be tiered to reduce contention. Phasers with large
+ * numbers of parties that would otherwise experience heavy
+ * synchronization contention costs may instead be arranged in trees.
+ * This will typically greatly increase throughput even though it
+ * incurs somewhat greater per-operation overhead.
+ *
+ * <li> By default, {@code awaitAdvance} continues to wait even if
+ * the waiting thread is interrupted. And unlike the case in
+ * CyclicBarriers, exceptions encountered while tasks wait
+ * interruptibly or with timeout do not change the state of the
+ * barrier. If necessary, you can perform any associated recovery
+ * within handlers of those exceptions, often after invoking
+ * {@code forceTermination}.
+ *
+ * <li>Phasers ensure lack of starvation when used by ForkJoinTasks.
+ *
+ * </ul>
+ *
+ * <p><b>Sample usages:</b>
+ *
+ * <p>A Phaser may be used instead of a {@code CountDownLatch} to control
+ * a one-shot action serving a variable number of parties. The typical
+ * idiom is for the method setting this up to first register, then
+ * start the actions, then deregister, as in:
+ *
+ * <pre>
+ * void runTasks(List&lt;Runnable&gt; list) {
+ * final Phaser phaser = new Phaser(1); // "1" to register self
+ * for (Runnable r : list) {
+ * phaser.register();
+ * new Thread() {
+ * public void run() {
+ * phaser.arriveAndAwaitAdvance(); // await all creation
+ * r.run();
+ * phaser.arriveAndDeregister(); // signal completion
+ * }
+ * }.start();
+ * }
+ *
+ * doSomethingOnBehalfOfWorkers();
+ * phaser.arrive(); // allow threads to start
+ * int p = phaser.arriveAndDeregister(); // deregister self ...
+ * p = phaser.awaitAdvance(p); // ... and await arrival
+ * otherActions(); // do other things while tasks execute
+ * phaser.awaitAdvance(p); // await final completion
+ * }
+ * </pre>
+ *
+ * <p>One way to cause a set of threads to repeatedly perform actions
+ * for a given number of iterations is to override {@code onAdvance}:
+ *
+ * <pre>
+ * void startTasks(List&lt;Runnable&gt; list, final int iterations) {
+ * final Phaser phaser = new Phaser() {
+ * public boolean onAdvance(int phase, int registeredParties) {
+ * return phase &gt;= iterations || registeredParties == 0;
+ * }
+ * };
+ * phaser.register();
+ * for (Runnable r : list) {
+ * phaser.register();
+ * new Thread() {
+ * public void run() {
+ * do {
+ * r.run();
+ * phaser.arriveAndAwaitAdvance();
+ * } while(!phaser.isTerminated();
+ * }
+ * }.start();
+ * }
+ * phaser.arriveAndDeregister(); // deregister self, don't wait
+ * }
+ * </pre>
+ *
+ * <p> To create a set of tasks using a tree of Phasers,
+ * you could use code of the following form, assuming a
+ * Task class with a constructor accepting a Phaser that
+ * it registers for upon construction:
+ * <pre>
+ * void build(Task[] actions, int lo, int hi, Phaser b) {
+ * int step = (hi - lo) / TASKS_PER_PHASER;
+ * if (step &gt; 1) {
+ * int i = lo;
+ * while (i &lt; hi) {
+ * int r = Math.min(i + step, hi);
+ * build(actions, i, r, new Phaser(b));
+ * i = r;
+ * }
+ * }
+ * else {
+ * for (int i = lo; i &lt; hi; ++i)
+ * actions[i] = new Task(b);
+ * // assumes new Task(b) performs b.register()
+ * }
+ * }
+ * // .. initially called, for n tasks via
+ * build(new Task[n], 0, n, new Phaser());
+ * </pre>
+ *
+ * The best value of {@code TASKS_PER_PHASER} depends mainly on
+ * expected barrier synchronization rates. A value as low as four may
+ * be appropriate for extremely small per-barrier task bodies (thus
+ * high rates), or up to hundreds for extremely large ones.
+ *
+ * </pre>
+ *
+ * <p><b>Implementation notes</b>: This implementation restricts the
+ * maximum number of parties to 65535. Attempts to register additional
+ * parties result in IllegalStateExceptions. However, you can and
+ * should create tiered phasers to accommodate arbitrarily large sets
+ * of participants.
+ */
+public class Phaser {
+ /*
+ * This class implements an extension of X10 "clocks". Thanks to
+ * Vijay Saraswat for the idea, and to Vivek Sarkar for
+ * enhancements to extend functionality.
+ */
+
+ /**
+ * Barrier state representation. Conceptually, a barrier contains
+ * four values:
+ *
+ * * parties -- the number of parties to wait (16 bits)
+ * * unarrived -- the number of parties yet to hit barrier (16 bits)
+ * * phase -- the generation of the barrier (31 bits)
+ * * terminated -- set if barrier is terminated (1 bit)
+ *
+ * However, to efficiently maintain atomicity, these values are
+ * packed into a single (atomic) long. Termination uses the sign
+ * bit of 32 bit representation of phase, so phase is set to -1 on
+ * termination. Good performance relies on keeping state decoding
+ * and encoding simple, and keeping race windows short.
+ *
+ * Note: there are some cheats in arrive() that rely on unarrived
+ * count being lowest 16 bits.
+ */
+ private volatile long state;
+
+ private static final int ushortBits = 16;
+ private static final int ushortMask = 0xffff;
+ private static final int phaseMask = 0x7fffffff;
+
+ private static int unarrivedOf(long s) {
+ return (int)(s & ushortMask);
+ }
+
+ private static int partiesOf(long s) {
+ return ((int)s) >>> 16;
+ }
+
+ private static int phaseOf(long s) {
+ return (int)(s >>> 32);
+ }
+
+ private static int arrivedOf(long s) {
+ return partiesOf(s) - unarrivedOf(s);
+ }
+
+ private static long stateFor(int phase, int parties, int unarrived) {
+ return ((((long)phase) << 32) | (((long)parties) << 16) |
+ (long)unarrived);
+ }
+
+ private static long trippedStateFor(int phase, int parties) {
+ long lp = (long)parties;
+ return (((long)phase) << 32) | (lp << 16) | lp;
+ }
+
+ /**
+ * Returns message string for bad bounds exceptions
+ */
+ private static String badBounds(int parties, int unarrived) {
+ return ("Attempt to set " + unarrived +
+ " unarrived of " + parties + " parties");
+ }
+
+ /**
+ * The parent of this phaser, or null if none
+ */
+ private final Phaser parent;
+
+ /**
+ * The root of Phaser tree. Equals this if not in a tree. Used to
+ * support faster state push-down.
+ */
+ private final Phaser root;
+
+ // Wait queues
+
+ /**
+ * Heads of Treiber stacks for waiting threads. To eliminate
+ * contention while releasing some threads while adding others, we
+ * use two of them, alternating across even and odd phases.
+ */
+ private final AtomicReference<QNode> evenQ = new AtomicReference<QNode>();
+ private final AtomicReference<QNode> oddQ = new AtomicReference<QNode>();
+
+ private AtomicReference<QNode> queueFor(int phase) {
+ return (phase & 1) == 0? evenQ : oddQ;
+ }
+
+ /**
+ * Returns current state, first resolving lagged propagation from
+ * root if necessary.
+ */
+ private long getReconciledState() {
+ return parent == null? state : reconcileState();
+ }
+
+ /**
+ * Recursively resolves state.
+ */
+ private long reconcileState() {
+ Phaser p = parent;
+ long s = state;
+ if (p != null) {
+ while (unarrivedOf(s) == 0 && phaseOf(s) != phaseOf(root.state)) {
+ long parentState = p.getReconciledState();
+ int parentPhase = phaseOf(parentState);
+ int phase = phaseOf(s = state);
+ if (phase != parentPhase) {
+ long next = trippedStateFor(parentPhase, partiesOf(s));
+ if (casState(s, next)) {
+ releaseWaiters(phase);
+ s = next;
+ }
+ }
+ }
+ }
+ return s;
+ }
+
+ /**
+ * Creates a new Phaser without any initially registered parties,
+ * initial phase number 0, and no parent. Any thread using this
+ * Phaser will need to first register for it.
+ */
+ public Phaser() {
+ this(null);
+ }
+
+ /**
+ * Creates a new Phaser with the given numbers of registered
+ * unarrived parties, initial phase number 0, and no parent.
+ * @param parties the number of parties required to trip barrier.
+ * @throws IllegalArgumentException if parties less than zero
+ * or greater than the maximum number of parties supported.
+ */
+ public Phaser(int parties) {
+ this(null, parties);
+ }
+
+ /**
+ * Creates a new Phaser with the given parent, without any
+ * initially registered parties. If parent is non-null this phaser
+ * is registered with the parent and its initial phase number is
+ * the same as that of parent phaser.
+ * @param parent the parent phaser.
+ */
+ public Phaser(Phaser parent) {
+ int phase = 0;
+ this.parent = parent;
+ if (parent != null) {
+ this.root = parent.root;
+ phase = parent.register();
+ }
+ else
+ this.root = this;
+ this.state = trippedStateFor(phase, 0);
+ }
+
+ /**
+ * Creates a new Phaser with the given parent and numbers of
+ * registered unarrived parties. If parent is non-null this phaser
+ * is registered with the parent and its initial phase number is
+ * the same as that of parent phaser.
+ * @param parent the parent phaser.
+ * @param parties the number of parties required to trip barrier.
+ * @throws IllegalArgumentException if parties less than zero
+ * or greater than the maximum number of parties supported.
+ */
+ public Phaser(Phaser parent, int parties) {
+ if (parties < 0 || parties > ushortMask)
+ throw new IllegalArgumentException("Illegal number of parties");
+ int phase = 0;
+ this.parent = parent;
+ if (parent != null) {
+ this.root = parent.root;
+ phase = parent.register();
+ }
+ else
+ this.root = this;
+ this.state = trippedStateFor(phase, parties);
+ }
+
+ /**
+ * Adds a new unarrived party to this phaser.
+ * @return the current barrier phase number upon registration
+ * @throws IllegalStateException if attempting to register more
+ * than the maximum supported number of parties.
+ */
+ public int register() {
+ return doRegister(1);
+ }
+
+ /**
+ * Adds the given number of new unarrived parties to this phaser.
+ * @param parties the number of parties required to trip barrier.
+ * @return the current barrier phase number upon registration
+ * @throws IllegalStateException if attempting to register more
+ * than the maximum supported number of parties.
+ */
+ public int bulkRegister(int parties) {
+ if (parties < 0)
+ throw new IllegalArgumentException();
+ if (parties == 0)
+ return getPhase();
+ return doRegister(parties);
+ }
+
+ /**
+ * Shared code for register, bulkRegister
+ */
+ private int doRegister(int registrations) {
+ int phase;
+ for (;;) {
+ long s = getReconciledState();
+ phase = phaseOf(s);
+ int unarrived = unarrivedOf(s) + registrations;
+ int parties = partiesOf(s) + registrations;
+ if (phase < 0)
+ break;
+ if (parties > ushortMask || unarrived > ushortMask)
+ throw new IllegalStateException(badBounds(parties, unarrived));
+ if (phase == phaseOf(root.state) &&
+ casState(s, stateFor(phase, parties, unarrived)))
+ break;
+ }
+ return phase;
+ }
+
+ /**
+ * Arrives at the barrier, but does not wait for others. (You can
+ * in turn wait for others via {@link #awaitAdvance}).
+ *
+ * @return the barrier phase number upon entry to this method, or a
+ * negative value if terminated;
+ * @throws IllegalStateException if not terminated and the number
+ * of unarrived parties would become negative.
+ */
+ public int arrive() {
+ int phase;
+ for (;;) {
+ long s = state;
+ phase = phaseOf(s);
+ if (phase < 0)
+ break;
+ int parties = partiesOf(s);
+ int unarrived = unarrivedOf(s) - 1;
+ if (unarrived > 0) { // Not the last arrival
+ if (casState(s, s - 1)) // s-1 adds one arrival
+ break;
+ }
+ else if (unarrived == 0) { // the last arrival
+ Phaser par = parent;
+ if (par == null) { // directly trip
+ if (casState
+ (s,
+ trippedStateFor(onAdvance(phase, parties)? -1 :
+ ((phase + 1) & phaseMask), parties))) {
+ releaseWaiters(phase);
+ break;
+ }
+ }
+ else { // cascade to parent
+ if (casState(s, s - 1)) { // zeroes unarrived
+ par.arrive();
+ reconcileState();
+ break;
+ }
+ }
+ }
+ else if (phase != phaseOf(root.state)) // or if unreconciled
+ reconcileState();
+ else
+ throw new IllegalStateException(badBounds(parties, unarrived));
+ }
+ return phase;
+ }
+
+ /**
+ * Arrives at the barrier, and deregisters from it, without
+ * waiting for others. Deregistration reduces number of parties
+ * required to trip the barrier in future phases. If this phaser
+ * has a parent, and deregistration causes this phaser to have
+ * zero parties, this phaser is also deregistered from its parent.
+ *
+ * @return the current barrier phase number upon entry to
+ * this method, or a negative value if terminated;
+ * @throws IllegalStateException if not terminated and the number
+ * of registered or unarrived parties would become negative.
+ */
+ public int arriveAndDeregister() {
+ // similar code to arrive, but too different to merge
+ Phaser par = parent;
+ int phase;
+ for (;;) {
+ long s = state;
+ phase = phaseOf(s);
+ if (phase < 0)
+ break;
+ int parties = partiesOf(s) - 1;
+ int unarrived = unarrivedOf(s) - 1;
+ if (parties >= 0) {
+ if (unarrived > 0 || (unarrived == 0 && par != null)) {
+ if (casState
+ (s,
+ stateFor(phase, parties, unarrived))) {
+ if (unarrived == 0) {
+ par.arriveAndDeregister();
+ reconcileState();
+ }
+ break;
+ }
+ continue;
+ }
+ if (unarrived == 0) {
+ if (casState
+ (s,
+ trippedStateFor(onAdvance(phase, parties)? -1 :
+ ((phase + 1) & phaseMask), parties))) {
+ releaseWaiters(phase);
+ break;
+ }
+ continue;
+ }
+ if (par != null && phase != phaseOf(root.state)) {
+ reconcileState();
+ continue;
+ }
+ }
+ throw new IllegalStateException(badBounds(parties, unarrived));
+ }
+ return phase;
+ }
+
+ /**
+ * Arrives at the barrier and awaits others. Equivalent in effect
+ * to {@code awaitAdvance(arrive())}. If you instead need to
+ * await with interruption of timeout, and/or deregister upon
+ * arrival, you can arrange them using analogous constructions.
+ * @return the phase on entry to this method
+ * @throws IllegalStateException if not terminated and the number
+ * of unarrived parties would become negative.
+ */
+ public int arriveAndAwaitAdvance() {
+ return awaitAdvance(arrive());
+ }
+
+ /**
+ * Awaits the phase of the barrier to advance from the given
+ * value, or returns immediately if argument is negative or this
+ * barrier is terminated.
+ * @param phase the phase on entry to this method
+ * @return the phase on exit from this method
+ */
+ public int awaitAdvance(int phase) {
+ if (phase < 0)
+ return phase;
+ long s = getReconciledState();
+ int p = phaseOf(s);
+ if (p != phase)
+ return p;
+ if (unarrivedOf(s) == 0 && parent != null)
+ parent.awaitAdvance(phase);
+ // Fall here even if parent waited, to reconcile and help release
+ return untimedWait(phase);
+ }
+
+ /**
+ * Awaits the phase of the barrier to advance from the given
+ * value, or returns immediately if argument is negative or this
+ * barrier is terminated, or throws InterruptedException if
+ * interrupted while waiting.
+ * @param phase the phase on entry to this method
+ * @return the phase on exit from this method
+ * @throws InterruptedException if thread interrupted while waiting
+ */
+ public int awaitAdvanceInterruptibly(int phase)
+ throws InterruptedException {
+ if (phase < 0)
+ return phase;
+ long s = getReconciledState();
+ int p = phaseOf(s);
+ if (p != phase)
+ return p;
+ if (unarrivedOf(s) == 0 && parent != null)
+ parent.awaitAdvanceInterruptibly(phase);
+ return interruptibleWait(phase);
+ }
+
+ /**
+ * Awaits the phase of the barrier to advance from the given value
+ * or the given timeout elapses, or returns immediately if
+ * argument is negative or this barrier is terminated.
+ * @param phase the phase on entry to this method
+ * @return the phase on exit from this method
+ * @throws InterruptedException if thread interrupted while waiting
+ * @throws TimeoutException if timed out while waiting
+ */
+ public int awaitAdvanceInterruptibly(int phase, long timeout, TimeUnit unit)
+ throws InterruptedException, TimeoutException {
+ if (phase < 0)
+ return phase;
+ long s = getReconciledState();
+ int p = phaseOf(s);
+ if (p != phase)
+ return p;
+ if (unarrivedOf(s) == 0 && parent != null)
+ parent.awaitAdvanceInterruptibly(phase, timeout, unit);
+ return timedWait(phase, unit.toNanos(timeout));
+ }
+
+ /**
+ * Forces this barrier to enter termination state. Counts of
+ * arrived and registered parties are unaffected. If this phaser
+ * has a parent, it too is terminated. This method may be useful
+ * for coordinating recovery after one or more tasks encounter
+ * unexpected exceptions.
+ */
+ public void forceTermination() {
+ for (;;) {
+ long s = getReconciledState();
+ int phase = phaseOf(s);
+ int parties = partiesOf(s);
+ int unarrived = unarrivedOf(s);
+ if (phase < 0 ||
+ casState(s, stateFor(-1, parties, unarrived))) {
+ releaseWaiters(0);
+ releaseWaiters(1);
+ if (parent != null)
+ parent.forceTermination();
+ return;
+ }
+ }
+ }
+
+ /**
+ * Returns the current phase number. The maximum phase number is
+ * {@code Integer.MAX_VALUE}, after which it restarts at
+ * zero. Upon termination, the phase number is negative.
+ * @return the phase number, or a negative value if terminated
+ */
+ public final int getPhase() {
+ return phaseOf(getReconciledState());
+ }
+
+ /**
+ * Returns {@code true} if the current phase number equals the given phase.
+ * @param phase the phase
+ * @return {@code true} if the current phase number equals the given phase
+ */
+ public final boolean hasPhase(int phase) {
+ return phaseOf(getReconciledState()) == phase;
+ }
+
+ /**
+ * Returns the number of parties registered at this barrier.
+ * @return the number of parties
+ */
+ public int getRegisteredParties() {
+ return partiesOf(state);
+ }
+
+ /**
+ * Returns the number of parties that have arrived at the current
+ * phase of this barrier.
+ * @return the number of arrived parties
+ */
+ public int getArrivedParties() {
+ return arrivedOf(state);
+ }
+
+ /**
+ * Returns the number of registered parties that have not yet
+ * arrived at the current phase of this barrier.
+ * @return the number of unarrived parties
+ */
+ public int getUnarrivedParties() {
+ return unarrivedOf(state);
+ }
+
+ /**
+ * Returns the parent of this phaser, or null if none.
+ * @return the parent of this phaser, or null if none
+ */
+ public Phaser getParent() {
+ return parent;
+ }
+
+ /**
+ * Returns the root ancestor of this phaser, which is the same as
+ * this phaser if it has no parent.
+ * @return the root ancestor of this phaser
+ */
+ public Phaser getRoot() {
+ return root;
+ }
+
+ /**
+ * Returns {@code true} if this barrier has been terminated.
+ * @return {@code true} if this barrier has been terminated
+ */
+ public boolean isTerminated() {
+ return getPhase() < 0;
+ }
+
+ /**
+ * Overridable method to perform an action upon phase advance, and
+ * to control termination. This method is invoked whenever the
+ * barrier is tripped (and thus all other waiting parties are
+ * dormant). If it returns true, then, rather than advance the
+ * phase number, this barrier will be set to a final termination
+ * state, and subsequent calls to {@code isTerminated} will
+ * return true.
+ *
+ * <p> The default version returns true when the number of
+ * registered parties is zero. Normally, overrides that arrange
+ * termination for other reasons should also preserve this
+ * property.
+ *
+ * <p> You may override this method to perform an action with side
+ * effects visible to participating tasks, but it is in general
+ * only sensible to do so in designs where all parties register
+ * before any arrive, and all {@code awaitAdvance} at each phase.
+ * Otherwise, you cannot ensure lack of interference. In
+ * particular, this method may be invoked more than once per
+ * transition if other parties successfully register while the
+ * invocation of this method is in progress, thus postponing the
+ * transition until those parties also arrive, re-triggering this
+ * method.
+ *
+ * @param phase the phase number on entering the barrier
+ * @param registeredParties the current number of registered parties
+ * @return {@code true} if this barrier should terminate
+ */
+ protected boolean onAdvance(int phase, int registeredParties) {
+ return registeredParties <= 0;
+ }
+
+ /**
+ * Returns a string identifying this phaser, as well as its
+ * state. The state, in brackets, includes the String {@code
+ * "phase = "} followed by the phase number, {@code "parties = "}
+ * followed by the number of registered parties, and {@code
+ * "arrived = "} followed by the number of arrived parties.
+ *
+ * @return a string identifying this barrier, as well as its state
+ */
+ public String toString() {
+ long s = getReconciledState();
+ return super.toString() +
+ "[phase = " + phaseOf(s) +
+ " parties = " + partiesOf(s) +
+ " arrived = " + arrivedOf(s) + "]";
+ }
+
+ // methods for waiting
+
+ /**
+ * Wait nodes for Treiber stack representing wait queue
+ */
+ static final class QNode implements ForkJoinPool.ManagedBlocker {
+ final Phaser phaser;
+ final int phase;
+ final long startTime;
+ final long nanos;
+ final boolean timed;
+ final boolean interruptible;
+ volatile boolean wasInterrupted = false;
+ volatile Thread thread; // nulled to cancel wait
+ QNode next;
+ QNode(Phaser phaser, int phase, boolean interruptible,
+ boolean timed, long startTime, long nanos) {
+ this.phaser = phaser;
+ this.phase = phase;
+ this.timed = timed;
+ this.interruptible = interruptible;
+ this.startTime = startTime;
+ this.nanos = nanos;
+ thread = Thread.currentThread();
+ }
+ public boolean isReleasable() {
+ return (thread == null ||
+ phaser.getPhase() != phase ||
+ (interruptible && wasInterrupted) ||
+ (timed && (nanos - (System.nanoTime() - startTime)) <= 0));
+ }
+ public boolean block() {
+ if (Thread.interrupted()) {
+ wasInterrupted = true;
+ if (interruptible)
+ return true;
+ }
+ if (!timed)
+ LockSupport.park();//TR (this);
+ else {
+ long waitTime = nanos - (System.nanoTime() - startTime);
+ if (waitTime <= 0)
+ return true;
+ LockSupport.parkNanos(waitTime);//TR (this, waitTime);
+ }
+ return isReleasable();
+ }
+ void signal() {
+ Thread t = thread;
+ if (t != null) {
+ thread = null;
+ LockSupport.unpark(t);
+ }
+ }
+ boolean doWait() {
+ if (thread != null) {
+ try {
+ ForkJoinPool.managedBlock(this, false);
+ } catch (InterruptedException ie) {
+ }
+ }
+ return wasInterrupted;
+ }
+
+ }
+
+ /**
+ * Removes and signals waiting threads from wait queue
+ */
+ private void releaseWaiters(int phase) {
+ AtomicReference<QNode> head = queueFor(phase);
+ QNode q;
+ while ((q = head.get()) != null) {
+ if (head.compareAndSet(q, q.next))
+ q.signal();
+ }
+ }
+
+ /**
+ * Tries to enqueue given node in the appropriate wait queue
+ * @return true if successful
+ */
+ private boolean tryEnqueue(QNode node) {
+ AtomicReference<QNode> head = queueFor(node.phase);
+ return head.compareAndSet(node.next = head.get(), node);
+ }
+
+ /**
+ * Enqueues node and waits unless aborted or signalled.
+ * @return current phase
+ */
+ private int untimedWait(int phase) {
+ QNode node = null;
+ boolean queued = false;
+ boolean interrupted = false;
+ int p;
+ while ((p = getPhase()) == phase) {
+ if (Thread.interrupted())
+ interrupted = true;
+ else if (node == null)
+ node = new QNode(this, phase, false, false, 0, 0);
+ else if (!queued)
+ queued = tryEnqueue(node);
+ else
+ interrupted = node.doWait();
+ }
+ if (node != null)
+ node.thread = null;
+ releaseWaiters(phase);
+ if (interrupted)
+ Thread.currentThread().interrupt();
+ return p;
+ }
+
+ /**
+ * Interruptible version
+ * @return current phase
+ */
+ private int interruptibleWait(int phase) throws InterruptedException {
+ QNode node = null;
+ boolean queued = false;
+ boolean interrupted = false;
+ int p;
+ while ((p = getPhase()) == phase && !interrupted) {
+ if (Thread.interrupted())
+ interrupted = true;
+ else if (node == null)
+ node = new QNode(this, phase, true, false, 0, 0);
+ else if (!queued)
+ queued = tryEnqueue(node);
+ else
+ interrupted = node.doWait();
+ }
+ if (node != null)
+ node.thread = null;
+ if (p != phase || (p = getPhase()) != phase)
+ releaseWaiters(phase);
+ if (interrupted)
+ throw new InterruptedException();
+ return p;
+ }
+
+ /**
+ * Timeout version.
+ * @return current phase
+ */
+ private int timedWait(int phase, long nanos)
+ throws InterruptedException, TimeoutException {
+ long startTime = System.nanoTime();
+ QNode node = null;
+ boolean queued = false;
+ boolean interrupted = false;
+ int p;
+ while ((p = getPhase()) == phase && !interrupted) {
+ if (Thread.interrupted())
+ interrupted = true;
+ else if (nanos - (System.nanoTime() - startTime) <= 0)
+ break;
+ else if (node == null)
+ node = new QNode(this, phase, true, true, startTime, nanos);
+ else if (!queued)
+ queued = tryEnqueue(node);
+ else
+ interrupted = node.doWait();
+ }
+ if (node != null)
+ node.thread = null;
+ if (p != phase || (p = getPhase()) != phase)
+ releaseWaiters(phase);
+ if (interrupted)
+ throw new InterruptedException();
+ if (p == phase)
+ throw new TimeoutException();
+ return p;
+ }
+
+ // Temporary Unsafe mechanics for preliminary release
+ private static Unsafe getUnsafe() throws Throwable {
+ try {
+ return Unsafe.getUnsafe();
+ } catch (SecurityException se) {
+ try {
+ return java.security.AccessController.doPrivileged
+ (new java.security.PrivilegedExceptionAction<Unsafe>() {
+ public Unsafe run() throws Exception {
+ return getUnsafePrivileged();
+ }});
+ } catch (java.security.PrivilegedActionException e) {
+ throw e.getCause();
+ }
+ }
+ }
+
+ private static Unsafe getUnsafePrivileged()
+ throws NoSuchFieldException, IllegalAccessException {
+ Field f = Unsafe.class.getDeclaredField("theUnsafe");
+ f.setAccessible(true);
+ return (Unsafe) f.get(null);
+ }
+
+ private static long fieldOffset(String fieldName)
+ throws NoSuchFieldException {
+ return _unsafe.objectFieldOffset
+ (Phaser.class.getDeclaredField(fieldName));
+ }
+
+ static final Unsafe _unsafe;
+ static final long stateOffset;
+
+ static {
+ try {
+ _unsafe = getUnsafe();
+ stateOffset = fieldOffset("state");
+ } catch (Throwable e) {
+ throw new RuntimeException("Could not initialize intrinsics", e);
+ }
+ }
+
+ final boolean casState(long cmp, long val) {
+ return _unsafe.compareAndSwapLong(this, stateOffset, cmp, val);
+ }
+}
diff --git a/src/actors/scala/actors/forkjoin/RecursiveAction.java b/src/actors/scala/actors/forkjoin/RecursiveAction.java
new file mode 100644
index 0000000000..fc813ec8b6
--- /dev/null
+++ b/src/actors/scala/actors/forkjoin/RecursiveAction.java
@@ -0,0 +1,151 @@
+/*
+ * Written by Doug Lea with assistance from members of JCP JSR-166
+ * Expert Group and released to the public domain, as explained at
+ * http://creativecommons.org/licenses/publicdomain
+ */
+
+package scala.actors.forkjoin;
+
+/**
+ * Recursive resultless ForkJoinTasks. This class establishes
+ * conventions to parameterize resultless actions as <tt>Void</tt>
+ * ForkJoinTasks. Because <tt>null</tt> is the only valid value of
+ * <tt>Void</tt>, methods such as join always return <tt>null</tt>
+ * upon completion.
+ *
+ * <p><b>Sample Usages.</b> Here is a sketch of a ForkJoin sort that
+ * sorts a given <tt>long[]</tt> array:
+ *
+ * <pre>
+ * class SortTask extends RecursiveAction {
+ * final long[] array; final int lo; final int hi;
+ * SortTask(long[] array, int lo, int hi) {
+ * this.array = array; this.lo = lo; this.hi = hi;
+ * }
+ * protected void compute() {
+ * if (hi - lo &lt; THRESHOLD)
+ * sequentiallySort(array, lo, hi);
+ * else {
+ * int mid = (lo + hi) &gt;&gt;&gt; 1;
+ * invokeAll(new SortTask(array, lo, mid),
+ * new SortTask(array, mid, hi));
+ * merge(array, lo, hi);
+ * }
+ * }
+ * }
+ * </pre>
+ *
+ * You could then sort anArray by creating <tt>new SortTask(anArray, 0,
+ * anArray.length-1) </tt> and invoking it in a ForkJoinPool.
+ * As a more concrete simple example, the following task increments
+ * each element of an array:
+ * <pre>
+ * class IncrementTask extends RecursiveAction {
+ * final long[] array; final int lo; final int hi;
+ * IncrementTask(long[] array, int lo, int hi) {
+ * this.array = array; this.lo = lo; this.hi = hi;
+ * }
+ * protected void compute() {
+ * if (hi - lo &lt; THRESHOLD) {
+ * for (int i = lo; i &lt; hi; ++i)
+ * array[i]++;
+ * }
+ * else {
+ * int mid = (lo + hi) &gt;&gt;&gt; 1;
+ * invokeAll(new IncrementTask(array, lo, mid),
+ * new IncrementTask(array, mid, hi));
+ * }
+ * }
+ * }
+ * </pre>
+ *
+ *
+ * <p>The following example illustrates some refinements and idioms
+ * that may lead to better performance: RecursiveActions need not be
+ * fully recursive, so long as they maintain the basic
+ * divide-and-conquer approach. Here is a class that sums the squares
+ * of each element of a double array, by subdividing out only the
+ * right-hand-sides of repeated divisions by two, and keeping track of
+ * them with a chain of <tt>next</tt> references. It uses a dynamic
+ * threshold based on method <tt>surplus</tt>, but counterbalances
+ * potential excess partitioning by directly performing leaf actions
+ * on unstolen tasks rather than further subdividing.
+ *
+ * <pre>
+ * double sumOfSquares(ForkJoinPool pool, double[] array) {
+ * int n = array.length;
+ * int seqSize = 1 + n / (8 * pool.getParallelism());
+ * Applyer a = new Applyer(array, 0, n, seqSize, null);
+ * pool.invoke(a);
+ * return a.result;
+ * }
+ *
+ * class Applyer extends RecursiveAction {
+ * final double[] array;
+ * final int lo, hi, seqSize;
+ * double result;
+ * Applyer next; // keeps track of right-hand-side tasks
+ * Applyer(double[] array, int lo, int hi, int seqSize, Applyer next) {
+ * this.array = array; this.lo = lo; this.hi = hi;
+ * this.seqSize = seqSize; this.next = next;
+ * }
+ *
+ * double atLeaf(int l, int r) {
+ * double sum = 0;
+ * for (int i = l; i &lt; h; ++i) // perform leftmost base step
+ * sum += array[i] * array[i];
+ * return sum;
+ * }
+ *
+ * protected void compute() {
+ * int l = lo;
+ * int h = hi;
+ * Applyer right = null;
+ * while (h - l &gt; 1 &amp;&amp;
+ * ForkJoinWorkerThread.getEstimatedSurplusTaskCount() &lt;= 3) {
+ * int mid = (l + h) &gt;&gt;&gt; 1;
+ * right = new Applyer(array, mid, h, seqSize, right);
+ * right.fork();
+ * h = mid;
+ * }
+ * double sum = atLeaf(l, h);
+ * while (right != null) {
+ * if (right.tryUnfork()) // directly calculate if not stolen
+ * sum += right.atLeaf(right.lo, right.hi);
+ * else {
+ * right.helpJoin();
+ * sum += right.result;
+ * }
+ * right = right.next;
+ * }
+ * result = sum;
+ * }
+ * }
+ * </pre>
+ */
+public abstract class RecursiveAction extends ForkJoinTask<Void> {
+
+ /**
+ * The main computation performed by this task.
+ */
+ protected abstract void compute();
+
+ /**
+ * Always returns null
+ */
+ public final Void getRawResult() { return null; }
+
+ /**
+ * Requires null completion value.
+ */
+ protected final void setRawResult(Void mustBeNull) { }
+
+ /**
+ * Implements execution conventions for RecursiveActions
+ */
+ protected final boolean exec() {
+ compute();
+ return true;
+ }
+
+}
diff --git a/src/actors/scala/actors/forkjoin/RecursiveTask.java b/src/actors/scala/actors/forkjoin/RecursiveTask.java
new file mode 100644
index 0000000000..5286174fdf
--- /dev/null
+++ b/src/actors/scala/actors/forkjoin/RecursiveTask.java
@@ -0,0 +1,71 @@
+/*
+ * Written by Doug Lea with assistance from members of JCP JSR-166
+ * Expert Group and released to the public domain, as explained at
+ * http://creativecommons.org/licenses/publicdomain
+ */
+
+package scala.actors.forkjoin;
+
+/**
+ * Recursive result-bearing ForkJoinTasks.
+ * <p> For a classic example, here is a task computing Fibonacci numbers:
+ *
+ * <pre>
+ * class Fibonacci extends RecursiveTask&lt;Integer&gt; {
+ * final int n;
+ * Fibonnaci(int n) { this.n = n; }
+ * Integer compute() {
+ * if (n &lt;= 1)
+ * return n;
+ * Fibonacci f1 = new Fibonacci(n - 1);
+ * f1.fork();
+ * Fibonacci f2 = new Fibonacci(n - 2);
+ * return f2.compute() + f1.join();
+ * }
+ * }
+ * </pre>
+ *
+ * However, besides being a dumb way to compute Fibonacci functions
+ * (there is a simple fast linear algorithm that you'd use in
+ * practice), this is likely to perform poorly because the smallest
+ * subtasks are too small to be worthwhile splitting up. Instead, as
+ * is the case for nearly all fork/join applications, you'd pick some
+ * minimum granularity size (for example 10 here) for which you always
+ * sequentially solve rather than subdividing.
+ *
+ */
+public abstract class RecursiveTask<V> extends ForkJoinTask<V> {
+
+ /**
+ * Empty contructor for use by subclasses.
+ */
+ protected RecursiveTask() {
+ }
+
+ /**
+ * The result returned by compute method.
+ */
+ V result;
+
+ /**
+ * The main computation performed by this task.
+ */
+ protected abstract V compute();
+
+ public final V getRawResult() {
+ return result;
+ }
+
+ protected final void setRawResult(V value) {
+ result = value;
+ }
+
+ /**
+ * Implements execution conventions for RecursiveTask
+ */
+ protected final boolean exec() {
+ result = compute();
+ return true;
+ }
+
+}
diff --git a/src/actors/scala/actors/forkjoin/ThreadLocalRandom.java b/src/actors/scala/actors/forkjoin/ThreadLocalRandom.java
new file mode 100644
index 0000000000..1fa3bcd71e
--- /dev/null
+++ b/src/actors/scala/actors/forkjoin/ThreadLocalRandom.java
@@ -0,0 +1,186 @@
+/*
+ * Written by Doug Lea with assistance from members of JCP JSR-166
+ * Expert Group and released to the public domain, as explained at
+ * http://creativecommons.org/licenses/publicdomain
+ */
+
+package scala.actors.forkjoin;
+import java.util.*;
+
+/**
+ * A random number generator with the same properties as class {@link
+ * Random} but isolated to the current Thread. Like the global
+ * generator used by the {@link java.lang.Math} class, a
+ * ThreadLocalRandom is initialized with an internally generated seed
+ * that may not otherwise be modified. When applicable, use of
+ * ThreadLocalRandom rather than shared Random objects in concurrent
+ * programs will typically encounter much less overhead and
+ * contention. ThreadLocalRandoms are particularly appropriate when
+ * multiple tasks (for example, each a {@link ForkJoinTask}), use
+ * random numbers in parallel in thread pools.
+ *
+ * <p>Usages of this class should typically be of the form:
+ * <code>ThreadLocalRandom.current().nextX(...)</code> (where
+ * <code>X</code> is <code>Int</code>, <code>Long</code>, etc).
+ * When all usages are of this form, it is never possible to
+ * accidently share ThreadLocalRandoms across multiple threads.
+ *
+ * <p>This class also provides additional commonly used bounded random
+ * generation methods.
+ */
+public class ThreadLocalRandom extends Random {
+ // same constants as Random, but must be redeclared because private
+ private final static long multiplier = 0x5DEECE66DL;
+ private final static long addend = 0xBL;
+ private final static long mask = (1L << 48) - 1;
+
+ /**
+ * The random seed. We can't use super.seed
+ */
+ private long rnd;
+
+ /**
+ * Initialization flag to permit the first and only allowed call
+ * to setSeed (inside Random constructor) to succeed. We can't
+ * allow others since it would cause setting seed in one part of a
+ * program to unintentionally impact other usages by the thread.
+ */
+ boolean initialized;
+
+ // Padding to help avoid memory contention among seed updates in
+ // different TLRs in the common case that they are located near
+ // each other.
+ private long pad0, pad1, pad2, pad3, pad4, pad5, pad6, pad7;
+
+ /**
+ * The actual ThreadLocal
+ */
+ private static final ThreadLocal<ThreadLocalRandom> localRandom =
+ new ThreadLocal<ThreadLocalRandom>() {
+ protected ThreadLocalRandom initialValue() {
+ return new ThreadLocalRandom();
+ }
+ };
+
+
+ /**
+ * Constructor called only by localRandom.initialValue.
+ * We rely on the fact that the superclass no-arg constructor
+ * invokes setSeed exactly once to initialize.
+ */
+ ThreadLocalRandom() {
+ super();
+ }
+
+ /**
+ * Returns the current Thread's ThreadLocalRandom
+ * @return the current Thread's ThreadLocalRandom
+ */
+ public static ThreadLocalRandom current() {
+ return localRandom.get();
+ }
+
+ /**
+ * Throws UnsupportedOperationException. Setting seeds in this
+ * generator is unsupported.
+ * @throws UnsupportedOperationException always
+ */
+ public void setSeed(long seed) {
+ if (initialized)
+ throw new UnsupportedOperationException();
+ initialized = true;
+ rnd = (seed ^ multiplier) & mask;
+ }
+
+ protected int next(int bits) {
+ return (int)((rnd = (rnd * multiplier + addend) & mask) >>> (48-bits));
+ }
+
+ /**
+ * Returns a pseudorandom, uniformly distributed value between the
+ * given least value (inclusive) and bound (exclusive).
+ * @param least the least value returned
+ * @param bound the upper bound (exclusive)
+ * @throws IllegalArgumentException if least greater than or equal
+ * to bound
+ * @return the next value
+ */
+ public int nextInt(int least, int bound) {
+ if (least >= bound)
+ throw new IllegalArgumentException();
+ return nextInt(bound - least) + least;
+ }
+
+ /**
+ * Returns a pseudorandom, uniformly distributed value
+ * between 0 (inclusive) and the specified value (exclusive)
+ * @param n the bound on the random number to be returned. Must be
+ * positive.
+ * @return the next value
+ * @throws IllegalArgumentException if n is not positive
+ */
+ public long nextLong(long n) {
+ if (n <= 0)
+ throw new IllegalArgumentException("n must be positive");
+ // Divide n by two until small enough for nextInt. On each
+ // iteration (at most 31 of them but usually much less),
+ // randomly choose both whether to include high bit in result
+ // (offset) and whether to continue with the lower vs upper
+ // half (which makes a difference only if odd).
+ long offset = 0;
+ while (n >= Integer.MAX_VALUE) {
+ int bits = next(2);
+ long half = n >>> 1;
+ long nextn = ((bits & 2) == 0)? half : n - half;
+ if ((bits & 1) == 0)
+ offset += n - nextn;
+ n = nextn;
+ }
+ return offset + nextInt((int)n);
+ }
+
+ /**
+ * Returns a pseudorandom, uniformly distributed value between the
+ * given least value (inclusive) and bound (exclusive).
+ * @param least the least value returned
+ * @param bound the upper bound (exclusive)
+ * @return the next value
+ * @throws IllegalArgumentException if least greater than or equal
+ * to bound
+ */
+ public long nextLong(long least, long bound) {
+ if (least >= bound)
+ throw new IllegalArgumentException();
+ return nextLong(bound - least) + least;
+ }
+
+ /**
+ * Returns a pseudorandom, uniformly distributed {@code double} value
+ * between 0 (inclusive) and the specified value (exclusive)
+ * @param n the bound on the random number to be returned. Must be
+ * positive.
+ * @return the next value
+ * @throws IllegalArgumentException if n is not positive
+ */
+ public double nextDouble(double n) {
+ if (n <= 0)
+ throw new IllegalArgumentException("n must be positive");
+ return nextDouble() * n;
+ }
+
+ /**
+ * Returns a pseudorandom, uniformly distributed value between the
+ * given least value (inclusive) and bound (exclusive).
+ * @param least the least value returned
+ * @param bound the upper bound (exclusive)
+ * @return the next value
+ * @throws IllegalArgumentException if least greater than or equal
+ * to bound
+ */
+ public double nextDouble(double least, double bound) {
+ if (least >= bound)
+ throw new IllegalArgumentException();
+ return nextDouble() * (bound - least) + least;
+ }
+
+}
diff --git a/src/actors/scala/actors/forkjoin/TransferQueue.java b/src/actors/scala/actors/forkjoin/TransferQueue.java
new file mode 100644
index 0000000000..27ee9f463b
--- /dev/null
+++ b/src/actors/scala/actors/forkjoin/TransferQueue.java
@@ -0,0 +1,118 @@
+/*
+ * Written by Doug Lea with assistance from members of JCP JSR-166
+ * Expert Group and released to the public domain, as explained at
+ * http://creativecommons.org/licenses/publicdomain
+ */
+
+package scala.actors.forkjoin;
+import java.util.concurrent.*;
+
+/**
+ * A {@link BlockingQueue} in which producers may wait for consumers
+ * to receive elements. A {@code TransferQueue} may be useful for
+ * example in message passing applications in which producers
+ * sometimes (using method {@code transfer}) await receipt of
+ * elements by consumers invoking {@code take} or {@code poll},
+ * while at other times enqueue elements (via method {@code put})
+ * without waiting for receipt. Non-blocking and time-out versions of
+ * {@code tryTransfer} are also available. A TransferQueue may also
+ * be queried via {@code hasWaitingConsumer} whether there are any
+ * threads waiting for items, which is a converse analogy to a
+ * {@code peek} operation.
+ *
+ * <p>Like any {@code BlockingQueue}, a {@code TransferQueue} may be
+ * capacity bounded. If so, an attempted {@code transfer} operation
+ * may initially block waiting for available space, and/or
+ * subsequently block waiting for reception by a consumer. Note that
+ * in a queue with zero capacity, such as {@link SynchronousQueue},
+ * {@code put} and {@code transfer} are effectively synonymous.
+ *
+ * <p>This interface is a member of the
+ * <a href="{@docRoot}/../technotes/guides/collections/index.html">
+ * Java Collections Framework</a>.
+ *
+ * @since 1.7
+ * @author Doug Lea
+ * @param <E> the type of elements held in this collection
+ */
+public interface TransferQueue<E> extends BlockingQueue<E> {
+ /**
+ * Transfers the specified element if there exists a consumer
+ * already waiting to receive it, otherwise returning {@code false}
+ * without enqueuing the element.
+ *
+ * @param e the element to transfer
+ * @return {@code true} if the element was transferred, else
+ * {@code false}
+ * @throws ClassCastException if the class of the specified element
+ * prevents it from being added to this queue
+ * @throws NullPointerException if the specified element is null
+ * @throws IllegalArgumentException if some property of the specified
+ * element prevents it from being added to this queue
+ */
+ boolean tryTransfer(E e);
+
+ /**
+ * Inserts the specified element into this queue, waiting if
+ * necessary for space to become available and the element to be
+ * dequeued by a consumer invoking {@code take} or {@code poll}.
+ *
+ * @param e the element to transfer
+ * @throws InterruptedException if interrupted while waiting,
+ * in which case the element is not enqueued.
+ * @throws ClassCastException if the class of the specified element
+ * prevents it from being added to this queue
+ * @throws NullPointerException if the specified element is null
+ * @throws IllegalArgumentException if some property of the specified
+ * element prevents it from being added to this queue
+ */
+ void transfer(E e) throws InterruptedException;
+
+ /**
+ * Inserts the specified element into this queue, waiting up to
+ * the specified wait time if necessary for space to become
+ * available and the element to be dequeued by a consumer invoking
+ * {@code take} or {@code poll}.
+ *
+ * @param e the element to transfer
+ * @param timeout how long to wait before giving up, in units of
+ * {@code unit}
+ * @param unit a {@code TimeUnit} determining how to interpret the
+ * {@code timeout} parameter
+ * @return {@code true} if successful, or {@code false} if
+ * the specified waiting time elapses before completion,
+ * in which case the element is not enqueued.
+ * @throws InterruptedException if interrupted while waiting,
+ * in which case the element is not enqueued.
+ * @throws ClassCastException if the class of the specified element
+ * prevents it from being added to this queue
+ * @throws NullPointerException if the specified element is null
+ * @throws IllegalArgumentException if some property of the specified
+ * element prevents it from being added to this queue
+ */
+ boolean tryTransfer(E e, long timeout, TimeUnit unit)
+ throws InterruptedException;
+
+ /**
+ * Returns {@code true} if there is at least one consumer waiting
+ * to dequeue an element via {@code take} or {@code poll}.
+ * The return value represents a momentary state of affairs.
+ *
+ * @return {@code true} if there is at least one waiting consumer
+ */
+ boolean hasWaitingConsumer();
+
+ /**
+ * Returns an estimate of the number of consumers waiting to
+ * dequeue elements via {@code take} or {@code poll}. The return
+ * value is an approximation of a momentary state of affairs, that
+ * may be inaccurate if consumers have completed or given up
+ * waiting. The value may be useful for monitoring and heuristics,
+ * but not for synchronization control. Implementations of this
+ * method are likely to be noticeably slower than those for
+ * {@link #hasWaitingConsumer}.
+ *
+ * @return the number of consumers waiting to dequeue elements
+ */
+ int getWaitingConsumerCount();
+}
diff --git a/src/actors/scala/actors/forkjoin/package-info.java b/src/actors/scala/actors/forkjoin/package-info.java
new file mode 100644
index 0000000000..4945bc80fc
--- /dev/null
+++ b/src/actors/scala/actors/forkjoin/package-info.java
@@ -0,0 +1,29 @@
+/*
+ * Written by Doug Lea with assistance from members of JCP JSR-166
+ * Expert Group and released to the public domain, as explained at
+ * http://creativecommons.org/licenses/publicdomain
+ */
+
+
+/**
+ * Preview versions of classes targeted for Java 7. Includes a
+ * fine-grained parallel computation framework: ForkJoinTasks and
+ * their related support classes provide a very efficient basis for
+ * obtaining platform-independent parallel speed-ups of
+ * computation-intensive operations. They are not a full substitute
+ * for the kinds of arbitrary processing supported by Executors or
+ * Threads. However, when applicable, they typically provide
+ * significantly greater performance on multiprocessor platforms.
+ *
+ * <p> Candidates for fork/join processing mainly include those that
+ * can be expressed using parallel divide-and-conquer techniques: To
+ * solve a problem, break it in two (or more) parts, and then solve
+ * those parts in parallel, continuing on in this way until the
+ * problem is too small to be broken up, so is solved directly. The
+ * underlying <em>work-stealing</em> framework makes subtasks
+ * available to other threads (normally one per CPU), that help
+ * complete the tasks. In general, the most efficient ForkJoinTasks
+ * are those that directly implement this algorithmic design pattern.
+ *
+ */
+package jsr166y;
generated by cgit v1.2.3 (git 2.39.1) at 2024-06-26 08:54:20 +0000