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.52 by dl, Sat Nov 13 01:27:13 2010 UTC vs.
Revision 1.56 by dl, Wed Nov 17 10:48:59 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_SHIFT;
262		}
263
264		private static int phaseOf(long s) {
#	Line 302 | Line 303 | public class Phaser {
303		* ONE_ARRIVAL|ONE_PARTY (for arriveAndDeregister)
304		*/
305		private int doArrive(long adj) {
306	<	long s;
307	<	int phase, unarrived;
308	<	while ((phase = (int)((s = state) >>> PHASE_SHIFT)) >= 0) {
309	<	if ((unarrived = (int)(s & UNARRIVED_MASK)) != 0) {
310	<	if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s -= adj)) {
311	<	if (unarrived == 1) {
312	<	Phaser par;
313	<	long p = s & PARTIES_MASK; // unshifted parties field
314	<	long lu = p >>> PARTIES_SHIFT;
315	<	int u = (int)lu;
316	<	int nextPhase = (phase + 1) & MAX_PHASE;
317	<	long next = ((long)nextPhase << PHASE_SHIFT) | p | lu;
318	<	if ((par = parent) == null) {
319	<	UNSAFE.compareAndSwapLong
320	<	(this, stateOffset, s, onAdvance(phase, u)?
321	<	next | TERMINATION_PHASE : next);
322	<	releaseWaiters(phase);
323	<	}
324	<	else {
325	<	par.doArrive(u == 0?
326	<	ONE_ARRIVAL|ONE_PARTY : ONE_ARRIVAL);
327	<	if ((int)(par.state >>> PHASE_SHIFT) != nextPhase ||
328	<	((int)(state >>> PHASE_SHIFT) != nextPhase &&
329	<	!UNSAFE.compareAndSwapLong(this, stateOffset,
330	<	s, next)))
331	<	reconcileState();
332	<	}
306	>	for (;;) {
307	>	long s;
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)
312	>	checkBadArrive(s);
313	>	else if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s-=adj)) {
314	>	if (unarrived == 1) {
315	>	Phaser par;
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	>	if (onAdvance(phase, u))
323	>	next |= TERMINATION_PHASE; // obliterate phase
324	>	UNSAFE.compareAndSwapLong(this, stateOffset, s, next);
325	>	releaseWaiters(phase);
326	>	}
327	>	else {
328	>	par.doArrive(u == 0?
329	>	ONE_ARRIVAL|ONE_PARTY : ONE_ARRIVAL);
330	>	if ((int)(par.state >>> PHASE_SHIFT) != nextPhase ||
331	>	((int)(state >>> PHASE_SHIFT) != nextPhase &&
332	>	!UNSAFE.compareAndSwapLong(this, stateOffset,
333	>	s, next)))
334	>	reconcileState();
335		}
333	–	break;
336		}
337	+	return phase;
338		}
336	–	else if (state == s && reconcileState() == s) // recheck
337	–	throw new IllegalStateException(badArrive());
339		}
339	–	return phase;
340		}
341
342		/**
343	<	* Returns message string for bounds exceptions on arrival.
344	<	* Declared out of-line from doArrive to reduce string op bulk.
343	>	* Rechecks state and throws bounds exceptions on arrival -- called
344	>	* only if unarrived is apparently zero.
345		*/
346	<	private String badArrive() {
347	<	return ("Attempted arrival of unregistered party for " +
348	<	this.toString());
346	>	private void checkBadArrive(long s) {
347	>	if (reconcileState() == s)
348	>	throw new IllegalStateException
349	>	("Attempted arrival of unregistered party for " +
350	>	stateToString(s));
351		}
352
353		/**
#	Line 357 | Line 359 | public class Phaser {
359		long adj = (long)registrations; // adjustment to state
360		adj |= adj << PARTIES_SHIFT;
361		Phaser par = parent;
362	<	long s;
363	<	int phase;
364	<	while ((phase = (int)((s = (par == null? state : reconcileState()))
365	<	>>> PHASE_SHIFT)) >= 0) {
366	<	int parties = ((int)(s & PARTIES_MASK)) >>> PARTIES_SHIFT;
367	<	if (parties != 0 && (s & UNARRIVED_MASK) == 0)
362	>	for (;;) {
363	>	int phase, parties;
364	>	long s = par == null? state : reconcileState();
365	>	if ((phase = (int)(s >>> PHASE_SHIFT)) < 0)
366	>	return phase;
367	>	if ((parties = (int)s >>> PARTIES_SHIFT) != 0 &&
368	>	((int)s & UNARRIVED_MASK) == 0)
369		internalAwaitAdvance(phase, null); // wait for onAdvance
370	<	else if (parties + registrations > MAX_COUNT)
371	<	throw new IllegalStateException(badRegister());
370	>	else if (parties + registrations > MAX_PARTIES)
371	>	throw new IllegalStateException(badRegister(s));
372		else if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s + adj))
373	<	break;
373	>	return phase;
374		}
372	–	return phase;
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() {
381	<	return ("Attempt to register more than " + MAX_COUNT + " parties for "+
382	<	this.toString());
380	>	private String badRegister(long s) {
381	>	return "Attempt to register more than " +
382	>	MAX_PARTIES + " parties for " + stateToString(s);
383		}
384
385		/**
386	<	* Recursively resolves lagged phase propagation from root if
385	<	* necessary.
386	>	* Recursively resolves lagged phase propagation from root if necessary.
387		*/
388		private long reconcileState() {
389		Phaser par = parent;
390		if (par == null)
391		return state;
392		Phaser rt = root;
393	<	long s;
394	<	int phase, rPhase;
395	<	while ((phase = (int)((s = state) >>> PHASE_SHIFT)) >= 0 &&
396	<	(rPhase = (int)(rt.state >>> PHASE_SHIFT)) != phase) {
397	<	if (rPhase < 0 || (s & UNARRIVED_MASK) == 0) {
398	<	long ps = par.parent == null? par.state : par.reconcileState();
399	<	int pPhase = (int)(ps >>> PHASE_SHIFT);
400	<	if (pPhase < 0 || pPhase == ((phase + 1) & MAX_PHASE)) {
401	<	if (state != s)
402	<	continue;
403	<	long p = s & PARTIES_MASK;
404	<	long next = ((((long) pPhase) << PHASE_SHIFT) |
405	<	(p >>> PARTIES_SHIFT) | p);
406	<	if (UNSAFE.compareAndSwapLong(this, stateOffset, s, next))
407	<	return next;
408	<	}
393	>	for (;;) {
394	>	long s, u;
395	>	int phase, rPhase, pPhase;
396	>	if ((phase = (int)((s = state)>>> PHASE_SHIFT)) < 0 ||
397	>	(rPhase = (int)(rt.state >>> PHASE_SHIFT)) == phase)
398	>	return s;
399	>	long pState = par.parent == null? par.state : par.reconcileState();
400	>	if (state == s) {
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 & LPARTIES_MASK) |
408	>	(u >>> PARTIES_SHIFT)); // reset unarrived to parties
409	>	else
410	>	releaseWaiters(phase); // help release others
411		}
409	–	if (state == s)
410	–	releaseWaiters(phase); // help release others
412		}
412	–	return s;
413		}
414
415		/**
#	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)
509	<	throw new IllegalStateException(badRegister());
508	>	if (parties > MAX_PARTIES)
509	>	throw new IllegalStateException(badRegister(state));
510		if (parties == 0)
511		return getPhase();
512		return doRegister(parties);
#	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 674 | Line 682 | public class Phaser {
682		* @return the phase number, or a negative value if terminated
683		*/
684		public final int getPhase() {
685	<	return (int)((parent == null? state : reconcileState()) >>> PHASE_SHIFT);
685	>	return (int)((parent==null? state : reconcileState()) >>> PHASE_SHIFT);
686		}
687
688		/**
#	Line 683 | Line 691 | public class Phaser {
691		* @return the number of parties
692		*/
693		public int getRegisteredParties() {
694	<	return partiesOf(parent == null? state : reconcileState());
694	>	return partiesOf(parent==null? state : reconcileState());
695		}
696
697		/**
#	Line 693 | Line 701 | public class Phaser {
701		* @return the number of arrived parties
702		*/
703		public int getArrivedParties() {
704	<	return arrivedOf(parent == null? state : reconcileState());
704	>	return arrivedOf(parent==null? state : reconcileState());
705		}
706
707		/**
#	Line 703 | Line 711 | public class Phaser {
711		* @return the number of unarrived parties
712		*/
713		public int getUnarrivedParties() {
714	<	return unarrivedOf(parent == null? state : reconcileState());
714	>	return unarrivedOf(parent==null? state : reconcileState());
715		}
716
717		/**
#	Line 779 | Line 787 | public class Phaser {
787		* @return a string identifying this barrier, as well as its state
788		*/
789		public String toString() {
790	<	long s = reconcileState();
790	>	return stateToString(reconcileState());
791	>	}
792	>
793	>	/**
794	>	* Implementation of toString and string-based error messages
795	>	*/
796	>	private String stateToString(long s) {
797		return super.toString() +
798		"[phase = " + phaseOf(s) +
799		" parties = " + partiesOf(s) +
800		" arrived = " + arrivedOf(s) + "]";
801		}
802
803	+	// Waiting mechanics
804	+
805		/**
806		* Removes and signals threads from queue for phase
807		*/
#	Line 826 | 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.
854		*
855		* @param phase current phase
856	<	* @param node if nonnull, the wait node to track interrupt and timeout;
856	>	* @param node if non-null, the wait node to track interrupt and timeout;
857		* if null, denotes noninterruptible wait
858		* @return current phase
859		*/
860		private int internalAwaitAdvance(int phase, QNode node) {
861		Phaser current = this;       // to eventually wait at root if tiered
862	<	Phaser par = parent;
846	<	boolean queued = false;
847	<	int spins = SPINS_PER_ARRIVAL;
862	>	boolean queued = false;      // true when node is enqueued
863		int lastUnarrived = -1;      // to increase spins upon change
864	<	long s;
865	<	int p;
866	<	while ((p = (int)((s = current.state) >>> PHASE_SHIFT)) == phase) {
867	<	int unarrived = (int)(s & UNARRIVED_MASK);
868	<	if (unarrived != lastUnarrived) {
869	<	if ((lastUnarrived = unarrived) < NCPU)
870	<	spins += SPINS_PER_ARRIVAL;
864	>	int spins = SPINS_PER_ARRIVAL;
865	>	for (;;) {
866	>	int p, unarrived;
867	>	Phaser par;
868	>	long s = current.state;
869	>	if ((p = (int)(s >>> PHASE_SHIFT)) != phase) {
870	>	if (node != null)
871	>	node.onRelease();
872	>	releaseWaiters(phase);
873	>	return p;
874		}
875	<	else if (unarrived == 0 && par != 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 (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 (spins > 0)
862	<	--spins;
863	<	else if (node == null)
889	>	else if (node == null)   // must be noninterruptible
890		node = new QNode(this, phase, false, false, 0L);
891	<	else if (node.isReleasable())
892	<	break;
891	>	else if (node.isReleasable()) {
892	>	if ((int)(reconcileState() >>> PHASE_SHIFT) == phase)
893	>	return phase;    // aborted
894	>	}
895		else if (!queued)
896		queued = tryEnqueue(phase, node);
897		else {
#	Line 874 | Line 902 | public class Phaser {
902		}
903		}
904		}
877	–	if (node != null) {
878	–	if (node.thread != null)
879	–	node.thread = null;
880	–	if (!node.interruptible && node.wasInterrupted)
881	–	Thread.currentThread().interrupt();
882	–	}
883	–	if (p == phase)
884	–	p = (int)(reconcileState() >>> PHASE_SHIFT);
885	–	if (p != phase)
886	–	releaseWaiters(phase);
887	–	return p;
905		}
906
907		/**
#	Line 956 | Line 973 | public class Phaser {
973		LockSupport.unpark(t);
974		}
975		}
976	+
977	+	void onRelease() { // actions upon return from internalAwaitAdvance
978	+	if (!interruptible && wasInterrupted)
979	+	Thread.currentThread().interrupt();
980	+	if (thread != null)
981	+	thread = null;
982	+	}
983	+
984		}
985
986		// Unsafe mechanics

Diff Legend

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