292 |
|
* "terminate" status, cancels all unprocessed tasks, and wakes up |
293 |
|
* all waiting workers. Detecting whether termination should |
294 |
|
* commence after a non-abrupt shutdown() call requires more work |
295 |
< |
* and bookkeeping. We need consensus about quiesence (i.e., that |
295 |
> |
* and bookkeeping. We need consensus about quiescence (i.e., that |
296 |
|
* there is no more work) which is reflected in active counts so |
297 |
|
* long as there are no current blockers, as well as possible |
298 |
|
* re-evaluations during independent changes in blocking or |
762 |
|
|
763 |
|
/** |
764 |
|
* Tries to enqueue worker w in wait queue and await change in |
765 |
< |
* worker's eventCount. If the pool is quiescent, possibly |
766 |
< |
* terminates worker upon exit. Otherwise, before blocking, |
767 |
< |
* rescans queues to avoid missed signals. Upon finding work, |
768 |
< |
* releases at least one worker (which may be the current |
769 |
< |
* worker). Rescans restart upon detected staleness or failure to |
770 |
< |
* release due to contention. Note the unusual conventions about |
771 |
< |
* Thread.interrupt here and elsewhere: Because interrupts are |
772 |
< |
* used solely to alert threads to check termination, which is |
773 |
< |
* checked here anyway, we clear status (using Thread.interrupted) |
774 |
< |
* before any call to park, so that park does not immediately |
775 |
< |
* return due to status being set via some other unrelated call to |
776 |
< |
* interrupt in user code. |
765 |
> |
* worker's eventCount. If the pool is quiescent and there is |
766 |
> |
* more than one worker, possibly terminates worker upon exit. |
767 |
> |
* Otherwise, before blocking, rescans queues to avoid missed |
768 |
> |
* signals. Upon finding work, releases at least one worker |
769 |
> |
* (which may be the current worker). Rescans restart upon |
770 |
> |
* detected staleness or failure to release due to |
771 |
> |
* contention. Note the unusual conventions about Thread.interrupt |
772 |
> |
* here and elsewhere: Because interrupts are used solely to alert |
773 |
> |
* threads to check termination, which is checked here anyway, we |
774 |
> |
* clear status (using Thread.interrupted) before any call to |
775 |
> |
* park, so that park does not immediately return due to status |
776 |
> |
* being set via some other unrelated call to interrupt in user |
777 |
> |
* code. |
778 |
|
* |
779 |
|
* @param w the calling worker |
780 |
|
* @param c the ctl value on entry |
795 |
|
else if (w.eventCount != v) |
796 |
|
return true; // update next time |
797 |
|
} |
798 |
< |
if (parallelism + (int)(nc >> AC_SHIFT) == 0 && |
798 |
> |
if ((!shutdown || !tryTerminate(false)) && |
799 |
> |
(int)c != 0 && parallelism + (int)(nc >> AC_SHIFT) == 0 && |
800 |
|
blockedCount == 0 && quiescerCount == 0) |
801 |
|
idleAwaitWork(w, nc, c, v); // quiescent |
802 |
|
for (boolean rescanned = false;;) { |
866 |
|
w.parked = false; |
867 |
|
if (w.eventCount != v) |
868 |
|
break; |
869 |
< |
else if (System.nanoTime() - startTime < SHRINK_RATE) |
869 |
> |
else if (System.nanoTime() - startTime < |
870 |
> |
SHRINK_RATE - (SHRINK_RATE / 10)) // timing slop |
871 |
|
Thread.interrupted(); // spurious wakeup |
872 |
|
else if (UNSAFE.compareAndSwapLong(this, ctlOffset, |
873 |
|
currentCtl, prevCtl)) { |
996 |
|
do {} while (!UNSAFE.compareAndSwapLong(this, ctlOffset, // no mask |
997 |
|
c = ctl, c + AC_UNIT)); |
998 |
|
int b; |
999 |
< |
do {} while(!UNSAFE.compareAndSwapInt(this, blockedCountOffset, |
1000 |
< |
b = blockedCount, b - 1)); |
999 |
> |
do {} while (!UNSAFE.compareAndSwapInt(this, blockedCountOffset, |
1000 |
> |
b = blockedCount, b - 1)); |
1001 |
|
} |
1002 |
|
|
1003 |
|
/** |
1149 |
|
ws[k] = w; |
1150 |
|
nextWorkerIndex = k + 1; |
1151 |
|
int m = g & SMASK; |
1152 |
< |
g = k > m? ((m << 1) + 1) & SMASK : g + (SG_UNIT<<1); |
1152 |
> |
g = (k > m) ? ((m << 1) + 1) & SMASK : g + (SG_UNIT<<1); |
1153 |
|
} |
1154 |
|
} finally { |
1155 |
|
scanGuard = g; |
1332 |
|
*/ |
1333 |
|
final void addQuiescerCount(int delta) { |
1334 |
|
int c; |
1335 |
< |
do {} while(!UNSAFE.compareAndSwapInt(this, quiescerCountOffset, |
1336 |
< |
c = quiescerCount, c + delta)); |
1335 |
> |
do {} while (!UNSAFE.compareAndSwapInt(this, quiescerCountOffset, |
1336 |
> |
c = quiescerCount, c + delta)); |
1337 |
|
} |
1338 |
|
|
1339 |
|
/** |
1358 |
|
final int idlePerActive() { |
1359 |
|
// Approximate at powers of two for small values, saturate past 4 |
1360 |
|
int p = parallelism; |
1361 |
< |
int a = p + (int)(ctl >> AC_SHIFT); |
1362 |
< |
return (a > (p >>>= 1) ? 0 : |
1363 |
< |
a > (p >>>= 1) ? 1 : |
1364 |
< |
a > (p >>>= 1) ? 2 : |
1365 |
< |
a > (p >>>= 1) ? 4 : |
1366 |
< |
8); |
1361 |
> |
int a = p + (int)(ctl >> AC_SHIFT); |
1362 |
> |
return (a > (p >>>= 1) ? 0 : |
1363 |
> |
a > (p >>>= 1) ? 1 : |
1364 |
> |
a > (p >>>= 1) ? 2 : |
1365 |
> |
a > (p >>>= 1) ? 4 : |
1366 |
> |
8); |
1367 |
|
} |
1368 |
|
|
1369 |
|
// Exported methods |
1686 |
|
*/ |
1687 |
|
public int getRunningThreadCount() { |
1688 |
|
int r = parallelism + (int)(ctl >> AC_SHIFT); |
1689 |
< |
return r <= 0? 0 : r; // suppress momentarily negative values |
1689 |
> |
return (r <= 0) ? 0 : r; // suppress momentarily negative values |
1690 |
|
} |
1691 |
|
|
1692 |
|
/** |
1698 |
|
*/ |
1699 |
|
public int getActiveThreadCount() { |
1700 |
|
int r = parallelism + (int)(ctl >> AC_SHIFT) + blockedCount; |
1701 |
< |
return r <= 0? 0 : r; // suppress momentarily negative values |
1701 |
> |
return (r <= 0) ? 0 : r; // suppress momentarily negative values |
1702 |
|
} |
1703 |
|
|
1704 |
|
/** |
1853 |
|
int ac = rc + blockedCount; |
1854 |
|
String level; |
1855 |
|
if ((c & STOP_BIT) != 0) |
1856 |
< |
level = (tc == 0)? "Terminated" : "Terminating"; |
1856 |
> |
level = (tc == 0) ? "Terminated" : "Terminating"; |
1857 |
|
else |
1858 |
< |
level = shutdown? "Shutting down" : "Running"; |
1858 |
> |
level = shutdown ? "Shutting down" : "Running"; |
1859 |
|
return super.toString() + |
1860 |
|
"[" + level + |
1861 |
|
", parallelism = " + pc + |
2121 |
|
int s; |
2122 |
|
try { |
2123 |
|
UNSAFE = getUnsafe(); |
2124 |
< |
Class k = ForkJoinPool.class; |
2124 |
> |
Class<?> k = ForkJoinPool.class; |
2125 |
|
ctlOffset = UNSAFE.objectFieldOffset |
2126 |
|
(k.getDeclaredField("ctl")); |
2127 |
|
stealCountOffset = UNSAFE.objectFieldOffset |
2134 |
|
(k.getDeclaredField("scanGuard")); |
2135 |
|
nextWorkerNumberOffset = UNSAFE.objectFieldOffset |
2136 |
|
(k.getDeclaredField("nextWorkerNumber")); |
2137 |
< |
Class a = ForkJoinTask[].class; |
2137 |
> |
Class<?> a = ForkJoinTask[].class; |
2138 |
|
ABASE = UNSAFE.arrayBaseOffset(a); |
2139 |
|
s = UNSAFE.arrayIndexScale(a); |
2140 |
|
} catch (Exception e) { |