ViewVC Help

View File | Revision Log | Show Annotations | Download File | Root Listing

root/jsr166/jsr166/src/main/java/util/concurrent/Phaser.java

Comparing jsr166/src/main/java/util/concurrent/Phaser.java (file contents):
Revision 1.49 by jsr166, Fri Dec 3 23:18:12 2010 UTC vs.
Revision 1.50 by dl, Sat Dec 4 15:25:12 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 100 | Line 101 | import java.util.concurrent.locks.LockSu
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}), this
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}, this child phaser is deregistered
107	>	* {@link #arriveAndDeregister}, the child phaser is deregistered
108		* from its parent.
109		*
110		* <p><b>Monitoring.</b> While synchronization methods may be invoked
#	Line 348 | 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
375	<	else
376	<	n |= (long)u;           // reset unarr to parties
377	<	// assert state == s || isTerminated();
378	<	UNSAFE.compareAndSwapLong(this, stateOffset, s, n);
379	<	releaseWaiters(phase);
380	<	}
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		}
385	–	return phase;
383		}
384
385		/**
#	Line 499 | Line 496 | public class Phaser {
496
497		/**
498		* Creates a new phaser with the given parent and number of
499	<	* registered unarrived parties.  If the given parent is non-null
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		*
#	Line 538 | 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 child 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 552 | 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
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 626 | Line 631 | public class Phaser {
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 642 | Line 681 | public class Phaser {
681		* if terminated
682		*/
683		public int awaitAdvance(int phase) {
684	+	final Phaser root = this.root;
685	+	int p = (int)((root == this? state : reconcileState()) >>> PHASE_SHIFT);
686		if (phase < 0)
687		return phase;
688	<	int p = (int)(state >>> PHASE_SHIFT);
689	<	if (p == phase) {
649	<	final Phaser root = this.root;
650	<	p = (int)(root.state >>> PHASE_SHIFT);
651	<	if (p == phase)
652	<	return root.internalAwaitAdvance(phase, null);
653	<	reconcileState();
654	<	}
688	>	if (p == phase)
689	>	return root.internalAwaitAdvance(phase, null);
690		return p;
691		}
692
#	Line 672 | Line 707 | public class Phaser {
707		*/
708		public int awaitAdvanceInterruptibly(int phase)
709		throws InterruptedException {
710	+	final Phaser root = this.root;
711	+	int p = (int)((root == this? state : reconcileState()) >>> PHASE_SHIFT);
712		if (phase < 0)
713		return phase;
677	–	int p = (int)(state >>> PHASE_SHIFT);
714		if (p == phase) {
715	<	final Phaser root = this.root;
716	<	p = (int)(root.state >>> PHASE_SHIFT);
717	<	if (p == phase) {
718	<	QNode node = new QNode(this, phase, true, false, 0L);
683	<	p = root.internalAwaitAdvance(phase, node);
684	<	if (node.wasInterrupted)
685	<	throw new InterruptedException();
686	<	}
687	<	else
688	<	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 713 | Line 743 | public class Phaser {
743		public int awaitAdvanceInterruptibly(int phase,
744		long timeout, TimeUnit unit)
745		throws InterruptedException, TimeoutException {
746	+	long nanos = unit.toNanos(timeout);
747	+	final Phaser root = this.root;
748	+	int p = (int)((root == this? state : reconcileState()) >>> PHASE_SHIFT);
749		if (phase < 0)
750		return phase;
718	–	long nanos = unit.toNanos(timeout);
719	–	int p = (int)(state >>> PHASE_SHIFT);
751		if (p == phase) {
752	<	final Phaser root = this.root;
753	<	p = (int)(root.state >>> PHASE_SHIFT);
754	<	if (p == phase) {
755	<	QNode node = new QNode(this, phase, true, true, nanos);
756	<	p = root.internalAwaitAdvance(phase, node);
757	<	if (node.wasInterrupted)
727	<	throw new InterruptedException();
728	<	else if (p == phase)
729	<	throw new TimeoutException();
730	<	}
731	<	else
732	<	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 902 | Line 927 | public class Phaser {
927		*/
928		private void releaseWaiters(int phase) {
929		QNode q;   // first element of queue
905	–	int p;     // its phase
930		Thread t;  // its thread
907	–	//        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 916 | 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 984 | 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