| 43 |
|
* tasks that are never joined. |
| 44 |
|
* |
| 45 |
|
* <p>A static {@link #commonPool} is available and appropriate for |
| 46 |
< |
* most applications. The common pool is constructed upon first |
| 47 |
< |
* access, or upon usage by any ForkJoinTask that is not explictly |
| 48 |
< |
* submitted to a specified pool. Using the common pool normally |
| 49 |
< |
* reduces resource usage (its threads are slowly reclaimed during |
| 50 |
< |
* periods of non-use, and reinstated upon subsequent use). The |
| 51 |
< |
* common pool is by default constructed with default parameters, but |
| 52 |
< |
* these may be controlled by setting any or all of the three |
| 53 |
< |
* properties {@code |
| 46 |
> |
* most applications. The common pool is used by any ForkJoinTask that |
| 47 |
> |
* is not explicitly submitted to a specified pool. Using the common |
| 48 |
> |
* pool normally reduces resource usage (its threads are slowly |
| 49 |
> |
* reclaimed during periods of non-use, and reinstated upon subsequent |
| 50 |
> |
* use). The common pool is by default constructed with default |
| 51 |
> |
* parameters, but these may be controlled by setting any or all of |
| 52 |
> |
* the three properties {@code |
| 53 |
|
* java.util.concurrent.ForkJoinPool.common.{parallelism, |
| 54 |
|
* threadFactory, exceptionHandler}}. |
| 55 |
|
* |
| 235 |
|
* when locked remains available to check consistency. |
| 236 |
|
* |
| 237 |
|
* Recording WorkQueues. WorkQueues are recorded in the |
| 238 |
< |
* "workQueues" array that is created upon pool construction and |
| 239 |
< |
* expanded if necessary. Updates to the array while recording |
| 240 |
< |
* new workers and unrecording terminated ones are protected from |
| 241 |
< |
* each other by a lock but the array is otherwise concurrently |
| 242 |
< |
* readable, and accessed directly. To simplify index-based |
| 243 |
< |
* operations, the array size is always a power of two, and all |
| 244 |
< |
* readers must tolerate null slots. Shared (submission) queues |
| 245 |
< |
* are at even indices, worker queues at odd indices. Grouping |
| 246 |
< |
* them together in this way simplifies and speeds up task |
| 248 |
< |
* scanning. |
| 238 |
> |
* "workQueues" array that is created upon first use and expanded |
| 239 |
> |
* if necessary. Updates to the array while recording new workers |
| 240 |
> |
* and unrecording terminated ones are protected from each other |
| 241 |
> |
* by a lock but the array is otherwise concurrently readable, and |
| 242 |
> |
* accessed directly. To simplify index-based operations, the |
| 243 |
> |
* array size is always a power of two, and all readers must |
| 244 |
> |
* tolerate null slots. Shared (submission) queues are at even |
| 245 |
> |
* indices, worker queues at odd indices. Grouping them together |
| 246 |
> |
* in this way simplifies and speeds up task scanning. |
| 247 |
|
* |
| 248 |
|
* All worker thread creation is on-demand, triggered by task |
| 249 |
|
* submissions, replacement of terminated workers, and/or |
| 501 |
|
} |
| 502 |
|
|
| 503 |
|
/** |
| 506 |
– |
* A simple non-reentrant lock used for exclusion when managing |
| 507 |
– |
* queues and workers. We use a custom lock so that we can readily |
| 508 |
– |
* probe lock state in constructions that check among alternative |
| 509 |
– |
* actions. The lock is normally only very briefly held, and |
| 510 |
– |
* sometimes treated as a spinlock, but other usages block to |
| 511 |
– |
* reduce overall contention in those cases where locked code |
| 512 |
– |
* bodies perform allocation/resizing. |
| 513 |
– |
*/ |
| 514 |
– |
static final class Mutex extends AbstractQueuedSynchronizer { |
| 515 |
– |
public final boolean tryAcquire(int ignore) { |
| 516 |
– |
return compareAndSetState(0, 1); |
| 517 |
– |
} |
| 518 |
– |
public final boolean tryRelease(int ignore) { |
| 519 |
– |
setState(0); |
| 520 |
– |
return true; |
| 521 |
– |
} |
| 522 |
– |
public final void lock() { acquire(0); } |
| 523 |
– |
public final void unlock() { release(0); } |
| 524 |
– |
public final boolean isHeldExclusively() { return getState() == 1; } |
| 525 |
– |
public final Condition newCondition() { return new ConditionObject(); } |
| 526 |
– |
} |
| 527 |
– |
|
| 528 |
– |
/** |
| 504 |
|
* Class for artificial tasks that are used to replace the target |
| 505 |
|
* of local joins if they are removed from an interior queue slot |
| 506 |
|
* in WorkQueue.tryRemoveAndExec. We don't need the proxy to |
| 788 |
|
|
| 789 |
|
/** |
| 790 |
|
* Version of tryUnpush for shared queues; called by non-FJ |
| 791 |
< |
* submitters. Conservatively fails to unpush if all workers |
| 817 |
< |
* are active unless there are multiple tasks in queue. |
| 791 |
> |
* submitters after prechecking that task probably exists. |
| 792 |
|
*/ |
| 793 |
< |
final boolean trySharedUnpush(ForkJoinTask<?> task, ForkJoinPool p) { |
| 793 |
> |
final boolean trySharedUnpush(ForkJoinTask<?> t) { |
| 794 |
|
boolean success = false; |
| 795 |
< |
if (task != null && top != base && runState == 0 && |
| 822 |
< |
U.compareAndSwapInt(this, RUNSTATE, 0, 1)) { |
| 795 |
> |
if (runState == 0 && U.compareAndSwapInt(this, RUNSTATE, 0, 1)) { |
| 796 |
|
try { |
| 797 |
< |
ForkJoinTask<?>[] a; int n, s; |
| 798 |
< |
if ((a = array) != null && (n = (s = top) - base) > 0 && |
| 799 |
< |
(n > 1 || p == null || (int)(p.ctl >> AC_SHIFT) < 0)) { |
| 800 |
< |
int j = (((a.length - 1) & --s) << ASHIFT) + ABASE; |
| 801 |
< |
if (U.getObjectVolatile(a, j) == task && |
| 802 |
< |
U.compareAndSwapObject(a, j, task, null)) { |
| 830 |
< |
top = s; |
| 831 |
< |
success = true; |
| 832 |
< |
} |
| 797 |
> |
ForkJoinTask<?>[] a; int s; |
| 798 |
> |
if ((a = array) != null && (s = top) != base && |
| 799 |
> |
U.compareAndSwapObject |
| 800 |
> |
(a, (((a.length - 1) & --s) << ASHIFT) + ABASE, t, null)) { |
| 801 |
> |
top = s; |
| 802 |
> |
success = true; |
| 803 |
|
} |
| 804 |
|
} finally { |
| 805 |
|
runState = 0; // unlock |
| 1082 |
|
public static final ForkJoinWorkerThreadFactory |
| 1083 |
|
defaultForkJoinWorkerThreadFactory; |
| 1084 |
|
|
| 1085 |
+ |
|
| 1086 |
+ |
/** Property prefix for constructing common pool */ |
| 1087 |
+ |
private static final String propPrefix = |
| 1088 |
+ |
"java.util.concurrent.ForkJoinPool.common."; |
| 1089 |
+ |
|
| 1090 |
+ |
/** |
| 1091 |
+ |
* Common (static) pool. Non-null for public use unless a static |
| 1092 |
+ |
* construction exception, but internal usages must null-check on |
| 1093 |
+ |
* use. |
| 1094 |
+ |
*/ |
| 1095 |
+ |
static final ForkJoinPool commonPool; |
| 1096 |
+ |
|
| 1097 |
+ |
/** |
| 1098 |
+ |
* Common pool parallelism. Must equal commonPool.parallelism. |
| 1099 |
+ |
*/ |
| 1100 |
+ |
static final int commonPoolParallelism; |
| 1101 |
+ |
|
| 1102 |
|
/** |
| 1103 |
|
* Generator for assigning sequence numbers as pool names. |
| 1104 |
|
*/ |
| 1123 |
|
*/ |
| 1124 |
|
private static final ThreadSubmitter submitters; |
| 1125 |
|
|
| 1139 |
– |
/** Common default pool */ |
| 1140 |
– |
static volatile ForkJoinPool commonPool; |
| 1141 |
– |
|
| 1142 |
– |
// commonPool construction parameters |
| 1143 |
– |
private static final String propPrefix = |
| 1144 |
– |
"java.util.concurrent.ForkJoinPool.common."; |
| 1145 |
– |
private static final Thread.UncaughtExceptionHandler commonPoolUEH; |
| 1146 |
– |
private static final ForkJoinWorkerThreadFactory commonPoolFactory; |
| 1147 |
– |
static final int commonPoolParallelism; |
| 1148 |
– |
|
| 1149 |
– |
/** Static initialization lock */ |
| 1150 |
– |
private static final Mutex initializationLock; |
| 1151 |
– |
|
| 1126 |
|
// static constants |
| 1127 |
|
|
| 1128 |
|
/** |
| 1129 |
< |
* Initial timeout value (in nanoseconds) for the tread triggering |
| 1129 |
> |
* Initial timeout value (in nanoseconds) for the thread triggering |
| 1130 |
|
* quiescence to park waiting for new work. On timeout, the thread |
| 1131 |
|
* will instead try to shrink the number of workers. |
| 1132 |
|
*/ |
| 1255 |
|
* empirically works OK on current JVMs. |
| 1256 |
|
*/ |
| 1257 |
|
|
| 1258 |
+ |
volatile long stealCount; // collects worker counts |
| 1259 |
|
volatile long ctl; // main pool control |
| 1260 |
|
final int parallelism; // parallelism level |
| 1261 |
|
final int localMode; // per-worker scheduling mode |
| 1262 |
+ |
volatile int nextWorkerNumber; // to create worker name string |
| 1263 |
|
final int submitMask; // submit queue index bound |
| 1264 |
|
int nextSeed; // for initializing worker seeds |
| 1265 |
+ |
volatile int mainLock; // spinlock for array updates |
| 1266 |
|
volatile int runState; // shutdown status and seq |
| 1267 |
|
WorkQueue[] workQueues; // main registry |
| 1291 |
– |
final Mutex lock; // for registration |
| 1292 |
– |
final Condition termination; // for awaitTermination |
| 1268 |
|
final ForkJoinWorkerThreadFactory factory; // factory for new workers |
| 1269 |
|
final Thread.UncaughtExceptionHandler ueh; // per-worker UEH |
| 1270 |
< |
final AtomicLong stealCount; // collect counts when terminated |
| 1271 |
< |
final AtomicInteger nextWorkerNumber; // to create worker name string |
| 1272 |
< |
String workerNamePrefix; // to create worker name string |
| 1270 |
> |
final String workerNamePrefix; // to create worker name string |
| 1271 |
> |
|
| 1272 |
> |
/* |
| 1273 |
> |
* Mechanics for main lock protecting worker array updates. Uses |
| 1274 |
> |
* the same strategy as ConcurrentHashMap bins -- a spinLock for |
| 1275 |
> |
* normal cases, but falling back to builtin lock when (rarely) |
| 1276 |
> |
* needed. See internal ConcurrentHashMap documentation for |
| 1277 |
> |
* explanation. |
| 1278 |
> |
*/ |
| 1279 |
> |
|
| 1280 |
> |
static final int LOCK_WAITING = 2; // bit to indicate need for signal |
| 1281 |
> |
static final int MAX_LOCK_SPINS = 1 << 8; |
| 1282 |
> |
|
| 1283 |
> |
private void tryAwaitMainLock() { |
| 1284 |
> |
int spins = MAX_LOCK_SPINS, r = 0, h; |
| 1285 |
> |
while (((h = mainLock) & 1) != 0) { |
| 1286 |
> |
if (r == 0) |
| 1287 |
> |
r = ThreadLocalRandom.current().nextInt(); // randomize spins |
| 1288 |
> |
else if (spins >= 0) { |
| 1289 |
> |
r ^= r << 1; r ^= r >>> 3; r ^= r << 10; // xorshift |
| 1290 |
> |
if (r >= 0) |
| 1291 |
> |
--spins; |
| 1292 |
> |
} |
| 1293 |
> |
else if (U.compareAndSwapInt(this, MAINLOCK, h, h | LOCK_WAITING)) { |
| 1294 |
> |
synchronized (this) { |
| 1295 |
> |
if ((mainLock & LOCK_WAITING) != 0) { |
| 1296 |
> |
try { |
| 1297 |
> |
wait(); |
| 1298 |
> |
} catch (InterruptedException ie) { |
| 1299 |
> |
Thread.currentThread().interrupt(); |
| 1300 |
> |
} |
| 1301 |
> |
} |
| 1302 |
> |
else |
| 1303 |
> |
notifyAll(); // possibly won race vs signaller |
| 1304 |
> |
} |
| 1305 |
> |
break; |
| 1306 |
> |
} |
| 1307 |
> |
} |
| 1308 |
> |
} |
| 1309 |
|
|
| 1310 |
|
// Creating, registering, and deregistering workers |
| 1311 |
|
|
| 1333 |
|
* ForkJoinWorkerThread. |
| 1334 |
|
*/ |
| 1335 |
|
final String nextWorkerName() { |
| 1336 |
< |
return workerNamePrefix.concat |
| 1337 |
< |
(Integer.toString(nextWorkerNumber.addAndGet(1))); |
| 1336 |
> |
int n; |
| 1337 |
> |
do {} while(!U.compareAndSwapInt(this, NEXTWORKERNUMBER, |
| 1338 |
> |
n = nextWorkerNumber, ++n)); |
| 1339 |
> |
return workerNamePrefix.concat(Integer.toString(n)); |
| 1340 |
|
} |
| 1341 |
|
|
| 1342 |
|
/** |
| 1349 |
|
* @param w the worker's queue |
| 1350 |
|
*/ |
| 1351 |
|
final void registerWorker(WorkQueue w) { |
| 1352 |
< |
Mutex lock = this.lock; |
| 1353 |
< |
lock.lock(); |
| 1352 |
> |
while (!U.compareAndSwapInt(this, MAINLOCK, 0, 1)) |
| 1353 |
> |
tryAwaitMainLock(); |
| 1354 |
|
try { |
| 1355 |
< |
WorkQueue[] ws = workQueues; |
| 1356 |
< |
if (w != null && ws != null) { // skip on shutdown/failure |
| 1357 |
< |
int rs, n = ws.length, m = n - 1; |
| 1355 |
> |
WorkQueue[] ws; |
| 1356 |
> |
if ((ws = workQueues) == null) |
| 1357 |
> |
ws = workQueues = new WorkQueue[submitMask + 1]; |
| 1358 |
> |
if (w != null) { |
| 1359 |
> |
int rs, n = ws.length, m = n - 1; |
| 1360 |
|
int s = nextSeed += SEED_INCREMENT; // rarely-colliding sequence |
| 1361 |
|
w.seed = (s == 0) ? 1 : s; // ensure non-zero seed |
| 1362 |
|
int r = (s << 1) | 1; // use odd-numbered indices |
| 1376 |
|
runState = ((rs = runState) & SHUTDOWN) | ((rs + 2) & ~SHUTDOWN); |
| 1377 |
|
} |
| 1378 |
|
} finally { |
| 1379 |
< |
lock.unlock(); |
| 1379 |
> |
if (!U.compareAndSwapInt(this, MAINLOCK, 1, 0)) { |
| 1380 |
> |
mainLock = 0; |
| 1381 |
> |
synchronized (this) { notifyAll(); }; |
| 1382 |
> |
} |
| 1383 |
|
} |
| 1384 |
+ |
|
| 1385 |
|
} |
| 1386 |
|
|
| 1387 |
|
/** |
| 1394 |
|
* @param ex the exception causing failure, or null if none |
| 1395 |
|
*/ |
| 1396 |
|
final void deregisterWorker(ForkJoinWorkerThread wt, Throwable ex) { |
| 1378 |
– |
Mutex lock = this.lock; |
| 1397 |
|
WorkQueue w = null; |
| 1398 |
|
if (wt != null && (w = wt.workQueue) != null) { |
| 1399 |
|
w.runState = -1; // ensure runState is set |
| 1400 |
< |
stealCount.getAndAdd(w.totalSteals + w.nsteals); |
| 1400 |
> |
long steals = w.totalSteals + w.nsteals, sc; |
| 1401 |
> |
do {} while(!U.compareAndSwapLong(this, STEALCOUNT, |
| 1402 |
> |
sc = stealCount, sc + steals)); |
| 1403 |
|
int idx = w.poolIndex; |
| 1404 |
< |
lock.lock(); |
| 1405 |
< |
try { // remove record from array |
| 1404 |
> |
while (!U.compareAndSwapInt(this, MAINLOCK, 0, 1)) |
| 1405 |
> |
tryAwaitMainLock(); |
| 1406 |
> |
try { |
| 1407 |
|
WorkQueue[] ws = workQueues; |
| 1408 |
|
if (ws != null && idx >= 0 && idx < ws.length && ws[idx] == w) |
| 1409 |
|
ws[idx] = null; |
| 1410 |
|
} finally { |
| 1411 |
< |
lock.unlock(); |
| 1411 |
> |
if (!U.compareAndSwapInt(this, MAINLOCK, 1, 0)) { |
| 1412 |
> |
mainLock = 0; |
| 1413 |
> |
synchronized (this) { notifyAll(); }; |
| 1414 |
> |
} |
| 1415 |
|
} |
| 1416 |
|
} |
| 1417 |
|
|
| 1450 |
|
for (int r = s.seed, m = submitMask;;) { |
| 1451 |
|
WorkQueue[] ws; WorkQueue q; |
| 1452 |
|
int k = r & m & SQMASK; // use only even indices |
| 1453 |
< |
if (runState < 0 || (ws = workQueues) == null || ws.length <= k) |
| 1453 |
> |
if (runState < 0) |
| 1454 |
|
throw new RejectedExecutionException(); // shutting down |
| 1455 |
+ |
else if ((ws = workQueues) == null || ws.length <= k) { |
| 1456 |
+ |
while (!U.compareAndSwapInt(this, MAINLOCK, 0, 1)) |
| 1457 |
+ |
tryAwaitMainLock(); |
| 1458 |
+ |
try { |
| 1459 |
+ |
if (workQueues == null) |
| 1460 |
+ |
workQueues = new WorkQueue[submitMask + 1]; |
| 1461 |
+ |
} finally { |
| 1462 |
+ |
if (!U.compareAndSwapInt(this, MAINLOCK, 1, 0)) { |
| 1463 |
+ |
mainLock = 0; |
| 1464 |
+ |
synchronized (this) { notifyAll(); }; |
| 1465 |
+ |
} |
| 1466 |
+ |
} |
| 1467 |
+ |
} |
| 1468 |
|
else if ((q = ws[k]) == null) { // create new queue |
| 1469 |
|
WorkQueue nq = new WorkQueue(this, null, SHARED_QUEUE); |
| 1470 |
< |
Mutex lock = this.lock; // construct outside lock |
| 1471 |
< |
lock.lock(); |
| 1472 |
< |
try { // recheck under lock |
| 1470 |
> |
while (!U.compareAndSwapInt(this, MAINLOCK, 0, 1)) |
| 1471 |
> |
tryAwaitMainLock(); |
| 1472 |
> |
try { |
| 1473 |
|
int rs = runState; // to update seq |
| 1474 |
|
if (ws == workQueues && ws[k] == null) { |
| 1475 |
|
ws[k] = nq; |
| 1476 |
|
runState = ((rs & SHUTDOWN) | ((rs + 2) & ~SHUTDOWN)); |
| 1477 |
|
} |
| 1478 |
|
} finally { |
| 1479 |
< |
lock.unlock(); |
| 1479 |
> |
if (!U.compareAndSwapInt(this, MAINLOCK, 1, 0)) { |
| 1480 |
> |
mainLock = 0; |
| 1481 |
> |
synchronized (this) { notifyAll(); }; |
| 1482 |
> |
} |
| 1483 |
|
} |
| 1484 |
|
} |
| 1485 |
|
else if (q.trySharedPush(task)) { |
| 1502 |
|
static void submitToCommonPool(ForkJoinTask<?> task) { |
| 1503 |
|
ForkJoinPool p; |
| 1504 |
|
if ((p = commonPool) == null) |
| 1505 |
< |
p = ensureCommonPool(); |
| 1505 |
> |
throw new RejectedExecutionException("Common Pool Unavailable"); |
| 1506 |
|
p.doSubmit(task); |
| 1507 |
|
} |
| 1508 |
|
|
| 1516 |
|
* @return true if successful |
| 1517 |
|
*/ |
| 1518 |
|
static boolean tryUnsubmitFromCommonPool(ForkJoinTask<?> task) { |
| 1519 |
+ |
// Peek, looking for task and eligibility before |
| 1520 |
+ |
// using trySharedUnpush to actually take it under lock |
| 1521 |
|
ForkJoinPool p; WorkQueue[] ws; WorkQueue q; |
| 1522 |
+ |
ForkJoinTask<?>[] a; int t, s, n; |
| 1523 |
|
int k = submitters.get().seed & SQMASK; |
| 1524 |
|
return ((p = commonPool) != null && |
| 1525 |
|
(ws = p.workQueues) != null && |
| 1526 |
|
ws.length > (k &= p.submitMask) && |
| 1527 |
|
(q = ws[k]) != null && |
| 1528 |
< |
q.trySharedUnpush(task, p)); |
| 1528 |
> |
(a = q.array) != null && |
| 1529 |
> |
(n = (t = q.top) - q.base) > 0 && |
| 1530 |
> |
(n > 1 || (int)(p.ctl >> AC_SHIFT) < 0) && |
| 1531 |
> |
(s = t - 1) >= 0 && s < a.length && a[s] == task && |
| 1532 |
> |
q.trySharedUnpush(task)); |
| 1533 |
|
} |
| 1534 |
|
|
| 1535 |
|
// Maintaining ctl counts |
| 2003 |
|
*/ |
| 2004 |
|
private WorkQueue findNonEmptyStealQueue(WorkQueue w) { |
| 2005 |
|
// Similar to loop in scan(), but ignoring submissions |
| 2006 |
< |
int r; |
| 1960 |
< |
if (w == null) // allow external callers |
| 1961 |
< |
r = ThreadLocalRandom.current().nextInt(); |
| 1962 |
< |
else { |
| 1963 |
< |
r = w.seed; r ^= r << 13; r ^= r >>> 17; w.seed = r ^= r << 5; |
| 1964 |
< |
} |
| 2006 |
> |
int r = w.seed; r ^= r << 13; r ^= r >>> 17; w.seed = r ^= r << 5; |
| 2007 |
|
int step = (r >>> 16) | 1; |
| 2008 |
|
for (WorkQueue[] ws;;) { |
| 2009 |
|
int rs = runState, m; |
| 2144 |
|
* @return true if now terminating or terminated |
| 2145 |
|
*/ |
| 2146 |
|
private boolean tryTerminate(boolean now, boolean enable) { |
| 2105 |
– |
Mutex lock = this.lock; |
| 2147 |
|
for (long c;;) { |
| 2148 |
|
if (((c = ctl) & STOP_BIT) != 0) { // already terminating |
| 2149 |
|
if ((short)(c >>> TC_SHIFT) == -parallelism) { |
| 2150 |
< |
lock.lock(); // don't need try/finally |
| 2151 |
< |
termination.signalAll(); // signal when 0 workers |
| 2152 |
< |
lock.unlock(); |
| 2150 |
> |
synchronized(this) { |
| 2151 |
> |
notifyAll(); // signal when 0 workers |
| 2152 |
> |
} |
| 2153 |
|
} |
| 2154 |
|
return true; |
| 2155 |
|
} |
| 2156 |
|
if (runState >= 0) { // not yet enabled |
| 2157 |
|
if (!enable) |
| 2158 |
|
return false; |
| 2159 |
< |
lock.lock(); |
| 2160 |
< |
runState |= SHUTDOWN; |
| 2161 |
< |
lock.unlock(); |
| 2159 |
> |
while (!U.compareAndSwapInt(this, MAINLOCK, 0, 1)) |
| 2160 |
> |
tryAwaitMainLock(); |
| 2161 |
> |
try { |
| 2162 |
> |
runState |= SHUTDOWN; |
| 2163 |
> |
} finally { |
| 2164 |
> |
if (!U.compareAndSwapInt(this, MAINLOCK, 1, 0)) { |
| 2165 |
> |
mainLock = 0; |
| 2166 |
> |
synchronized (this) { notifyAll(); }; |
| 2167 |
> |
} |
| 2168 |
> |
} |
| 2169 |
|
} |
| 2170 |
|
if (!now) { // check if idle & no tasks |
| 2171 |
|
if ((int)(c >> AC_SHIFT) != -parallelism || |
| 2298 |
|
// Use nearest power 2 for workQueues size. See Hackers Delight sec 3.2. |
| 2299 |
|
int n = parallelism - 1; |
| 2300 |
|
n |= n >>> 1; n |= n >>> 2; n |= n >>> 4; n |= n >>> 8; n |= n >>> 16; |
| 2301 |
< |
int size = (n + 1) << 1; // #slots = 2*#workers |
| 2254 |
< |
this.submitMask = size - 1; // room for max # of submit queues |
| 2255 |
< |
this.workQueues = new WorkQueue[size]; |
| 2256 |
< |
this.termination = (this.lock = new Mutex()).newCondition(); |
| 2257 |
< |
this.stealCount = new AtomicLong(); |
| 2258 |
< |
this.nextWorkerNumber = new AtomicInteger(); |
| 2301 |
> |
this.submitMask = ((n + 1) << 1) - 1; |
| 2302 |
|
int pn = poolNumberGenerator.incrementAndGet(); |
| 2303 |
|
StringBuilder sb = new StringBuilder("ForkJoinPool-"); |
| 2304 |
|
sb.append(Integer.toString(pn)); |
| 2305 |
|
sb.append("-worker-"); |
| 2306 |
|
this.workerNamePrefix = sb.toString(); |
| 2264 |
– |
lock.lock(); |
| 2307 |
|
this.runState = 1; // set init flag |
| 2266 |
– |
lock.unlock(); |
| 2308 |
|
} |
| 2309 |
|
|
| 2310 |
|
/** |
| 2311 |
< |
* Returns the common pool instance |
| 2311 |
> |
* Constructor for common pool, suitable only for static initialization. |
| 2312 |
> |
* Basically the same as above, but uses smallest possible initial footprint. |
| 2313 |
> |
*/ |
| 2314 |
> |
ForkJoinPool(int parallelism, int submitMask, |
| 2315 |
> |
ForkJoinWorkerThreadFactory factory, |
| 2316 |
> |
Thread.UncaughtExceptionHandler handler) { |
| 2317 |
> |
this.factory = factory; |
| 2318 |
> |
this.ueh = handler; |
| 2319 |
> |
this.submitMask = submitMask; |
| 2320 |
> |
this.parallelism = parallelism; |
| 2321 |
> |
long np = (long)(-parallelism); |
| 2322 |
> |
this.ctl = ((np << AC_SHIFT) & AC_MASK) | ((np << TC_SHIFT) & TC_MASK); |
| 2323 |
> |
this.localMode = LIFO_QUEUE; |
| 2324 |
> |
this.workerNamePrefix = "ForkJoinPool.commonPool-worker-"; |
| 2325 |
> |
this.runState = 1; |
| 2326 |
> |
} |
| 2327 |
> |
|
| 2328 |
> |
/** |
| 2329 |
> |
* Returns the common pool instance. |
| 2330 |
|
* |
| 2331 |
|
* @return the common pool instance |
| 2332 |
|
*/ |
| 2333 |
|
public static ForkJoinPool commonPool() { |
| 2334 |
|
ForkJoinPool p; |
| 2335 |
< |
return (p = commonPool) != null? p : ensureCommonPool(); |
| 2336 |
< |
} |
| 2278 |
< |
|
| 2279 |
< |
private static ForkJoinPool ensureCommonPool() { |
| 2280 |
< |
ForkJoinPool p; |
| 2281 |
< |
if ((p = commonPool) == null) { |
| 2282 |
< |
final Mutex lock = initializationLock; |
| 2283 |
< |
lock.lock(); |
| 2284 |
< |
try { |
| 2285 |
< |
if ((p = commonPool) == null) { |
| 2286 |
< |
p = commonPool = new ForkJoinPool(commonPoolParallelism, |
| 2287 |
< |
commonPoolFactory, |
| 2288 |
< |
commonPoolUEH, false); |
| 2289 |
< |
// use a more informative name string for workers |
| 2290 |
< |
p.workerNamePrefix = "ForkJoinPool.commonPool-worker-"; |
| 2291 |
< |
} |
| 2292 |
< |
} finally { |
| 2293 |
< |
lock.unlock(); |
| 2294 |
< |
} |
| 2295 |
< |
} |
| 2335 |
> |
if ((p = commonPool) == null) |
| 2336 |
> |
throw new Error("Common Pool Unavailable"); |
| 2337 |
|
return p; |
| 2338 |
|
} |
| 2339 |
|
|
| 2599 |
|
* @return the number of steals |
| 2600 |
|
*/ |
| 2601 |
|
public long getStealCount() { |
| 2602 |
< |
long count = stealCount.get(); |
| 2602 |
> |
long count = stealCount; |
| 2603 |
|
WorkQueue[] ws; WorkQueue w; |
| 2604 |
|
if ((ws = workQueues) != null) { |
| 2605 |
|
for (int i = 1; i < ws.length; i += 2) { |
| 2729 |
|
public String toString() { |
| 2730 |
|
// Use a single pass through workQueues to collect counts |
| 2731 |
|
long qt = 0L, qs = 0L; int rc = 0; |
| 2732 |
< |
long st = stealCount.get(); |
| 2732 |
> |
long st = stealCount; |
| 2733 |
|
long c = ctl; |
| 2734 |
|
WorkQueue[] ws; WorkQueue w; |
| 2735 |
|
if ((ws = workQueues) != null) { |
| 2867 |
|
public boolean awaitTermination(long timeout, TimeUnit unit) |
| 2868 |
|
throws InterruptedException { |
| 2869 |
|
long nanos = unit.toNanos(timeout); |
| 2870 |
< |
final Mutex lock = this.lock; |
| 2871 |
< |
lock.lock(); |
| 2872 |
< |
try { |
| 2873 |
< |
for (;;) { |
| 2874 |
< |
if (isTerminated()) |
| 2875 |
< |
return true; |
| 2876 |
< |
if (nanos <= 0) |
| 2877 |
< |
return false; |
| 2878 |
< |
nanos = termination.awaitNanos(nanos); |
| 2870 |
> |
if (isTerminated()) |
| 2871 |
> |
return true; |
| 2872 |
> |
long startTime = System.nanoTime(); |
| 2873 |
> |
boolean terminated = false; |
| 2874 |
> |
synchronized(this) { |
| 2875 |
> |
for (long waitTime = nanos, millis = 0L;;) { |
| 2876 |
> |
if (terminated = isTerminated() || |
| 2877 |
> |
waitTime <= 0L || |
| 2878 |
> |
(millis = unit.toMillis(waitTime)) <= 0L) |
| 2879 |
> |
break; |
| 2880 |
> |
wait(millis); |
| 2881 |
> |
waitTime = nanos - (System.nanoTime() - startTime); |
| 2882 |
|
} |
| 2839 |
– |
} finally { |
| 2840 |
– |
lock.unlock(); |
| 2883 |
|
} |
| 2884 |
+ |
return terminated; |
| 2885 |
|
} |
| 2886 |
|
|
| 2887 |
|
/** |
| 3013 |
|
private static final long PARKBLOCKER; |
| 3014 |
|
private static final int ABASE; |
| 3015 |
|
private static final int ASHIFT; |
| 3016 |
+ |
private static final long NEXTWORKERNUMBER; |
| 3017 |
+ |
private static final long STEALCOUNT; |
| 3018 |
+ |
private static final long MAINLOCK; |
| 3019 |
|
|
| 3020 |
|
static { |
| 3021 |
|
poolNumberGenerator = new AtomicInteger(); |
| 3024 |
|
defaultForkJoinWorkerThreadFactory = |
| 3025 |
|
new DefaultForkJoinWorkerThreadFactory(); |
| 3026 |
|
submitters = new ThreadSubmitter(); |
| 2981 |
– |
initializationLock = new Mutex(); |
| 3027 |
|
int s; |
| 3028 |
|
try { |
| 3029 |
|
U = getUnsafe(); |
| 3031 |
|
Class<?> ak = ForkJoinTask[].class; |
| 3032 |
|
CTL = U.objectFieldOffset |
| 3033 |
|
(k.getDeclaredField("ctl")); |
| 3034 |
+ |
NEXTWORKERNUMBER = U.objectFieldOffset |
| 3035 |
+ |
(k.getDeclaredField("nextWorkerNumber")); |
| 3036 |
+ |
STEALCOUNT = U.objectFieldOffset |
| 3037 |
+ |
(k.getDeclaredField("stealCount")); |
| 3038 |
+ |
MAINLOCK = U.objectFieldOffset |
| 3039 |
+ |
(k.getDeclaredField("mainLock")); |
| 3040 |
|
Class<?> tk = Thread.class; |
| 3041 |
|
PARKBLOCKER = U.objectFieldOffset |
| 3042 |
|
(tk.getDeclaredField("parkBlocker")); |
| 3043 |
|
ABASE = U.arrayBaseOffset(ak); |
| 3044 |
|
s = U.arrayIndexScale(ak); |
| 3045 |
+ |
ASHIFT = 31 - Integer.numberOfLeadingZeros(s); |
| 3046 |
|
} catch (Exception e) { |
| 3047 |
|
throw new Error(e); |
| 3048 |
|
} |
| 3049 |
|
if ((s & (s-1)) != 0) |
| 3050 |
|
throw new Error("data type scale not a power of two"); |
| 3051 |
< |
ASHIFT = 31 - Integer.numberOfLeadingZeros(s); |
| 3000 |
< |
|
| 3001 |
< |
// Establish configuration for default pool |
| 3002 |
< |
try { |
| 3051 |
> |
try { // Establish common pool |
| 3052 |
|
String pp = System.getProperty(propPrefix + "parallelism"); |
| 3053 |
|
String fp = System.getProperty(propPrefix + "threadFactory"); |
| 3054 |
|
String up = System.getProperty(propPrefix + "exceptionHandler"); |
| 3055 |
+ |
ForkJoinWorkerThreadFactory fac = (fp == null) ? |
| 3056 |
+ |
defaultForkJoinWorkerThreadFactory : |
| 3057 |
+ |
((ForkJoinWorkerThreadFactory)ClassLoader. |
| 3058 |
+ |
getSystemClassLoader().loadClass(fp).newInstance()); |
| 3059 |
+ |
Thread.UncaughtExceptionHandler ueh = (up == null)? null : |
| 3060 |
+ |
((Thread.UncaughtExceptionHandler)ClassLoader. |
| 3061 |
+ |
getSystemClassLoader().loadClass(up).newInstance()); |
| 3062 |
|
int par; |
| 3063 |
|
if ((pp == null || (par = Integer.parseInt(pp)) <= 0)) |
| 3064 |
|
par = Runtime.getRuntime().availableProcessors(); |
| 3065 |
+ |
if (par > MAX_CAP) |
| 3066 |
+ |
par = MAX_CAP; |
| 3067 |
|
commonPoolParallelism = par; |
| 3068 |
< |
if (fp != null) |
| 3069 |
< |
commonPoolFactory = (ForkJoinWorkerThreadFactory) |
| 3070 |
< |
ClassLoader.getSystemClassLoader().loadClass(fp).newInstance(); |
| 3071 |
< |
else |
| 3072 |
< |
commonPoolFactory = defaultForkJoinWorkerThreadFactory; |
| 3015 |
< |
if (up != null) |
| 3016 |
< |
commonPoolUEH = (Thread.UncaughtExceptionHandler) |
| 3017 |
< |
ClassLoader.getSystemClassLoader().loadClass(up).newInstance(); |
| 3018 |
< |
else |
| 3019 |
< |
commonPoolUEH = null; |
| 3068 |
> |
int n = par - 1; // precompute submit mask |
| 3069 |
> |
n |= n >>> 1; n |= n >>> 2; n |= n >>> 4; |
| 3070 |
> |
n |= n >>> 8; n |= n >>> 16; |
| 3071 |
> |
int mask = ((n + 1) << 1) - 1; |
| 3072 |
> |
commonPool = new ForkJoinPool(par, mask, fac, ueh); |
| 3073 |
|
} catch (Exception e) { |
| 3074 |
|
throw new Error(e); |
| 3075 |
|
} |
