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.68 by dl, Sat Dec 4 15:25:08 2010 UTC vs.
Revision 1.74 by jsr166, Wed Jun 1 21:04:30 2011 UTC

#	Line 1 | Line 1
1		/*
2		* Written by Doug Lea with assistance from members of JCP JSR-166
3		* Expert Group and released to the public domain, as explained at
4	<	* http://creativecommons.org/licenses/publicdomain
4	>	* http://creativecommons.org/publicdomain/zero/1.0/
5		*/
6
7		package jsr166y;
#	Line 77 | Line 77 | import java.util.concurrent.locks.LockSu
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
81	<	* value. Similarly, attempts to register upon termination have no
82	<	* effect.  Termination is triggered when an invocation of {@code
83	<	* onAdvance} returns {@code true}. The default implementation returns
84	<	* {@code true} if a deregistration has caused the number of
85	<	* registered parties to become zero.  As illustrated below, when
86	<	* phasers control actions with a fixed number of iterations, it is
87	<	* often convenient to override this method to cause termination when
88	<	* the current phase number reaches a threshold. Method {@link
89	<	* #forceTermination} is also available to abruptly release waiting
90	<	* threads and allow them to terminate.
80	>	* waiting for advance, as indicated by a negative return value.
81	>	* Similarly, attempts to register upon termination have no effect.
82	>	* Termination is triggered when an invocation of {@code onAdvance}
83	>	* returns {@code true}. The default implementation returns {@code
84	>	* true} if a deregistration has caused the number of registered
85	>	* parties to become zero.  As illustrated below, when phasers control
86	>	* actions with a fixed number of iterations, it is often convenient
87	>	* to override this method to cause termination when the current phase
88	>	* number reaches a threshold. Method {@link #forceTermination} is
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.,
93		* constructed in tree structures) to reduce contention. Phasers with
#	Line 130 | Line 130 | import java.util.concurrent.locks.LockSu
130		* void runTasks(List<Runnable> tasks) {
131		*   final Phaser phaser = new Phaser(1); // "1" to register self
132		*   // create and start threads
133	<	*   for (Runnable task : tasks) {
133	>	*   for (final Runnable task : tasks) {
134		*     phaser.register();
135		*     new Thread() {
136		*       public void run() {
#	Line 237 | Line 237 | public class Phaser {
237		*/
238
239		/**
240	<	* Primary state representation, holding four fields:
240	>	* Primary state representation, holding four bit-fields:
241		*
242	<	* * unarrived -- the number of parties yet to hit barrier (bits  0-15)
243	<	* * parties -- the number of parties to wait              (bits 16-31)
244	<	* * phase -- the generation of the barrier                (bits 32-62)
245	<	* * terminated -- set if barrier is terminated            (bit  63 / sign)
242	>	* unarrived  -- the number of parties yet to hit barrier (bits  0-15)
243	>	* parties    -- the number of parties to wait            (bits 16-31)
244	>	* phase      -- the generation of the barrier            (bits 32-62)
245	>	* terminated -- set if barrier is terminated             (bit  63 / sign)
246		*
247		* Except that a phaser with no registered parties is
248	<	* distinguished with the otherwise illegal state of having zero
248	>	* distinguished by the otherwise illegal state of having zero
249		* parties and one unarrived parties (encoded as EMPTY below).
250		*
251		* To efficiently maintain atomicity, these values are packed into
#	Line 260 | Line 260 | public class Phaser {
260		* parent.
261		*
262		* The phase of a subphaser is allowed to lag that of its
263	<	* ancestors until it is actually accessed.  Method reconcileState
264	<	* is usually attempted only only when the number of unarrived
265	<	* parties appears to be zero, which indicates a potential lag in
266	<	* updating phase after the root advanced.
263	>	* ancestors until it is actually accessed -- see method
264	>	* reconcileState.
265		*/
266		private volatile long state;
267
268		private static final int  MAX_PARTIES     = 0xffff;
269	<	private static final int  MAX_PHASE       = 0x7fffffff;
269	>	private static final int  MAX_PHASE       = Integer.MAX_VALUE;
270		private static final int  PARTIES_SHIFT   = 16;
271		private static final int  PHASE_SHIFT     = 32;
272		private static final int  UNARRIVED_MASK  = 0xffff;      // to mask ints
#	Line 292 | Line 290 | public class Phaser {
290		}
291
292		private static int phaseOf(long s) {
293	<	return (int) (s >>> PHASE_SHIFT);
293	>	return (int)(s >>> PHASE_SHIFT);
294		}
295
296		private static int arrivedOf(long s) {
#	Line 364 | Line 362 | public class Phaser {
362		else if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s-=adj)) {
363		if (unarrived == 0) {
364		long n = s & PARTIES_MASK;  // base of next state
365	<	int nextUnarrived = ((int)n) >>> PARTIES_SHIFT;
365	>	int nextUnarrived = (int)n >>> PARTIES_SHIFT;
366		if (root != this)
367		return parent.doArrive(nextUnarrived == 0);
368		if (onAdvance(phase, nextUnarrived))
#	Line 373 | Line 371 | public class Phaser {
371		n |= EMPTY;
372		else
373		n |= nextUnarrived;
374	<	n |= ((long)((phase + 1) & MAX_PHASE)) << PHASE_SHIFT;
374	>	n |= (long)((phase + 1) & MAX_PHASE) << PHASE_SHIFT;
375		UNSAFE.compareAndSwapLong(this, stateOffset, s, n);
376		releaseWaiters(phase);
377		}
#	Line 391 | Line 389 | public class Phaser {
389		private int doRegister(int registrations) {
390		// adjustment to state
391		long adj = ((long)registrations << PARTIES_SHIFT) | registrations;
392	<	Phaser par = parent;
392	>	final Phaser parent = this.parent;
393		int phase;
394		for (;;) {
395		long s = state;
#	Line 403 | Line 401 | public class Phaser {
401		else if ((phase = (int)(s >>> PHASE_SHIFT)) < 0)
402		break;
403		else if (counts != EMPTY) {             // not 1st registration
404	<	if (par == null || reconcileState() == s) {
404	>	if (parent == null || reconcileState() == s) {
405		if (unarrived == 0)             // wait out advance
406		root.internalAwaitAdvance(phase, null);
407		else if (UNSAFE.compareAndSwapLong(this, stateOffset,
#	Line 411 | Line 409 | public class Phaser {
409		break;
410		}
411		}
412	<	else if (par == null) {                 // 1st root registration
413	<	long next = (((long) phase) << PHASE_SHIFT) | adj;
412	>	else if (parent == null) {              // 1st root registration
413	>	long next = ((long)phase << PHASE_SHIFT) | adj;
414		if (UNSAFE.compareAndSwapLong(this, stateOffset, s, next))
415		break;
416		}
417		else {
418		synchronized (this) {               // 1st sub registration
419		if (state == s) {               // recheck under lock
420	<	par.doRegister(1);
420	>	parent.doRegister(1);
421		do {                        // force current phase
422		phase = (int)(root.state >>> PHASE_SHIFT);
423		// assert phase < 0 || (int)state == EMPTY;
424		} while (!UNSAFE.compareAndSwapLong
425		(this, stateOffset, state,
426	<	(((long) phase) << PHASE_SHIFT) | adj));
426	>	((long)phase << PHASE_SHIFT) | adj));
427		break;
428		}
429		}
#	Line 436 | Line 434 | public class Phaser {
434
435		/**
436		* Resolves lagged phase propagation from root if necessary.
437	+	* Reconciliation normally occurs when root has advanced but
438	+	* subphasers have not yet done so, in which case they must finish
439	+	* their own advance by setting unarrived to parties (or if
440	+	* 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.
445	+	*
446	+	* @return reconciled state
447		*/
448		private long reconcileState() {
449	<	Phaser rt = root;
449	>	final Phaser root = this.root;
450		long s = state;
451	<	if (rt != this) {
452	<	int phase;
453	<	while ((phase = (int)(rt.state >>> PHASE_SHIFT)) !=
454	<	(int)(s >>> PHASE_SHIFT)) {
455	<	// assert phase < 0 || unarrivedOf(s) == 0
456	<	long t;                             // to reread s
457	<	long p = s & PARTIES_MASK;          // unshifted parties field
458	<	long n = (((long) phase) << PHASE_SHIFT) | p;
459	<	if (phase >= 0) {
460	<	if (p == 0L)
461	<	n |= EMPTY;                 // reset to empty
462	<	else
455	<	n |= p >>> PARTIES_SHIFT;   // set unarr to parties
456	<	}
457	<	if ((t = state) == s &&
458	<	UNSAFE.compareAndSwapLong(this, stateOffset, s, s = n))
459	<	break;
460	<	s = t;
461	<	}
451	>	if (root != this) {
452	>	int phase, u, p;
453	>	// CAS root phase with current parties; possibly trip unarrived
454	>	while ((phase = (int)(root.state >>> PHASE_SHIFT)) !=
455	>	(int)(s >>> PHASE_SHIFT) &&
456	>	!UNSAFE.compareAndSwapLong
457	>	(this, stateOffset, s,
458	>	s = (((long)phase << PHASE_SHIFT) |
459	>	(s & PARTIES_MASK) |
460	>	((p = (int)s >>> PARTIES_SHIFT) == 0 ? EMPTY :
461	>	(u = (int)s & UNARRIVED_MASK) == 0 ? p : u))))
462	>	s = state;
463		}
464		return s;
465		}
#	Line 524 | Line 525 | public class Phaser {
525		this.evenQ = new AtomicReference<QNode>();
526		this.oddQ = new AtomicReference<QNode>();
527		}
528	<	this.state = (parties == 0) ? (long) EMPTY :
529	<	((((long) phase) << PHASE_SHIFT) |
530	<	(((long) parties) << PARTIES_SHIFT) |
531	<	((long) parties));
528	>	this.state = (parties == 0) ? (long)EMPTY :
529	>	((long)phase << PHASE_SHIFT) |
530	>	((long)parties << PARTIES_SHIFT) |
531	>	((long)parties);
532		}
533
534		/**
#	Line 541 | Line 542 | public class Phaser {
542		*
543		* @return the arrival phase number to which this registration
544		* applied.  If this value is negative, then this phaser has
545	<	* terminated, in which casem registration has no effect.
545	>	* terminated, in which case registration has no effect.
546		* @throws IllegalStateException if attempting to register more
547		* than the maximum supported number of parties
548		*/
#	Line 563 | Line 564 | public class Phaser {
564		* advance to the next phase
565		* @return the arrival phase number to which this registration
566		* applied.  If this value is negative, then this phaser has
567	<	* terminated, in which casem registration has no effect.
567	>	* terminated, in which case registration has no effect.
568		* @throws IllegalStateException if attempting to register more
569		* than the maximum supported number of parties
570		* @throws IllegalArgumentException if {@code parties < 0}
#	Line 651 | Line 652 | public class Phaser {
652		if (root != this)
653		return parent.arriveAndAwaitAdvance();
654		long n = s & PARTIES_MASK;  // base of next state
655	<	int nextUnarrived = ((int)n) >>> PARTIES_SHIFT;
655	>	int nextUnarrived = (int)n >>> PARTIES_SHIFT;
656		if (onAdvance(phase, nextUnarrived))
657		n |= TERMINATION_BIT;
658		else if (nextUnarrived == 0)
#	Line 682 | Line 683 | public class Phaser {
683		*/
684		public int awaitAdvance(int phase) {
685		final Phaser root = this.root;
686	<	int p = (int)((root == this? state : reconcileState()) >>> PHASE_SHIFT);
686	>	long s = (root == this) ? state : reconcileState();
687	>	int p = (int)(s >>> PHASE_SHIFT);
688		if (phase < 0)
689		return phase;
690		if (p == phase)
#	Line 708 | Line 710 | public class Phaser {
710		public int awaitAdvanceInterruptibly(int phase)
711		throws InterruptedException {
712		final Phaser root = this.root;
713	<	int p = (int)((root == this? state : reconcileState()) >>> PHASE_SHIFT);
713	>	long s = (root == this) ? state : reconcileState();
714	>	int p = (int)(s >>> PHASE_SHIFT);
715		if (phase < 0)
716		return phase;
717		if (p == phase) {
#	Line 745 | Line 748 | public class Phaser {
748		throws InterruptedException, TimeoutException {
749		long nanos = unit.toNanos(timeout);
750		final Phaser root = this.root;
751	<	int p = (int)((root == this? state : reconcileState()) >>> PHASE_SHIFT);
751	>	long s = (root == this) ? state : reconcileState();
752	>	int p = (int)(s >>> PHASE_SHIFT);
753		if (phase < 0)
754		return phase;
755		if (p == phase) {
#	Line 773 | Line 777 | public class Phaser {
777		final Phaser root = this.root;
778		long s;
779		while ((s = root.state) >= 0) {
780	<	long next = (s & ~((long)UNARRIVED_MASK)) | TERMINATION_BIT;
781	<	if (UNSAFE.compareAndSwapLong(root, stateOffset, s, next)) {
780	>	if (UNSAFE.compareAndSwapLong(root, stateOffset,
781	>	s, s | TERMINATION_BIT)) {
782		// signal all threads
783		releaseWaiters(0);
784		releaseWaiters(1);
#	Line 807 | Line 811 | public class Phaser {
811
812		/**
813		* Returns the number of registered parties that have arrived at
814	<	* the current phase of this phaser.
814	>	* the current phase of this phaser. If this phaser has terminated,
815	>	* the returned value is meaningless and arbitrary.
816		*
817		* @return the number of arrived parties
818		*/
#	Line 817 | Line 822 | public class Phaser {
822
823		/**
824		* Returns the number of registered parties that have not yet
825	<	* arrived at the current phase of this phaser.
825	>	* arrived at the current phase of this phaser. If this phaser has
826	>	* terminated, the returned value is meaningless and arbitrary.
827		*
828		* @return the number of unarrived parties
829		*/
#	Line 1102 | Line 1108 | public class Phaser {
1108
1109		// Unsafe mechanics
1110
1111	<	private static final sun.misc.Unsafe UNSAFE = getUnsafe();
1112	<	private static final long stateOffset =
1113	<	objectFieldOffset("state", Phaser.class);
1108	<
1109	<	private static long objectFieldOffset(String field, Class<?> klazz) {
1111	>	private static final sun.misc.Unsafe UNSAFE;
1112	>	private static final long stateOffset;
1113	>	static {
1114		try {
1115	<	return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
1116	<	} catch (NoSuchFieldException e) {
1117	<	// Convert Exception to corresponding Error
1118	<	NoSuchFieldError error = new NoSuchFieldError(field);
1119	<	error.initCause(e);
1120	<	throw error;
1115	>	UNSAFE = getUnsafe();
1116	>	Class<?> k = Phaser.class;
1117	>	stateOffset = UNSAFE.objectFieldOffset
1118	>	(k.getDeclaredField("state"));
1119	>	} catch (Exception e) {
1120	>	throw new Error(e);
1121		}
1122		}
1123

Diff Legend

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