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.54 by dl, Sat Nov 13 13:10:04 2010 UTC vs.
Revision 1.55 by dl, Mon Nov 15 12:51:54 2010 UTC

#	Line 240 | Line 240 | public class Phaser {
240		*/
241		private volatile long state;
242
243	<	private static final int  MAX_COUNT      = 0xffff;
243	>	private static final int  MAX_PARTIES    = 0xffff;
244		private static final int  MAX_PHASE      = 0x7fffffff;
245		private static final int  PARTIES_SHIFT  = 16;
246		private static final int  PHASE_SHIFT    = 32;
247	<	private static final long UNARRIVED_MASK = 0xffffL;
248	<	private static final long PARTIES_MASK   = 0xffff0000L;
247	>	private static final int  UNARRIVED_MASK = 0xffff;
248	>	private static final int  PARTIES_MASK   = 0xffff0000;
249	>	private static final long LPARTIES_MASK  = 0xffff0000L; // long version
250		private static final long ONE_ARRIVAL    = 1L;
251		private static final long ONE_PARTY      = 1L << PARTIES_SHIFT;
252		private static final long TERMINATION_PHASE  = -1L << PHASE_SHIFT;
#	Line 253 | Line 254 | public class Phaser {
254		// The following unpacking methods are usually manually inlined
255
256		private static int unarrivedOf(long s) {
257	<	return (int) (s & UNARRIVED_MASK);
257	>	return ((int) s) & UNARRIVED_MASK;
258		}
259
260		private static int partiesOf(long s) {
261	<	return ((int) (s & PARTIES_MASK)) >>> PARTIES_SHIFT;
261	>	return (((int) s) & PARTIES_MASK) >>> PARTIES_SHIFT;
262		}
263
264		private static int phaseOf(long s) {
#	Line 307 | Line 308 | public class Phaser {
308		int phase, unarrived;
309		if ((phase = (int)((s = state) >>> PHASE_SHIFT)) < 0)
310		return phase;
311	<	else if ((unarrived = (int)(s & UNARRIVED_MASK)) == 0)
311	>	else if ((unarrived = ((int)s) & UNARRIVED_MASK) == 0)
312		checkBadArrive(s);
313	<	else if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s -= adj)){
313	>	else if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s-=adj)) {
314		if (unarrived == 1) {
315		Phaser par;
316	<	long p = s & PARTIES_MASK; // unshifted parties field
316	>	long p = s & LPARTIES_MASK; // unshifted parties field
317		long lu = p >>> PARTIES_SHIFT;
318		int u = (int)lu;
319		int nextPhase = (phase + 1) & MAX_PHASE;
320		long next = ((long)nextPhase << PHASE_SHIFT) | p | lu;
321		if ((par = parent) == null) {
322	<	UNSAFE.compareAndSwapLong
323	<	(this, stateOffset, s, onAdvance(phase, u)?
324	<	next | TERMINATION_PHASE : next);
322	>	if (onAdvance(phase, u))
323	>	next |= TERMINATION_PHASE; // obliterate phase
324	>	UNSAFE.compareAndSwapLong(this, stateOffset, s, next);
325		releaseWaiters(phase);
326		}
327		else {
#	Line 363 | Line 364 | public class Phaser {
364		long s = par == null? state : reconcileState();
365		if ((phase = (int)(s >>> PHASE_SHIFT)) < 0)
366		return phase;
367	<	if ((parties = ((int)(s & PARTIES_MASK)) >>> PARTIES_SHIFT) != 0 &&
368	<	(s & UNARRIVED_MASK) == 0)
367	>	if ((parties = (((int)s) & PARTIES_MASK) >>> PARTIES_SHIFT) != 0 &&
368	>	(((int)s) & UNARRIVED_MASK) == 0)
369		internalAwaitAdvance(phase, null); // wait for onAdvance
370	<	else if (parties + registrations > MAX_COUNT)
370	>	else if (parties + registrations > MAX_PARTIES)
371		throw new IllegalStateException(badRegister(s));
372		else if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s + adj))
373		return phase;
#	Line 374 | Line 375 | public class Phaser {
375		}
376
377		/**
378	<	* Returns message string for bounds exceptions on registration
378	>	* Returns message string for out of bounds exceptions on registration.
379		*/
380		private String badRegister(long s) {
381		return "Attempt to register more than " +
382	<	MAX_COUNT + " parties for " + stateToString(s);
382	>	MAX_PARTIES + " parties for " + stateToString(s);
383		}
384
385		/**
386	<	* Recursively resolves lagged phase propagation from root if
386	<	* necessary.
386	>	* Recursively resolves lagged phase propagation from root if necessary.
387		*/
388		private long reconcileState() {
389		Phaser par = parent;
#	Line 398 | Line 398 | public class Phaser {
398		return s;
399		long pState = par.parent == null? par.state : par.reconcileState();
400		if (state == s) {
401	<	if ((rPhase < 0 || (s & UNARRIVED_MASK) == 0) &&
401	>	if ((rPhase < 0 || (((int)s) & UNARRIVED_MASK) == 0) &&
402		((pPhase = (int)(pState >>> PHASE_SHIFT)) < 0 ||
403		pPhase == ((phase + 1) & MAX_PHASE)))
404		UNSAFE.compareAndSwapLong
405		(this, stateOffset, s,
406		(((long) pPhase) << PHASE_SHIFT) |
407	<	(u = s & PARTIES_MASK) |
407	>	(u = s & LPARTIES_MASK) |
408		(u >>> PARTIES_SHIFT)); // reset unarrived to parties
409		else
410		releaseWaiters(phase); // help release others
#	Line 457 | Line 457 | public class Phaser {
457		* or greater than the maximum number of parties supported
458		*/
459		public Phaser(Phaser parent, int parties) {
460	<	if (parties < 0 || parties > MAX_COUNT)
460	>	if (parties >>> PARTIES_SHIFT != 0)
461		throw new IllegalArgumentException("Illegal number of parties");
462		int phase;
463		this.parent = parent;
#	Line 475 | Line 475 | public class Phaser {
475		phase = 0;
476		}
477		long p = (long)parties;
478	<	this.state = (((long) phase) << PHASE_SHIFT) | p | (p << PARTIES_SHIFT);
478	>	this.state = (((long)phase) << PHASE_SHIFT) | p | (p << PARTIES_SHIFT);
479		}
480
481		/**
#	Line 505 | Line 505 | public class Phaser {
505		public int bulkRegister(int parties) {
506		if (parties < 0)
507		throw new IllegalArgumentException();
508	<	if (parties > MAX_COUNT)
508	>	if (parties > MAX_PARTIES)
509		throw new IllegalStateException(badRegister(state));
510		if (parties == 0)
511		return getPhase();
#	Line 573 | Line 573 | public class Phaser {
573		* if terminated or argument is negative
574		*/
575		public int awaitAdvance(int phase) {
576	+	int p;
577		if (phase < 0)
578		return phase;
579	<	int p = (int)((parent==null? state : reconcileState()) >>> PHASE_SHIFT);
580	<	if (p != phase)
579	>	else if ((p = (int)((parent == null? state : reconcileState())
580	>	>>> PHASE_SHIFT)) == phase)
581	>	return internalAwaitAdvance(phase, null);
582	>	else
583		return p;
581	–	return internalAwaitAdvance(phase, null);
584		}
585
586		/**
#	Line 597 | Line 599 | public class Phaser {
599		*/
600		public int awaitAdvanceInterruptibly(int phase)
601		throws InterruptedException {
602	+	int p;
603		if (phase < 0)
604		return phase;
605	<	int p = (int)((parent==null? state : reconcileState()) >>> PHASE_SHIFT);
606	<	if (p != phase)
607	<	return p;
608	<	QNode node = new QNode(this, phase, true, false, 0L);
609	<	p = internalAwaitAdvance(phase, node);
610	<	if (node.wasInterrupted)
611	<	throw new InterruptedException();
612	<	else
610	<	return p;
605	>	if ((p = (int)((parent == null? state : reconcileState())
606	>	>>> PHASE_SHIFT)) == phase) {
607	>	QNode node = new QNode(this, phase, true, false, 0L);
608	>	p = internalAwaitAdvance(phase, node);
609	>	if (node.wasInterrupted)
610	>	throw new InterruptedException();
611	>	}
612	>	return p;
613		}
614
615		/**
#	Line 634 | Line 636 | public class Phaser {
636		long timeout, TimeUnit unit)
637		throws InterruptedException, TimeoutException {
638		long nanos = unit.toNanos(timeout);
639	+	int p;
640		if (phase < 0)
641		return phase;
642	<	int p = (int)((parent==null? state : reconcileState()) >>> PHASE_SHIFT);
643	<	if (p != phase)
644	<	return p;
645	<	QNode node = new QNode(this, phase, true, true, nanos);
646	<	p = internalAwaitAdvance(phase, node);
647	<	if (node.wasInterrupted)
648	<	throw new InterruptedException();
649	<	else if (p == phase)
650	<	throw new TimeoutException();
651	<	else
649	<	return p;
642	>	if ((p = (int)((parent == null? state : reconcileState())
643	>	>>> PHASE_SHIFT)) == phase) {
644	>	QNode node = new QNode(this, phase, true, true, nanos);
645	>	p = internalAwaitAdvance(phase, node);
646	>	if (node.wasInterrupted)
647	>	throw new InterruptedException();
648	>	else if (p == phase)
649	>	throw new TimeoutException();
650	>	}
651	>	return p;
652		}
653
654		/**
655	<	* Forces this barrier to enter termination state. Counts of
656	<	* arrived and registered parties are unaffected. If this phaser
657	<	* has a parent, it too is terminated. This method may be useful
658	<	* for coordinating recovery after one or more tasks encounter
659	<	* unexpected exceptions.
655	>	* Forces this barrier to enter termination state.  Counts of
656	>	* arrived and registered parties are unaffected.  If this phaser
657	>	* is a member of a tiered set of phasers, then all of the phasers
658	>	* in the set are terminated.  If this phaser is already
659	>	* terminated, this method has no effect.  This method may be
660	>	* useful for coordinating recovery after one or more tasks
661	>	* encounter unexpected exceptions.
662		*/
663		public void forceTermination() {
664	<	Phaser r = root;    // force at root then reconcile
664	>	// Only need to change root state
665	>	final Phaser root = this.root;
666		long s;
667	<	while ((s = r.state) >= 0)
668	<	UNSAFE.compareAndSwapLong(r, stateOffset, s, s | TERMINATION_PHASE);
669	<	reconcileState();
670	<	releaseWaiters(0); // signal all threads
671	<	releaseWaiters(1);
667	>	while ((s = root.state) >= 0) {
668	>	if (UNSAFE.compareAndSwapLong(root, stateOffset,
669	>	s, s | TERMINATION_PHASE)) {
670	>	releaseWaiters(0); // signal all threads
671	>	releaseWaiters(1);
672	>	return;
673	>	}
674	>	}
675		}
676
677		/**
#	Line 834 | Line 842 | public class Phaser {
842		* avoid it when threads regularly arrive: When a thread in
843		* internalAwaitAdvance notices another arrival before blocking,
844		* and there appear to be enough CPUs available, it spins
845	<	* SPINS_PER_ARRIVAL more times before continuing to try to
846	<	* block. The value trades off good-citizenship vs big unnecessary
847	<	* slowdowns.
845	>	* SPINS_PER_ARRIVAL more times before blocking. Plus, even on
846	>	* uniprocessors, there is at least one intervening Thread.yield
847	>	* before blocking. The value trades off good-citizenship vs big
848	>	* unnecessary slowdowns.
849		*/
850	<	static final int SPINS_PER_ARRIVAL = NCPU < 2? 1 : 1 << 8;
850	>	static final int SPINS_PER_ARRIVAL = (NCPU < 2) ? 1 : 1 << 8;
851
852		/**
853		* Possibly blocks and waits for phase to advance unless aborted.
#	Line 863 | Line 872 | public class Phaser {
872		releaseWaiters(phase);
873		return p;
874		}
875	<	else if ((unarrived = (int)(s & UNARRIVED_MASK)) != lastUnarrived) {
876	<	if ((lastUnarrived = unarrived) < NCPU)
868	<	spins += SPINS_PER_ARRIVAL;
869	<	}
870	<	else if (unarrived == 0 && (par = current.parent) != null) {
875	>	else if ((unarrived = ((int)s) & UNARRIVED_MASK) == 0 &&
876	>	(par = current.parent) != null) {
877		current = par;       // if all arrived, use parent
878		par = par.parent;
879		lastUnarrived = -1;
880		}
881	<	else if (spins > 0)
882	<	--spins;
881	>	else if (unarrived != lastUnarrived) {
882	>	if ((lastUnarrived = unarrived) < NCPU)
883	>	spins += SPINS_PER_ARRIVAL;
884	>	}
885	>	else if (spins > 0) {
886	>	if (--spins == (SPINS_PER_ARRIVAL >>> 1))
887	>	Thread.yield();  // yield midway through spin
888	>	}
889		else if (node == null)   // must be noninterruptible
890		node = new QNode(this, phase, false, false, 0L);
891		else if (node.isReleasable()) {

Diff Legend

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