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.69 by jsr166, Sat Dec 4 22:00:05 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.
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 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 case 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 case 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	>	long s = (root == this) ? state : reconcileState();
686	>	int p = (int)(s >>> PHASE_SHIFT);
687		if (phase < 0)
688		return phase;
689	<	if (p == phase) {
690	<	if ((p = (int)((rt = root).state >>> PHASE_SHIFT)) == phase)
646	<	return rt.internalAwaitAdvance(phase, null);
647	<	reconcileState();
648	<	}
689	>	if (p == phase)
690	>	return root.internalAwaitAdvance(phase, null);
691		return p;
692		}
693
#	Line 659 | Line 701 | public class Phaser {
701		* @param phase an arrival phase number, or negative value if
702		* terminated; this argument is normally the value returned by a
703		* previous call to {@code arrive} or {@code arriveAndDeregister}.
704	<	* @return the next arrival phase number, or a negative value
705	<	* if terminated or argument is negative
704	>	* @return the next arrival phase number, or the argument if it is
705	>	* negative, or the (negative) {@linkplain #getPhase() current phase}
706	>	* if terminated
707		* @throws InterruptedException if thread interrupted while waiting
708		*/
709		public int awaitAdvanceInterruptibly(int phase)
710		throws InterruptedException {
711	<	Phaser rt;
712	<	int p = (int)(state >>> PHASE_SHIFT);
711	>	final Phaser root = this.root;
712	>	long s = (root == this) ? state : reconcileState();
713	>	int p = (int)(s >>> PHASE_SHIFT);
714		if (phase < 0)
715		return phase;
716		if (p == phase) {
717	<	if ((p = (int)((rt = root).state >>> PHASE_SHIFT)) == phase) {
718	<	QNode node = new QNode(this, phase, true, false, 0L);
719	<	p = rt.internalAwaitAdvance(phase, node);
720	<	if (node.wasInterrupted)
677	<	throw new InterruptedException();
678	<	}
679	<	else
680	<	reconcileState();
717	>	QNode node = new QNode(this, phase, true, false, 0L);
718	>	p = root.internalAwaitAdvance(phase, node);
719	>	if (node.wasInterrupted)
720	>	throw new InterruptedException();
721		}
722		return p;
723		}
#	Line 696 | Line 736 | public class Phaser {
736		*        {@code unit}
737		* @param unit a {@code TimeUnit} determining how to interpret the
738		*        {@code timeout} parameter
739	<	* @return the next arrival phase number, or a negative value
740	<	* if terminated or argument is negative
739	>	* @return the next arrival phase number, or the argument if it is
740	>	* negative, or the (negative) {@linkplain #getPhase() current phase}
741	>	* if terminated
742		* @throws InterruptedException if thread interrupted while waiting
743		* @throws TimeoutException if timed out while waiting
744		*/
#	Line 705 | Line 746 | public class Phaser {
746		long timeout, TimeUnit unit)
747		throws InterruptedException, TimeoutException {
748		long nanos = unit.toNanos(timeout);
749	<	Phaser rt;
750	<	int p = (int)(state >>> PHASE_SHIFT);
749	>	final Phaser root = this.root;
750	>	long s = (root == this) ? state : reconcileState();
751	>	int p = (int)(s >>> PHASE_SHIFT);
752		if (phase < 0)
753		return phase;
754		if (p == phase) {
755	<	if ((p = (int)((rt = root).state >>> PHASE_SHIFT)) == phase) {
756	<	QNode node = new QNode(this, phase, true, true, nanos);
757	<	p = rt.internalAwaitAdvance(phase, node);
758	<	if (node.wasInterrupted)
759	<	throw new InterruptedException();
760	<	else if (p == phase)
719	<	throw new TimeoutException();
720	<	}
721	<	else
722	<	reconcileState();
755	>	QNode node = new QNode(this, phase, true, true, nanos);
756	>	p = root.internalAwaitAdvance(phase, node);
757	>	if (node.wasInterrupted)
758	>	throw new InterruptedException();
759	>	else if (p == phase)
760	>	throw new TimeoutException();
761		}
762		return p;
763		}
#	Line 738 | Line 776 | public class Phaser {
776		final Phaser root = this.root;
777		long s;
778		while ((s = root.state) >= 0) {
779	<	long next = (s & ~(long)(MAX_PARTIES)) | TERMINATION_BIT;
779	>	long next = (s & ~((long)UNARRIVED_MASK)) | TERMINATION_BIT;
780		if (UNSAFE.compareAndSwapLong(root, stateOffset, s, next)) {
781	<	releaseWaiters(0); // signal all threads
781	>	// signal all threads
782	>	releaseWaiters(0);
783		releaseWaiters(1);
784		return;
785		}
#	Line 891 | Line 930 | public class Phaser {
930		*/
931		private void releaseWaiters(int phase) {
932		QNode q;   // first element of queue
894	–	int p;     // its phase
933		Thread t;  // its thread
896	–	//        assert phase != phaseOf(root.state);
934		AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ;
935		while ((q = head.get()) != null &&
936		q.phase != (int)(root.state >>> PHASE_SHIFT)) {
#	Line 905 | Line 942 | public class Phaser {
942		}
943		}
944
945	+	/**
946	+	* Variant of releaseWaiters that additionally tries to remove any
947	+	* nodes no longer waiting for advance due to timeout or
948	+	* interrupt. Currently, nodes are removed only if they are at
949	+	* head of queue, which suffices to reduce memory footprint in
950	+	* most usages.
951	+	*
952	+	* @return current phase on exit
953	+	*/
954	+	private int abortWait(int phase) {
955	+	AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ;
956	+	for (;;) {
957	+	Thread t;
958	+	QNode q = head.get();
959	+	int p = (int)(root.state >>> PHASE_SHIFT);
960	+	if (q == null || ((t = q.thread) != null && q.phase == p))
961	+	return p;
962	+	if (head.compareAndSet(q, q.next) && t != null) {
963	+	q.thread = null;
964	+	LockSupport.unpark(t);
965	+	}
966	+	}
967	+	}
968	+
969		/** The number of CPUs, for spin control */
970		private static final int NCPU = Runtime.getRuntime().availableProcessors();
971
#	Line 973 | Line 1034 | public class Phaser {
1034		if (node.wasInterrupted && !node.interruptible)
1035		Thread.currentThread().interrupt();
1036		if (p == phase && (p = (int)(state >>> PHASE_SHIFT)) == phase)
1037	<	return p;                 // recheck abort
1037	>	return abortWait(phase); // possibly clean up on abort
1038		}
1039		releaseWaiters(phase);
1040		return p;

Diff Legend

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