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.72 by dl, Mon May 16 11:41:14 2011 UTC vs.
Revision 1.77 by jsr166, Mon Oct 17 23:37:19 2011 UTC

#	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 266 | Line 266 | public class Phaser {
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 long PHASE_MASK      = -1L << PHASE_SHIFT;
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 291 | Line 292 | public class Phaser {
292		}
293
294		private static int phaseOf(long s) {
295	<	return (int) (s >>> PHASE_SHIFT);
295	>	return (int)(s >>> PHASE_SHIFT);
296		}
297
298		private static int arrivedOf(long s) {
#	Line 344 | 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) {
350	<	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);
355	–	int counts = (int)s;
356	–	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	<	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	<	return parent.doArrive(nextUnarrived == 0);
369	<	if (onAdvance(phase, nextUnarrived))
370	<	n |= TERMINATION_BIT;
371	<	else if (nextUnarrived == 0)
372	<	n |= EMPTY;
366	>	int nextUnarrived = (int)n >>> PARTIES_SHIFT;
367	>	if (root == this) {
368	>	if (onAdvance(phase, nextUnarrived))
369	>	n |= TERMINATION_BIT;
370	>	else if (nextUnarrived == 0)
371	>	n |= EMPTY;
372	>	else
373	>	n |= nextUnarrived;
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(ONE_DEREGISTER);
381	>	UNSAFE.compareAndSwapLong(this, stateOffset,
382	>	s, s | EMPTY);
383	>	}
384		else
385	<	n |= nextUnarrived;
375	<	n |= ((long)((phase + 1) & MAX_PHASE)) << PHASE_SHIFT;
376	<	UNSAFE.compareAndSwapLong(this, stateOffset, s, n);
377	<	releaseWaiters(phase);
385	>	phase = parent.doArrive(ONE_ARRIVAL);
386		}
387		return phase;
388		}
#	Line 389 | Line 397 | public class Phaser {
397		*/
398		private int doRegister(int registrations) {
399		// adjustment to state
400	<	long adj = ((long)registrations << PARTIES_SHIFT) | registrations;
401	<	Phaser par = parent;
400	>	long adjust = ((long)registrations << PARTIES_SHIFT) | registrations;
401	>	final Phaser parent = this.parent;
402		int phase;
403		for (;;) {
404	<	long s = state;
404	>	long s = (parent == null) ? state : reconcileState();
405		int counts = (int)s;
406		int parties = counts >>> PARTIES_SHIFT;
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
414	<	if (par == null || reconcileState() == s) {
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 (par == null) {                 // 1st root registration
423	<	long next = (((long) phase) << PHASE_SHIFT) | adj;
422	>	else if (parent == null) {              // 1st root registration
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	<	par.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
426	<	(this, stateOffset, state,
427	<	(((long) phase) << PHASE_SHIFT) | adj));
441	>	// assert (int)s == EMPTY;
442	>	}
443		break;
444		}
445		}
#	Line 439 | 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).
442	–	* However, this method may also be called when "floating"
443	–	* subphasers with possibly some unarrived parties are merely
444	–	* catching up to current phase, in which case counts are
445	–	* unaffected.
457		*
458		* @return reconciled state
459		*/
#	Line 450 | 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) | (s & PARTIES_MASK) |
471	<	((p = (int)s >>> PARTIES_SHIFT) == 0 ? EMPTY :
472	<	(u = (int)s & UNARRIVED_MASK) == 0 ? p : u))))
470	>	s = (((long)phase << PHASE_SHIFT) |
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 525 | Line 537 | public class Phaser {
537		this.evenQ = new AtomicReference<QNode>();
538		this.oddQ = new AtomicReference<QNode>();
539		}
540	<	this.state = (parties == 0) ? (long) EMPTY :
541	<	((((long) phase) << PHASE_SHIFT) |
542	<	(((long) parties) << PARTIES_SHIFT) |
543	<	((long) parties));
540	>	this.state = (parties == 0) ? (long)EMPTY :
541	>	((long)phase << PHASE_SHIFT) |
542	>	((long)parties << PARTIES_SHIFT) |
543	>	((long)parties);
544		}
545
546		/**
#	Line 590 | 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 610 | 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 637 | Line 649 | public class Phaser {
649		for (;;) {
650		long s = (root == this) ? state : reconcileState();
651		int phase = (int)(s >>> PHASE_SHIFT);
640	–	int counts = (int)s;
641	–	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();
664		long n = s & PARTIES_MASK;  // base of next state
665	<	int nextUnarrived = ((int)n) >>> PARTIES_SHIFT;
665	>	int nextUnarrived = (int)n >>> PARTIES_SHIFT;
666		if (onAdvance(phase, nextUnarrived))
667		n |= TERMINATION_BIT;
668		else if (nextUnarrived == 0)
#	Line 780 | 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 987 | 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 995 | 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 1108 | Line 1119 | public class Phaser {
1119
1120		// Unsafe mechanics
1121
1122	<	private static final sun.misc.Unsafe UNSAFE = getUnsafe();
1123	<	private static final long stateOffset =
1124	<	objectFieldOffset("state", Phaser.class);
1114	<
1115	<	private static long objectFieldOffset(String field, Class<?> klazz) {
1122	>	private static final sun.misc.Unsafe UNSAFE;
1123	>	private static final long stateOffset;
1124	>	static {
1125		try {
1126	<	return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
1127	<	} catch (NoSuchFieldException e) {
1128	<	// Convert Exception to corresponding Error
1129	<	NoSuchFieldError error = new NoSuchFieldError(field);
1130	<	error.initCause(e);
1131	<	throw error;
1126	>	UNSAFE = getUnsafe();
1127	>	Class<?> k = Phaser.class;
1128	>	stateOffset = UNSAFE.objectFieldOffset
1129	>	(k.getDeclaredField("state"));
1130	>	} catch (Exception e) {
1131	>	throw new Error(e);
1132		}
1133		}
1134

Diff Legend

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