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.13 by jsr166, Mon Jul 20 22:40:09 2009 UTC vs.
Revision 1.22 by jsr166, Sun Jul 26 17:33:37 2009 UTC

#	Line 7 | Line 7
7		package jsr166y;
8
9		import java.util.concurrent.*;
10	<	import java.util.concurrent.atomic.*;
10	>
11	>	import java.util.concurrent.atomic.AtomicReference;
12		import java.util.concurrent.locks.LockSupport;
12	–	import sun.misc.Unsafe;
13	–	import java.lang.reflect.*;
13
14		/**
15		* A reusable synchronization barrier, similar in functionality to a
#	Line 175 | Line 174 | import java.lang.reflect.*;
174		* parties result in IllegalStateExceptions. However, you can and
175		* should create tiered phasers to accommodate arbitrarily large sets
176		* of participants.
177	+	*
178	+	* @since 1.7
179	+	* @author Doug Lea
180		*/
181		public class Phaser {
182		/*
#	Line 208 | Line 210 | public class Phaser {
210		private static final int phaseMask  = 0x7fffffff;
211
212		private static int unarrivedOf(long s) {
213	<	return (int)(s & ushortMask);
213	>	return (int) (s & ushortMask);
214		}
215
216		private static int partiesOf(long s) {
217	<	return ((int)s) >>> 16;
217	>	return ((int) s) >>> 16;
218		}
219
220		private static int phaseOf(long s) {
221	<	return (int)(s >>> 32);
221	>	return (int) (s >>> 32);
222		}
223
224		private static int arrivedOf(long s) {
#	Line 224 | Line 226 | public class Phaser {
226		}
227
228		private static long stateFor(int phase, int parties, int unarrived) {
229	<	return ((((long)phase) << 32) | (((long)parties) << 16) |
230	<	(long)unarrived);
229	>	return ((((long) phase) << 32) | (((long) parties) << 16) |
230	>	(long) unarrived);
231		}
232
233		private static long trippedStateFor(int phase, int parties) {
234	<	long lp = (long)parties;
235	<	return (((long)phase) << 32) | (lp << 16) | lp;
234	>	long lp = (long) parties;
235	>	return (((long) phase) << 32) | (lp << 16) | lp;
236		}
237
238		/**
239	<	* Returns message string for bad bounds exceptions
239	>	* Returns message string for bad bounds exceptions.
240		*/
241		private static String badBounds(int parties, int unarrived) {
242		return ("Attempt to set " + unarrived +
#	Line 263 | Line 265 | public class Phaser {
265		private final AtomicReference<QNode> oddQ  = new AtomicReference<QNode>();
266
267		private AtomicReference<QNode> queueFor(int phase) {
268	<	return (phase & 1) == 0? evenQ : oddQ;
268	>	return ((phase & 1) == 0) ? evenQ : oddQ;
269		}
270
271		/**
#	Line 271 | Line 273 | public class Phaser {
273		* root if necessary.
274		*/
275		private long getReconciledState() {
276	<	return parent == null? state : reconcileState();
276	>	return (parent == null) ? state : reconcileState();
277		}
278
279		/**
#	Line 309 | Line 311 | public class Phaser {
311		/**
312		* Creates a new Phaser with the given numbers of registered
313		* unarrived parties, initial phase number 0, and no parent.
314	<	* @param parties the number of parties required to trip barrier.
314	>	*
315	>	* @param parties the number of parties required to trip barrier
316		* @throws IllegalArgumentException if parties less than zero
317	<	* or greater than the maximum number of parties supported.
317	>	* or greater than the maximum number of parties supported
318		*/
319		public Phaser(int parties) {
320		this(null, parties);
#	Line 322 | Line 325 | public class Phaser {
325		* initially registered parties. If parent is non-null this phaser
326		* is registered with the parent and its initial phase number is
327		* the same as that of parent phaser.
328	<	* @param parent the parent phaser.
328	>	*
329	>	* @param parent the parent phaser
330		*/
331		public Phaser(Phaser parent) {
332		int phase = 0;
#	Line 338 | Line 342 | public class Phaser {
342
343		/**
344		* Creates a new Phaser with the given parent and numbers of
345	<	* registered unarrived parties. If parent is non-null this phaser
345	>	* registered unarrived parties. If parent is non-null, this phaser
346		* is registered with the parent and its initial phase number is
347		* the same as that of parent phaser.
348	<	* @param parent the parent phaser.
349	<	* @param parties the number of parties required to trip barrier.
348	>	*
349	>	* @param parent the parent phaser
350	>	* @param parties the number of parties required to trip barrier
351		* @throws IllegalArgumentException if parties less than zero
352	<	* or greater than the maximum number of parties supported.
352	>	* or greater than the maximum number of parties supported
353		*/
354		public Phaser(Phaser parent, int parties) {
355		if (parties < 0 || parties > ushortMask)
#	Line 362 | Line 367 | public class Phaser {
367
368		/**
369		* Adds a new unarrived party to this phaser.
370	+	*
371		* @return the current barrier phase number upon registration
372		* @throws IllegalStateException if attempting to register more
373	<	* than the maximum supported number of parties.
373	>	* than the maximum supported number of parties
374		*/
375		public int register() {
376		return doRegister(1);
#	Line 372 | Line 378 | public class Phaser {
378
379		/**
380		* Adds the given number of new unarrived parties to this phaser.
381	<	* @param parties the number of parties required to trip barrier.
381	>	*
382	>	* @param parties the number of parties required to trip barrier
383		* @return the current barrier phase number upon registration
384		* @throws IllegalStateException if attempting to register more
385	<	* than the maximum supported number of parties.
385	>	* than the maximum supported number of parties
386		*/
387		public int bulkRegister(int parties) {
388		if (parties < 0)
#	Line 411 | Line 418 | public class Phaser {
418		* in turn wait for others via {@link #awaitAdvance}).
419		*
420		* @return the barrier phase number upon entry to this method, or a
421	<	* negative value if terminated;
421	>	* negative value if terminated
422		* @throws IllegalStateException if not terminated and the number
423	<	* of unarrived parties would become negative.
423	>	* of unarrived parties would become negative
424		*/
425		public int arrive() {
426		int phase;
#	Line 433 | Line 440 | public class Phaser {
440		if (par == null) {      // directly trip
441		if (casState
442		(s,
443	<	trippedStateFor(onAdvance(phase, parties)? -1 :
443	>	trippedStateFor(onAdvance(phase, parties) ? -1 :
444		((phase + 1) & phaseMask), parties))) {
445		releaseWaiters(phase);
446		break;
#	Line 463 | Line 470 | public class Phaser {
470		* zero parties, this phaser is also deregistered from its parent.
471		*
472		* @return the current barrier phase number upon entry to
473	<	* this method, or a negative value if terminated;
473	>	* this method, or a negative value if terminated
474		* @throws IllegalStateException if not terminated and the number
475	<	* of registered or unarrived parties would become negative.
475	>	* of registered or unarrived parties would become negative
476		*/
477		public int arriveAndDeregister() {
478		// similar code to arrive, but too different to merge
#	Line 494 | Line 501 | public class Phaser {
501		if (unarrived == 0) {
502		if (casState
503		(s,
504	<	trippedStateFor(onAdvance(phase, parties)? -1 :
504	>	trippedStateFor(onAdvance(phase, parties) ? -1 :
505		((phase + 1) & phaseMask), parties))) {
506		releaseWaiters(phase);
507		break;
#	Line 516 | Line 523 | public class Phaser {
523		* to {@code awaitAdvance(arrive())}.  If you instead need to
524		* await with interruption of timeout, and/or deregister upon
525		* arrival, you can arrange them using analogous constructions.
526	+	*
527		* @return the phase on entry to this method
528		* @throws IllegalStateException if not terminated and the number
529	<	* of unarrived parties would become negative.
529	>	* of unarrived parties would become negative
530		*/
531		public int arriveAndAwaitAdvance() {
532		return awaitAdvance(arrive());
#	Line 528 | Line 536 | public class Phaser {
536		* Awaits the phase of the barrier to advance from the given
537		* value, or returns immediately if argument is negative or this
538		* barrier is terminated.
539	+	*
540		* @param phase the phase on entry to this method
541		* @return the phase on exit from this method
542		*/
#	Line 549 | Line 558 | public class Phaser {
558		* value, or returns immediately if argument is negative or this
559		* barrier is terminated, or throws InterruptedException if
560		* interrupted while waiting.
561	+	*
562		* @param phase the phase on entry to this method
563		* @return the phase on exit from this method
564		* @throws InterruptedException if thread interrupted while waiting
#	Line 570 | Line 580 | public class Phaser {
580		* Awaits the phase of the barrier to advance from the given value
581		* or the given timeout elapses, or returns immediately if
582		* argument is negative or this barrier is terminated.
583	+	*
584		* @param phase the phase on entry to this method
585		* @return the phase on exit from this method
586		* @throws InterruptedException if thread interrupted while waiting
587		* @throws TimeoutException if timed out while waiting
588		*/
589	<	public int awaitAdvanceInterruptibly(int phase, long timeout, TimeUnit unit)
589	>	public int awaitAdvanceInterruptibly(int phase,
590	>	long timeout, TimeUnit unit)
591		throws InterruptedException, TimeoutException {
592		if (phase < 0)
593		return phase;
#	Line 616 | Line 628 | public class Phaser {
628		* Returns the current phase number. The maximum phase number is
629		* {@code Integer.MAX_VALUE}, after which it restarts at
630		* zero. Upon termination, the phase number is negative.
631	+	*
632		* @return the phase number, or a negative value if terminated
633		*/
634		public final int getPhase() {
#	Line 624 | Line 637 | public class Phaser {
637
638		/**
639		* Returns {@code true} if the current phase number equals the given phase.
640	+	*
641		* @param phase the phase
642		* @return {@code true} if the current phase number equals the given phase
643		*/
#	Line 633 | Line 647 | public class Phaser {
647
648		/**
649		* Returns the number of parties registered at this barrier.
650	+	*
651		* @return the number of parties
652		*/
653		public int getRegisteredParties() {
#	Line 642 | Line 657 | public class Phaser {
657		/**
658		* Returns the number of parties that have arrived at the current
659		* phase of this barrier.
660	+	*
661		* @return the number of arrived parties
662		*/
663		public int getArrivedParties() {
#	Line 651 | Line 667 | public class Phaser {
667		/**
668		* Returns the number of registered parties that have not yet
669		* arrived at the current phase of this barrier.
670	+	*
671		* @return the number of unarrived parties
672		*/
673		public int getUnarrivedParties() {
#	Line 659 | Line 676 | public class Phaser {
676
677		/**
678		* Returns the parent of this phaser, or null if none.
679	+	*
680		* @return the parent of this phaser, or null if none
681		*/
682		public Phaser getParent() {
#	Line 668 | Line 686 | public class Phaser {
686		/**
687		* Returns the root ancestor of this phaser, which is the same as
688		* this phaser if it has no parent.
689	+	*
690		* @return the root ancestor of this phaser
691		*/
692		public Phaser getRoot() {
#	Line 676 | Line 695 | public class Phaser {
695
696		/**
697		* Returns {@code true} if this barrier has been terminated.
698	+	*
699		* @return {@code true} if this barrier has been terminated
700		*/
701		public boolean isTerminated() {
#	Line 799 | Line 819 | public class Phaser {
819		}
820
821		/**
822	<	* Removes and signals waiting threads from wait queue
822	>	* Removes and signals waiting threads from wait queue.
823		*/
824		private void releaseWaiters(int phase) {
825		AtomicReference<QNode> head = queueFor(phase);
#	Line 811 | Line 831 | public class Phaser {
831		}
832
833		/**
834	<	* Tries to enqueue given node in the appropriate wait queue
834	>	* Tries to enqueue given node in the appropriate wait queue.
835	>	*
836		* @return true if successful
837		*/
838		private boolean tryEnqueue(QNode node) {
#	Line 821 | Line 842 | public class Phaser {
842
843		/**
844		* Enqueues node and waits unless aborted or signalled.
845	+	*
846		* @return current phase
847		*/
848		private int untimedWait(int phase) {
#	Line 908 | Line 930 | public class Phaser {
930		return p;
931		}
932
933	<	// Temporary Unsafe mechanics for preliminary release
912	<	private static Unsafe getUnsafe() throws Throwable {
913	<	try {
914	<	return Unsafe.getUnsafe();
915	<	} catch (SecurityException se) {
916	<	try {
917	<	return java.security.AccessController.doPrivileged
918	<	(new java.security.PrivilegedExceptionAction<Unsafe>() {
919	<	public Unsafe run() throws Exception {
920	<	return getUnsafePrivileged();
921	<	}});
922	<	} catch (java.security.PrivilegedActionException e) {
923	<	throw e.getCause();
924	<	}
925	<	}
926	<	}
933	>	// Unsafe mechanics
934
935	<	private static Unsafe getUnsafePrivileged()
936	<	throws NoSuchFieldException, IllegalAccessException {
937	<	Field f = Unsafe.class.getDeclaredField("theUnsafe");
931	<	f.setAccessible(true);
932	<	return (Unsafe) f.get(null);
933	<	}
935	>	private static final sun.misc.Unsafe UNSAFE = getUnsafe();
936	>	private static final long stateOffset =
937	>	objectFieldOffset("state", Phaser.class);
938
939	<	private static long fieldOffset(String fieldName)
940	<	throws NoSuchFieldException {
937	<	return _unsafe.objectFieldOffset
938	<	(Phaser.class.getDeclaredField(fieldName));
939	>	private final boolean casState(long cmp, long val) {
940	>	return UNSAFE.compareAndSwapLong(this, stateOffset, cmp, val);
941		}
942
943	<	static final Unsafe _unsafe;
942	<	static final long stateOffset;
943	<
944	<	static {
943	>	private static long objectFieldOffset(String field, Class<?> klazz) {
944		try {
945	<	_unsafe = getUnsafe();
946	<	stateOffset = fieldOffset("state");
947	<	} catch (Throwable e) {
948	<	throw new RuntimeException("Could not initialize intrinsics", e);
945	>	return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
946	>	} catch (NoSuchFieldException e) {
947	>	// Convert Exception to corresponding Error
948	>	NoSuchFieldError error = new NoSuchFieldError(field);
949	>	error.initCause(e);
950	>	throw error;
951		}
952		}
953
954	<	final boolean casState(long cmp, long val) {
955	<	return _unsafe.compareAndSwapLong(this, stateOffset, cmp, val);
954	>	/**
955	>	* Returns a sun.misc.Unsafe.  Suitable for use in a 3rd party package.
956	>	* Replace with a simple call to Unsafe.getUnsafe when integrating
957	>	* into a jdk.
958	>	*
959	>	* @return a sun.misc.Unsafe
960	>	*/
961	>	private static sun.misc.Unsafe getUnsafe() {
962	>	try {
963	>	return sun.misc.Unsafe.getUnsafe();
964	>	} catch (SecurityException se) {
965	>	try {
966	>	return java.security.AccessController.doPrivileged
967	>	(new java.security
968	>	.PrivilegedExceptionAction<sun.misc.Unsafe>() {
969	>	public sun.misc.Unsafe run() throws Exception {
970	>	java.lang.reflect.Field f = sun.misc
971	>	.Unsafe.class.getDeclaredField("theUnsafe");
972	>	f.setAccessible(true);
973	>	return (sun.misc.Unsafe) f.get(null);
974	>	}});
975	>	} catch (java.security.PrivilegedActionException e) {
976	>	throw new RuntimeException("Could not initialize intrinsics",
977	>	e.getCause());
978	>	}
979	>	}
980		}
981		}

Diff Legend

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