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.75 by dl, Wed Sep 21 12:30:39 2011 UTC vs.
Revision 1.80 by jsr166, Sun Sep 13 16:28:14 2015 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 271 | Line 271 | public class Phaser {
271		private static final int  PHASE_SHIFT     = 32;
272		private static final int  UNARRIVED_MASK  = 0xffff;      // to mask ints
273		private static final long PARTIES_MASK    = 0xffff0000L; // to mask longs
274	+	private static final long COUNTS_MASK     = 0xffffffffL;
275		private static final long TERMINATION_BIT = 1L << 63;
276
277		// some special values
278		private static final int  ONE_ARRIVAL     = 1;
279		private static final int  ONE_PARTY       = 1 << PARTIES_SHIFT;
280	+	private static final int  ONE_DEREGISTER  = ONE_ARRIVAL|ONE_PARTY;
281		private static final int  EMPTY           = 1;
282
283		// The following unpacking methods are usually manually inlined
284
285		private static int unarrivedOf(long s) {
286		int counts = (int)s;
287	<	return (counts == EMPTY) ? 0 : counts & UNARRIVED_MASK;
287	>	return (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
288		}
289
290		private static int partiesOf(long s) {
#	Line 343 | Line 345 | public class Phaser {
345		* Manually tuned to speed up and minimize race windows for the
346		* common case of just decrementing unarrived field.
347		*
348	<	* @param deregister false for arrive, true for arriveAndDeregister
348	>	* @param adjust value to subtract from state;
349	>	*               ONE_ARRIVAL for arrive,
350	>	*               ONE_DEREGISTER for arriveAndDeregister
351		*/
352	<	private int doArrive(boolean deregister) {
349	<	int adj = deregister ? ONE_ARRIVAL|ONE_PARTY : ONE_ARRIVAL;
352	>	private int doArrive(int adjust) {
353		final Phaser root = this.root;
354		for (;;) {
355		long s = (root == this) ? state : reconcileState();
356		int phase = (int)(s >>> PHASE_SHIFT);
354	–	int counts = (int)s;
355	–	int unarrived = (counts & UNARRIVED_MASK) - 1;
357		if (phase < 0)
358		return phase;
359	<	else if (counts == EMPTY || unarrived < 0) {
360	<	if (root == this || reconcileState() == s)
361	<	throw new IllegalStateException(badArrive(s));
362	<	}
363	<	else if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s-=adj)) {
364	<	long n = s & PARTIES_MASK;  // base of next state
365	<	int nextUnarrived = (int)n >>> PARTIES_SHIFT;
366	<	if (unarrived == 0) {
359	>	int counts = (int)s;
360	>	int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
361	>	if (unarrived <= 0)
362	>	throw new IllegalStateException(badArrive(s));
363	>	if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s-=adjust)) {
364	>	if (unarrived == 1) {
365	>	long n = s & PARTIES_MASK;  // base of next state
366	>	int nextUnarrived = (int)n >>> PARTIES_SHIFT;
367		if (root == this) {
368		if (onAdvance(phase, nextUnarrived))
369		n |= TERMINATION_BIT;
#	Line 370 | Line 371 | public class Phaser {
371		n |= EMPTY;
372		else
373		n |= nextUnarrived;
374	<	n |= (long)((phase + 1) & MAX_PHASE) << PHASE_SHIFT;
374	>	int nextPhase = (phase + 1) & MAX_PHASE;
375	>	n |= (long)nextPhase << PHASE_SHIFT;
376		UNSAFE.compareAndSwapLong(this, stateOffset, s, n);
377	+	releaseWaiters(phase);
378		}
379		else if (nextUnarrived == 0) { // propagate deregistration
380	<	phase = parent.doArrive(true);
380	>	phase = parent.doArrive(ONE_DEREGISTER);
381		UNSAFE.compareAndSwapLong(this, stateOffset,
382		s, s | EMPTY);
383		}
384		else
385	<	phase = parent.doArrive(false);
383	<	releaseWaiters(phase);
385	>	phase = parent.doArrive(ONE_ARRIVAL);
386		}
387		return phase;
388		}
#	Line 395 | Line 397 | public class Phaser {
397		*/
398		private int doRegister(int registrations) {
399		// adjustment to state
400	<	long adj = ((long)registrations << PARTIES_SHIFT) | registrations;
400	>	long adjust = ((long)registrations << PARTIES_SHIFT) | registrations;
401		final Phaser parent = this.parent;
402		int phase;
403		for (;;) {
#	Line 405 | Line 407 | public class Phaser {
407		int unarrived = counts & UNARRIVED_MASK;
408		if (registrations > MAX_PARTIES - parties)
409		throw new IllegalStateException(badRegister(s));
410	<	else if ((phase = (int)(s >>> PHASE_SHIFT)) < 0)
410	>	phase = (int)(s >>> PHASE_SHIFT);
411	>	if (phase < 0)
412		break;
413	<	else if (counts != EMPTY) {             // not 1st registration
413	>	if (counts != EMPTY) {                  // not 1st registration
414		if (parent == null || reconcileState() == s) {
415		if (unarrived == 0)             // wait out advance
416		root.internalAwaitAdvance(phase, null);
417		else if (UNSAFE.compareAndSwapLong(this, stateOffset,
418	<	s, s + adj))
418	>	s, s + adjust))
419		break;
420		}
421		}
422		else if (parent == null) {              // 1st root registration
423	<	long next = ((long)phase << PHASE_SHIFT) | adj;
423	>	long next = ((long)phase << PHASE_SHIFT) | adjust;
424		if (UNSAFE.compareAndSwapLong(this, stateOffset, s, next))
425		break;
426		}
427		else {
428		synchronized (this) {               // 1st sub registration
429		if (state == s) {               // recheck under lock
430	<	parent.doRegister(1);
431	<	do {                        // force current phase
430	>	phase = parent.doRegister(1);
431	>	if (phase < 0)
432	>	break;
433	>	// finish registration whenever parent registration
434	>	// succeeded, even when racing with termination,
435	>	// since these are part of the same "transaction".
436	>	while (!UNSAFE.compareAndSwapLong
437	>	(this, stateOffset, s,
438	>	((long)phase << PHASE_SHIFT) | adjust)) {
439	>	s = state;
440		phase = (int)(root.state >>> PHASE_SHIFT);
441	<	// assert phase < 0 || (int)state == EMPTY;
442	<	} while (!UNSAFE.compareAndSwapLong
432	<	(this, stateOffset, state,
433	<	((long)phase << PHASE_SHIFT) | adj));
441	>	// assert (int)s == EMPTY;
442	>	}
443		break;
444		}
445		}
#	Line 445 | Line 454 | public class Phaser {
454		* subphasers have not yet done so, in which case they must finish
455		* their own advance by setting unarrived to parties (or if
456		* parties is zero, resetting to unregistered EMPTY state).
448	–	* However, this method may also be called when "floating"
449	–	* subphasers with possibly some unarrived parties are merely
450	–	* catching up to current phase, in which case counts are
451	–	* unaffected.
457		*
458		* @return reconciled state
459		*/
#	Line 456 | Line 461 | public class Phaser {
461		final Phaser root = this.root;
462		long s = state;
463		if (root != this) {
464	<	int phase, u, p;
465	<	// CAS root phase with current parties; possibly trip unarrived
464	>	int phase, p;
465	>	// CAS to root phase with current parties, tripping unarrived
466		while ((phase = (int)(root.state >>> PHASE_SHIFT)) !=
467		(int)(s >>> PHASE_SHIFT) &&
468		!UNSAFE.compareAndSwapLong
469		(this, stateOffset, s,
470		s = (((long)phase << PHASE_SHIFT) |
471	<	(s & PARTIES_MASK) |
472	<	((p = (int)s >>> PARTIES_SHIFT) == 0 ? EMPTY :
473	<	((u = (int)s & UNARRIVED_MASK) == 0 && phase >= 0) ?
469	<	p : u))))
471	>	((phase < 0) ? (s & COUNTS_MASK) :
472	>	(((p = (int)s >>> PARTIES_SHIFT) == 0) ? EMPTY :
473	>	((s & PARTIES_MASK) | p))))))
474		s = state;
475		}
476		return s;
#	Line 598 | Line 602 | public class Phaser {
602		* of unarrived parties would become negative
603		*/
604		public int arrive() {
605	<	return doArrive(false);
605	>	return doArrive(ONE_ARRIVAL);
606		}
607
608		/**
#	Line 618 | Line 622 | public class Phaser {
622		* of registered or unarrived parties would become negative
623		*/
624		public int arriveAndDeregister() {
625	<	return doArrive(true);
625	>	return doArrive(ONE_DEREGISTER);
626		}
627
628		/**
#	Line 645 | Line 649 | public class Phaser {
649		for (;;) {
650		long s = (root == this) ? state : reconcileState();
651		int phase = (int)(s >>> PHASE_SHIFT);
648	–	int counts = (int)s;
649	–	int unarrived = (counts & UNARRIVED_MASK) - 1;
652		if (phase < 0)
653		return phase;
654	<	else if (counts == EMPTY || unarrived < 0) {
655	<	if (reconcileState() == s)
656	<	throw new IllegalStateException(badArrive(s));
657	<	}
658	<	else if (UNSAFE.compareAndSwapLong(this, stateOffset, s,
659	<	s -= ONE_ARRIVAL)) {
660	<	if (unarrived != 0)
654	>	int counts = (int)s;
655	>	int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
656	>	if (unarrived <= 0)
657	>	throw new IllegalStateException(badArrive(s));
658	>	if (UNSAFE.compareAndSwapLong(this, stateOffset, s,
659	>	s -= ONE_ARRIVAL)) {
660	>	if (unarrived > 1)
661		return root.internalAwaitAdvance(phase, null);
662		if (root != this)
663		return parent.arriveAndAwaitAdvance();
#	Line 788 | Line 790 | public class Phaser {
790		if (UNSAFE.compareAndSwapLong(root, stateOffset,
791		s, s | TERMINATION_BIT)) {
792		// signal all threads
793	<	releaseWaiters(0);
794	<	releaseWaiters(1);
793	>	releaseWaiters(0); // Waiters on evenQ
794	>	releaseWaiters(1); // Waiters on oddQ
795		return;
796		}
797		}
#	Line 995 | Line 997 | public class Phaser {
997
998		/**
999		* Possibly blocks and waits for phase to advance unless aborted.
1000	<	* Call only from root node.
1000	>	* Call only on root phaser.
1001		*
1002		* @param phase current phase
1003		* @param node if non-null, the wait node to track interrupt and timeout;
#	Line 1003 | Line 1005 | public class Phaser {
1005		* @return current phase
1006		*/
1007		private int internalAwaitAdvance(int phase, QNode node) {
1008	+	// assert root == this;
1009		releaseWaiters(phase-1);          // ensure old queue clean
1010		boolean queued = false;           // true when node is enqueued
1011		int lastUnarrived = 0;            // to increase spins upon change
#	Line 1139 | Line 1142 | public class Phaser {
1142		private static sun.misc.Unsafe getUnsafe() {
1143		try {
1144		return sun.misc.Unsafe.getUnsafe();
1145	<	} catch (SecurityException se) {
1146	<	try {
1147	<	return java.security.AccessController.doPrivileged
1148	<	(new java.security
1149	<	.PrivilegedExceptionAction<sun.misc.Unsafe>() {
1150	<	public sun.misc.Unsafe run() throws Exception {
1151	<	java.lang.reflect.Field f = sun.misc
1152	<	.Unsafe.class.getDeclaredField("theUnsafe");
1153	<	f.setAccessible(true);
1154	<	return (sun.misc.Unsafe) f.get(null);
1155	<	}});
1156	<	} catch (java.security.PrivilegedActionException e) {
1157	<	throw new RuntimeException("Could not initialize intrinsics",
1158	<	e.getCause());
1159	<	}
1145	>	} catch (SecurityException tryReflectionInstead) {}
1146	>	try {
1147	>	return java.security.AccessController.doPrivileged
1148	>	(new java.security.PrivilegedExceptionAction<sun.misc.Unsafe>() {
1149	>	public sun.misc.Unsafe run() throws Exception {
1150	>	Class<sun.misc.Unsafe> k = sun.misc.Unsafe.class;
1151	>	for (java.lang.reflect.Field f : k.getDeclaredFields()) {
1152	>	f.setAccessible(true);
1153	>	Object x = f.get(null);
1154	>	if (k.isInstance(x))
1155	>	return k.cast(x);
1156	>	}
1157	>	throw new NoSuchFieldError("the Unsafe");
1158	>	}});
1159	>	} catch (java.security.PrivilegedActionException e) {
1160	>	throw new RuntimeException("Could not initialize intrinsics",
1161	>	e.getCause());
1162		}
1163		}
1164		}

Diff Legend

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