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.67 by jsr166, Fri Dec 3 21:29:34 2010 UTC vs.
Revision 1.68 by dl, Sat Dec 4 15:25:08 2010 UTC

#	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.
81	<	* Termination is triggered when an invocation of {@code onAdvance}
82	<	* returns {@code true}. The default implementation returns {@code
83	<	* true} if a deregistration has caused the number of registered
84	<	* parties to become zero.  As illustrated below, when phasers control
85	<	* actions with a fixed number of iterations, it is often convenient
86	<	* to override this method to cause termination when the current phase
87	<	* number reaches a threshold. Method {@link #forceTermination} is
88	<	* also available to abruptly release waiting threads and allow them
89	<	* to terminate.
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
81	>	* value. Similarly, attempts to register upon termination have no
82	>	* effect.  Termination is triggered when an invocation of {@code
83	>	* onAdvance} returns {@code true}. The default implementation returns
84	>	* {@code true} if a deregistration has caused the number of
85	>	* registered parties to become zero.  As illustrated below, when
86	>	* phasers control actions with a fixed number of iterations, it is
87	>	* often convenient to override this method to cause termination when
88	>	* the current phase number reaches a threshold. Method {@link
89	>	* #forceTermination} is also available to abruptly release waiting
90	>	* threads and allow them to terminate.
91		*
92		* <p> <b>Tiering.</b> Phasers may be <em>tiered</em> (i.e.,
93		* constructed in tree structures) to reduce contention. Phasers with
#	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 277 | Line 288 | public class Phaser {
288		}
289
290		private static int partiesOf(long s) {
291	<	int counts = (int)s;
281	<	return (counts == EMPTY) ? 0 : counts >>> PARTIES_SHIFT;
291	>	return (int)s >>> PARTIES_SHIFT;
292		}
293
294		private static int phaseOf(long s) {
#	Line 339 | Line 349 | public class Phaser {
349		*/
350		private int doArrive(boolean deregister) {
351		int adj = deregister ? ONE_ARRIVAL|ONE_PARTY : ONE_ARRIVAL;
352	<	long s;
353	<	int phase;
354	<	while ((phase = (int)((s = state) >>> PHASE_SHIFT)) >= 0) {
352	>	final Phaser root = this.root;
353	>	for (;;) {
354	>	long s = (root == this) ? state : reconcileState();
355	>	int phase = (int)(s >>> PHASE_SHIFT);
356		int counts = (int)s;
357	<	int unarrived = counts & UNARRIVED_MASK;
358	<	if (counts == EMPTY || unarrived == 0) {
359	<	if (reconcileState() == s)
357	>	int unarrived = (counts & UNARRIVED_MASK) - 1;
358	>	if (phase < 0)
359	>	return phase;
360	>	else if (counts == EMPTY || unarrived < 0) {
361	>	if (root == this || reconcileState() == s)
362		throw new IllegalStateException(badArrive(s));
363		}
364		else if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s-=adj)) {
365	<	if (unarrived == 1) {
366	<	long n = s & PARTIES_MASK;       // unshifted parties field
367	<	int u = ((int)n) >>> PARTIES_SHIFT;
368	<	Phaser par = parent;
369	<	if (par != null) {
370	<	par.doArrive(u == 0);
371	<	reconcileState();
372	<	}
373	<	else {
374	<	n |= (((long)((phase+1) & MAX_PHASE)) << PHASE_SHIFT);
375	<	if (onAdvance(phase, u))
376	<	n |= TERMINATION_BIT;
377	<	else if (u == 0)
378	<	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	<	}
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;
374	>	else
375	>	n |= nextUnarrived;
376	>	n |= ((long)((phase + 1) & MAX_PHASE)) << PHASE_SHIFT;
377	>	UNSAFE.compareAndSwapLong(this, stateOffset, s, n);
378	>	releaseWaiters(phase);
379		}
380	<	break;
380	>	return phase;
381		}
382		}
376	–	return phase;
383		}
384
385		/**
#	Line 490 | Line 496 | public class Phaser {
496
497		/**
498		* Creates a new phaser with the given parent and number of
499	<	* registered unarrived parties. Registration and deregistration
500	<	* of this child phaser with its parent are managed automatically.
501	<	* 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.
499	>	* registered unarrived parties.  When the given parent is non-null
500	>	* and the given number of parties is greater than zero, this
501	>	* child phaser is registered with its parent.
502		*
503		* @param parent the parent phaser
504		* @param parties the number of parties required to advance to the
#	Line 512 | Line 512 | public class Phaser {
512		int phase = 0;
513		this.parent = parent;
514		if (parent != null) {
515	<	Phaser r = parent.root;
516	<	this.root = r;
517	<	this.evenQ = r.evenQ;
518	<	this.oddQ = r.oddQ;
515	>	final Phaser root = parent.root;
516	>	this.root = root;
517	>	this.evenQ = root.evenQ;
518	>	this.oddQ = root.oddQ;
519		if (parties != 0)
520		phase = parent.doRegister(1);
521		}
#	Line 524 | Line 524 | public class Phaser {
524		this.evenQ = new AtomicReference<QNode>();
525		this.oddQ = new AtomicReference<QNode>();
526		}
527	<	this.state = (parties == 0) ? ((long) EMPTY) :
527	>	this.state = (parties == 0) ? (long) EMPTY :
528		((((long) phase) << PHASE_SHIFT) |
529		(((long) parties) << PARTIES_SHIFT) |
530		((long) parties));
#	Line 535 | Line 535 | public class Phaser {
535		* invocation of {@link #onAdvance} is in progress, this method
536		* may await its completion before returning.  If this phaser has
537		* a parent, and this phaser previously had no registered parties,
538	<	* this phaser is also registered with its parent.
539	<	*
540	<	* @return the arrival phase number to which this registration applied
538	>	* this child phaser is also registered with its parent. If
539	>	* this phaser is terminated, the attempt to register has
540	>	* no effect, and a negative value is returned.
541	>	*
542	>	* @return the arrival phase number to which this registration
543	>	* applied.  If this value is negative, then this phaser has
544	>	* terminated, in which casem registration has no effect.
545		* @throws IllegalStateException if attempting to register more
546		* than the maximum supported number of parties
547		*/
#	Line 549 | Line 553 | public class Phaser {
553		* Adds the given number of new unarrived parties to this phaser.
554		* If an ongoing invocation of {@link #onAdvance} is in progress,
555		* this method may await its completion before returning.  If this
556	<	* phaser has a parent, and the given number of parties is
557	<	* greater than zero, and this phaser previously had no registered
558	<	* parties, this phaser is also registered with its parent.
556	>	* phaser has a parent, and the given number of parties is greater
557	>	* than zero, and this phaser previously had no registered
558	>	* parties, this child phaser is also registered with its parent.
559	>	* If this phaser is terminated, the attempt to register has no
560	>	* effect, and a negative value is returned.
561		*
562		* @param parties the number of additional parties required to
563		* advance to the next phase
564	<	* @return the arrival phase number to which this registration applied
564	>	* @return the arrival phase number to which this registration
565	>	* applied.  If this value is negative, then this phaser has
566	>	* terminated, in which casem registration has no effect.
567		* @throws IllegalStateException if attempting to register more
568		* than the maximum supported number of parties
569		* @throws IllegalArgumentException if {@code parties < 0}
#	Line 617 | Line 625 | public class Phaser {
625		* IllegalStateException} only upon some subsequent operation on
626		* this phaser, if ever.
627		*
628	<	* @return the arrival phase number, or a negative number if terminated
628	>	* @return the arrival phase number, or the (negative)
629	>	* {@linkplain #getPhase() current phase} if terminated
630		* @throws IllegalStateException if not terminated and the number
631		* of unarrived parties would become negative
632		*/
633		public int arriveAndAwaitAdvance() {
634	<	return awaitAdvance(doArrive(false));
634	>	// Specialization of doArrive+awaitAdvance eliminating some reads/paths
635	>	final Phaser root = this.root;
636	>	for (;;) {
637	>	long s = (root == this) ? state : reconcileState();
638	>	int phase = (int)(s >>> PHASE_SHIFT);
639	>	int counts = (int)s;
640	>	int unarrived = (counts & UNARRIVED_MASK) - 1;
641	>	if (phase < 0)
642	>	return phase;
643	>	else if (counts == EMPTY || unarrived < 0) {
644	>	if (reconcileState() == s)
645	>	throw new IllegalStateException(badArrive(s));
646	>	}
647	>	else if (UNSAFE.compareAndSwapLong(this, stateOffset, s,
648	>	s -= ONE_ARRIVAL)) {
649	>	if (unarrived != 0)
650	>	return root.internalAwaitAdvance(phase, null);
651	>	if (root != this)
652	>	return parent.arriveAndAwaitAdvance();
653	>	long n = s & PARTIES_MASK;  // base of next state
654	>	int nextUnarrived = ((int)n) >>> PARTIES_SHIFT;
655	>	if (onAdvance(phase, nextUnarrived))
656	>	n |= TERMINATION_BIT;
657	>	else if (nextUnarrived == 0)
658	>	n |= EMPTY;
659	>	else
660	>	n |= nextUnarrived;
661	>	int nextPhase = (phase + 1) & MAX_PHASE;
662	>	n |= (long)nextPhase << PHASE_SHIFT;
663	>	if (!UNSAFE.compareAndSwapLong(this, stateOffset, s, n))
664	>	return (int)(state >>> PHASE_SHIFT); // terminated
665	>	releaseWaiters(phase);
666	>	return nextPhase;
667	>	}
668	>	}
669		}
670
671		/**
#	Line 633 | Line 676 | public class Phaser {
676		* @param phase an arrival phase number, or negative value if
677		* terminated; this argument is normally the value returned by a
678		* previous call to {@code arrive} or {@code arriveAndDeregister}.
679	<	* @return the next arrival phase number, or a negative value
680	<	* if terminated or argument is negative
679	>	* @return the next arrival phase number, or the argument if it is
680	>	* negative, or the (negative) {@linkplain #getPhase() current phase}
681	>	* if terminated
682		*/
683		public int awaitAdvance(int phase) {
684	<	Phaser rt;
685	<	int p = (int)(state >>> PHASE_SHIFT);
684	>	final Phaser root = this.root;
685	>	int p = (int)((root == this? state : reconcileState()) >>> PHASE_SHIFT);
686		if (phase < 0)
687		return phase;
688	<	if (p == phase) {
689	<	if ((p = (int)((rt = root).state >>> PHASE_SHIFT)) == phase)
646	<	return rt.internalAwaitAdvance(phase, null);
647	<	reconcileState();
648	<	}
688	>	if (p == phase)
689	>	return root.internalAwaitAdvance(phase, null);
690		return p;
691		}
692
#	Line 659 | Line 700 | public class Phaser {
700		* @param phase an arrival phase number, or negative value if
701		* terminated; this argument is normally the value returned by a
702		* previous call to {@code arrive} or {@code arriveAndDeregister}.
703	<	* @return the next arrival phase number, or a negative value
704	<	* if terminated or argument is negative
703	>	* @return the next arrival phase number, or the argument if it is
704	>	* negative, or the (negative) {@linkplain #getPhase() current phase}
705	>	* if terminated
706		* @throws InterruptedException if thread interrupted while waiting
707		*/
708		public int awaitAdvanceInterruptibly(int phase)
709		throws InterruptedException {
710	<	Phaser rt;
711	<	int p = (int)(state >>> PHASE_SHIFT);
710	>	final Phaser root = this.root;
711	>	int p = (int)((root == this? state : reconcileState()) >>> PHASE_SHIFT);
712		if (phase < 0)
713		return phase;
714		if (p == phase) {
715	<	if ((p = (int)((rt = root).state >>> PHASE_SHIFT)) == phase) {
716	<	QNode node = new QNode(this, phase, true, false, 0L);
717	<	p = rt.internalAwaitAdvance(phase, node);
718	<	if (node.wasInterrupted)
677	<	throw new InterruptedException();
678	<	}
679	<	else
680	<	reconcileState();
715	>	QNode node = new QNode(this, phase, true, false, 0L);
716	>	p = root.internalAwaitAdvance(phase, node);
717	>	if (node.wasInterrupted)
718	>	throw new InterruptedException();
719		}
720		return p;
721		}
#	Line 696 | Line 734 | public class Phaser {
734		*        {@code unit}
735		* @param unit a {@code TimeUnit} determining how to interpret the
736		*        {@code timeout} parameter
737	<	* @return the next arrival phase number, or a negative value
738	<	* if terminated or argument is negative
737	>	* @return the next arrival phase number, or the argument if it is
738	>	* negative, or the (negative) {@linkplain #getPhase() current phase}
739	>	* if terminated
740		* @throws InterruptedException if thread interrupted while waiting
741		* @throws TimeoutException if timed out while waiting
742		*/
#	Line 705 | Line 744 | public class Phaser {
744		long timeout, TimeUnit unit)
745		throws InterruptedException, TimeoutException {
746		long nanos = unit.toNanos(timeout);
747	<	Phaser rt;
748	<	int p = (int)(state >>> PHASE_SHIFT);
747	>	final Phaser root = this.root;
748	>	int p = (int)((root == this? state : reconcileState()) >>> PHASE_SHIFT);
749		if (phase < 0)
750		return phase;
751		if (p == phase) {
752	<	if ((p = (int)((rt = root).state >>> PHASE_SHIFT)) == phase) {
753	<	QNode node = new QNode(this, phase, true, true, nanos);
754	<	p = rt.internalAwaitAdvance(phase, node);
755	<	if (node.wasInterrupted)
756	<	throw new InterruptedException();
757	<	else if (p == phase)
719	<	throw new TimeoutException();
720	<	}
721	<	else
722	<	reconcileState();
752	>	QNode node = new QNode(this, phase, true, true, nanos);
753	>	p = root.internalAwaitAdvance(phase, node);
754	>	if (node.wasInterrupted)
755	>	throw new InterruptedException();
756	>	else if (p == phase)
757	>	throw new TimeoutException();
758		}
759		return p;
760		}
#	Line 738 | Line 773 | public class Phaser {
773		final Phaser root = this.root;
774		long s;
775		while ((s = root.state) >= 0) {
776	<	long next = (s & ~(long)(MAX_PARTIES)) | TERMINATION_BIT;
776	>	long next = (s & ~((long)UNARRIVED_MASK)) | TERMINATION_BIT;
777		if (UNSAFE.compareAndSwapLong(root, stateOffset, s, next)) {
778	<	releaseWaiters(0); // signal all threads
778	>	// signal all threads
779	>	releaseWaiters(0);
780		releaseWaiters(1);
781		return;
782		}
#	Line 891 | Line 927 | public class Phaser {
927		*/
928		private void releaseWaiters(int phase) {
929		QNode q;   // first element of queue
894	–	int p;     // its phase
930		Thread t;  // its thread
896	–	//        assert phase != phaseOf(root.state);
931		AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ;
932		while ((q = head.get()) != null &&
933		q.phase != (int)(root.state >>> PHASE_SHIFT)) {
#	Line 905 | Line 939 | public class Phaser {
939		}
940		}
941
942	+	/**
943	+	* Variant of releaseWaiters that additionally tries to remove any
944	+	* nodes no longer waiting for advance due to timeout or
945	+	* interrupt. Currently, nodes are removed only if they are at
946	+	* head of queue, which suffices to reduce memory footprint in
947	+	* most usages.
948	+	*
949	+	* @return current phase on exit
950	+	*/
951	+	private int abortWait(int phase) {
952	+	AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ;
953	+	for (;;) {
954	+	Thread t;
955	+	QNode q = head.get();
956	+	int p = (int)(root.state >>> PHASE_SHIFT);
957	+	if (q == null || ((t = q.thread) != null && q.phase == p))
958	+	return p;
959	+	if (head.compareAndSet(q, q.next) && t != null) {
960	+	q.thread = null;
961	+	LockSupport.unpark(t);
962	+	}
963	+	}
964	+	}
965	+
966		/** The number of CPUs, for spin control */
967		private static final int NCPU = Runtime.getRuntime().availableProcessors();
968
#	Line 973 | Line 1031 | public class Phaser {
1031		if (node.wasInterrupted && !node.interruptible)
1032		Thread.currentThread().interrupt();
1033		if (p == phase && (p = (int)(state >>> PHASE_SHIFT)) == phase)
1034	<	return p;                 // recheck abort
1034	>	return abortWait(phase); // possibly clean up on abort
1035		}
1036		releaseWaiters(phase);
1037		return p;

Diff Legend

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