ViewVC Help

View File | Revision Log | Show Annotations | Download File | Root Listing

root/jsr166/jsr166/src/jsr166y/Phaser.java

Comparing jsr166/src/jsr166y/Phaser.java (file contents):
Revision 1.74 by jsr166, Wed Jun 1 21:04:30 2011 UTC vs.
Revision 1.81 by jsr166, Fri Jul 15 18:49:12 2016 UTC

#	Line 17 | Line 17 | import java.util.concurrent.locks.LockSu
17		* {@link java.util.concurrent.CountDownLatch CountDownLatch}
18		* but supporting more flexible usage.
19		*
20	<	* <p> <b>Registration.</b> Unlike the case for other barriers, the
20	>	* <p><b>Registration.</b> Unlike the case for other barriers, the
21		* number of parties <em>registered</em> to synchronize on a phaser
22		* may vary over time.  Tasks may be registered at any time (using
23		* methods {@link #register}, {@link #bulkRegister}, or forms of
#	Line 30 | Line 30 | import java.util.concurrent.locks.LockSu
30		* (However, you can introduce such bookkeeping by subclassing this
31		* class.)
32		*
33	<	* <p> <b>Synchronization.</b> Like a {@code CyclicBarrier}, a {@code
33	>	* <p><b>Synchronization.</b> Like a {@code CyclicBarrier}, a {@code
34		* Phaser} may be repeatedly awaited.  Method {@link
35		* #arriveAndAwaitAdvance} has effect analogous to {@link
36		* java.util.concurrent.CyclicBarrier#await CyclicBarrier.await}. Each
#	Line 44 | Line 44 | import java.util.concurrent.locks.LockSu
44		*
45		* <ul>
46		*
47	<	*   <li> <b>Arrival.</b> Methods {@link #arrive} and
47	>	*   <li><b>Arrival.</b> Methods {@link #arrive} and
48		*       {@link #arriveAndDeregister} record arrival.  These methods
49		*       do not block, but return an associated <em>arrival phase
50		*       number</em>; that is, the phase number of the phaser to which
#	Line 57 | Line 57 | import java.util.concurrent.locks.LockSu
57		*       flexible than, providing a barrier action to a {@code
58		*       CyclicBarrier}.
59		*
60	<	*   <li> <b>Waiting.</b> Method {@link #awaitAdvance} requires an
60	>	*   <li><b>Waiting.</b> Method {@link #awaitAdvance} requires an
61		*       argument indicating an arrival phase number, and returns when
62		*       the phaser advances to (or is already at) a different phase.
63		*       Unlike similar constructions using {@code CyclicBarrier},
#	Line 74 | Line 74 | import java.util.concurrent.locks.LockSu
74		*
75		* </ul>
76		*
77	<	* <p> <b>Termination.</b> A phaser may enter a <em>termination</em>
77	>	* <p><b>Termination.</b> A phaser may enter a <em>termination</em>
78		* state, that may be checked using method {@link #isTerminated}. Upon
79		* termination, all synchronization methods immediately return without
80		* waiting for advance, as indicated by a negative return value.
#	Line 89 | Line 89 | import java.util.concurrent.locks.LockSu
89		* also available to abruptly release waiting threads and allow them
90		* to terminate.
91		*
92	<	* <p> <b>Tiering.</b> Phasers may be <em>tiered</em> (i.e.,
92	>	* <p><b>Tiering.</b> Phasers may be <em>tiered</em> (i.e.,
93		* constructed in tree structures) to reduce contention. Phasers with
94		* large numbers of parties that would otherwise experience heavy
95		* synchronization contention costs may instead be set up so that
#	Line 193 | Line 193 | import java.util.concurrent.locks.LockSu
193		*   phaser.arriveAndDeregister();
194		* }}</pre>
195		*
196	–	*
196		* <p>To create a set of {@code n} tasks using a tree of phasers, you
197		* could use code of the following form, assuming a Task class with a
198		* constructor accepting a {@code Phaser} that it registers with upon
#	Line 271 | Line 270 | public class Phaser {
270		private static final int  PHASE_SHIFT     = 32;
271		private static final int  UNARRIVED_MASK  = 0xffff;      // to mask ints
272		private static final long PARTIES_MASK    = 0xffff0000L; // to mask longs
273	+	private static final long COUNTS_MASK     = 0xffffffffL;
274		private static final long TERMINATION_BIT = 1L << 63;
275
276		// some special values
277		private static final int  ONE_ARRIVAL     = 1;
278		private static final int  ONE_PARTY       = 1 << PARTIES_SHIFT;
279	+	private static final int  ONE_DEREGISTER  = ONE_ARRIVAL|ONE_PARTY;
280		private static final int  EMPTY           = 1;
281
282		// The following unpacking methods are usually manually inlined
283
284		private static int unarrivedOf(long s) {
285		int counts = (int)s;
286	<	return (counts == EMPTY) ? 0 : counts & UNARRIVED_MASK;
286	>	return (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
287		}
288
289		private static int partiesOf(long s) {
#	Line 343 | Line 344 | public class Phaser {
344		* Manually tuned to speed up and minimize race windows for the
345		* common case of just decrementing unarrived field.
346		*
347	<	* @param deregister false for arrive, true for arriveAndDeregister
347	>	* @param adjust value to subtract from state;
348	>	*               ONE_ARRIVAL for arrive,
349	>	*               ONE_DEREGISTER for arriveAndDeregister
350		*/
351	<	private int doArrive(boolean deregister) {
349	<	int adj = deregister ? ONE_ARRIVAL|ONE_PARTY : ONE_ARRIVAL;
351	>	private int doArrive(int adjust) {
352		final Phaser root = this.root;
353		for (;;) {
354		long s = (root == this) ? state : reconcileState();
355		int phase = (int)(s >>> PHASE_SHIFT);
354	–	int counts = (int)s;
355	–	int unarrived = (counts & UNARRIVED_MASK) - 1;
356		if (phase < 0)
357		return phase;
358	<	else if (counts == EMPTY || unarrived < 0) {
359	<	if (root == this || reconcileState() == s)
360	<	throw new IllegalStateException(badArrive(s));
361	<	}
362	<	else if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s-=adj)) {
363	<	if (unarrived == 0) {
358	>	int counts = (int)s;
359	>	int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
360	>	if (unarrived <= 0)
361	>	throw new IllegalStateException(badArrive(s));
362	>	if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s-=adjust)) {
363	>	if (unarrived == 1) {
364		long n = s & PARTIES_MASK;  // base of next state
365		int nextUnarrived = (int)n >>> PARTIES_SHIFT;
366	<	if (root != this)
367	<	return parent.doArrive(nextUnarrived == 0);
368	<	if (onAdvance(phase, nextUnarrived))
369	<	n |= TERMINATION_BIT;
370	<	else if (nextUnarrived == 0)
371	<	n |= EMPTY;
366	>	if (root == this) {
367	>	if (onAdvance(phase, nextUnarrived))
368	>	n |= TERMINATION_BIT;
369	>	else if (nextUnarrived == 0)
370	>	n |= EMPTY;
371	>	else
372	>	n |= nextUnarrived;
373	>	int nextPhase = (phase + 1) & MAX_PHASE;
374	>	n |= (long)nextPhase << PHASE_SHIFT;
375	>	UNSAFE.compareAndSwapLong(this, stateOffset, s, n);
376	>	releaseWaiters(phase);
377	>	}
378	>	else if (nextUnarrived == 0) { // propagate deregistration
379	>	phase = parent.doArrive(ONE_DEREGISTER);
380	>	UNSAFE.compareAndSwapLong(this, stateOffset,
381	>	s, s | EMPTY);
382	>	}
383		else
384	<	n |= nextUnarrived;
374	<	n |= (long)((phase + 1) & MAX_PHASE) << PHASE_SHIFT;
375	<	UNSAFE.compareAndSwapLong(this, stateOffset, s, n);
376	<	releaseWaiters(phase);
384	>	phase = parent.doArrive(ONE_ARRIVAL);
385		}
386		return phase;
387		}
#	Line 388 | Line 396 | public class Phaser {
396		*/
397		private int doRegister(int registrations) {
398		// adjustment to state
399	<	long adj = ((long)registrations << PARTIES_SHIFT) | registrations;
399	>	long adjust = ((long)registrations << PARTIES_SHIFT) | registrations;
400		final Phaser parent = this.parent;
401		int phase;
402		for (;;) {
403	<	long s = state;
403	>	long s = (parent == null) ? state : reconcileState();
404		int counts = (int)s;
405		int parties = counts >>> PARTIES_SHIFT;
406		int unarrived = counts & UNARRIVED_MASK;
407		if (registrations > MAX_PARTIES - parties)
408		throw new IllegalStateException(badRegister(s));
409	<	else if ((phase = (int)(s >>> PHASE_SHIFT)) < 0)
409	>	phase = (int)(s >>> PHASE_SHIFT);
410	>	if (phase < 0)
411		break;
412	<	else if (counts != EMPTY) {             // not 1st registration
412	>	if (counts != EMPTY) {                  // not 1st registration
413		if (parent == null || reconcileState() == s) {
414		if (unarrived == 0)             // wait out advance
415		root.internalAwaitAdvance(phase, null);
416		else if (UNSAFE.compareAndSwapLong(this, stateOffset,
417	<	s, s + adj))
417	>	s, s + adjust))
418		break;
419		}
420		}
421		else if (parent == null) {              // 1st root registration
422	<	long next = ((long)phase << PHASE_SHIFT) | adj;
422	>	long next = ((long)phase << PHASE_SHIFT) | adjust;
423		if (UNSAFE.compareAndSwapLong(this, stateOffset, s, next))
424		break;
425		}
426		else {
427		synchronized (this) {               // 1st sub registration
428		if (state == s) {               // recheck under lock
429	<	parent.doRegister(1);
430	<	do {                        // force current phase
429	>	phase = parent.doRegister(1);
430	>	if (phase < 0)
431	>	break;
432	>	// finish registration whenever parent registration
433	>	// succeeded, even when racing with termination,
434	>	// since these are part of the same "transaction".
435	>	while (!UNSAFE.compareAndSwapLong
436	>	(this, stateOffset, s,
437	>	((long)phase << PHASE_SHIFT) | adjust)) {
438	>	s = state;
439		phase = (int)(root.state >>> PHASE_SHIFT);
440	<	// assert phase < 0 || (int)state == EMPTY;
441	<	} while (!UNSAFE.compareAndSwapLong
425	<	(this, stateOffset, state,
426	<	((long)phase << PHASE_SHIFT) | adj));
440	>	// assert (int)s == EMPTY;
441	>	}
442		break;
443		}
444		}
#	Line 438 | Line 453 | public class Phaser {
453		* subphasers have not yet done so, in which case they must finish
454		* their own advance by setting unarrived to parties (or if
455		* parties is zero, resetting to unregistered EMPTY state).
441	–	* However, this method may also be called when "floating"
442	–	* subphasers with possibly some unarrived parties are merely
443	–	* catching up to current phase, in which case counts are
444	–	* unaffected.
456		*
457		* @return reconciled state
458		*/
#	Line 449 | Line 460 | public class Phaser {
460		final Phaser root = this.root;
461		long s = state;
462		if (root != this) {
463	<	int phase, u, p;
464	<	// CAS root phase with current parties; possibly trip unarrived
463	>	int phase, p;
464	>	// CAS to root phase with current parties, tripping unarrived
465		while ((phase = (int)(root.state >>> PHASE_SHIFT)) !=
466		(int)(s >>> PHASE_SHIFT) &&
467		!UNSAFE.compareAndSwapLong
468		(this, stateOffset, s,
469		s = (((long)phase << PHASE_SHIFT) |
470	<	(s & PARTIES_MASK) |
471	<	((p = (int)s >>> PARTIES_SHIFT) == 0 ? EMPTY :
472	<	(u = (int)s & UNARRIVED_MASK) == 0 ? p : u))))
470	>	((phase < 0) ? (s & COUNTS_MASK) :
471	>	(((p = (int)s >>> PARTIES_SHIFT) == 0) ? EMPTY :
472	>	((s & PARTIES_MASK) | p))))))
473		s = state;
474		}
475		return s;
#	Line 590 | Line 601 | public class Phaser {
601		* of unarrived parties would become negative
602		*/
603		public int arrive() {
604	<	return doArrive(false);
604	>	return doArrive(ONE_ARRIVAL);
605		}
606
607		/**
#	Line 610 | Line 621 | public class Phaser {
621		* of registered or unarrived parties would become negative
622		*/
623		public int arriveAndDeregister() {
624	<	return doArrive(true);
624	>	return doArrive(ONE_DEREGISTER);
625		}
626
627		/**
#	Line 637 | Line 648 | public class Phaser {
648		for (;;) {
649		long s = (root == this) ? state : reconcileState();
650		int phase = (int)(s >>> PHASE_SHIFT);
640	–	int counts = (int)s;
641	–	int unarrived = (counts & UNARRIVED_MASK) - 1;
651		if (phase < 0)
652		return phase;
653	<	else if (counts == EMPTY || unarrived < 0) {
654	<	if (reconcileState() == s)
655	<	throw new IllegalStateException(badArrive(s));
656	<	}
657	<	else if (UNSAFE.compareAndSwapLong(this, stateOffset, s,
658	<	s -= ONE_ARRIVAL)) {
659	<	if (unarrived != 0)
653	>	int counts = (int)s;
654	>	int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
655	>	if (unarrived <= 0)
656	>	throw new IllegalStateException(badArrive(s));
657	>	if (UNSAFE.compareAndSwapLong(this, stateOffset, s,
658	>	s -= ONE_ARRIVAL)) {
659	>	if (unarrived > 1)
660		return root.internalAwaitAdvance(phase, null);
661		if (root != this)
662		return parent.arriveAndAwaitAdvance();
#	Line 780 | Line 789 | public class Phaser {
789		if (UNSAFE.compareAndSwapLong(root, stateOffset,
790		s, s | TERMINATION_BIT)) {
791		// signal all threads
792	<	releaseWaiters(0);
793	<	releaseWaiters(1);
792	>	releaseWaiters(0); // Waiters on evenQ
793	>	releaseWaiters(1); // Waiters on oddQ
794		return;
795		}
796		}
#	Line 987 | Line 996 | public class Phaser {
996
997		/**
998		* Possibly blocks and waits for phase to advance unless aborted.
999	<	* Call only from root node.
999	>	* Call only on root phaser.
1000		*
1001		* @param phase current phase
1002		* @param node if non-null, the wait node to track interrupt and timeout;
#	Line 995 | Line 1004 | public class Phaser {
1004		* @return current phase
1005		*/
1006		private int internalAwaitAdvance(int phase, QNode node) {
1007	+	// assert root == this;
1008		releaseWaiters(phase-1);          // ensure old queue clean
1009		boolean queued = false;           // true when node is enqueued
1010		int lastUnarrived = 0;            // to increase spins upon change
#	Line 1131 | Line 1141 | public class Phaser {
1141		private static sun.misc.Unsafe getUnsafe() {
1142		try {
1143		return sun.misc.Unsafe.getUnsafe();
1144	<	} catch (SecurityException se) {
1145	<	try {
1146	<	return java.security.AccessController.doPrivileged
1147	<	(new java.security
1148	<	.PrivilegedExceptionAction<sun.misc.Unsafe>() {
1149	<	public sun.misc.Unsafe run() throws Exception {
1150	<	java.lang.reflect.Field f = sun.misc
1151	<	.Unsafe.class.getDeclaredField("theUnsafe");
1152	<	f.setAccessible(true);
1153	<	return (sun.misc.Unsafe) f.get(null);
1154	<	}});
1155	<	} catch (java.security.PrivilegedActionException e) {
1156	<	throw new RuntimeException("Could not initialize intrinsics",
1157	<	e.getCause());
1158	<	}
1144	>	} catch (SecurityException tryReflectionInstead) {}
1145	>	try {
1146	>	return java.security.AccessController.doPrivileged
1147	>	(new java.security.PrivilegedExceptionAction<sun.misc.Unsafe>() {
1148	>	public sun.misc.Unsafe run() throws Exception {
1149	>	Class<sun.misc.Unsafe> k = sun.misc.Unsafe.class;
1150	>	for (java.lang.reflect.Field f : k.getDeclaredFields()) {
1151	>	f.setAccessible(true);
1152	>	Object x = f.get(null);
1153	>	if (k.isInstance(x))
1154	>	return k.cast(x);
1155	>	}
1156	>	throw new NoSuchFieldError("the Unsafe");
1157	>	}});
1158	>	} catch (java.security.PrivilegedActionException e) {
1159	>	throw new RuntimeException("Could not initialize intrinsics",
1160	>	e.getCause());
1161		}
1162		}
1163		}

Diff Legend

–	Removed lines
+	Added lines
<	Changed lines
>	Changed lines