Switching to the embedded jsr166y in scala.concurrent and dropping akka.jsr166y
This commit is contained in:
Viktor Klang
parent
5f335cb8c8
commit
702b5d9c19
10 changed files with 4 additions and 4916 deletions
File diff suppressed because it is too large
Load diff
File diff suppressed because it is too large
Load diff
|
|
@ -1,119 +0,0 @@
|
|||
/*
|
||||
* 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/publicdomain/zero/1.0/
|
||||
*/
|
||||
|
||||
package akka.jsr166y;
|
||||
|
||||
/**
|
||||
* A thread managed by a {@link ForkJoinPool}, which executes
|
||||
* {@link ForkJoinTask}s.
|
||||
* 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 methods surrounding the main task processing loop.
|
||||
* If you do create such a subclass, you will also need to supply a
|
||||
* custom {@link ForkJoinPool.ForkJoinWorkerThreadFactory} to use it
|
||||
* in a {@code ForkJoinPool}.
|
||||
*
|
||||
* @since 1.7
|
||||
* @author Doug Lea
|
||||
*/
|
||||
public class ForkJoinWorkerThread extends Thread {
|
||||
/*
|
||||
* ForkJoinWorkerThreads are managed by ForkJoinPools and perform
|
||||
* ForkJoinTasks. For explanation, see the internal documentation
|
||||
* of class ForkJoinPool.
|
||||
*/
|
||||
|
||||
final ForkJoinPool.WorkQueue workQueue; // Work-stealing mechanics
|
||||
final ForkJoinPool pool; // the pool this thread works in
|
||||
|
||||
/**
|
||||
* 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) {
|
||||
super(pool.nextWorkerName());
|
||||
setDaemon(true);
|
||||
Thread.UncaughtExceptionHandler ueh = pool.ueh;
|
||||
if (ueh != null)
|
||||
setUncaughtExceptionHandler(ueh);
|
||||
this.pool = pool;
|
||||
pool.registerWorker(this.workQueue = new ForkJoinPool.WorkQueue
|
||||
(pool, this, pool.localMode));
|
||||
}
|
||||
|
||||
/**
|
||||
* 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 workQueue.poolIndex;
|
||||
}
|
||||
|
||||
/**
|
||||
* Initializes internal state after construction but before
|
||||
* processing any tasks. If you override this method, you must
|
||||
* invoke {@code 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() {
|
||||
}
|
||||
|
||||
/**
|
||||
* Performs cleanup associated with termination of this worker
|
||||
* thread. If you override this method, you must invoke
|
||||
* {@code super.onTermination} at the end of the overridden method.
|
||||
*
|
||||
* @param exception the exception causing this thread to abort due
|
||||
* to an unrecoverable error, or {@code null} if completed normally
|
||||
*/
|
||||
protected void onTermination(Throwable exception) {
|
||||
}
|
||||
|
||||
/**
|
||||
* This method is required to be public, but should never be
|
||||
* called explicitly. It performs the main run loop to execute
|
||||
* {@link ForkJoinTask}s.
|
||||
*/
|
||||
public void run() {
|
||||
Throwable exception = null;
|
||||
try {
|
||||
onStart();
|
||||
pool.runWorker(workQueue);
|
||||
} catch (Throwable ex) {
|
||||
exception = ex;
|
||||
} finally {
|
||||
try {
|
||||
onTermination(exception);
|
||||
} catch (Throwable ex) {
|
||||
if (exception == null)
|
||||
exception = ex;
|
||||
} finally {
|
||||
pool.deregisterWorker(this, exception);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1,164 +0,0 @@
|
|||
/*
|
||||
* 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/publicdomain/zero/1.0/
|
||||
*/
|
||||
|
||||
package akka.jsr166y;
|
||||
|
||||
/**
|
||||
* A recursive resultless {@link ForkJoinTask}. This class
|
||||
* establishes conventions to parameterize resultless actions as
|
||||
* {@code Void} {@code ForkJoinTask}s. Because {@code null} is the
|
||||
* only valid value of type {@code Void}, methods such as {@code join}
|
||||
* always return {@code null} upon completion.
|
||||
*
|
||||
* <p><b>Sample Usages.</b> Here is a simple but complete ForkJoin
|
||||
* sort that sorts a given {@code long[]} array:
|
||||
*
|
||||
* <pre> {@code
|
||||
* static class SortTask extends RecursiveAction {
|
||||
* final long[] array; final int lo, hi;
|
||||
* SortTask(long[] array, int lo, int hi) {
|
||||
* this.array = array; this.lo = lo; this.hi = hi;
|
||||
* }
|
||||
* SortTask(long[] array) { this(array, 0, array.length); }
|
||||
* protected void compute() {
|
||||
* if (hi - lo < THRESHOLD)
|
||||
* sortSequentially(lo, hi);
|
||||
* else {
|
||||
* int mid = (lo + hi) >>> 1;
|
||||
* invokeAll(new SortTask(array, lo, mid),
|
||||
* new SortTask(array, mid, hi));
|
||||
* merge(lo, mid, hi);
|
||||
* }
|
||||
* }
|
||||
* // implementation details follow:
|
||||
* final static int THRESHOLD = 1000;
|
||||
* void sortSequentially(int lo, int hi) {
|
||||
* Arrays.sort(array, lo, hi);
|
||||
* }
|
||||
* void merge(int lo, int mid, int hi) {
|
||||
* long[] buf = Arrays.copyOfRange(array, lo, mid);
|
||||
* for (int i = 0, j = lo, k = mid; i < buf.length; j++)
|
||||
* array[j] = (k == hi || buf[i] < array[k]) ?
|
||||
* buf[i++] : array[k++];
|
||||
* }
|
||||
* }}</pre>
|
||||
*
|
||||
* You could then sort {@code anArray} by creating {@code new
|
||||
* SortTask(anArray)} and invoking it in a ForkJoinPool. As a more
|
||||
* concrete simple example, the following task increments each element
|
||||
* of an array:
|
||||
* <pre> {@code
|
||||
* class IncrementTask extends RecursiveAction {
|
||||
* final long[] array; final int lo, hi;
|
||||
* IncrementTask(long[] array, int lo, int hi) {
|
||||
* this.array = array; this.lo = lo; this.hi = hi;
|
||||
* }
|
||||
* protected void compute() {
|
||||
* if (hi - lo < THRESHOLD) {
|
||||
* for (int i = lo; i < hi; ++i)
|
||||
* array[i]++;
|
||||
* }
|
||||
* else {
|
||||
* int mid = (lo + hi) >>> 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 {@code next} references. It uses a dynamic
|
||||
* threshold based on method {@code getSurplusQueuedTaskCount}, but
|
||||
* counterbalances potential excess partitioning by directly
|
||||
* performing leaf actions on unstolen tasks rather than further
|
||||
* subdividing.
|
||||
*
|
||||
* <pre> {@code
|
||||
* double sumOfSquares(ForkJoinPool pool, double[] array) {
|
||||
* int n = array.length;
|
||||
* Applyer a = new Applyer(array, 0, n, null);
|
||||
* pool.invoke(a);
|
||||
* return a.result;
|
||||
* }
|
||||
*
|
||||
* class Applyer extends RecursiveAction {
|
||||
* final double[] array;
|
||||
* final int lo, hi;
|
||||
* double result;
|
||||
* Applyer next; // keeps track of right-hand-side tasks
|
||||
* Applyer(double[] array, int lo, int hi, Applyer next) {
|
||||
* this.array = array; this.lo = lo; this.hi = hi;
|
||||
* this.next = next;
|
||||
* }
|
||||
*
|
||||
* double atLeaf(int l, int h) {
|
||||
* double sum = 0;
|
||||
* for (int i = l; i < 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 > 1 && getSurplusQueuedTaskCount() <= 3) {
|
||||
* int mid = (l + h) >>> 1;
|
||||
* right = new Applyer(array, mid, h, 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.join();
|
||||
* sum += right.result;
|
||||
* }
|
||||
* right = right.next;
|
||||
* }
|
||||
* result = sum;
|
||||
* }
|
||||
* }}</pre>
|
||||
*
|
||||
* @since 1.7
|
||||
* @author Doug Lea
|
||||
*/
|
||||
public abstract class RecursiveAction extends ForkJoinTask<Void> {
|
||||
private static final long serialVersionUID = 5232453952276485070L;
|
||||
|
||||
/**
|
||||
* The main computation performed by this task.
|
||||
*/
|
||||
protected abstract void compute();
|
||||
|
||||
/**
|
||||
* Always returns {@code null}.
|
||||
*
|
||||
* @return {@code null} always
|
||||
*/
|
||||
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;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -1,68 +0,0 @@
|
|||
/*
|
||||
* 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/publicdomain/zero/1.0/
|
||||
*/
|
||||
|
||||
package akka.jsr166y;
|
||||
|
||||
/**
|
||||
* A recursive result-bearing {@link ForkJoinTask}.
|
||||
*
|
||||
* <p>For a classic example, here is a task computing Fibonacci numbers:
|
||||
*
|
||||
* <pre> {@code
|
||||
* class Fibonacci extends RecursiveTask<Integer> {
|
||||
* final int n;
|
||||
* Fibonacci(int n) { this.n = n; }
|
||||
* Integer compute() {
|
||||
* if (n <= 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.
|
||||
*
|
||||
* @since 1.7
|
||||
* @author Doug Lea
|
||||
*/
|
||||
public abstract class RecursiveTask<V> extends ForkJoinTask<V> {
|
||||
private static final long serialVersionUID = 5232453952276485270L;
|
||||
|
||||
/**
|
||||
* The result of the computation.
|
||||
*/
|
||||
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;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -1,197 +0,0 @@
|
|||
/*
|
||||
* 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/publicdomain/zero/1.0/
|
||||
*/
|
||||
|
||||
package akka.jsr166y;
|
||||
|
||||
import java.util.Random;
|
||||
|
||||
/**
|
||||
* A random number generator isolated to the current thread. Like the
|
||||
* global {@link java.util.Random} generator used by the {@link
|
||||
* java.lang.Math} class, a {@code ThreadLocalRandom} is initialized
|
||||
* with an internally generated seed that may not otherwise be
|
||||
* modified. When applicable, use of {@code ThreadLocalRandom} rather
|
||||
* than shared {@code Random} objects in concurrent programs will
|
||||
* typically encounter much less overhead and contention. Use of
|
||||
* {@code ThreadLocalRandom} is 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(...)} (where
|
||||
* {@code X} is {@code Int}, {@code Long}, etc).
|
||||
* When all usages are of this form, it is never possible to
|
||||
* accidently share a {@code ThreadLocalRandom} across multiple threads.
|
||||
*
|
||||
* <p>This class also provides additional commonly used bounded random
|
||||
* generation methods.
|
||||
*
|
||||
* @since 1.7
|
||||
* @author Doug Lea
|
||||
*/
|
||||
public class ThreadLocalRandom extends Random {
|
||||
// same constants as Random, but must be redeclared because private
|
||||
private static final long multiplier = 0x5DEECE66DL;
|
||||
private static final long addend = 0xBL;
|
||||
private static final long mask = (1L << 48) - 1;
|
||||
|
||||
/**
|
||||
* The random seed. We can't use super.seed.
|
||||
*/
|
||||
private long rnd;
|
||||
|
||||
/**
|
||||
* Initialization flag to permit calls to setSeed to succeed only
|
||||
* while executing the Random constructor. 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.
|
||||
*/
|
||||
ThreadLocalRandom() {
|
||||
super();
|
||||
initialized = true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the current thread's {@code ThreadLocalRandom}.
|
||||
*
|
||||
* @return the current thread's {@code ThreadLocalRandom}
|
||||
*/
|
||||
public static ThreadLocalRandom current() {
|
||||
return localRandom.get();
|
||||
}
|
||||
|
||||
/**
|
||||
* Throws {@code UnsupportedOperationException}. Setting seeds in
|
||||
* this generator is not supported.
|
||||
*
|
||||
* @throws UnsupportedOperationException always
|
||||
*/
|
||||
public void setSeed(long seed) {
|
||||
if (initialized)
|
||||
throw new UnsupportedOperationException();
|
||||
rnd = (seed ^ multiplier) & mask;
|
||||
}
|
||||
|
||||
protected int next(int bits) {
|
||||
rnd = (rnd * multiplier + addend) & mask;
|
||||
return (int) (rnd >>> (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;
|
||||
}
|
||||
|
||||
private static final long serialVersionUID = -5851777807851030925L;
|
||||
}
|
||||
|
|
@ -13,7 +13,7 @@ import akka.event.EventStream
|
|||
import com.typesafe.config.Config
|
||||
import akka.serialization.SerializationExtension
|
||||
import akka.event.Logging.LogEventException
|
||||
import akka.jsr166y.{ ForkJoinTask, ForkJoinPool }
|
||||
import scala.concurrent.forkjoin.{ ForkJoinTask, ForkJoinPool }
|
||||
import akka.util.{ Unsafe, Duration, NonFatal, Index }
|
||||
|
||||
final case class Envelope private (val message: Any, val sender: ActorRef)
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ package akka.dispatch
|
|||
|
||||
import java.util.Collection
|
||||
import akka.util.Duration
|
||||
import akka.jsr166y._
|
||||
import scala.concurrent.forkjoin._
|
||||
import java.util.concurrent.atomic.AtomicLong
|
||||
import java.util.concurrent.ArrayBlockingQueue
|
||||
import java.util.concurrent.BlockingQueue
|
||||
|
|
|
|||
|
|
@ -15,7 +15,7 @@ import com.typesafe.config.Config
|
|||
import scala.collection.JavaConversions.iterableAsScalaIterable
|
||||
import java.util.concurrent.atomic.{ AtomicLong, AtomicBoolean }
|
||||
import java.util.concurrent.TimeUnit
|
||||
import akka.jsr166y.ThreadLocalRandom
|
||||
import scala.concurrent.forkjoin.ThreadLocalRandom
|
||||
import akka.util.Unsafe
|
||||
import akka.dispatch.Dispatchers
|
||||
import scala.annotation.tailrec
|
||||
|
|
|
|||
|
|
@ -10,7 +10,7 @@ import akka.ConfigurationException
|
|||
import akka.dispatch.Await
|
||||
import akka.dispatch.MonitorableThreadFactory
|
||||
import akka.event.Logging
|
||||
import akka.jsr166y.ThreadLocalRandom
|
||||
import scala.concurrent.forkjoin.ThreadLocalRandom
|
||||
import akka.pattern._
|
||||
import akka.remote._
|
||||
import akka.routing._
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue