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.53 by jsr166, Sat Nov 13 05:59:25 2010 UTC vs.
Revision 1.63 by dl, Mon Nov 29 15:47:19 2010 UTC

#	Line 34 | Line 34 | import java.util.concurrent.locks.LockSu
34		* Phaser} may be repeatedly awaited.  Method {@link
35		* #arriveAndAwaitAdvance} has effect analogous to {@link
36		* java.util.concurrent.CyclicBarrier#await CyclicBarrier.await}. Each
37	<	* generation of a {@code Phaser} has an associated phase number. The
38	<	* phase number starts at zero, and advances when all parties arrive
39	<	* at the barrier, wrapping around to zero after reaching {@code
37	>	* generation of a phaser has an associated phase number. The phase
38	>	* number starts at zero, and advances when all parties arrive at the
39	>	* phaser, wrapping around to zero after reaching {@code
40		* Integer.MAX_VALUE}. The use of phase numbers enables independent
41	<	* control of actions upon arrival at a barrier and upon awaiting
41	>	* control of actions upon arrival at a phaser and upon awaiting
42		* others, via two kinds of methods that may be invoked by any
43		* registered party:
44		*
45		* <ul>
46		*
47		*   <li> <b>Arrival.</b> Methods {@link #arrive} and
48	<	*       {@link #arriveAndDeregister} record arrival at a
49	<	*       barrier. These methods do not block, but return an associated
50	<	*       <em>arrival phase number</em>; that is, the phase number of
51	<	*       the barrier to which the arrival applied. When the final
52	<	*       party for a given phase arrives, an optional barrier action
53	<	*       is performed and the phase advances.  Barrier actions,
54	<	*       performed by the party triggering a phase advance, are
55	<	*       arranged by overriding method {@link #onAdvance(int, int)},
56	<	*       which also controls termination. Overriding this method is
57	<	*       similar to, but more flexible than, providing a barrier
58	<	*       action to a {@code CyclicBarrier}.
48	>	*       {@link #arriveAndDeregister} record arrival.  These methods
49	>	*       do not block, but return an associated <em>arrival phase
50	>	*       number</em>; that is, the phase number of the phaser to which
51	>	*       the arrival applied. When the final party for a given phase
52	>	*       arrives, an optional action is performed and the phase
53	>	*       advances.  These actions are performed by the party
54	>	*       triggering a phase advance, and are arranged by overriding
55	>	*       method {@link #onAdvance(int, int)}, which also controls
56	>	*       termination. Overriding this method is similar to, but more
57	>	*       flexible than, providing a barrier action to a {@code
58	>	*       CyclicBarrier}.
59		*
60		*   <li> <b>Waiting.</b> Method {@link #awaitAdvance} requires an
61		*       argument indicating an arrival phase number, and returns when
62	<	*       the barrier advances to (or is already at) a different phase.
62	>	*       the phaser advances to (or is already at) a different phase.
63		*       Unlike similar constructions using {@code CyclicBarrier},
64		*       method {@code awaitAdvance} continues to wait even if the
65		*       waiting thread is interrupted. Interruptible and timeout
66		*       versions are also available, but exceptions encountered while
67		*       tasks wait interruptibly or with timeout do not change the
68	<	*       state of the barrier. If necessary, you can perform any
68	>	*       state of the phaser. If necessary, you can perform any
69		*       associated recovery within handlers of those exceptions,
70		*       often after invoking {@code forceTermination}.  Phasers may
71		*       also be used by tasks executing in a {@link ForkJoinPool},
#	Line 74 | Line 74 | import java.util.concurrent.locks.LockSu
74		*
75		* </ul>
76		*
77	<	* <p> <b>Termination.</b> A {@code Phaser} may enter a
78	<	* <em>termination</em> state in which all synchronization methods
79	<	* immediately return without updating phaser state or waiting for
80	<	* advance, and indicating (via a negative phase value) that execution
81	<	* is complete.  Termination is triggered when an invocation of {@code
82	<	* onAdvance} returns {@code true}.  As illustrated below, when
83	<	* phasers control actions with a fixed number of iterations, it is
84	<	* often convenient to override this method to cause termination when
85	<	* the current phase number reaches a threshold. Method {@link
86	<	* #forceTermination} is also available to abruptly release waiting
87	<	* threads and allow them to terminate.
88	<	*
89	<	* <p> <b>Tiering.</b> Phasers may be <em>tiered</em> (i.e., arranged
90	<	* in tree structures) to reduce contention. Phasers with large
91	<	* numbers of parties that would otherwise experience heavy
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.
90	>	*
91	>	* <p> <b>Tiering.</b> Phasers may be <em>tiered</em> (i.e.,
92	>	* constructed in tree structures) to reduce contention. Phasers with
93	>	* large numbers of parties that would otherwise experience heavy
94		* synchronization contention costs may instead be set up so that
95		* groups of sub-phasers share a common parent.  This may greatly
96		* increase throughput even though it incurs greater per-operation
#	Line 183 | Line 185 | import java.util.concurrent.locks.LockSu
185		*
186		* <p>To create a set of tasks using a tree of phasers,
187		* you could use code of the following form, assuming a
188	<	* Task class with a constructor accepting a phaser that
188	>	* Task class with a constructor accepting a {@code Phaser} that
189		* it registers with upon construction:
190		*
191		*  <pre> {@code
#	Line 203 | Line 205 | import java.util.concurrent.locks.LockSu
205		* build(new Task[n], 0, n, new Phaser());}</pre>
206		*
207		* The best value of {@code TASKS_PER_PHASER} depends mainly on
208	<	* expected barrier synchronization rates. A value as low as four may
209	<	* be appropriate for extremely small per-barrier task bodies (thus
208	>	* expected synchronization rates. A value as low as four may
209	>	* be appropriate for extremely small per-phase task bodies (thus
210		* high rates), or up to hundreds for extremely large ones.
211		*
212		* <p><b>Implementation notes</b>: This implementation restricts the
#	Line 224 | Line 226 | public class Phaser {
226		*/
227
228		/**
229	<	* Barrier state representation. Conceptually, a barrier contains
228	<	* four values:
229	>	* Primary state representation, holding four fields:
230		*
231		* * unarrived -- the number of parties yet to hit barrier (bits  0-15)
232		* * parties -- the number of parties to wait              (bits 16-31)
#	Line 240 | Line 241 | public class Phaser {
241		*/
242		private volatile long state;
243
244	<	private static final int  MAX_COUNT      = 0xffff;
245	<	private static final int  MAX_PHASE      = 0x7fffffff;
246	<	private static final int  PARTIES_SHIFT  = 16;
247	<	private static final int  PHASE_SHIFT    = 32;
248	<	private static final long UNARRIVED_MASK = 0xffffL;
249	<	private static final long PARTIES_MASK   = 0xffff0000L;
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;
244	>	private static final int  MAX_PARTIES     = 0xffff;
245	>	private static final int  MAX_PHASE       = 0x7fffffff;
246	>	private static final int  PARTIES_SHIFT   = 16;
247	>	private static final int  PHASE_SHIFT     = 32;
248	>	private static final int  UNARRIVED_MASK  = 0xffff;      // to mask ints
249	>	private static final long PARTIES_MASK    = 0xffff0000L; // to mask longs
250	>	private static final long ONE_ARRIVAL     = 1L;
251	>	private static final long ONE_PARTY       = 1L << PARTIES_SHIFT;
252	>	private static final long TERMINATION_BIT = 1L << 63;
253
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 293 | Line 294 | public class Phaser {
294		}
295
296		/**
297	+	* Returns message string for bounds exceptions on arrival.
298	+	*/
299	+	private String badArrive(long s) {
300	+	return "Attempted arrival of unregistered party for " +
301	+	stateToString(s);
302	+	}
303	+
304	+	/**
305	+	* Returns message string for bounds exceptions on registration.
306	+	*/
307	+	private String badRegister(long s) {
308	+	return "Attempt to register more than " +
309	+	MAX_PARTIES + " parties for " + stateToString(s);
310	+	}
311	+
312	+	/**
313		* Main implementation for methods arrive and arriveAndDeregister.
314		* Manually tuned to speed up and minimize race windows for the
315		* common case of just decrementing unarrived field.
#	Line 302 | Line 319 | public class Phaser {
319		* ONE_ARRIVAL|ONE_PARTY (for arriveAndDeregister)
320		*/
321		private int doArrive(long adj) {
322	<	long s;
323	<	int phase, unarrived;
324	<	while ((phase = (int)((s = state) >>> PHASE_SHIFT)) >= 0) {
325	<	if ((unarrived = (int)(s & UNARRIVED_MASK)) != 0) {
326	<	if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s -= adj)) {
327	<	if (unarrived == 1) {
328	<	Phaser par;
329	<	long p = s & PARTIES_MASK; // unshifted parties field
330	<	long lu = p >>> PARTIES_SHIFT;
331	<	int u = (int)lu;
332	<	int nextPhase = (phase + 1) & MAX_PHASE;
333	<	long next = ((long)nextPhase << PHASE_SHIFT) | p | lu;
334	<	if ((par = parent) == null) {
335	<	UNSAFE.compareAndSwapLong
336	<	(this, stateOffset, s, onAdvance(phase, u)?
337	<	next | TERMINATION_PHASE : next);
338	<	releaseWaiters(phase);
339	<	}
340	<	else {
341	<	par.doArrive(u == 0?
342	<	ONE_ARRIVAL|ONE_PARTY : ONE_ARRIVAL);
343	<	if ((int)(par.state >>> PHASE_SHIFT) != nextPhase ||
344	<	((int)(state >>> PHASE_SHIFT) != nextPhase &&
345	<	!UNSAFE.compareAndSwapLong(this, stateOffset,
346	<	s, next)))
347	<	reconcileState();
348	<	}
322	>	for (;;) {
323	>	long s = state;
324	>	int unarrived = (int)s & UNARRIVED_MASK;
325	>	int phase = (int)(s >>> PHASE_SHIFT);
326	>	if (phase < 0)
327	>	return phase;
328	>	else if (unarrived == 0) {
329	>	if (reconcileState() == s)     // recheck
330	>	throw new IllegalStateException(badArrive(s));
331	>	}
332	>	else if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s-=adj)) {
333	>	if (unarrived == 1) {
334	>	long p = s & PARTIES_MASK; // unshifted parties field
335	>	long lu = p >>> PARTIES_SHIFT;
336	>	int u = (int)lu;
337	>	int nextPhase = (phase + 1) & MAX_PHASE;
338	>	long next = ((long)nextPhase << PHASE_SHIFT) | p | lu;
339	>	final Phaser parent = this.parent;
340	>	if (parent == null) {
341	>	if (onAdvance(phase, u))
342	>	next |= TERMINATION_BIT;
343	>	UNSAFE.compareAndSwapLong(this, stateOffset, s, next);
344	>	releaseWaiters(phase);
345	>	}
346	>	else {
347	>	parent.doArrive((u == 0) ?
348	>	ONE_ARRIVAL|ONE_PARTY : ONE_ARRIVAL);
349	>	if ((int)(parent.state >>> PHASE_SHIFT) != nextPhase)
350	>	reconcileState();
351	>	else if (state == s)
352	>	UNSAFE.compareAndSwapLong(this, stateOffset, s,
353	>	next);
354		}
333	–	break;
355		}
356	+	return phase;
357		}
336	–	else if (state == s && reconcileState() == s) // recheck
337	–	throw new IllegalStateException(badArrive());
358		}
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.
345	–	*/
346	–	private String badArrive() {
347	–	return ("Attempted arrival of unregistered party for " +
348	–	this.toString());
359		}
360
361		/**
362		* Implementation of register, bulkRegister
363		*
364	<	* @param registrations number to add to both parties and unarrived fields
364	>	* @param registrations number to add to both parties and
365	>	* unarrived fields. Must be greater than zero.
366		*/
367		private int doRegister(int registrations) {
368	<	long adj = (long)registrations; // adjustment to state
369	<	adj |= adj << PARTIES_SHIFT;
370	<	Phaser par = parent;
371	<	long s;
372	<	int phase;
373	<	while ((phase = (int)((s = (par == null? state : reconcileState()))
374	<	>>> PHASE_SHIFT)) >= 0) {
375	<	int parties = ((int)(s & PARTIES_MASK)) >>> PARTIES_SHIFT;
376	<	if (parties != 0 && (s & UNARRIVED_MASK) == 0)
377	<	internalAwaitAdvance(phase, null); // wait for onAdvance
378	<	else if (parties + registrations > MAX_COUNT)
379	<	throw new IllegalStateException(badRegister());
380	<	else if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s + adj))
381	<	break;
368	>	// adjustment to state
369	>	long adj = ((long)registrations << PARTIES_SHIFT) | registrations;
370	>	final Phaser parent = this.parent;
371	>	for (;;) {
372	>	long s = (parent == null) ? state : reconcileState();
373	>	int parties = (int)s >>> PARTIES_SHIFT;
374	>	int phase = (int)(s >>> PHASE_SHIFT);
375	>	if (phase < 0)
376	>	return phase;
377	>	else if (registrations > MAX_PARTIES - parties)
378	>	throw new IllegalStateException(badRegister(s));
379	>	else if ((parties == 0 && parent == null) || // first reg of root
380	>	((int)s & UNARRIVED_MASK) != 0) {   // not advancing
381	>	if (UNSAFE.compareAndSwapLong(this, stateOffset, s, s + adj))
382	>	return phase;
383	>	}
384	>	else if (parties != 0)               // wait for onAdvance
385	>	root.internalAwaitAdvance(phase, null);
386	>	else {                               // 1st registration of child
387	>	synchronized (this) {            // register parent first
388	>	if (reconcileState() == s) { // recheck under lock
389	>	parent.doRegister(1);    // OK if throws IllegalState
390	>	for (;;) {               // simpler form of outer loop
391	>	s = reconcileState();
392	>	phase = (int)(s >>> PHASE_SHIFT);
393	>	if (phase < 0 ||
394	>	UNSAFE.compareAndSwapLong(this, stateOffset,
395	>	s, s + adj))
396	>	return phase;
397	>	}
398	>	}
399	>	}
400	>	}
401		}
372	–	return phase;
402		}
403
404		/**
405	<	* Returns message string for bounds exceptions on registration
377	<	*/
378	<	private String badRegister() {
379	<	return ("Attempt to register more than " + MAX_COUNT + " parties for "+
380	<	this.toString());
381	<	}
382	<
383	<	/**
384	<	* Recursively resolves lagged phase propagation from root if
385	<	* necessary.
405	>	* Recursively resolves lagged phase propagation from root if necessary.
406		*/
407		private long reconcileState() {
408		Phaser par = parent;
409	<	if (par == null)
410	<	return state;
411	<	Phaser rt = root;
412	<	long s;
413	<	int phase, rPhase;
414	<	while ((phase = (int)((s = state) >>> PHASE_SHIFT)) >= 0 &&
415	<	(rPhase = (int)(rt.state >>> PHASE_SHIFT)) != phase) {
416	<	if (rPhase < 0 || (s & UNARRIVED_MASK) == 0) {
417	<	long ps = par.parent == null? par.state : par.reconcileState();
418	<	int pPhase = (int)(ps >>> PHASE_SHIFT);
419	<	if (pPhase < 0 || pPhase == ((phase + 1) & MAX_PHASE)) {
420	<	if (state != s)
421	<	continue;
402	<	long p = s & PARTIES_MASK;
403	<	long next = ((((long) pPhase) << PHASE_SHIFT) |
404	<	(p >>> PARTIES_SHIFT) | p);
405	<	if (UNSAFE.compareAndSwapLong(this, stateOffset, s, next))
406	<	return next;
409	>	long s = state;
410	>	if (par != null) {
411	>	Phaser rt = root;
412	>	int phase, rPhase;
413	>	while ((phase = (int)(s >>> PHASE_SHIFT)) >= 0 &&
414	>	(rPhase = (int)(rt.state >>> PHASE_SHIFT)) != phase) {
415	>	if (par != rt && (int)(par.state >>> PHASE_SHIFT) != rPhase)
416	>	par.reconcileState();
417	>	else if (rPhase < 0 || ((int)s & UNARRIVED_MASK) == 0) {
418	>	long u = s & PARTIES_MASK; // reset unarrived to parties
419	>	long next = ((((long) rPhase) << PHASE_SHIFT) | u |
420	>	(u >>> PARTIES_SHIFT));
421	>	UNSAFE.compareAndSwapLong(this, stateOffset, s, next);
422		}
423	+	s = state;
424		}
409	–	if (state == s)
410	–	releaseWaiters(phase); // help release others
425		}
426		return s;
427		}
428
429		/**
430	<	* Creates a new phaser without any initially registered parties,
431	<	* initial phase number 0, and no parent. Any thread using this
430	>	* Creates a new phaser with no initially registered parties, no
431	>	* parent, and initial phase number 0. Any thread using this
432		* phaser will need to first register for it.
433		*/
434		public Phaser() {
#	Line 423 | Line 437 | public class Phaser {
437
438		/**
439		* Creates a new phaser with the given number of registered
440	<	* unarrived parties, initial phase number 0, and no parent.
440	>	* unarrived parties, no parent, and initial phase number 0.
441		*
442	<	* @param parties the number of parties required to trip barrier
442	>	* @param parties the number of parties required to advance to the
443	>	* next phase
444		* @throws IllegalArgumentException if parties less than zero
445		* or greater than the maximum number of parties supported
446		*/
#	Line 434 | Line 449 | public class Phaser {
449		}
450
451		/**
452	<	* Creates a new phaser with the given parent, without any
438	<	* initially registered parties. If parent is non-null this phaser
439	<	* is registered with the parent and its initial phase number is
440	<	* the same as that of parent phaser.
452	>	* Equivalent to {@link #Phaser(Phaser, int) Phaser(parent, 0)}.
453		*
454		* @param parent the parent phaser
455		*/
#	Line 447 | Line 459 | public class Phaser {
459
460		/**
461		* Creates a new phaser with the given parent and number of
462	<	* registered unarrived parties. If parent is non-null, this phaser
463	<	* is registered with the parent and its initial phase number is
464	<	* the same as that of parent phaser.
462	>	* registered unarrived parties. Registration and deregistration
463	>	* of this child phaser with its parent are managed automatically.
464	>	* If the given parent is non-null, whenever this child phaser has
465	>	* any registered parties (as established in this constructor,
466	>	* {@link #register}, or {@link #bulkRegister}), this child phaser
467	>	* is registered with its parent. Whenever the number of
468	>	* registered parties becomes zero as the result of an invocation
469	>	* of {@link #arriveAndDeregister}, this child phaser is
470	>	* deregistered from its parent.
471		*
472		* @param parent the parent phaser
473	<	* @param parties the number of parties required to trip barrier
473	>	* @param parties the number of parties required to advance to the
474	>	* next phase
475		* @throws IllegalArgumentException if parties less than zero
476		* or greater than the maximum number of parties supported
477		*/
478		public Phaser(Phaser parent, int parties) {
479	<	if (parties < 0 || parties > MAX_COUNT)
479	>	if (parties >>> PARTIES_SHIFT != 0)
480		throw new IllegalArgumentException("Illegal number of parties");
481	<	int phase;
481	>	long s = ((long) parties) | (((long) parties) << PARTIES_SHIFT);
482		this.parent = parent;
483		if (parent != null) {
484		Phaser r = parent.root;
485		this.root = r;
486		this.evenQ = r.evenQ;
487		this.oddQ = r.oddQ;
488	<	phase = parent.register();
488	>	if (parties != 0)
489	>	s |= ((long)(parent.doRegister(1))) << PHASE_SHIFT;
490		}
491		else {
492		this.root = this;
493		this.evenQ = new AtomicReference<QNode>();
494		this.oddQ = new AtomicReference<QNode>();
475	–	phase = 0;
495		}
496	<	long p = (long)parties;
478	<	this.state = (((long) phase) << PHASE_SHIFT) | p | (p << PARTIES_SHIFT);
496	>	this.state = s;
497		}
498
499		/**
500	<	* Adds a new unarrived party to this phaser.
501	<	* If an ongoing invocation of {@link #onAdvance} is in progress,
502	<	* this method may wait until its completion before registering.
500	>	* Adds a new unarrived party to this phaser.  If an ongoing
501	>	* invocation of {@link #onAdvance} is in progress, this method
502	>	* may await its completion before returning.  If this phaser has
503	>	* a parent, and this phaser previously had no registered parties,
504	>	* this phaser is also registered with its parent.
505		*
506		* @return the arrival phase number to which this registration applied
507		* @throws IllegalStateException if attempting to register more
#	Line 494 | Line 514 | public class Phaser {
514		/**
515		* Adds the given number of new unarrived parties to this phaser.
516		* If an ongoing invocation of {@link #onAdvance} is in progress,
517	<	* this method may wait until its completion before registering.
517	>	* this method may await its completion before returning.  If this
518	>	* phaser has a parent, and the given number of parities is
519	>	* greater than zero, and this phaser previously had no registered
520	>	* parties, this phaser is also registered with its parent.
521		*
522	<	* @param parties the number of additional parties required to trip barrier
522	>	* @param parties the number of additional parties required to
523	>	* advance to the next phase
524		* @return the arrival phase number to which this registration applied
525		* @throws IllegalStateException if attempting to register more
526		* than the maximum supported number of parties
#	Line 505 | Line 529 | public class Phaser {
529		public int bulkRegister(int parties) {
530		if (parties < 0)
531		throw new IllegalArgumentException();
508	–	if (parties > MAX_COUNT)
509	–	throw new IllegalStateException(badRegister());
532		if (parties == 0)
533		return getPhase();
534		return doRegister(parties);
535		}
536
537		/**
538	<	* Arrives at the barrier, but does not wait for others.  (You can
539	<	* in turn wait for others via {@link #awaitAdvance}).  It is an
540	<	* unenforced usage error for an unregistered party to invoke this
541	<	* method.
538	>	* Arrives at this phaser, without waiting for others to arrive.
539	>	*
540	>	* <p>It is a usage error for an unregistered party to invoke this
541	>	* method.  However, this error may result in an {@code
542	>	* IllegalStateException} only upon some subsequent operation on
543	>	* this phaser, if ever.
544		*
545		* @return the arrival phase number, or a negative value if terminated
546		* @throws IllegalStateException if not terminated and the number
#	Line 527 | Line 551 | public class Phaser {
551		}
552
553		/**
554	<	* Arrives at the barrier and deregisters from it without waiting
555	<	* for others. Deregistration reduces the number of parties
556	<	* required to trip the barrier in future phases.  If this phaser
554	>	* Arrives at this phaser and deregisters from it without waiting
555	>	* for others to arrive. Deregistration reduces the number of
556	>	* parties required to advance in future phases.  If this phaser
557		* has a parent, and deregistration causes this phaser to have
558	<	* zero parties, this phaser also arrives at and is deregistered
559	<	* from its parent.  It is an unenforced usage error for an
560	<	* unregistered party to invoke this method.
558	>	* zero parties, this phaser is also deregistered from its parent.
559	>	*
560	>	* <p>It is a usage error for an unregistered party to invoke this
561	>	* method.  However, this error may result in an {@code
562	>	* IllegalStateException} only upon some subsequent operation on
563	>	* this phaser, if ever.
564		*
565		* @return the arrival phase number, or a negative value if terminated
566		* @throws IllegalStateException if not terminated and the number
#	Line 544 | Line 571 | public class Phaser {
571		}
572
573		/**
574	<	* Arrives at the barrier and awaits others. Equivalent in effect
574	>	* Arrives at this phaser and awaits others. Equivalent in effect
575		* to {@code awaitAdvance(arrive())}.  If you need to await with
576		* interruption or timeout, you can arrange this with an analogous
577		* construction using one of the other forms of the {@code
578		* awaitAdvance} method.  If instead you need to deregister upon
579	<	* arrival, use {@link #arriveAndDeregister}. It is an unenforced
580	<	* usage error for an unregistered party to invoke this method.
579	>	* arrival, use {@code awaitAdvance(arriveAndDeregister())}.
580	>	*
581	>	* <p>It is a usage error for an unregistered party to invoke this
582	>	* method.  However, this error may result in an {@code
583	>	* IllegalStateException} only upon some subsequent operation on
584	>	* this phaser, if ever.
585		*
586		* @return the arrival phase number, or a negative number if terminated
587		* @throws IllegalStateException if not terminated and the number
588		* of unarrived parties would become negative
589		*/
590		public int arriveAndAwaitAdvance() {
591	<	return awaitAdvance(arrive());
591	>	return awaitAdvance(doArrive(ONE_ARRIVAL));
592		}
593
594		/**
595	<	* Awaits the phase of the barrier to advance from the given phase
596	<	* value, returning immediately if the current phase of the
597	<	* barrier is not equal to the given phase value or this barrier
567	<	* is terminated.
595	>	* Awaits the phase of this phaser to advance from the given phase
596	>	* value, returning immediately if the current phase is not equal
597	>	* to the given phase value or this phaser is terminated.
598		*
599		* @param phase an arrival phase number, or negative value if
600		* terminated; this argument is normally the value returned by a
601	<	* previous call to {@code arrive} or its variants
601	>	* previous call to {@code arrive} or {@code arriveAndDeregister}.
602		* @return the next arrival phase number, or a negative value
603		* if terminated or argument is negative
604		*/
605		public int awaitAdvance(int phase) {
606	+	Phaser rt;
607	+	int p = (int)(state >>> PHASE_SHIFT);
608		if (phase < 0)
609		return phase;
610	<	int p = (int)((parent==null? state : reconcileState()) >>> PHASE_SHIFT);
611	<	if (p != phase)
612	<	return p;
613	<	return internalAwaitAdvance(phase, null);
610	>	if (p == phase &&
611	>	(p = (int)((rt = root).state >>> PHASE_SHIFT)) == phase)
612	>	return rt.internalAwaitAdvance(phase, null);
613	>	return p;
614		}
615
616		/**
617	<	* Awaits the phase of the barrier to advance from the given phase
617	>	* Awaits the phase of this phaser to advance from the given phase
618		* value, throwing {@code InterruptedException} if interrupted
619	<	* while waiting, or returning immediately if the current phase of
620	<	* the barrier is not equal to the given phase value or this
621	<	* barrier is terminated.
619	>	* while waiting, or returning immediately if the current phase is
620	>	* not equal to the given phase value or this phaser is
621	>	* terminated.
622		*
623		* @param phase an arrival phase number, or negative value if
624		* terminated; this argument is normally the value returned by a
625	<	* previous call to {@code arrive} or its variants
625	>	* previous call to {@code arrive} or {@code arriveAndDeregister}.
626		* @return the next arrival phase number, or a negative value
627		* if terminated or argument is negative
628		* @throws InterruptedException if thread interrupted while waiting
629		*/
630		public int awaitAdvanceInterruptibly(int phase)
631		throws InterruptedException {
632	+	Phaser rt;
633	+	int p = (int)(state >>> PHASE_SHIFT);
634		if (phase < 0)
635		return phase;
636	<	int p = (int)((parent==null? state : reconcileState()) >>> PHASE_SHIFT);
637	<	if (p != phase)
638	<	return p;
639	<	QNode node = new QNode(this, phase, true, false, 0L);
640	<	p = internalAwaitAdvance(phase, node);
641	<	if (node.wasInterrupted)
642	<	throw new InterruptedException();
643	<	else
610	<	return p;
636	>	if (p == phase &&
637	>	(p = (int)((rt = root).state >>> PHASE_SHIFT)) == phase) {
638	>	QNode node = new QNode(this, phase, true, false, 0L);
639	>	p = rt.internalAwaitAdvance(phase, node);
640	>	if (node.wasInterrupted)
641	>	throw new InterruptedException();
642	>	}
643	>	return p;
644		}
645
646		/**
647	<	* Awaits the phase of the barrier to advance from the given phase
647	>	* Awaits the phase of this phaser to advance from the given phase
648		* value or the given timeout to elapse, throwing {@code
649		* InterruptedException} if interrupted while waiting, or
650	<	* returning immediately if the current phase of the barrier is
651	<	* not equal to the given phase value or this barrier is
619	<	* terminated.
650	>	* returning immediately if the current phase is not equal to the
651	>	* given phase value or this phaser is terminated.
652		*
653		* @param phase an arrival phase number, or negative value if
654		* terminated; this argument is normally the value returned by a
655	<	* previous call to {@code arrive} or its variants
655	>	* previous call to {@code arrive} or {@code arriveAndDeregister}.
656		* @param timeout how long to wait before giving up, in units of
657		*        {@code unit}
658		* @param unit a {@code TimeUnit} determining how to interpret the
#	Line 634 | Line 666 | public class Phaser {
666		long timeout, TimeUnit unit)
667		throws InterruptedException, TimeoutException {
668		long nanos = unit.toNanos(timeout);
669	+	Phaser rt;
670	+	int p = (int)(state >>> PHASE_SHIFT);
671		if (phase < 0)
672		return phase;
673	<	int p = (int)((parent==null? state : reconcileState()) >>> PHASE_SHIFT);
674	<	if (p != phase)
675	<	return p;
676	<	QNode node = new QNode(this, phase, true, true, nanos);
677	<	p = internalAwaitAdvance(phase, node);
678	<	if (node.wasInterrupted)
679	<	throw new InterruptedException();
680	<	else if (p == phase)
681	<	throw new TimeoutException();
682	<	else
649	<	return p;
673	>	if (p == phase &&
674	>	(p = (int)((rt = root).state >>> PHASE_SHIFT)) == phase) {
675	>	QNode node = new QNode(this, phase, true, true, nanos);
676	>	p = rt.internalAwaitAdvance(phase, node);
677	>	if (node.wasInterrupted)
678	>	throw new InterruptedException();
679	>	else if (p == phase)
680	>	throw new TimeoutException();
681	>	}
682	>	return p;
683		}
684
685		/**
686	<	* Forces this barrier to enter termination state. Counts of
687	<	* arrived and registered parties are unaffected. If this phaser
688	<	* has a parent, it too is terminated. This method may be useful
689	<	* for coordinating recovery after one or more tasks encounter
690	<	* unexpected exceptions.
686	>	* Forces this phaser to enter termination state.  Counts of
687	>	* arrived and registered parties are unaffected.  If this phaser
688	>	* is a member of a tiered set of phasers, then all of the phasers
689	>	* in the set are terminated.  If this phaser is already
690	>	* terminated, this method has no effect.  This method may be
691	>	* useful for coordinating recovery after one or more tasks
692	>	* encounter unexpected exceptions.
693		*/
694		public void forceTermination() {
695	<	Phaser r = root;    // force at root then reconcile
695	>	// Only need to change root state
696	>	final Phaser root = this.root;
697		long s;
698	<	while ((s = r.state) >= 0)
699	<	UNSAFE.compareAndSwapLong(r, stateOffset, s, s | TERMINATION_PHASE);
700	<	reconcileState();
701	<	releaseWaiters(0); // signal all threads
702	<	releaseWaiters(1);
698	>	while ((s = root.state) >= 0) {
699	>	if (UNSAFE.compareAndSwapLong(root, stateOffset,
700	>	s, s | TERMINATION_BIT)) {
701	>	releaseWaiters(0); // signal all threads
702	>	releaseWaiters(1);
703	>	return;
704	>	}
705	>	}
706		}
707
708		/**
709		* Returns the current phase number. The maximum phase number is
710		* {@code Integer.MAX_VALUE}, after which it restarts at
711	<	* zero. Upon termination, the phase number is negative.
711	>	* zero. Upon termination, the phase number is negative,
712	>	* in which case the prevailing phase prior to termination
713	>	* may be obtained via {@code getPhase() + Integer.MIN_VALUE}.
714		*
715		* @return the phase number, or a negative value if terminated
716		*/
717		public final int getPhase() {
718	<	return (int)((parent == null? state : reconcileState()) >>> PHASE_SHIFT);
718	>	return (int)(root.state >>> PHASE_SHIFT);
719		}
720
721		/**
722	<	* Returns the number of parties registered at this barrier.
722	>	* Returns the number of parties registered at this phaser.
723		*
724		* @return the number of parties
725		*/
726		public int getRegisteredParties() {
727	<	return partiesOf(parent == null? state : reconcileState());
727	>	return partiesOf(state);
728		}
729
730		/**
731		* Returns the number of registered parties that have arrived at
732	<	* the current phase of this barrier.
732	>	* the current phase of this phaser.
733		*
734		* @return the number of arrived parties
735		*/
736		public int getArrivedParties() {
737	<	return arrivedOf(parent == null? state : reconcileState());
737	>	long s = state;
738	>	int u = unarrivedOf(s); // only reconcile if possibly needed
739	>	return (u != 0 || parent == null) ?
740	>	partiesOf(s) - u :
741	>	arrivedOf(reconcileState());
742		}
743
744		/**
745		* Returns the number of registered parties that have not yet
746	<	* arrived at the current phase of this barrier.
746	>	* arrived at the current phase of this phaser.
747		*
748		* @return the number of unarrived parties
749		*/
750		public int getUnarrivedParties() {
751	<	return unarrivedOf(parent == null? state : reconcileState());
751	>	int u = unarrivedOf(state);
752	>	return (u != 0 || parent == null) ? u : unarrivedOf(reconcileState());
753		}
754
755		/**
#	Line 726 | Line 772 | public class Phaser {
772		}
773
774		/**
775	<	* Returns {@code true} if this barrier has been terminated.
775	>	* Returns {@code true} if this phaser has been terminated.
776		*
777	<	* @return {@code true} if this barrier has been terminated
777	>	* @return {@code true} if this phaser has been terminated
778		*/
779		public boolean isTerminated() {
780	<	return (parent == null? state : reconcileState()) < 0;
780	>	return root.state < 0L;
781		}
782
783		/**
784		* Overridable method to perform an action upon impending phase
785		* advance, and to control termination. This method is invoked
786	<	* upon arrival of the party tripping the barrier (when all other
786	>	* upon arrival of the party advancing this phaser (when all other
787		* waiting parties are dormant).  If this method returns {@code
788	<	* true}, then, rather than advance the phase number, this barrier
788	>	* true}, then, rather than advance the phase number, this phaser
789		* will be set to a final termination state, and subsequent calls
790		* to {@link #isTerminated} will return true. Any (unchecked)
791		* Exception or Error thrown by an invocation of this method is
792	<	* propagated to the party attempting to trip the barrier, in
792	>	* propagated to the party attempting to advance this phaser, in
793		* which case no advance occurs.
794		*
795		* <p>The arguments to this method provide the state of the phaser
796		* prevailing for the current transition.  The effects of invoking
797	<	* arrival, registration, and waiting methods on this Phaser from
797	>	* arrival, registration, and waiting methods on this phaser from
798		* within {@code onAdvance} are unspecified and should not be
799		* relied on.
800		*
801	<	* <p>If this Phaser is a member of a tiered set of Phasers, then
802	<	* {@code onAdvance} is invoked only for its root Phaser on each
801	>	* <p>If this phaser is a member of a tiered set of phasers, then
802	>	* {@code onAdvance} is invoked only for its root phaser on each
803		* advance.
804		*
805	<	* <p>The default version returns {@code true} when the number of
806	<	* registered parties is zero. Normally, overrides that arrange
807	<	* termination for other reasons should also preserve this
808	<	* property.
805	>	* <p>To support the most common use cases, the default
806	>	* implementation of this method returns {@code true} when the
807	>	* number of registered parties has become zero as the result of a
808	>	* party invoking {@code arriveAndDeregister}.  You can disable
809	>	* this behavior, thus enabling continuation upon future
810	>	* registrations, by overriding this method to always return
811	>	* {@code false}:
812	>	*
813	>	* <pre> {@code
814	>	* Phaser phaser = new Phaser() {
815	>	*   protected boolean onAdvance(int phase, int parties) { return false; }
816	>	* }}</pre>
817		*
818	<	* @param phase the phase number on entering the barrier
818	>	* @param phase the current phase number on entry to this method,
819	>	* before this phaser is advanced
820		* @param registeredParties the current number of registered parties
821	<	* @return {@code true} if this barrier should terminate
821	>	* @return {@code true} if this phaser should terminate
822		*/
823		protected boolean onAdvance(int phase, int registeredParties) {
824	<	return registeredParties <= 0;
824	>	return registeredParties == 0;
825		}
826
827		/**
#	Line 776 | Line 831 | public class Phaser {
831		* followed by the number of registered parties, and {@code
832		* "arrived = "} followed by the number of arrived parties.
833		*
834	<	* @return a string identifying this barrier, as well as its state
834	>	* @return a string identifying this phaser, as well as its state
835		*/
836		public String toString() {
837	<	long s = reconcileState();
837	>	return stateToString(reconcileState());
838	>	}
839	>
840	>	/**
841	>	* Implementation of toString and string-based error messages
842	>	*/
843	>	private String stateToString(long s) {
844		return super.toString() +
845		"[phase = " + phaseOf(s) +
846		" parties = " + partiesOf(s) +
847		" arrived = " + arrivedOf(s) + "]";
848		}
849
850	+	// Waiting mechanics
851	+
852		/**
853	<	* Removes and signals threads from queue for phase
853	>	* Removes and signals threads from queue for phase.
854		*/
855		private void releaseWaiters(int phase) {
856	<	AtomicReference<QNode> head = queueFor(phase);
857	<	QNode q;
858	<	int p;
856	>	QNode q;   // first element of queue
857	>	int p;     // its phase
858	>	Thread t;  // its thread
859	>	AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ;
860		while ((q = head.get()) != null &&
861		((p = q.phase) == phase ||
862		(int)(root.state >>> PHASE_SHIFT) != p)) {
863	<	if (head.compareAndSet(q, q.next))
864	<	q.signal();
863	>	if (head.compareAndSet(q, q.next) &&
864	>	(t = q.thread) != null) {
865	>	q.thread = null;
866	>	LockSupport.unpark(t);
867	>	}
868		}
869		}
870
804	–	/**
805	–	* Tries to enqueue given node in the appropriate wait queue.
806	–	*
807	–	* @return true if successful
808	–	*/
809	–	private boolean tryEnqueue(int phase, QNode node) {
810	–	releaseWaiters(phase-1); // ensure old queue clean
811	–	AtomicReference<QNode> head = queueFor(phase);
812	–	QNode q = head.get();
813	–	return ((q == null || q.phase == phase) &&
814	–	(int)(root.state >>> PHASE_SHIFT) == phase &&
815	–	head.compareAndSet(node.next = q, node));
816	–	}
817	–
871		/** The number of CPUs, for spin control */
872		private static final int NCPU = Runtime.getRuntime().availableProcessors();
873
#	Line 826 | Line 879 | public class Phaser {
879		* avoid it when threads regularly arrive: When a thread in
880		* internalAwaitAdvance notices another arrival before blocking,
881		* and there appear to be enough CPUs available, it spins
882	<	* SPINS_PER_ARRIVAL more times before continuing to try to
883	<	* block. The value trades off good-citizenship vs big unnecessary
831	<	* slowdowns.
882	>	* SPINS_PER_ARRIVAL more times before blocking. The value trades
883	>	* off good-citizenship vs big unnecessary slowdowns.
884		*/
885	<	static final int SPINS_PER_ARRIVAL = NCPU < 2? 1 : 1 << 8;
885	>	static final int SPINS_PER_ARRIVAL = (NCPU < 2) ? 1 : 1 << 8;
886
887		/**
888		* Possibly blocks and waits for phase to advance unless aborted.
889	+	* Call only from root node.
890		*
891		* @param phase current phase
892		* @param node if non-null, the wait node to track interrupt and timeout;
#	Line 841 | Line 894 | public class Phaser {
894		* @return current phase
895		*/
896		private int internalAwaitAdvance(int phase, QNode node) {
897	<	Phaser current = this;       // to eventually wait at root if tiered
898	<	Phaser par = parent;
899	<	boolean queued = false;
897	>	releaseWaiters(phase-1);          // ensure old queue clean
898	>	boolean queued = false;           // true when node is enqueued
899	>	int lastUnarrived = 0;            // to increase spins upon change
900		int spins = SPINS_PER_ARRIVAL;
848	–	int lastUnarrived = -1;      // to increase spins upon change
901		long s;
902		int p;
903	<	while ((p = (int)((s = current.state) >>> PHASE_SHIFT)) == phase) {
904	<	int unarrived = (int)(s & UNARRIVED_MASK);
905	<	if (unarrived != lastUnarrived) {
906	<	if ((lastUnarrived = unarrived) < NCPU)
903	>	while ((p = (int)((s = state) >>> PHASE_SHIFT)) == phase) {
904	>	if (node == null) {           // spinning in noninterruptible mode
905	>	int unarrived = (int)s & UNARRIVED_MASK;
906	>	if (unarrived != lastUnarrived &&
907	>	(lastUnarrived = unarrived) < NCPU)
908		spins += SPINS_PER_ARRIVAL;
909	+	boolean interrupted = Thread.interrupted();
910	+	if (interrupted || --spins < 0) { // need node to record intr
911	+	node = new QNode(this, phase, false, false, 0L);
912	+	node.wasInterrupted = interrupted;
913	+	}
914		}
915	<	else if (unarrived == 0 && par != null) {
858	<	current = par;       // if all arrived, use parent
859	<	par = par.parent;
860	<	}
861	<	else if (spins > 0)
862	<	--spins;
863	<	else if (node == null)
864	<	node = new QNode(this, phase, false, false, 0L);
865	<	else if (node.isReleasable())
915	>	else if (node.isReleasable()) // done or aborted
916		break;
917	<	else if (!queued)
918	<	queued = tryEnqueue(phase, node);
917	>	else if (!queued) {           // push onto queue
918	>	AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ;
919	>	QNode q = node.next = head.get();
920	>	if ((q == null || q.phase == phase) &&
921	>	(int)(state >>> PHASE_SHIFT) == phase) // avoid stale enq
922	>	queued = head.compareAndSet(q, node);
923	>	}
924		else {
925		try {
926		ForkJoinPool.managedBlock(node);
#	Line 874 | Line 929 | public class Phaser {
929		}
930		}
931		}
932	+
933		if (node != null) {
934		if (node.thread != null)
935	<	node.thread = null;
936	<	if (!node.interruptible && node.wasInterrupted)
935	>	node.thread = null;       // avoid need for unpark()
936	>	if (node.wasInterrupted && !node.interruptible)
937		Thread.currentThread().interrupt();
938	+	if ((p = (int)(state >>> PHASE_SHIFT)) == phase)
939	+	return p;                 // recheck abort
940		}
941	<	if (p == phase)
884	<	p = (int)(reconcileState() >>> PHASE_SHIFT);
885	<	if (p != phase)
886	<	releaseWaiters(phase);
941	>	releaseWaiters(phase);
942		return p;
943		}
944
#	Line 913 | Line 968 | public class Phaser {
968		}
969
970		public boolean isReleasable() {
971	<	Thread t = thread;
972	<	if (t != null) {
973	<	if (phaser.getPhase() != phase)
919	<	t = null;
920	<	else {
921	<	if (Thread.interrupted())
922	<	wasInterrupted = true;
923	<	if (interruptible && wasInterrupted)
924	<	t = null;
925	<	else if (timed) {
926	<	if (nanos > 0) {
927	<	long now = System.nanoTime();
928	<	nanos -= now - lastTime;
929	<	lastTime = now;
930	<	}
931	<	if (nanos <= 0)
932	<	t = null;
933	<	}
934	<	}
935	<	if (t != null)
936	<	return false;
971	>	if (thread == null)
972	>	return true;
973	>	if (phaser.getPhase() != phase) {
974		thread = null;
975	+	return true;
976		}
977	<	return true;
977	>	if (Thread.interrupted())
978	>	wasInterrupted = true;
979	>	if (wasInterrupted && interruptible) {
980	>	thread = null;
981	>	return true;
982	>	}
983	>	if (timed) {
984	>	if (nanos > 0L) {
985	>	long now = System.nanoTime();
986	>	nanos -= now - lastTime;
987	>	lastTime = now;
988	>	}
989	>	if (nanos <= 0L) {
990	>	thread = null;
991	>	return true;
992	>	}
993	>	}
994	>	return false;
995		}
996
997		public boolean block() {
#	Line 948 | Line 1003 | public class Phaser {
1003		LockSupport.parkNanos(this, nanos);
1004		return isReleasable();
1005		}
951	–
952	–	void signal() {
953	–	Thread t = thread;
954	–	if (t != null) {
955	–	thread = null;
956	–	LockSupport.unpark(t);
957	–	}
958	–	}
1006		}
1007
1008		// Unsafe mechanics

Diff Legend

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