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.57 by jsr166, Wed Oct 28 09:28:30 2009 UTC vs.
Revision 1.76 by jsr166, Wed Sep 1 22:49:09 2010 UTC

#	Line 15 | Line 15 | import java.util.Iterator;
15		import java.util.NoSuchElementException;
16		import java.util.Queue;
17		import java.util.concurrent.locks.LockSupport;
18	+
19		/**
20		* An unbounded {@link TransferQueue} based on linked nodes.
21		* This queue orders elements FIFO (first-in-first-out) with respect
#	Line 206 | Line 207 | public class LinkedTransferQueue<E> exte
207		* additional GC bookkeeping ("write barriers") that are sometimes
208		* more costly than the writes themselves because of contention).
209		*
209	–	* Removal of interior nodes (due to timed out or interrupted
210	–	* waits, or calls to remove(x) or Iterator.remove) can use a
211	–	* scheme roughly similar to that described in Scherer, Lea, and
212	–	* Scott's SynchronousQueue. Given a predecessor, we can unsplice
213	–	* any node except the (actual) tail of the queue. To avoid
214	–	* build-up of cancelled trailing nodes, upon a request to remove
215	–	* a trailing node, it is placed in field "cleanMe" to be
216	–	* unspliced upon the next call to unsplice any other node.
217	–	* Situations needing such mechanics are not common but do occur
218	–	* in practice; for example when an unbounded series of short
219	–	* timed calls to poll repeatedly time out but never otherwise
220	–	* fall off the list because of an untimed call to take at the
221	–	* front of the queue. Note that maintaining field cleanMe does
222	–	* not otherwise much impact garbage retention even if never
223	–	* cleared by some other call because the held node will
224	–	* eventually either directly or indirectly lead to a self-link
225	–	* once off the list.
226	–	*
210		* *** Overview of implementation ***
211		*
212		* We use a threshold-based approach to updates, with a slack
#	Line 239 | Line 222 | public class LinkedTransferQueue<E> exte
222		* per-thread one available, but even ThreadLocalRandom is too
223		* heavy for these purposes.
224		*
225	<	* With such a small slack threshold value, it is rarely
226	<	* worthwhile to augment this with path short-circuiting; i.e.,
227	<	* unsplicing nodes between head and the first unmatched node, or
228	<	* similarly for tail, rather than advancing head or tail
246	<	* proper. However, it is used (in awaitMatch) immediately before
247	<	* a waiting thread starts to block, as a final bit of helping at
248	<	* a point when contention with others is extremely unlikely
249	<	* (since if other threads that could release it are operating,
250	<	* then the current thread wouldn't be blocking).
225	>	* With such a small slack threshold value, it is not worthwhile
226	>	* to augment this with path short-circuiting (i.e., unsplicing
227	>	* interior nodes) except in the case of cancellation/removal (see
228	>	* below).
229		*
230		* We allow both the head and tail fields to be null before any
231		* nodes are enqueued; initializing upon first append.  This
#	Line 329 | Line 307 | public class LinkedTransferQueue<E> exte
307		*    versa) compared to their predecessors receive additional
308		*    chained spins, reflecting longer paths typically required to
309		*    unblock threads during phase changes.
310	+	*
311	+	*
312	+	* ** Unlinking removed interior nodes **
313	+	*
314	+	* In addition to minimizing garbage retention via self-linking
315	+	* described above, we also unlink removed interior nodes. These
316	+	* may arise due to timed out or interrupted waits, or calls to
317	+	* remove(x) or Iterator.remove.  Normally, given a node that was
318	+	* at one time known to be the predecessor of some node s that is
319	+	* to be removed, we can unsplice s by CASing the next field of
320	+	* its predecessor if it still points to s (otherwise s must
321	+	* already have been removed or is now offlist). But there are two
322	+	* situations in which we cannot guarantee to make node s
323	+	* unreachable in this way: (1) If s is the trailing node of list
324	+	* (i.e., with null next), then it is pinned as the target node
325	+	* for appends, so can only be removed later after other nodes are
326	+	* appended. (2) We cannot necessarily unlink s given a
327	+	* predecessor node that is matched (including the case of being
328	+	* cancelled): the predecessor may already be unspliced, in which
329	+	* case some previous reachable node may still point to s.
330	+	* (For further explanation see Herlihy & Shavit "The Art of
331	+	* Multiprocessor Programming" chapter 9).  Although, in both
332	+	* cases, we can rule out the need for further action if either s
333	+	* or its predecessor are (or can be made to be) at, or fall off
334	+	* from, the head of list.
335	+	*
336	+	* Without taking these into account, it would be possible for an
337	+	* unbounded number of supposedly removed nodes to remain
338	+	* reachable.  Situations leading to such buildup are uncommon but
339	+	* can occur in practice; for example when a series of short timed
340	+	* calls to poll repeatedly time out but never otherwise fall off
341	+	* the list because of an untimed call to take at the front of the
342	+	* queue.
343	+	*
344	+	* When these cases arise, rather than always retraversing the
345	+	* entire list to find an actual predecessor to unlink (which
346	+	* won't help for case (1) anyway), we record a conservative
347	+	* estimate of possible unsplice failures (in "sweepVotes").
348	+	* We trigger a full sweep when the estimate exceeds a threshold
349	+	* ("SWEEP_THRESHOLD") indicating the maximum number of estimated
350	+	* removal failures to tolerate before sweeping through, unlinking
351	+	* cancelled nodes that were not unlinked upon initial removal.
352	+	* We perform sweeps by the thread hitting threshold (rather than
353	+	* background threads or by spreading work to other threads)
354	+	* because in the main contexts in which removal occurs, the
355	+	* caller is already timed-out, cancelled, or performing a
356	+	* potentially O(n) operation (e.g. remove(x)), none of which are
357	+	* time-critical enough to warrant the overhead that alternatives
358	+	* would impose on other threads.
359	+	*
360	+	* Because the sweepVotes estimate is conservative, and because
361	+	* nodes become unlinked "naturally" as they fall off the head of
362	+	* the queue, and because we allow votes to accumulate even while
363	+	* sweeps are in progress, there are typically significantly fewer
364	+	* such nodes than estimated.  Choice of a threshold value
365	+	* balances the likelihood of wasted effort and contention, versus
366	+	* providing a worst-case bound on retention of interior nodes in
367	+	* quiescent queues. The value defined below was chosen
368	+	* empirically to balance these under various timeout scenarios.
369	+	*
370	+	* Note that we cannot self-link unlinked interior nodes during
371	+	* sweeps. However, the associated garbage chains terminate when
372	+	* some successor ultimately falls off the head of the list and is
373	+	* self-linked.
374		*/
375
376		/** True if on multiprocessor */
#	Line 355 | Line 397 | public class LinkedTransferQueue<E> exte
397		private static final int CHAINED_SPINS = FRONT_SPINS >>> 1;
398
399		/**
400	+	* The maximum number of estimated removal failures (sweepVotes)
401	+	* to tolerate before sweeping through the queue unlinking
402	+	* cancelled nodes that were not unlinked upon initial
403	+	* removal. See above for explanation. The value must be at least
404	+	* two to avoid useless sweeps when removing trailing nodes.
405	+	*/
406	+	static final int SWEEP_THRESHOLD = 32;
407	+
408	+	/**
409		* Queue nodes. Uses Object, not E, for items to allow forgetting
410		* them after use.  Relies heavily on Unsafe mechanics to minimize
411	<	* unnecessary ordering constraints: Writes that intrinsically
412	<	* precede or follow CASes use simple relaxed forms.  Other
362	<	* cleanups use releasing/lazy writes.
411	>	* unnecessary ordering constraints: Writes that are intrinsically
412	>	* ordered wrt other accesses or CASes use simple relaxed forms.
413		*/
414	<	static final class Node<E> {
414	>	static final class Node {
415		final boolean isData;   // false if this is a request node
416		volatile Object item;   // initially non-null if isData; CASed to match
417	<	volatile Node<E> next;
417	>	volatile Node next;
418		volatile Thread waiter; // null until waiting
419
420		// CAS methods for fields
421	<	final boolean casNext(Node<E> cmp, Node<E> val) {
421	>	final boolean casNext(Node cmp, Node val) {
422		return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
423		}
424
425		final boolean casItem(Object cmp, Object val) {
426	<	assert cmp == null || cmp.getClass() != Node.class;
426	>	//            assert cmp == null || cmp.getClass() != Node.class;
427		return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
428		}
429
430		/**
431	<	* Creates a new node. Uses relaxed write because item can only
432	<	* be seen if followed by CAS.
431	>	* Constructs a new node.  Uses relaxed write because item can
432	>	* only be seen after publication via casNext.
433		*/
434	<	Node(E item, boolean isData) {
434	>	Node(Object item, boolean isData) {
435		UNSAFE.putObject(this, itemOffset, item); // relaxed write
436		this.isData = isData;
437		}
#	Line 395 | Line 445 | public class LinkedTransferQueue<E> exte
445		}
446
447		/**
448	<	* Sets item to self (using a releasing/lazy write) and waiter
449	<	* to null, to avoid garbage retention after extracting or
450	<	* cancelling.
448	>	* Sets item to self and waiter to null, to avoid garbage
449	>	* retention after matching or cancelling. Uses relaxed writes
450	>	* because order is already constrained in the only calling
451	>	* contexts: item is forgotten only after volatile/atomic
452	>	* mechanics that extract items.  Similarly, clearing waiter
453	>	* follows either CAS or return from park (if ever parked;
454	>	* else we don't care).
455		*/
456		final void forgetContents() {
457	<	UNSAFE.putOrderedObject(this, itemOffset, this);
458	<	UNSAFE.putOrderedObject(this, waiterOffset, null);
457	>	UNSAFE.putObject(this, itemOffset, this);
458	>	UNSAFE.putObject(this, waiterOffset, null);
459		}
460
461		/**
#	Line 414 | Line 468 | public class LinkedTransferQueue<E> exte
468		}
469
470		/**
471	+	* Returns true if this is an unmatched request node.
472	+	*/
473	+	final boolean isUnmatchedRequest() {
474	+	return !isData && item == null;
475	+	}
476	+
477	+	/**
478		* Returns true if a node with the given mode cannot be
479		* appended to this node because this node is unmatched and
480		* has opposite data mode.
#	Line 428 | Line 489 | public class LinkedTransferQueue<E> exte
489		* Tries to artificially match a data node -- used by remove.
490		*/
491		final boolean tryMatchData() {
492	+	//            assert isData;
493		Object x = item;
494		if (x != null && x != this && casItem(x, null)) {
495		LockSupport.unpark(waiter);
#	Line 449 | Line 511 | public class LinkedTransferQueue<E> exte
511		}
512
513		/** head of the queue; null until first enqueue */
514	<	transient volatile Node<E> head;
453	<
454	<	/** predecessor of dangling unspliceable node */
455	<	private transient volatile Node<E> cleanMe; // decl here reduces contention
514	>	transient volatile Node head;
515
516		/** tail of the queue; null until first append */
517	<	private transient volatile Node<E> tail;
517	>	private transient volatile Node tail;
518	>
519	>	/** The number of apparent failures to unsplice removed nodes */
520	>	private transient volatile int sweepVotes;
521
522		// CAS methods for fields
523	<	private boolean casTail(Node<E> cmp, Node<E> val) {
523	>	private boolean casTail(Node cmp, Node val) {
524		return UNSAFE.compareAndSwapObject(this, tailOffset, cmp, val);
525		}
526
527	<	private boolean casHead(Node<E> cmp, Node<E> val) {
527	>	private boolean casHead(Node cmp, Node val) {
528		return UNSAFE.compareAndSwapObject(this, headOffset, cmp, val);
529		}
530
531	<	private boolean casCleanMe(Node<E> cmp, Node<E> val) {
532	<	return UNSAFE.compareAndSwapObject(this, cleanMeOffset, cmp, val);
531	>	private boolean casSweepVotes(int cmp, int val) {
532	>	return UNSAFE.compareAndSwapInt(this, sweepVotesOffset, cmp, val);
533		}
534
535		/*
536	<	* Possible values for "how" argument in xfer method. Beware that
475	<	* the order of assigned numerical values matters.
536	>	* Possible values for "how" argument in xfer method.
537		*/
538	<	private static final int NOW     = 0; // for untimed poll, tryTransfer
539	<	private static final int ASYNC   = 1; // for offer, put, add
540	<	private static final int SYNC    = 2; // for transfer, take
541	<	private static final int TIMEOUT = 3; // for timed poll, tryTransfer
538	>	private static final int NOW   = 0; // for untimed poll, tryTransfer
539	>	private static final int ASYNC = 1; // for offer, put, add
540	>	private static final int SYNC  = 2; // for transfer, take
541	>	private static final int TIMED = 3; // for timed poll, tryTransfer
542
543		@SuppressWarnings("unchecked")
544		static <E> E cast(Object item) {
545	<	assert item == null || item.getClass() != Node.class;
545	>	//        assert item == null || item.getClass() != Node.class;
546		return (E) item;
547		}
548
#	Line 490 | Line 551 | public class LinkedTransferQueue<E> exte
551		*
552		* @param e the item or null for take
553		* @param haveData true if this is a put, else a take
554	<	* @param how NOW, ASYNC, SYNC, or TIMEOUT
555	<	* @param nanos timeout in nanosecs, used only if mode is TIMEOUT
554	>	* @param how NOW, ASYNC, SYNC, or TIMED
555	>	* @param nanos timeout in nanosecs, used only if mode is TIMED
556		* @return an item if matched, else e
557		* @throws NullPointerException if haveData mode but e is null
558		*/
559		private E xfer(E e, boolean haveData, int how, long nanos) {
560		if (haveData && (e == null))
561		throw new NullPointerException();
562	<	Node<E> s = null;                     // the node to append, if needed
562	>	Node s = null;                        // the node to append, if needed
563
564		retry: for (;;) {                     // restart on append race
565
566	<	for (Node<E> h = head, p = h; p != null;) {
506	<	// find & match first node
566	>	for (Node h = head, p = h; p != null;) { // find & match first node
567		boolean isData = p.isData;
568		Object item = p.item;
569		if (item != p && (item != null) == isData) { // unmatched
570		if (isData == haveData)   // can't match
571		break;
572		if (p.casItem(item, e)) { // match
573	<	for (Node<E> q = p; q != h;) {
574	<	Node<E> n = q.next; // update head by 2
575	<	if (n != null)    // unless singleton
516	<	q = n;
517	<	if (head == h && casHead(h, q)) {
573	>	for (Node q = p; q != h;) {
574	>	Node n = q.next;  // update by 2 unless singleton
575	>	if (head == h && casHead(h, n == null? q : n)) {
576		h.forgetNext();
577		break;
578		}                 // advance and retry
#	Line 526 | Line 584 | public class LinkedTransferQueue<E> exte
584		return this.<E>cast(item);
585		}
586		}
587	<	Node<E> n = p.next;
587	>	Node n = p.next;
588		p = (p != n) ? n : (h = head); // Use head if p offlist
589		}
590
591	<	if (how >= ASYNC) {               // No matches available
591	>	if (how != NOW) {                 // No matches available
592		if (s == null)
593	<	s = new Node<E>(e, haveData);
594	<	Node<E> pred = tryAppend(s, haveData);
593	>	s = new Node(e, haveData);
594	>	Node pred = tryAppend(s, haveData);
595		if (pred == null)
596		continue retry;           // lost race vs opposite mode
597	<	if (how >= SYNC)
598	<	return awaitMatch(s, pred, e, how, nanos);
597	>	if (how != ASYNC)
598	>	return awaitMatch(s, pred, e, (how == TIMED), nanos);
599		}
600		return e; // not waiting
601		}
#	Line 552 | Line 610 | public class LinkedTransferQueue<E> exte
610		* different mode, else s's predecessor, or s itself if no
611		* predecessor
612		*/
613	<	private Node<E> tryAppend(Node<E> s, boolean haveData) {
614	<	for (Node<E> t = tail, p = t;;) { // move p to last node and append
615	<	Node<E> n, u;                     // temps for reads of next & tail
613	>	private Node tryAppend(Node s, boolean haveData) {
614	>	for (Node t = tail, p = t;;) {        // move p to last node and append
615	>	Node n, u;                        // temps for reads of next & tail
616		if (p == null && (p = head) == null) {
617		if (casHead(null, s))
618		return s;                 // initialize
#	Line 586 | Line 644 | public class LinkedTransferQueue<E> exte
644		* predecessor, or null if unknown (the null case does not occur
645		* in any current calls but may in possible future extensions)
646		* @param e the comparison value for checking match
647	<	* @param how either SYNC or TIMEOUT
648	<	* @param nanos timeout value
647	>	* @param timed if true, wait only until timeout elapses
648	>	* @param nanos timeout in nanosecs, used only if timed is true
649		* @return matched item, or e if unmatched on interrupt or timeout
650		*/
651	<	private E awaitMatch(Node<E> s, Node<E> pred, E e, int how, long nanos) {
652	<	long lastTime = (how == TIMEOUT) ? System.nanoTime() : 0L;
651	>	private E awaitMatch(Node s, Node pred, E e, boolean timed, long nanos) {
652	>	long lastTime = timed ? System.nanoTime() : 0L;
653		Thread w = Thread.currentThread();
654		int spins = -1; // initialized after first item and cancel checks
655		ThreadLocalRandom randomYields = null; // bound if needed
#	Line 599 | Line 657 | public class LinkedTransferQueue<E> exte
657		for (;;) {
658		Object item = s.item;
659		if (item != e) {                  // matched
660	<	assert item != s;
660	>	//                assert item != s;
661		s.forgetContents();           // avoid garbage
662		return this.<E>cast(item);
663		}
664	<	if ((w.isInterrupted() || (how == TIMEOUT && nanos <= 0)) &&
665	<	s.casItem(e, s)) {       // cancel
664	>	if ((w.isInterrupted() || (timed && nanos <= 0)) &&
665	>	s.casItem(e, s)) {        // cancel
666		unsplice(pred, s);
667		return e;
668		}
#	Line 614 | Line 672 | public class LinkedTransferQueue<E> exte
672		randomYields = ThreadLocalRandom.current();
673		}
674		else if (spins > 0) {             // spin
675	<	if (--spins == 0)
676	<	shortenHeadPath();        // reduce slack before blocking
619	<	else if (randomYields.nextInt(CHAINED_SPINS) == 0)
675	>	--spins;
676	>	if (randomYields.nextInt(CHAINED_SPINS) == 0)
677		Thread.yield();           // occasionally yield
678		}
679		else if (s.waiter == null) {
680		s.waiter = w;                 // request unpark then recheck
681		}
682	<	else if (how == TIMEOUT) {
682	>	else if (timed) {
683		long now = System.nanoTime();
684		if ((nanos -= now - lastTime) > 0)
685		LockSupport.parkNanos(this, nanos);
#	Line 630 | Line 687 | public class LinkedTransferQueue<E> exte
687		}
688		else {
689		LockSupport.park(this);
633	–	s.waiter = null;
634	–	spins = -1;                   // spin if front upon wakeup
690		}
691		}
692		}
#	Line 640 | Line 695 | public class LinkedTransferQueue<E> exte
695		* Returns spin/yield value for a node with given predecessor and
696		* data mode. See above for explanation.
697		*/
698	<	private static int spinsFor(Node<?> pred, boolean haveData) {
698	>	private static int spinsFor(Node pred, boolean haveData) {
699		if (MP && pred != null) {
700		if (pred.isData != haveData)      // phase change
701		return FRONT_SPINS + CHAINED_SPINS;
#	Line 652 | Line 707 | public class LinkedTransferQueue<E> exte
707		return 0;
708		}
709
710	+	/* -------------- Traversal methods -------------- */
711	+
712		/**
713	<	* Tries (once) to unsplice nodes between head and first unmatched
714	<	* or trailing node; failing on contention.
715	<	*/
716	<	private void shortenHeadPath() {
717	<	Node<E> h, hn, p, q;
718	<	if ((p = h = head) != null && h.isMatched() &&
719	<	(q = hn = h.next) != null) {
663	<	Node<E> n;
664	<	while ((n = q.next) != q) {
665	<	if (n == null || !q.isMatched()) {
666	<	if (hn != q && h.next == hn)
667	<	h.casNext(hn, q);
668	<	break;
669	<	}
670	<	p = q;
671	<	q = n;
672	<	}
673	<	}
713	>	* Returns the successor of p, or the head node if p.next has been
714	>	* linked to self, which will only be true if traversing with a
715	>	* stale pointer that is now off the list.
716	>	*/
717	>	final Node succ(Node p) {
718	>	Node next = p.next;
719	>	return (p == next) ? head : next;
720		}
721
676	–	/* -------------- Traversal methods -------------- */
677	–
722		/**
723		* Returns the first unmatched node of the given mode, or null if
724		* none.  Used by methods isEmpty, hasWaitingConsumer.
725		*/
726	<	private Node<E> firstOfMode(boolean data) {
727	<	for (Node<E> p = head; p != null; ) {
726	>	private Node firstOfMode(boolean isData) {
727	>	for (Node p = head; p != null; p = succ(p)) {
728		if (!p.isMatched())
729	<	return (p.isData == data) ? p : null;
686	<	Node<E> n = p.next;
687	<	p = (n != p) ? n : head;
729	>	return (p.isData == isData) ? p : null;
730		}
731		return null;
732		}
#	Line 694 | Line 736 | public class LinkedTransferQueue<E> exte
736		* null if none.  Used by peek.
737		*/
738		private E firstDataItem() {
739	<	for (Node<E> p = head; p != null; ) {
698	<	boolean isData = p.isData;
739	>	for (Node p = head; p != null; p = succ(p)) {
740		Object item = p.item;
741	<	if (item != p && (item != null) == isData)
742	<	return isData ? this.<E>cast(item) : null;
743	<	Node<E> n = p.next;
744	<	p = (n != p) ? n : head;
741	>	if (p.isData) {
742	>	if (item != null && item != p)
743	>	return this.<E>cast(item);
744	>	}
745	>	else if (item == null)
746	>	return null;
747		}
748		return null;
749		}
#	Line 711 | Line 754 | public class LinkedTransferQueue<E> exte
754		*/
755		private int countOfMode(boolean data) {
756		int count = 0;
757	<	for (Node<E> p = head; p != null; ) {
757	>	for (Node p = head; p != null; ) {
758		if (!p.isMatched()) {
759		if (p.isData != data)
760		return 0;
761		if (++count == Integer.MAX_VALUE) // saturated
762		break;
763		}
764	<	Node<E> n = p.next;
764	>	Node n = p.next;
765		if (n != p)
766		p = n;
767		else {
#	Line 730 | Line 773 | public class LinkedTransferQueue<E> exte
773		}
774
775		final class Itr implements Iterator<E> {
776	<	private Node<E> nextNode;   // next node to return item for
777	<	private E nextItem;         // the corresponding item
778	<	private Node<E> lastRet;    // last returned node, to support remove
776	>	private Node nextNode;   // next node to return item for
777	>	private E nextItem;      // the corresponding item
778	>	private Node lastRet;    // last returned node, to support remove
779	>	private Node lastPred;   // predecessor to unlink lastRet
780
781		/**
782		* Moves to next node after prev, or first node if prev null.
783		*/
784	<	private void advance(Node<E> prev) {
784	>	private void advance(Node prev) {
785	>	lastPred = lastRet;
786		lastRet = prev;
787	<	Node<E> p;
788	<	if (prev == null || (p = prev.next) == prev)
744	<	p = head;
745	<	while (p != null) {
787	>	for (Node p = (prev == null) ? head : succ(prev);
788	>	p != null; p = succ(p)) {
789		Object item = p.item;
790		if (p.isData) {
791		if (item != null && item != p) {
#	Line 753 | Line 796 | public class LinkedTransferQueue<E> exte
796		}
797		else if (item == null)
798		break;
756	–	Node<E> n = p.next;
757	–	p = (n != p) ? n : head;
799		}
800		nextNode = null;
801		}
#	Line 768 | Line 809 | public class LinkedTransferQueue<E> exte
809		}
810
811		public final E next() {
812	<	Node<E> p = nextNode;
812	>	Node p = nextNode;
813		if (p == null) throw new NoSuchElementException();
814		E e = nextItem;
815		advance(p);
#	Line 776 | Line 817 | public class LinkedTransferQueue<E> exte
817		}
818
819		public final void remove() {
820	<	Node<E> p = lastRet;
820	>	Node p = lastRet;
821		if (p == null) throw new IllegalStateException();
822	<	lastRet = null;
823	<	findAndRemoveNode(p);
822	>	if (p.tryMatchData())
823	>	unsplice(lastPred, p);
824		}
825		}
826
#	Line 789 | Line 830 | public class LinkedTransferQueue<E> exte
830		* Unsplices (now or later) the given deleted/cancelled node with
831		* the given predecessor.
832		*
833	<	* @param pred predecessor of node to be unspliced
833	>	* @param pred a node that was at one time known to be the
834	>	* predecessor of s, or null or s itself if s is/was at head
835		* @param s the node to be unspliced
836		*/
837	<	private void unsplice(Node<E> pred, Node<E> s) {
838	<	s.forgetContents(); // clear unneeded fields
837	>	final void unsplice(Node pred, Node s) {
838	>	s.forgetContents(); // forget unneeded fields
839		/*
840	<	* At any given time, exactly one node on list cannot be
841	<	* unlinked -- the last inserted node. To accommodate this, if
842	<	* we cannot unlink s, we save its predecessor as "cleanMe",
843	<	* processing the previously saved version first. Because only
844	<	* one node in the list can have a null next, at least one of
803	<	* node s or the node previously saved can always be
804	<	* processed, so this always terminates.
840	>	* See above for rationale. Briefly: if pred still points to
841	>	* s, try to unlink s.  If s cannot be unlinked, because it is
842	>	* trailing node or pred might be unlinked, and neither pred
843	>	* nor s are head or offlist, add to sweepVotes, and if enough
844	>	* votes have accumulated, sweep.
845		*/
846	<	if (pred != null && pred != s) {
847	<	while (pred.next == s) {
848	<	Node<E> oldpred = (cleanMe == null) ? null : reclean();
849	<	Node<E> n = s.next;
850	<	if (n != null) {
851	<	if (n != s)
852	<	pred.casNext(s, n);
853	<	break;
846	>	if (pred != null && pred != s && pred.next == s) {
847	>	Node n = s.next;
848	>	if (n == null ||
849	>	(n != s && pred.casNext(s, n) && pred.isMatched())) {
850	>	for (;;) {               // check if at, or could be, head
851	>	Node h = head;
852	>	if (h == pred || h == s || h == null)
853	>	return;          // at head or list empty
854	>	if (!h.isMatched())
855	>	break;
856	>	Node hn = h.next;
857	>	if (hn == null)
858	>	return;          // now empty
859	>	if (hn != h && casHead(h, hn))
860	>	h.forgetNext();  // advance head
861	>	}
862	>	if (pred.next != pred && s.next != s) { // recheck if offlist
863	>	for (;;) {           // sweep now if enough votes
864	>	int v = sweepVotes;
865	>	if (v < SWEEP_THRESHOLD) {
866	>	if (casSweepVotes(v, v + 1))
867	>	break;
868	>	}
869	>	else if (casSweepVotes(v, 0)) {
870	>	sweep();
871	>	break;
872	>	}
873	>	}
874		}
815	–	if (oldpred == pred ||      // Already saved
816	–	(oldpred == null && casCleanMe(null, pred)))
817	–	break;                  // Postpone cleaning
875		}
876		}
877		}
878
879		/**
880	<	* Tries to unsplice the deleted/cancelled node held in cleanMe
824	<	* that was previously uncleanable because it was at tail.
825	<	*
826	<	* @return current cleanMe node (or null)
880	>	* Unlinks matched nodes encountered in a traversal from head.
881		*/
882	<	private Node<E> reclean() {
883	<	/*
884	<	* cleanMe is, or at one time was, predecessor of a cancelled
885	<	* node s that was the tail so could not be unspliced.  If it
886	<	* is no longer the tail, try to unsplice if necessary and
887	<	* make cleanMe slot available.  This differs from similar
888	<	* code in unsplice() because we must check that pred still
835	<	* points to a matched node that can be unspliced -- if not,
836	<	* we can (must) clear cleanMe without unsplicing.  This can
837	<	* loop only due to contention.
838	<	*/
839	<	Node<E> pred;
840	<	while ((pred = cleanMe) != null) {
841	<	Node<E> s = pred.next;
842	<	Node<E> n;
843	<	if (s == null || s == pred || !s.isMatched())
844	<	casCleanMe(pred, null); // already gone
845	<	else if ((n = s.next) != null) {
846	<	if (n != s)
847	<	pred.casNext(s, n);
848	<	casCleanMe(pred, null);
849	<	}
850	<	else
882	>	private void sweep() {
883	>	for (Node p = head, s, n; p != null && (s = p.next) != null; ) {
884	>	if (p == s)                    // stale
885	>	p = head;
886	>	else if (!s.isMatched())
887	>	p = s;
888	>	else if ((n = s.next) == null) // trailing node is pinned
889		break;
890	<	}
891	<	return pred;
854	<	}
855	<
856	<	/**
857	<	* Main implementation of Iterator.remove(). Find
858	<	* and unsplice the given node.
859	<	*/
860	<	final void findAndRemoveNode(Node<E> s) {
861	<	if (s.tryMatchData()) {
862	<	Node<E> pred = null;
863	<	Node<E> p = head;
864	<	while (p != null) {
865	<	if (p == s) {
866	<	unsplice(pred, p);
867	<	break;
868	<	}
869	<	if (!p.isData && !p.isMatched())
870	<	break;
871	<	pred = p;
872	<	if ((p = p.next) == pred) { // stale
873	<	pred = null;
874	<	p = head;
875	<	}
876	<	}
890	>	else
891	>	p.casNext(s, n);
892		}
893		}
894
#	Line 882 | Line 897 | public class LinkedTransferQueue<E> exte
897		*/
898		private boolean findAndRemove(Object e) {
899		if (e != null) {
900	<	Node<E> pred = null;
886	<	Node<E> p = head;
887	<	while (p != null) {
900	>	for (Node pred = null, p = head; p != null; ) {
901		Object item = p.item;
902		if (p.isData) {
903		if (item != null && item != p && e.equals(item) &&
#	Line 896 | Line 909 | public class LinkedTransferQueue<E> exte
909		else if (item == null)
910		break;
911		pred = p;
912	<	if ((p = p.next) == pred) {
912	>	if ((p = p.next) == pred) { // stale
913		pred = null;
914		p = head;
915		}
#	Line 1024 | Line 1037 | public class LinkedTransferQueue<E> exte
1037		*/
1038		public boolean tryTransfer(E e, long timeout, TimeUnit unit)
1039		throws InterruptedException {
1040	<	if (xfer(e, true, TIMEOUT, unit.toNanos(timeout)) == null)
1040	>	if (xfer(e, true, TIMED, unit.toNanos(timeout)) == null)
1041		return true;
1042		if (!Thread.interrupted())
1043		return false;
#	Line 1040 | Line 1053 | public class LinkedTransferQueue<E> exte
1053		}
1054
1055		public E poll(long timeout, TimeUnit unit) throws InterruptedException {
1056	<	E e = xfer(null, false, TIMEOUT, unit.toNanos(timeout));
1056	>	E e = xfer(null, false, TIMED, unit.toNanos(timeout));
1057		if (e != null || !Thread.interrupted())
1058		return e;
1059		throw new InterruptedException();
#	Line 1113 | Line 1126 | public class LinkedTransferQueue<E> exte
1126		* @return {@code true} if this queue contains no elements
1127		*/
1128		public boolean isEmpty() {
1129	<	return firstOfMode(true) == null;
1129	>	for (Node p = head; p != null; p = succ(p)) {
1130	>	if (!p.isMatched())
1131	>	return !p.isData;
1132	>	}
1133	>	return true;
1134		}
1135
1136		public boolean hasWaitingConsumer() {
#	Line 1207 | Line 1224 | public class LinkedTransferQueue<E> exte
1224		objectFieldOffset(UNSAFE, "head", LinkedTransferQueue.class);
1225		private static final long tailOffset =
1226		objectFieldOffset(UNSAFE, "tail", LinkedTransferQueue.class);
1227	<	private static final long cleanMeOffset =
1228	<	objectFieldOffset(UNSAFE, "cleanMe", LinkedTransferQueue.class);
1227	>	private static final long sweepVotesOffset =
1228	>	objectFieldOffset(UNSAFE, "sweepVotes", LinkedTransferQueue.class);
1229
1230		static long objectFieldOffset(sun.misc.Unsafe UNSAFE,
1231		String field, Class<?> klazz) {

Diff Legend

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