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.69 by jsr166, Sat Dec 4 22:00:05 2010 UTC vs.
Revision 1.75 by dl, Wed Sep 21 12:30:39 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 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 362 | Line 360 | public class Phaser {
360		throw new IllegalStateException(badArrive(s));
361		}
362		else if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s-=adj)) {
363	+	long n = s & PARTIES_MASK;  // base of next state
364	+	int nextUnarrived = (int)n >>> PARTIES_SHIFT;
365		if (unarrived == 0) {
366	<	long n = s & PARTIES_MASK;  // base of next state
367	<	int nextUnarrived = ((int)n) >>> PARTIES_SHIFT;
368	<	if (root != this)
369	<	return parent.doArrive(nextUnarrived == 0);
370	<	if (onAdvance(phase, nextUnarrived))
371	<	n |= TERMINATION_BIT;
372	<	else if (nextUnarrived == 0)
373	<	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	>	n |= (long)((phase + 1) & MAX_PHASE) << PHASE_SHIFT;
374	>	UNSAFE.compareAndSwapLong(this, stateOffset, s, n);
375	>	}
376	>	else if (nextUnarrived == 0) { // propagate deregistration
377	>	phase = parent.doArrive(true);
378	>	UNSAFE.compareAndSwapLong(this, stateOffset,
379	>	s, s | EMPTY);
380	>	}
381		else
382	<	n |= nextUnarrived;
376	<	n |= ((long)((phase + 1) & MAX_PHASE)) << PHASE_SHIFT;
377	<	UNSAFE.compareAndSwapLong(this, stateOffset, s, n);
382	>	phase = parent.doArrive(false);
383		releaseWaiters(phase);
384		}
385		return phase;
#	Line 391 | Line 396 | public class Phaser {
396		private int doRegister(int registrations) {
397		// adjustment to state
398		long adj = ((long)registrations << PARTIES_SHIFT) | registrations;
399	<	Phaser par = parent;
399	>	final Phaser parent = this.parent;
400		int phase;
401		for (;;) {
402	<	long s = state;
402	>	long s = (parent == null) ? state : reconcileState();
403		int counts = (int)s;
404		int parties = counts >>> PARTIES_SHIFT;
405		int unarrived = counts & UNARRIVED_MASK;
#	Line 403 | Line 408 | public class Phaser {
408		else if ((phase = (int)(s >>> PHASE_SHIFT)) < 0)
409		break;
410		else if (counts != EMPTY) {             // not 1st registration
411	<	if (par == null || reconcileState() == s) {
411	>	if (parent == null || reconcileState() == s) {
412		if (unarrived == 0)             // wait out advance
413		root.internalAwaitAdvance(phase, null);
414		else if (UNSAFE.compareAndSwapLong(this, stateOffset,
#	Line 411 | Line 416 | public class Phaser {
416		break;
417		}
418		}
419	<	else if (par == null) {                 // 1st root registration
420	<	long next = (((long) phase) << PHASE_SHIFT) | adj;
419	>	else if (parent == null) {              // 1st root registration
420	>	long next = ((long)phase << PHASE_SHIFT) | adj;
421		if (UNSAFE.compareAndSwapLong(this, stateOffset, s, next))
422		break;
423		}
424		else {
425		synchronized (this) {               // 1st sub registration
426		if (state == s) {               // recheck under lock
427	<	par.doRegister(1);
427	>	parent.doRegister(1);
428		do {                        // force current phase
429		phase = (int)(root.state >>> PHASE_SHIFT);
430		// assert phase < 0 || (int)state == EMPTY;
431		} while (!UNSAFE.compareAndSwapLong
432		(this, stateOffset, state,
433	<	(((long) phase) << PHASE_SHIFT) | adj));
433	>	((long)phase << PHASE_SHIFT) | adj));
434		break;
435		}
436		}
#	Line 436 | Line 441 | public class Phaser {
441
442		/**
443		* Resolves lagged phase propagation from root if necessary.
444	+	* Reconciliation normally occurs when root has advanced but
445	+	* subphasers have not yet done so, in which case they must finish
446	+	* their own advance by setting unarrived to parties (or if
447	+	* 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.
452	+	*
453	+	* @return reconciled state
454		*/
455		private long reconcileState() {
456	<	Phaser rt = root;
456	>	final Phaser root = this.root;
457		long s = state;
458	<	if (rt != this) {
459	<	int phase;
460	<	while ((phase = (int)(rt.state >>> PHASE_SHIFT)) !=
461	<	(int)(s >>> PHASE_SHIFT)) {
462	<	// assert phase < 0 || unarrivedOf(s) == 0
463	<	long t;                             // to reread s
464	<	long p = s & PARTIES_MASK;          // unshifted parties field
465	<	long n = (((long) phase) << PHASE_SHIFT) | p;
466	<	if (phase >= 0) {
467	<	if (p == 0L)
468	<	n |= EMPTY;                 // reset to empty
469	<	else
470	<	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	<	}
458	>	if (root != this) {
459	>	int phase, u, p;
460	>	// CAS root phase with current parties; possibly trip unarrived
461	>	while ((phase = (int)(root.state >>> PHASE_SHIFT)) !=
462	>	(int)(s >>> PHASE_SHIFT) &&
463	>	!UNSAFE.compareAndSwapLong
464	>	(this, stateOffset, s,
465	>	s = (((long)phase << PHASE_SHIFT) |
466	>	(s & PARTIES_MASK) |
467	>	((p = (int)s >>> PARTIES_SHIFT) == 0 ? EMPTY :
468	>	((u = (int)s & UNARRIVED_MASK) == 0 && phase >= 0) ?
469	>	p : u))))
470	>	s = state;
471		}
472		return s;
473		}
#	Line 524 | Line 533 | public class Phaser {
533		this.evenQ = new AtomicReference<QNode>();
534		this.oddQ = new AtomicReference<QNode>();
535		}
536	<	this.state = (parties == 0) ? (long) EMPTY :
537	<	((((long) phase) << PHASE_SHIFT) |
538	<	(((long) parties) << PARTIES_SHIFT) |
539	<	((long) parties));
536	>	this.state = (parties == 0) ? (long)EMPTY :
537	>	((long)phase << PHASE_SHIFT) |
538	>	((long)parties << PARTIES_SHIFT) |
539	>	((long)parties);
540		}
541
542		/**
#	Line 651 | Line 660 | public class Phaser {
660		if (root != this)
661		return parent.arriveAndAwaitAdvance();
662		long n = s & PARTIES_MASK;  // base of next state
663	<	int nextUnarrived = ((int)n) >>> PARTIES_SHIFT;
663	>	int nextUnarrived = (int)n >>> PARTIES_SHIFT;
664		if (onAdvance(phase, nextUnarrived))
665		n |= TERMINATION_BIT;
666		else if (nextUnarrived == 0)
#	Line 776 | Line 785 | public class Phaser {
785		final Phaser root = this.root;
786		long s;
787		while ((s = root.state) >= 0) {
788	<	long next = (s & ~((long)UNARRIVED_MASK)) | TERMINATION_BIT;
789	<	if (UNSAFE.compareAndSwapLong(root, stateOffset, s, next)) {
788	>	if (UNSAFE.compareAndSwapLong(root, stateOffset,
789	>	s, s | TERMINATION_BIT)) {
790		// signal all threads
791		releaseWaiters(0);
792		releaseWaiters(1);
#	Line 810 | Line 819 | public class Phaser {
819
820		/**
821		* Returns the number of registered parties that have arrived at
822	<	* the current phase of this phaser.
822	>	* the current phase of this phaser. If this phaser has terminated,
823	>	* the returned value is meaningless and arbitrary.
824		*
825		* @return the number of arrived parties
826		*/
#	Line 820 | Line 830 | public class Phaser {
830
831		/**
832		* Returns the number of registered parties that have not yet
833	<	* arrived at the current phase of this phaser.
833	>	* arrived at the current phase of this phaser. If this phaser has
834	>	* terminated, the returned value is meaningless and arbitrary.
835		*
836		* @return the number of unarrived parties
837		*/
#	Line 1105 | Line 1116 | public class Phaser {
1116
1117		// Unsafe mechanics
1118
1119	<	private static final sun.misc.Unsafe UNSAFE = getUnsafe();
1120	<	private static final long stateOffset =
1121	<	objectFieldOffset("state", Phaser.class);
1111	<
1112	<	private static long objectFieldOffset(String field, Class<?> klazz) {
1119	>	private static final sun.misc.Unsafe UNSAFE;
1120	>	private static final long stateOffset;
1121	>	static {
1122		try {
1123	<	return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
1124	<	} catch (NoSuchFieldException e) {
1125	<	// Convert Exception to corresponding Error
1126	<	NoSuchFieldError error = new NoSuchFieldError(field);
1127	<	error.initCause(e);
1128	<	throw error;
1123	>	UNSAFE = getUnsafe();
1124	>	Class<?> k = Phaser.class;
1125	>	stateOffset = UNSAFE.objectFieldOffset
1126	>	(k.getDeclaredField("state"));
1127	>	} catch (Exception e) {
1128	>	throw new Error(e);
1129		}
1130		}
1131

Diff Legend

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