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.66 by jsr166, Wed Dec 1 19:12:53 2010 UTC vs.
Revision 1.72 by dl, Mon May 16 11:41:14 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 75 | Line 75 | import java.util.concurrent.locks.LockSu
75		* </ul>
76		*
77		* <p> <b>Termination.</b> A phaser may enter a <em>termination</em>
78	<	* state in which all synchronization methods immediately return
79	<	* without updating phaser state or waiting for advance, and
80	<	* indicating (via a negative phase value) that execution is complete.
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 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
#	Line 96 | Line 97 | import java.util.concurrent.locks.LockSu
97		* increase throughput even though it incurs greater per-operation
98		* overhead.
99		*
100	+	* <p>In a tree of tiered phasers, registration and deregistration of
101	+	* child phasers with their parent are managed automatically.
102	+	* Whenever the number of registered parties of a child phaser becomes
103	+	* non-zero (as established in the {@link #Phaser(Phaser,int)}
104	+	* constructor, {@link #register}, or {@link #bulkRegister}), the
105	+	* child phaser is registered with its parent.  Whenever the number of
106	+	* registered parties becomes zero as the result of an invocation of
107	+	* {@link #arriveAndDeregister}, the child phaser is deregistered
108	+	* from its parent.
109	+	*
110		* <p><b>Monitoring.</b> While synchronization methods may be invoked
111		* only by registered parties, the current state of a phaser may be
112		* monitored by any caller.  At any given moment there are {@link
#	Line 119 | 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 249 | 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
254	<	* parties appears to be zero, which indicates a potential lag in
255	<	* updating phase after the root advanced.
263	>	* ancestors until it is actually accessed -- see method
264	>	* reconcileState.
265		*/
266		private volatile long state;
267
#	Line 260 | Line 269 | public class Phaser {
269		private static final int  MAX_PHASE       = 0x7fffffff;
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;
273		private static final int  UNARRIVED_MASK  = 0xffff;      // to mask ints
274		private static final long PARTIES_MASK    = 0xffff0000L; // to mask longs
275		private static final long TERMINATION_BIT = 1L << 63;
#	Line 277 | Line 287 | public class Phaser {
287		}
288
289		private static int partiesOf(long s) {
290	<	int counts = (int)s;
281	<	return (counts == EMPTY) ? 0 : counts >>> PARTIES_SHIFT;
290	>	return (int)s >>> PARTIES_SHIFT;
291		}
292
293		private static int phaseOf(long s) {
#	Line 339 | Line 348 | public class Phaser {
348		*/
349		private int doArrive(boolean deregister) {
350		int adj = deregister ? ONE_ARRIVAL|ONE_PARTY : ONE_ARRIVAL;
351	<	long s;
352	<	int phase;
353	<	while ((phase = (int)((s = state) >>> PHASE_SHIFT)) >= 0) {
351	>	final Phaser root = this.root;
352	>	for (;;) {
353	>	long s = (root == this) ? state : reconcileState();
354	>	int phase = (int)(s >>> PHASE_SHIFT);
355		int counts = (int)s;
356	<	int unarrived = counts & UNARRIVED_MASK;
357	<	if (counts == EMPTY || unarrived == 0) {
358	<	if (reconcileState() == 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 == 1) {
365	<	long n = s & PARTIES_MASK;       // unshifted parties field
366	<	int u = ((int)n) >>> PARTIES_SHIFT;
367	<	Phaser par = parent;
368	<	if (par != null) {
369	<	par.doArrive(u == 0);
370	<	reconcileState();
371	<	}
372	<	else {
373	<	n |= (((long)((phase+1) & MAX_PHASE)) << PHASE_SHIFT);
374	<	if (onAdvance(phase, u))
375	<	n |= TERMINATION_BIT;
376	<	else if (u == 0)
377	<	n |= EMPTY;             // reset to unregistered
366	<	else
367	<	n |= (long)u;           // reset unarr to parties
368	<	// assert state == s || isTerminated();
369	<	UNSAFE.compareAndSwapLong(this, stateOffset, s, n);
370	<	releaseWaiters(phase);
371	<	}
364	>	if (unarrived == 0) {
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;
373	>	else
374	>	n |= nextUnarrived;
375	>	n |= ((long)((phase + 1) & MAX_PHASE)) << PHASE_SHIFT;
376	>	UNSAFE.compareAndSwapLong(this, stateOffset, s, n);
377	>	releaseWaiters(phase);
378		}
379	<	break;
379	>	return phase;
380		}
381		}
376	–	return phase;
382		}
383
384		/**
#	Line 430 | Line 435 | public class Phaser {
435
436		/**
437		* Resolves lagged phase propagation from root if necessary.
438	+	* Reconciliation normally occurs when root has advanced but
439	+	* subphasers have not yet done so, in which case they must finish
440	+	* their own advance by setting unarrived to parties (or if
441	+	* 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.
446	+	*
447	+	* @return reconciled state
448		*/
449		private long reconcileState() {
450	<	Phaser rt = root;
450	>	final Phaser root = this.root;
451		long s = state;
452	<	if (rt != this) {
453	<	int phase;
454	<	while ((phase = (int)(rt.state >>> PHASE_SHIFT)) !=
455	<	(int)(s >>> PHASE_SHIFT)) {
456	<	// assert phase < 0 || unarrivedOf(s) == 0
457	<	long t;                             // to reread s
458	<	long p = s & PARTIES_MASK;          // unshifted parties field
459	<	long n = (((long) phase) << PHASE_SHIFT) | p;
460	<	if (phase >= 0) {
461	<	if (p == 0L)
462	<	n |= EMPTY;                 // reset to empty
448	<	else
449	<	n |= p >>> PARTIES_SHIFT;   // set unarr to parties
450	<	}
451	<	if ((t = state) == s &&
452	<	UNSAFE.compareAndSwapLong(this, stateOffset, s, s = n))
453	<	break;
454	<	s = t;
455	<	}
452	>	if (root != this) {
453	>	int phase, u, p;
454	>	// CAS root phase with current parties; possibly trip unarrived
455	>	while ((phase = (int)(root.state >>> PHASE_SHIFT)) !=
456	>	(int)(s >>> PHASE_SHIFT) &&
457	>	!UNSAFE.compareAndSwapLong
458	>	(this, stateOffset, s,
459	>	s = ((((long) phase) << PHASE_SHIFT) | (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 490 | Line 497 | public class Phaser {
497
498		/**
499		* Creates a new phaser with the given parent and number of
500	<	* registered unarrived parties. Registration and deregistration
501	<	* of this child phaser with its parent are managed automatically.
502	<	* If the given parent is non-null, whenever this child phaser has
496	<	* any registered parties (as established in this constructor,
497	<	* {@link #register}, or {@link #bulkRegister}), this child phaser
498	<	* is registered with its parent. Whenever the number of
499	<	* registered parties becomes zero as the result of an invocation
500	<	* of {@link #arriveAndDeregister}, this child phaser is
501	<	* deregistered from its parent.
500	>	* registered unarrived parties.  When the given parent is non-null
501	>	* and the given number of parties is greater than zero, this
502	>	* child phaser is registered with its parent.
503		*
504		* @param parent the parent phaser
505		* @param parties the number of parties required to advance to the
#	Line 512 | Line 513 | public class Phaser {
513		int phase = 0;
514		this.parent = parent;
515		if (parent != null) {
516	<	Phaser r = parent.root;
517	<	this.root = r;
518	<	this.evenQ = r.evenQ;
519	<	this.oddQ = r.oddQ;
516	>	final Phaser root = parent.root;
517	>	this.root = root;
518	>	this.evenQ = root.evenQ;
519	>	this.oddQ = root.oddQ;
520		if (parties != 0)
521		phase = parent.doRegister(1);
522		}
#	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) :
528	>	this.state = (parties == 0) ? (long) EMPTY :
529		((((long) phase) << PHASE_SHIFT) |
530		(((long) parties) << PARTIES_SHIFT) |
531		((long) parties));
#	Line 535 | Line 536 | public class Phaser {
536		* invocation of {@link #onAdvance} is in progress, this method
537		* may await its completion before returning.  If this phaser has
538		* a parent, and this phaser previously had no registered parties,
539	<	* this phaser is also registered with its parent.
540	<	*
541	<	* @return the arrival phase number to which this registration applied
539	>	* this child phaser is also registered with its parent. If
540	>	* this phaser is terminated, the attempt to register has
541	>	* no effect, and a negative value is returned.
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 case registration has no effect.
546		* @throws IllegalStateException if attempting to register more
547		* than the maximum supported number of parties
548		*/
#	Line 549 | Line 554 | public class Phaser {
554		* Adds the given number of new unarrived parties to this phaser.
555		* If an ongoing invocation of {@link #onAdvance} is in progress,
556		* this method may await its completion before returning.  If this
557	<	* phaser has a parent, and the given number of parities is
558	<	* greater than zero, and this phaser previously had no registered
559	<	* parties, this phaser is also registered with its parent.
557	>	* phaser has a parent, and the given number of parties is greater
558	>	* than zero, and this phaser previously had no registered
559	>	* parties, this child phaser is also registered with its parent.
560	>	* If this phaser is terminated, the attempt to register has no
561	>	* effect, and a negative value is returned.
562		*
563		* @param parties the number of additional parties required to
564		* advance to the next phase
565	<	* @return the arrival phase number to which this registration applied
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 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 617 | Line 626 | public class Phaser {
626		* IllegalStateException} only upon some subsequent operation on
627		* this phaser, if ever.
628		*
629	<	* @return the arrival phase number, or a negative number if terminated
629	>	* @return the arrival phase number, or the (negative)
630	>	* {@linkplain #getPhase() current phase} if terminated
631		* @throws IllegalStateException if not terminated and the number
632		* of unarrived parties would become negative
633		*/
634		public int arriveAndAwaitAdvance() {
635	<	return awaitAdvance(doArrive(false));
635	>	// Specialization of doArrive+awaitAdvance eliminating some reads/paths
636	>	final Phaser root = this.root;
637	>	for (;;) {
638	>	long s = (root == this) ? state : reconcileState();
639	>	int phase = (int)(s >>> PHASE_SHIFT);
640	>	int counts = (int)s;
641	>	int unarrived = (counts & UNARRIVED_MASK) - 1;
642	>	if (phase < 0)
643	>	return phase;
644	>	else if (counts == EMPTY || unarrived < 0) {
645	>	if (reconcileState() == s)
646	>	throw new IllegalStateException(badArrive(s));
647	>	}
648	>	else if (UNSAFE.compareAndSwapLong(this, stateOffset, s,
649	>	s -= ONE_ARRIVAL)) {
650	>	if (unarrived != 0)
651	>	return root.internalAwaitAdvance(phase, null);
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;
656	>	if (onAdvance(phase, nextUnarrived))
657	>	n |= TERMINATION_BIT;
658	>	else if (nextUnarrived == 0)
659	>	n |= EMPTY;
660	>	else
661	>	n |= nextUnarrived;
662	>	int nextPhase = (phase + 1) & MAX_PHASE;
663	>	n |= (long)nextPhase << PHASE_SHIFT;
664	>	if (!UNSAFE.compareAndSwapLong(this, stateOffset, s, n))
665	>	return (int)(state >>> PHASE_SHIFT); // terminated
666	>	releaseWaiters(phase);
667	>	return nextPhase;
668	>	}
669	>	}
670		}
671
672		/**
#	Line 633 | Line 677 | public class Phaser {
677		* @param phase an arrival phase number, or negative value if
678		* terminated; this argument is normally the value returned by a
679		* previous call to {@code arrive} or {@code arriveAndDeregister}.
680	<	* @return the next arrival phase number, or a negative value
681	<	* if terminated or argument is negative
680	>	* @return the next arrival phase number, or the argument if it is
681	>	* negative, or the (negative) {@linkplain #getPhase() current phase}
682	>	* if terminated
683		*/
684		public int awaitAdvance(int phase) {
685	<	Phaser rt;
686	<	int p = (int)(state >>> PHASE_SHIFT);
685	>	final Phaser root = this.root;
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) {
691	<	if ((p = (int)((rt = root).state >>> PHASE_SHIFT)) == phase)
646	<	return rt.internalAwaitAdvance(phase, null);
647	<	reconcileState();
648	<	}
690	>	if (p == phase)
691	>	return root.internalAwaitAdvance(phase, null);
692		return p;
693		}
694
#	Line 659 | Line 702 | public class Phaser {
702		* @param phase an arrival phase number, or negative value if
703		* terminated; this argument is normally the value returned by a
704		* previous call to {@code arrive} or {@code arriveAndDeregister}.
705	<	* @return the next arrival phase number, or a negative value
706	<	* if terminated or argument is negative
705	>	* @return the next arrival phase number, or the argument if it is
706	>	* negative, or the (negative) {@linkplain #getPhase() current phase}
707	>	* if terminated
708		* @throws InterruptedException if thread interrupted while waiting
709		*/
710		public int awaitAdvanceInterruptibly(int phase)
711		throws InterruptedException {
712	<	Phaser rt;
713	<	int p = (int)(state >>> PHASE_SHIFT);
712	>	final Phaser root = this.root;
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) {
718	<	if ((p = (int)((rt = root).state >>> PHASE_SHIFT)) == phase) {
719	<	QNode node = new QNode(this, phase, true, false, 0L);
720	<	p = rt.internalAwaitAdvance(phase, node);
721	<	if (node.wasInterrupted)
677	<	throw new InterruptedException();
678	<	}
679	<	else
680	<	reconcileState();
718	>	QNode node = new QNode(this, phase, true, false, 0L);
719	>	p = root.internalAwaitAdvance(phase, node);
720	>	if (node.wasInterrupted)
721	>	throw new InterruptedException();
722		}
723		return p;
724		}
#	Line 696 | Line 737 | public class Phaser {
737		*        {@code unit}
738		* @param unit a {@code TimeUnit} determining how to interpret the
739		*        {@code timeout} parameter
740	<	* @return the next arrival phase number, or a negative value
741	<	* if terminated or argument is negative
740	>	* @return the next arrival phase number, or the argument if it is
741	>	* negative, or the (negative) {@linkplain #getPhase() current phase}
742	>	* if terminated
743		* @throws InterruptedException if thread interrupted while waiting
744		* @throws TimeoutException if timed out while waiting
745		*/
#	Line 705 | Line 747 | public class Phaser {
747		long timeout, TimeUnit unit)
748		throws InterruptedException, TimeoutException {
749		long nanos = unit.toNanos(timeout);
750	<	Phaser rt;
751	<	int p = (int)(state >>> PHASE_SHIFT);
750	>	final Phaser root = this.root;
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) {
756	<	if ((p = (int)((rt = root).state >>> PHASE_SHIFT)) == phase) {
757	<	QNode node = new QNode(this, phase, true, true, nanos);
758	<	p = rt.internalAwaitAdvance(phase, node);
759	<	if (node.wasInterrupted)
760	<	throw new InterruptedException();
761	<	else if (p == phase)
719	<	throw new TimeoutException();
720	<	}
721	<	else
722	<	reconcileState();
756	>	QNode node = new QNode(this, phase, true, true, nanos);
757	>	p = root.internalAwaitAdvance(phase, node);
758	>	if (node.wasInterrupted)
759	>	throw new InterruptedException();
760	>	else if (p == phase)
761	>	throw new TimeoutException();
762		}
763		return p;
764		}
#	Line 738 | 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)(MAX_PARTIES)) | TERMINATION_BIT;
781	<	if (UNSAFE.compareAndSwapLong(root, stateOffset, s, next)) {
782	<	releaseWaiters(0); // signal all threads
780	>	if (UNSAFE.compareAndSwapLong(root, stateOffset,
781	>	s, s | TERMINATION_BIT)) {
782	>	// signal all threads
783	>	releaseWaiters(0);
784		releaseWaiters(1);
785		return;
786		}
#	Line 771 | 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 781 | 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 891 | Line 933 | public class Phaser {
933		*/
934		private void releaseWaiters(int phase) {
935		QNode q;   // first element of queue
894	–	int p;     // its phase
936		Thread t;  // its thread
896	–	//        assert phase != phaseOf(root.state);
937		AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ;
938		while ((q = head.get()) != null &&
939		q.phase != (int)(root.state >>> PHASE_SHIFT)) {
#	Line 905 | Line 945 | public class Phaser {
945		}
946		}
947
948	+	/**
949	+	* Variant of releaseWaiters that additionally tries to remove any
950	+	* nodes no longer waiting for advance due to timeout or
951	+	* interrupt. Currently, nodes are removed only if they are at
952	+	* head of queue, which suffices to reduce memory footprint in
953	+	* most usages.
954	+	*
955	+	* @return current phase on exit
956	+	*/
957	+	private int abortWait(int phase) {
958	+	AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ;
959	+	for (;;) {
960	+	Thread t;
961	+	QNode q = head.get();
962	+	int p = (int)(root.state >>> PHASE_SHIFT);
963	+	if (q == null || ((t = q.thread) != null && q.phase == p))
964	+	return p;
965	+	if (head.compareAndSet(q, q.next) && t != null) {
966	+	q.thread = null;
967	+	LockSupport.unpark(t);
968	+	}
969	+	}
970	+	}
971	+
972		/** The number of CPUs, for spin control */
973		private static final int NCPU = Runtime.getRuntime().availableProcessors();
974
#	Line 973 | Line 1037 | public class Phaser {
1037		if (node.wasInterrupted && !node.interruptible)
1038		Thread.currentThread().interrupt();
1039		if (p == phase && (p = (int)(state >>> PHASE_SHIFT)) == phase)
1040	<	return p;                 // recheck abort
1040	>	return abortWait(phase); // possibly clean up on abort
1041		}
1042		releaseWaiters(phase);
1043		return p;

Diff Legend

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