ViewVC Help

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

root/jsr166/jsr166/src/jsr166y/LinkedTransferQueue.java

Comparing jsr166/src/jsr166y/LinkedTransferQueue.java (file contents):
Revision 1.51 by dl, Sat Oct 24 14:33:29 2009 UTC vs.
Revision 1.58 by jsr166, Wed Oct 28 10:23:38 2009 UTC

#	Line 361 | Line 361 | public class LinkedTransferQueue<E> exte
361		* precede or follow CASes use simple relaxed forms.  Other
362		* cleanups use releasing/lazy writes.
363		*/
364	<	static final class Node {
364	>	static final class Node<E> {
365		final boolean isData;   // false if this is a request node
366		volatile Object item;   // initially non-null if isData; CASed to match
367	<	volatile Node next;
367	>	volatile Node<E> next;
368		volatile Thread waiter; // null until waiting
369
370		// CAS methods for fields
371	<	final boolean casNext(Node cmp, Node val) {
371	>	final boolean casNext(Node<E> cmp, Node<E> val) {
372		return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
373		}
374
375		final boolean casItem(Object cmp, Object val) {
376	+	assert cmp == null || cmp.getClass() != Node.class;
377		return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
378		}
379
#	Line 380 | Line 381 | public class LinkedTransferQueue<E> exte
381		* Creates a new node. Uses relaxed write because item can only
382		* be seen if followed by CAS.
383		*/
384	<	Node(Object item, boolean isData) {
384	>	Node(E item, boolean isData) {
385		UNSAFE.putObject(this, itemOffset, item); // relaxed write
386		this.isData = isData;
387		}
#	Line 409 | Line 410 | public class LinkedTransferQueue<E> exte
410		*/
411		final boolean isMatched() {
412		Object x = item;
413	<	return x == this || (x != null) != isData;
413	>	return (x == this) || ((x == null) == isData);
414	>	}
415	>
416	>	/**
417	>	* Returns true if this is an unmatched request node.
418	>	*/
419	>	final boolean isUnmatchedRequest() {
420	>	return !isData && item == null;
421		}
422
423		/**
#	Line 427 | Line 435 | public class LinkedTransferQueue<E> exte
435		* Tries to artificially match a data node -- used by remove.
436		*/
437		final boolean tryMatchData() {
438	+	assert isData;
439		Object x = item;
440		if (x != null && x != this && casItem(x, null)) {
441		LockSupport.unpark(waiter);
#	Line 448 | Line 457 | public class LinkedTransferQueue<E> exte
457		}
458
459		/** head of the queue; null until first enqueue */
460	<	private transient volatile Node head;
460	>	transient volatile Node<E> head;
461
462		/** predecessor of dangling unspliceable node */
463	<	private transient volatile Node cleanMe; // decl here to reduce contention
463	>	private transient volatile Node<E> cleanMe; // decl here reduces contention
464
465		/** tail of the queue; null until first append */
466	<	private transient volatile Node tail;
466	>	private transient volatile Node<E> tail;
467
468		// CAS methods for fields
469	<	private boolean casTail(Node cmp, Node val) {
469	>	private boolean casTail(Node<E> cmp, Node<E> val) {
470		return UNSAFE.compareAndSwapObject(this, tailOffset, cmp, val);
471		}
472
473	<	private boolean casHead(Node cmp, Node val) {
473	>	private boolean casHead(Node<E> cmp, Node<E> val) {
474		return UNSAFE.compareAndSwapObject(this, headOffset, cmp, val);
475		}
476
477	<	private boolean casCleanMe(Node cmp, Node val) {
477	>	private boolean casCleanMe(Node<E> cmp, Node<E> val) {
478		return UNSAFE.compareAndSwapObject(this, cleanMeOffset, cmp, val);
479		}
480
#	Line 478 | Line 487 | public class LinkedTransferQueue<E> exte
487		private static final int SYNC    = 2; // for transfer, take
488		private static final int TIMEOUT = 3; // for timed poll, tryTransfer
489
490	+	@SuppressWarnings("unchecked")
491	+	static <E> E cast(Object item) {
492	+	assert item == null || item.getClass() != Node.class;
493	+	return (E) item;
494	+	}
495	+
496		/**
497		* Implements all queuing methods. See above for explanation.
498		*
#	Line 488 | Line 503 | public class LinkedTransferQueue<E> exte
503		* @return an item if matched, else e
504		* @throws NullPointerException if haveData mode but e is null
505		*/
506	<	private Object xfer(Object e, boolean haveData, int how, long nanos) {
506	>	private E xfer(E e, boolean haveData, int how, long nanos) {
507		if (haveData && (e == null))
508		throw new NullPointerException();
509	<	Node s = null;                        // the node to append, if needed
509	>	Node<E> s = null;                     // the node to append, if needed
510
511		retry: for (;;) {                     // restart on append race
512
513	<	for (Node h = head, p = h; p != null;) { // find & match first node
513	>	for (Node<E> h = head, p = h; p != null;) {
514	>	// find & match first node
515		boolean isData = p.isData;
516		Object item = p.item;
517		if (item != p && (item != null) == isData) { // unmatched
518		if (isData == haveData)   // can't match
519		break;
520		if (p.casItem(item, e)) { // match
521	<	LockSupport.unpark(p.waiter);
522	<	while (p != h) {      // update head
523	<	Node n = p.next;  // by 2 unless singleton
524	<	if (n != null)
525	<	p = n;
510	<	if (head == h && casHead(h, p)) {
521	>	for (Node<E> q = p; q != h;) {
522	>	Node<E> n = q.next; // update head by 2
523	>	if (n != null)    // unless singleton
524	>	q = n;
525	>	if (head == h && casHead(h, q)) {
526		h.forgetNext();
527		break;
528		}                 // advance and retry
529		if ((h = head)   == null ||
530	<	(p = h.next) == null || !p.isMatched())
530	>	(q = h.next) == null || !q.isMatched())
531		break;        // unless slack < 2
532		}
533	<	return item;
533	>	LockSupport.unpark(p.waiter);
534	>	return this.<E>cast(item);
535		}
536		}
537	<	Node n = p.next;
537	>	Node<E> n = p.next;
538		p = (p != n) ? n : (h = head); // Use head if p offlist
539		}
540
541		if (how >= ASYNC) {               // No matches available
542		if (s == null)
543	<	s = new Node(e, haveData);
544	<	Node pred = tryAppend(s, haveData);
543	>	s = new Node<E>(e, haveData);
544	>	Node<E> pred = tryAppend(s, haveData);
545		if (pred == null)
546		continue retry;           // lost race vs opposite mode
547		if (how >= SYNC)
#	Line 544 | Line 560 | public class LinkedTransferQueue<E> exte
560		* different mode, else s's predecessor, or s itself if no
561		* predecessor
562		*/
563	<	private Node tryAppend(Node s, boolean haveData) {
564	<	for (Node t = tail, p = t;;) { // move p to last node and append
565	<	Node n, u;                        // temps for reads of next & tail
563	>	private Node<E> tryAppend(Node<E> s, boolean haveData) {
564	>	for (Node<E> t = tail, p = t;;) { // move p to last node and append
565	>	Node<E> n, u;                     // temps for reads of next & tail
566		if (p == null && (p = head) == null) {
567		if (casHead(null, s))
568		return s;                 // initialize
#	Line 582 | Line 598 | public class LinkedTransferQueue<E> exte
598		* @param nanos timeout value
599		* @return matched item, or e if unmatched on interrupt or timeout
600		*/
601	<	private Object awaitMatch(Node s, Node pred, Object e,
586	<	int how, long nanos) {
601	>	private E awaitMatch(Node<E> s, Node<E> pred, E e, int how, long nanos) {
602		long lastTime = (how == TIMEOUT) ? System.nanoTime() : 0L;
603		Thread w = Thread.currentThread();
604		int spins = -1; // initialized after first item and cancel checks
#	Line 592 | Line 607 | public class LinkedTransferQueue<E> exte
607		for (;;) {
608		Object item = s.item;
609		if (item != e) {                  // matched
610	+	assert item != s;
611		s.forgetContents();           // avoid garbage
612	<	return item;
612	>	return this.<E>cast(item);
613		}
614		if ((w.isInterrupted() || (how == TIMEOUT && nanos <= 0)) &&
615	<	s.casItem(e, s)) {       // cancel
615	>	s.casItem(e, s)) {       // cancel
616		unsplice(pred, s);
617		return e;
618		}
#	Line 632 | Line 648 | public class LinkedTransferQueue<E> exte
648		* Returns spin/yield value for a node with given predecessor and
649		* data mode. See above for explanation.
650		*/
651	<	private static int spinsFor(Node pred, boolean haveData) {
651	>	private static int spinsFor(Node<?> pred, boolean haveData) {
652		if (MP && pred != null) {
653		if (pred.isData != haveData)      // phase change
654		return FRONT_SPINS + CHAINED_SPINS;
#	Line 649 | Line 665 | public class LinkedTransferQueue<E> exte
665		* or trailing node; failing on contention.
666		*/
667		private void shortenHeadPath() {
668	<	Node h, hn, p, q;
668	>	Node<E> h, hn, p, q;
669		if ((p = h = head) != null && h.isMatched() &&
670		(q = hn = h.next) != null) {
671	<	Node n;
671	>	Node<E> n;
672		while ((n = q.next) != q) {
673		if (n == null || !q.isMatched()) {
674		if (hn != q && h.next == hn)
#	Line 671 | Line 687 | public class LinkedTransferQueue<E> exte
687		* Returns the first unmatched node of the given mode, or null if
688		* none.  Used by methods isEmpty, hasWaitingConsumer.
689		*/
690	<	private Node firstOfMode(boolean data) {
691	<	for (Node p = head; p != null; ) {
690	>	private Node<E> firstOfMode(boolean data) {
691	>	for (Node<E> p = head; p != null; ) {
692		if (!p.isMatched())
693		return (p.isData == data) ? p : null;
694	<	Node n = p.next;
694	>	Node<E> n = p.next;
695		p = (n != p) ? n : head;
696		}
697		return null;
#	Line 683 | Line 699 | public class LinkedTransferQueue<E> exte
699
700		/**
701		* Returns the item in the first unmatched node with isData; or
702	<	* null if none. Used by peek.
702	>	* null if none.  Used by peek.
703		*/
704	<	private Object firstDataItem() {
705	<	for (Node p = head; p != null; ) {
704	>	private E firstDataItem() {
705	>	for (Node<E> p = head; p != null; ) {
706		boolean isData = p.isData;
707		Object item = p.item;
708		if (item != p && (item != null) == isData)
709	<	return isData ? item : null;
710	<	Node n = p.next;
709	>	return isData ? this.<E>cast(item) : null;
710	>	Node<E> n = p.next;
711		p = (n != p) ? n : head;
712		}
713		return null;
#	Line 703 | Line 719 | public class LinkedTransferQueue<E> exte
719		*/
720		private int countOfMode(boolean data) {
721		int count = 0;
722	<	for (Node p = head; p != null; ) {
722	>	for (Node<E> p = head; p != null; ) {
723		if (!p.isMatched()) {
724		if (p.isData != data)
725		return 0;
726		if (++count == Integer.MAX_VALUE) // saturated
727		break;
728		}
729	<	Node n = p.next;
729	>	Node<E> n = p.next;
730		if (n != p)
731		p = n;
732		else {
#	Line 722 | Line 738 | public class LinkedTransferQueue<E> exte
738		}
739
740		final class Itr implements Iterator<E> {
741	<	private Node nextNode;   // next node to return item for
742	<	private Object nextItem; // the corresponding item
743	<	private Node lastRet;    // last returned node, to support remove
741	>	private Node<E> nextNode;   // next node to return item for
742	>	private E nextItem;         // the corresponding item
743	>	private Node<E> lastRet;    // last returned node, to support remove
744
745		/**
746		* Moves to next node after prev, or first node if prev null.
747		*/
748	<	private void advance(Node prev) {
748	>	private void advance(Node<E> prev) {
749		lastRet = prev;
750	<	Node p;
750	>	Node<E> p;
751		if (prev == null || (p = prev.next) == prev)
752		p = head;
753		while (p != null) {
754		Object item = p.item;
755		if (p.isData) {
756		if (item != null && item != p) {
757	<	nextItem = item;
757	>	nextItem = LinkedTransferQueue.this.<E>cast(item);
758		nextNode = p;
759		return;
760		}
761		}
762		else if (item == null)
763		break;
764	<	Node n = p.next;
764	>	Node<E> n = p.next;
765		p = (n != p) ? n : head;
766		}
767		nextNode = null;
#	Line 760 | Line 776 | public class LinkedTransferQueue<E> exte
776		}
777
778		public final E next() {
779	<	Node p = nextNode;
779	>	Node<E> p = nextNode;
780		if (p == null) throw new NoSuchElementException();
781	<	Object e = nextItem;
781	>	E e = nextItem;
782		advance(p);
783	<	return (E) e;
783	>	return e;
784		}
785
786		public final void remove() {
787	<	Node p = lastRet;
787	>	Node<E> p = lastRet;
788		if (p == null) throw new IllegalStateException();
789		lastRet = null;
790	<	findAndRemoveNode(p);
790	>	findAndRemoveDataNode(p);
791		}
792		}
793
#	Line 784 | Line 800 | public class LinkedTransferQueue<E> exte
800		* @param pred predecessor of node to be unspliced
801		* @param s the node to be unspliced
802		*/
803	<	private void unsplice(Node pred, Node s) {
803	>	private void unsplice(Node<E> pred, Node<E> s) {
804		s.forgetContents(); // clear unneeded fields
805		/*
806		* At any given time, exactly one node on list cannot be
#	Line 797 | Line 813 | public class LinkedTransferQueue<E> exte
813		*/
814		if (pred != null && pred != s) {
815		while (pred.next == s) {
816	<	Node oldpred = (cleanMe == null) ? null : reclean();
817	<	Node n = s.next;
816	>	Node<E> oldpred = (cleanMe == null) ? null : reclean();
817	>	Node<E> n = s.next;
818		if (n != null) {
819		if (n != s)
820		pred.casNext(s, n);
#	Line 817 | Line 833 | public class LinkedTransferQueue<E> exte
833		*
834		* @return current cleanMe node (or null)
835		*/
836	<	private Node reclean() {
836	>	private Node<E> reclean() {
837		/*
838		* cleanMe is, or at one time was, predecessor of a cancelled
839		* node s that was the tail so could not be unspliced.  If it
#	Line 828 | Line 844 | public class LinkedTransferQueue<E> exte
844		* we can (must) clear cleanMe without unsplicing.  This can
845		* loop only due to contention.
846		*/
847	<	Node pred;
847	>	Node<E> pred;
848		while ((pred = cleanMe) != null) {
849	<	Node s = pred.next;
850	<	Node n;
849	>	Node<E> s = pred.next;
850	>	Node<E> n;
851		if (s == null || s == pred || !s.isMatched())
852		casCleanMe(pred, null); // already gone
853		else if ((n = s.next) != null) {
#	Line 847 | Line 863 | public class LinkedTransferQueue<E> exte
863
864		/**
865		* Main implementation of Iterator.remove(). Find
866	<	* and unsplice the given node.
866	>	* and unsplice the given data node.
867		*/
868	<	final void findAndRemoveNode(Node s) {
868	>	final void findAndRemoveDataNode(Node<E> s) {
869	>	assert s.isData;
870		if (s.tryMatchData()) {
871	<	Node pred = null;
855	<	Node p = head;
856	<	while (p != null) {
871	>	for (Node<E> pred = null, p = head; p != null; ) {
872		if (p == s) {
873		unsplice(pred, p);
874		break;
875		}
876	<	if (!p.isData && !p.isMatched())
876	>	if (p.isUnmatchedRequest())
877		break;
878		pred = p;
879		if ((p = p.next) == pred) { // stale
#	Line 874 | Line 889 | public class LinkedTransferQueue<E> exte
889		*/
890		private boolean findAndRemove(Object e) {
891		if (e != null) {
892	<	Node pred = null;
878	<	Node p = head;
879	<	while (p != null) {
892	>	for (Node<E> pred = null, p = head; p != null; ) {
893		Object item = p.item;
894		if (p.isData) {
895		if (item != null && item != p && e.equals(item) &&
#	Line 888 | Line 901 | public class LinkedTransferQueue<E> exte
901		else if (item == null)
902		break;
903		pred = p;
904	<	if ((p = p.next) == pred) {
904	>	if ((p = p.next) == pred) { // stale
905		pred = null;
906		p = head;
907		}
#	Line 1024 | Line 1037 | public class LinkedTransferQueue<E> exte
1037		}
1038
1039		public E take() throws InterruptedException {
1040	<	Object e = xfer(null, false, SYNC, 0);
1040	>	E e = xfer(null, false, SYNC, 0);
1041		if (e != null)
1042	<	return (E)e;
1042	>	return e;
1043		Thread.interrupted();
1044		throw new InterruptedException();
1045		}
1046
1047		public E poll(long timeout, TimeUnit unit) throws InterruptedException {
1048	<	Object e = xfer(null, false, TIMEOUT, unit.toNanos(timeout));
1048	>	E e = xfer(null, false, TIMEOUT, unit.toNanos(timeout));
1049		if (e != null || !Thread.interrupted())
1050	<	return (E)e;
1050	>	return e;
1051		throw new InterruptedException();
1052		}
1053
1054		public E poll() {
1055	<	return (E)xfer(null, false, NOW, 0);
1055	>	return xfer(null, false, NOW, 0);
1056		}
1057
1058		/**
#	Line 1096 | Line 1109 | public class LinkedTransferQueue<E> exte
1109		}
1110
1111		public E peek() {
1112	<	return (E) firstDataItem();
1112	>	return firstDataItem();
1113		}
1114
1115		/**
#	Line 1192 | Line 1205 | public class LinkedTransferQueue<E> exte
1205		}
1206		}
1207
1195	–
1208		// Unsafe mechanics
1209
1210		private static final sun.misc.Unsafe UNSAFE = getUnsafe();
#	Line 1215 | Line 1227 | public class LinkedTransferQueue<E> exte
1227		}
1228		}
1229
1230	<	private static sun.misc.Unsafe getUnsafe() {
1230	>	/**
1231	>	* Returns a sun.misc.Unsafe.  Suitable for use in a 3rd party package.
1232	>	* Replace with a simple call to Unsafe.getUnsafe when integrating
1233	>	* into a jdk.
1234	>	*
1235	>	* @return a sun.misc.Unsafe
1236	>	*/
1237	>	static sun.misc.Unsafe getUnsafe() {
1238		try {
1239		return sun.misc.Unsafe.getUnsafe();
1240		} catch (SecurityException se) {

Diff Legend

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