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.44 by jsr166, Tue Aug 4 20:32:16 2009 UTC vs.
Revision 1.45 by dl, Wed Oct 21 16:30:40 2009 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	–	import java.util.concurrent.atomic.AtomicReference;
19	–
18		/**
19		* An unbounded {@link TransferQueue} based on linked nodes.
20		* This queue orders elements FIFO (first-in-first-out) with respect
#	Line 54 | Line 52 | public class LinkedTransferQueue<E> exte
52		private static final long serialVersionUID = -3223113410248163686L;
53
54		/*
55	<	* This class extends the approach used in FIFO-mode
58	<	* SynchronousQueues. See the internal documentation, as well as
59	<	* the PPoPP 2006 paper "Scalable Synchronous Queues" by Scherer,
60	<	* Lea & Scott
61	<	* (http://www.cs.rice.edu/~wns1/papers/2006-PPoPP-SQ.pdf)
55	>	* *** Overview of Dual Queues with Slack ***
56		*
57	<	* The main extension is to provide different Wait modes for the
58	<	* main "xfer" method that puts or takes items.  These don't
59	<	* impact the basic dual-queue logic, but instead control whether
60	<	* or how threads block upon insertion of request or data nodes
61	<	* into the dual queue. It also uses slightly different
62	<	* conventions for tracking whether nodes are off-list or
63	<	* cancelled.
64	<	*/
65	<
66	<	// Wait modes for xfer method
67	<	static final int NOWAIT  = 0;
68	<	static final int TIMEOUT = 1;
69	<	static final int WAIT    = 2;
70	<
71	<	/** The number of CPUs, for spin control */
72	<	static final int NCPUS = Runtime.getRuntime().availableProcessors();
73	<
74	<	/**
75	<	* The number of times to spin before blocking in timed waits.
76	<	* The value is empirically derived -- it works well across a
77	<	* variety of processors and OSes. Empirically, the best value
78	<	* seems not to vary with number of CPUs (beyond 2) so is just
79	<	* a constant.
80	<	*/
81	<	static final int maxTimedSpins = (NCPUS < 2) ? 0 : 32;
57	>	* Dual Queues, introduced by Scherer and Scott
58	>	* (http://www.cs.rice.edu/~wns1/papers/2004-DISC-DDS.pdf) are
59	>	* (linked) queues in which nodes may represent either data or
60	>	* requests.  When a thread tries to enqueue a data node, but
61	>	* encounters a request node, it instead "matches" and removes it;
62	>	* and vice versa for enqueuing requests. Blocking Dual Queues
63	>	* arrange that threads enqueuing unmatched requests block until
64	>	* other threads provide the match. Dual Synchronous Queues (see
65	>	* Scherer, Lea, & Scott
66	>	* http://www.cs.rochester.edu/u/scott/papers/2009_Scherer_CACM_SSQ.pdf)
67	>	* additionally arrange that threads enqueuing unmatched data also
68	>	* block.  Dual Transfer Queues support all of these modes, as
69	>	* dictated by callers.
70	>	*
71	>	* A FIFO dual queue may be implemented using a variation of the
72	>	* Michael & Scott (M&S) lock-free queue algorithm
73	>	* (http://www.cs.rochester.edu/u/scott/papers/1996_PODC_queues.pdf).
74	>	* It maintains two pointer fields, "head", pointing to a
75	>	* (matched) node that in turn points to the first actual
76	>	* (unmatched) queue node (or null if empty); and "tail" that
77	>	* points to the last node on the queue (or again null if
78	>	* empty). For example, here is a possible queue with four data
79	>	* elements:
80	>	*
81	>	*  head                tail
82	>	*    |                   |
83	>	*    v                   v
84	>	*    M -> U -> U -> U -> U
85	>	*
86	>	* The M&S queue algorithm is known to be prone to scalability and
87	>	* overhead limitations when maintaining (via CAS) these head and
88	>	* tail pointers. This has led to the development of
89	>	* contention-reducing variants such as elimination arrays (see
90	>	* Moir et al http://portal.acm.org/citation.cfm?id=1074013) and
91	>	* optimistic back pointers (see Ladan-Mozes & Shavit
92	>	* http://people.csail.mit.edu/edya/publications/OptimisticFIFOQueue-journal.pdf).
93	>	* However, the nature of dual queues enables a simpler tactic for
94	>	* improving M&S-style implementations when dual-ness is needed.
95	>	*
96	>	* In a dual queue, each node must atomically maintain its match
97	>	* status. While there are other possible variants, we implement
98	>	* this here as: for a data-mode node, matching entails CASing an
99	>	* "item" field from a non-null data value to null upon match, and
100	>	* vice-versa for request nodes, CASing from null to a data
101	>	* value. (Note that the linearization properties of this style of
102	>	* queue are easy to verify -- elements are made available by
103	>	* linking, and unavailable by matching.) Compared to plain M&S
104	>	* queues, this property of dual queues requires one additional
105	>	* successful atomic operation per enq/deq pair. But it also
106	>	* enables lower cost variants of queue maintenance mechanics. (A
107	>	* variation of this idea applies even for non-dual queues that
108	>	* support deletion of embedded elements, such as
109	>	* j.u.c.ConcurrentLinkedQueue.)
110	>	*
111	>	* Once a node is matched, its item can never again change.  We
112	>	* may thus arrange that the linked list of them contains a prefix
113	>	* of zero or more matched nodes, followed by a suffix of zero or
114	>	* more unmatched nodes. (Note that we allow both the prefix and
115	>	* suffix to be zero length, which in turn means that we do not
116	>	* use a dummy header.)  If we were not concerned with either time
117	>	* or space efficiency, we could correctly perform enqueue and
118	>	* dequeue operations by traversing from a pointer to the initial
119	>	* node; CASing the item of the first unmatched node on match and
120	>	* CASing the next field of the trailing node on appends.  While
121	>	* this would be a terrible idea in itself, it does have the
122	>	* benefit of not requiring ANY atomic updates on head/tail
123	>	* fields.
124	>	*
125	>	* We introduce here an approach that lies between the extremes of
126	>	* never versus always updating queue (head and tail) pointers
127	>	* that reflects the tradeoff of sometimes require extra traversal
128	>	* steps to locate the first and/or last unmatched nodes, versus
129	>	* the reduced overhead and contention of fewer updates to queue
130	>	* pointers. For example, a possible snapshot of a queue is:
131	>	*
132	>	*  head           tail
133	>	*    |              |
134	>	*    v              v
135	>	*    M -> M -> U -> U -> U -> U
136	>	*
137	>	* The best value for this "slack" (the targeted maximum distance
138	>	* between the value of "head" and the first unmatched node, and
139	>	* similarly for "tail") is an empirical matter. We have found
140	>	* that using very small constants in the range of 1-3 work best
141	>	* over a range of platforms. Larger values introduce increasing
142	>	* costs of cache misses and risks of long traversal chains.
143	>	*
144	>	* Dual queues with slack differ from plain M&S dual queues by
145	>	* virtue of only sometimes updating head or tail pointers when
146	>	* matching, appending, or even traversing nodes; in order to
147	>	* maintain a targeted slack.  The idea of "sometimes" may be
148	>	* operationalized in several ways. The simplest is to use a
149	>	* per-operation counter incremented on each traversal step, and
150	>	* to try (via CAS) to update the associated queue pointer
151	>	* whenever the count exceeds a threshold. Another, that requires
152	>	* more overhead, is to use random number generators to update
153	>	* with a given probability per traversal step.
154	>	*
155	>	* In any strategy along these lines, because CASes updating
156	>	* fields may fail, the actual slack may exceed targeted
157	>	* slack. However, they may be retried at any time to maintain
158	>	* targets.  Even when using very small slack values, this
159	>	* approach works well for dual queues because it allows all
160	>	* operations up to the point of matching or appending an item
161	>	* (hence potentially releasing another thread) to be read-only,
162	>	* thus not introducing any further contention. As described
163	>	* below, we implement this by performing slack maintenance
164	>	* retries only after these points.
165	>	*
166	>	* As an accompaniment to such techniques, traversal overhead can
167	>	* be further reduced without increasing contention of head
168	>	* pointer updates.  During traversals, threads may sometimes
169	>	* shortcut the "next" link path from the current "head" node to
170	>	* be closer to the currently known first unmatched node. Again,
171	>	* this may be triggered with using thresholds or randomization.
172	>	*
173	>	* These ideas must be further extended to avoid unbounded amounts
174	>	* of costly-to-reclaim garbage caused by the sequential "next"
175	>	* links of nodes starting at old forgotten head nodes: As first
176	>	* described in detail by Boehm
177	>	* (http://portal.acm.org/citation.cfm?doid=503272.503282) if a GC
178	>	* delays noticing that any arbitrarily old node has become
179	>	* garbage, all newer dead nodes will also be unreclaimed.
180	>	* (Similar issues arise in non-GC environments.)  To cope with
181	>	* this in our implementation, upon CASing to advance the head
182	>	* pointer, we set the "next" link of the previous head to point
183	>	* only to itself; thus limiting the length connected dead lists.
184	>	* (We also take similar care to wipe out possibly garbage
185	>	* retaining values held in other Node fields.)  However, doing so
186	>	* adds some further complexity to traversal: If any "next"
187	>	* pointer links to itself, it indicates that the current thread
188	>	* has lagged behind a head-update, and so the traversal must
189	>	* continue from the "head".  Traversals trying to find the
190	>	* current tail starting from "tail" may also encounter
191	>	* self-links, in which case they also continue at "head".
192	>	*
193	>	* It is tempting in slack-based scheme to not even use CAS for
194	>	* updates (similarly to Ladan-Mozes & Shavit). However, this
195	>	* cannot be done for head updates under the above link-forgetting
196	>	* mechanics because an update may leave head at a detached node.
197	>	* And while direct writes are possible for tail updates, they
198	>	* increase the risk of long retraversals, and hence long garbage
199	>	* chains which can be much more costly than is worthwhile
200	>	* considering that the cost difference of performing a CAS vs
201	>	* write is smaller when they are not triggered on each operation
202	>	* (especially considering that writes and CASes equally require
203	>	* additional GC bookkeeping ("write barriers") that are sometimes
204	>	* more costly than the writes themselves because of contention).
205	>	*
206	>	* Removal of internal nodes (due to timed out or interrupted
207	>	* waits, or calls to remove or Iterator.remove) uses a scheme
208	>	* roughly similar to that in Scherer, Lea, and Scott
209	>	* SynchronousQueue. Given a predecessor, we can unsplice any node
210	>	* except the (actual) tail of the queue. To avoid build-up of
211	>	* cancelled trailing nodes, upon a request to remove a trailing
212	>	* node, it is placed in field "cleanMe" to be unspliced later.
213	>	*
214	>	* *** Overview of implementation ***
215	>	*
216	>	* We use a threshold-based approach to updates, with a target
217	>	* slack of two.  The slack value is hard-wired: a path greater
218	>	* than one is naturally implemented by checking equality of
219	>	* traversal pointers except when the list has only one element,
220	>	* in which case we keep max slack at one. Avoiding tracking
221	>	* explicit counts across situations slightly simplifies an
222	>	* already-messy implementation. Using randomization would
223	>	* probably work better if there were a low-quality dirt-cheap
224	>	* per-thread one available, but even ThreadLocalRandom is too
225	>	* heavy for these purposes.
226	>	*
227	>	* With such a small slack value, path short-circuiting is rarely
228	>	* worthwhile. However, it is used (in awaitMatch) immediately
229	>	* before a waiting thread starts to block, as a final bit of
230	>	* helping at a point when contention with others is extremely
231	>	* unlikely (since if other threads that could release it are
232	>	* operating, then the current thread wouldn't be blocking).
233	>	*
234	>	* All enqueue/dequeue operations are handled by the single method
235	>	* "xfer" with parameters indicating whether to act as some form
236	>	* of offer, put, poll, take, or transfer (each possibly with
237	>	* timeout). The relative complexity of using one monolithic
238	>	* method outweighs the code bulk and maintenance problems of
239	>	* using nine separate methods.
240	>	*
241	>	* Operation consists of up to three phases. The first is
242	>	* implemented within method xfer, the second in tryAppend, and
243	>	* the third in method awaitMatch.
244	>	*
245	>	* 1. Try to match an existing node
246	>	*
247	>	*    Starting at head, skip already-matched nodes until finding
248	>	*    an unmatched node of opposite mode, if one exists, in which
249	>	*    case matching it and returning, also if necessary updating
250	>	*    head to one past the matched node (or the node itself if the
251	>	*    list has no other unmatched nodes). If the CAS misses, then
252	>	*    a retry loops until the slack is at most two. Traversals
253	>	*    also check if the initial head is now off-list, in which
254	>	*    case they start at the new head.
255	>	*
256	>	*    If no candidates are found and the call was untimed
257	>	*    poll/offer, (argument "how" is NOW) return.
258	>	*
259	>	* 2. Try to append a new node (method tryAppend)
260	>	*
261	>	*    Starting at current tail pointer, try to append a new node
262	>	*    to the list (or if head was null, establish the first
263	>	*    node). Nodes can be appended only if their predecessors are
264	>	*    either already matched or are of the same mode. If we detect
265	>	*    otherwise, then a new node with opposite mode must have been
266	>	*    appended during traversal, so must restart at phase 1. The
267	>	*    traversal and update steps are otherwise similar to phase 1:
268	>	*    Retrying upon CAS misses and checking for staleness.  In
269	>	*    particular, if a self-link is encountered, then we can
270	>	*    safely jump to a node on the list by continuing the
271	>	*    traversal at current head.
272	>	*
273	>	*    On successful append, if the call was ASYNC, return
274	>	*
275	>	* 3. Await match or cancellation (method awaitMatch)
276	>	*
277	>	*    Wait for another thread to match node; instead cancelling if
278	>	*    current thread was interrupted or the wait timed out. On
279	>	*    multiprocessors, we use front-of-queue spinning: If a node
280	>	*    appears to be the first unmatched node in the queue, it
281	>	*    spins a bit before blocking. In either case, before blocking
282	>	*    it tries to unsplice any nodes between the current "head"
283	>	*    and the first unmatched node.
284	>	*
285	>	*    Front-of-queue spinning vastly improves performance of
286	>	*    heavily contended queues. And so long as it is relatively
287	>	*    brief and "quiet", spinning does not much impact performance
288	>	*    of less-contended queues.  During spins threads check their
289	>	*    interrupt status and generate a thread-local random number
290	>	*    to decide to occasionally perform a Thread.yield. While
291	>	*    yield has underdefined specs, we assume that might it help,
292	>	*    and will not hurt in limiting impact of spinning on busy
293	>	*    systems.  We also use much smaller (1/4) spins for nodes
294	>	*    that are not known to be front but whose predecessors have
295	>	*    not blocked -- these "chained" spins avoid artifacts of
296	>	*    front-of-queue rules which otherwise lead to alternating
297	>	*    nodes spinning vs blocking. Further, front threads that
298	>	*    represent phase changes (from data to request node or vice
299	>	*    versa) compared to their predecessors receive additional
300	>	*    spins, reflecting the longer code path lengths necessary to
301	>	*    release them under contention.
302	>	*/
303	>
304	>	/** True if on multiprocessor */
305	>	private static final boolean MP =
306	>	Runtime.getRuntime().availableProcessors() > 1;
307	>
308	>	/**
309	>	* The number of times to spin (with on average one randomly
310	>	* interspersed call to Thread.yield) on multiprocessor before
311	>	* blocking when a node is apparently the first waiter in the
312	>	* queue.  See above for explanation. Must be a power of two. The
313	>	* value is empirically derived -- it works pretty well across a
314	>	* variety of processors, numbers of CPUs, and OSes.
315	>	*/
316	>	private static final int FRONT_SPINS   = 1 << 7;
317	>
318	>	/**
319	>	* The number of times to spin before blocking when a node is
320	>	* preceded by another node that is apparently spinning.
321	>	*/
322	>	private static final int CHAINED_SPINS = FRONT_SPINS >>> 2;
323	>
324	>	/**
325	>	* Queue nodes. Uses Object, not E for items to allow forgetting
326	>	* them after use.  Relies heavily on Unsafe mechanics to minimize
327	>	* unecessary ordering constraints: Writes that intrinsically
328	>	* precede or follow CASes use simple relaxed forms.  Other
329	>	* cleanups use releasing/lazy writes.
330	>	*/
331	>	static final class Node {
332	>	final boolean isData;   // false if this is a request node
333	>	volatile Object item;   // initially nonnull if isData; CASed to match
334	>	volatile Node next;
335	>	volatile Thread waiter; // null until waiting
336
337	<	/**
338	<	* The number of times to spin before blocking in untimed waits.
339	<	* This is greater than timed value because untimed waits spin
340	<	* faster since they don't need to check times on each spin.
93	<	*/
94	<	static final int maxUntimedSpins = maxTimedSpins * 16;
95	<
96	<	/**
97	<	* The number of nanoseconds for which it is faster to spin
98	<	* rather than to use timed park. A rough estimate suffices.
99	<	*/
100	<	static final long spinForTimeoutThreshold = 1000L;
337	>	// CAS methods for fields
338	>	final boolean casNext(Node cmp, Node val) {
339	>	return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
340	>	}
341
342	<	/**
343	<	* Node class for LinkedTransferQueue. Opportunistically
344	<	* subclasses from AtomicReference to represent item. Uses Object,
105	<	* not E, to allow setting item to "this" after use, to avoid
106	<	* garbage retention. Similarly, setting the next field to this is
107	<	* used as sentinel that node is off list.
108	<	*/
109	<	static final class Node<E> extends AtomicReference<Object> {
110	<	volatile Node<E> next;
111	<	volatile Thread waiter;       // to control park/unpark
112	<	final boolean isData;
342	>	final boolean casItem(Object cmp, Object val) {
343	>	return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
344	>	}
345
346	<	Node(E item, boolean isData) {
347	<	super(item);
346	>	/**
347	>	* Create a new node. Uses relaxed write because item can only
348	>	* be seen if followed by CAS
349	>	*/
350	>	Node(Object item, boolean isData) {
351	>	UNSAFE.putObject(this, itemOffset, item); // relaxed write
352		this.isData = isData;
353		}
354
355	<	// Unsafe mechanics
355	>	/**
356	>	* Links node to itself to avoid garbage retention.  Called
357	>	* only after CASing head field, so uses relaxed write.
358	>	*/
359	>	final void forgetNext() {
360	>	UNSAFE.putObject(this, nextOffset, this);
361	>	}
362
363	<	private static final sun.misc.Unsafe UNSAFE = getUnsafe();
364	<	private static final long nextOffset =
365	<	objectFieldOffset(UNSAFE, "next", Node.class);
363	>	/**
364	>	* Sets item to self (using a releasing/lazy write) and waiter
365	>	* to null, to avoid garbage retention after extracting or
366	>	* cancelling.
367	>	*/
368	>	final void forgetContents() {
369	>	UNSAFE.putOrderedObject(this, itemOffset, this);
370	>	UNSAFE.putOrderedObject(this, waiterOffset, null);
371	>	}
372
373	<	final boolean casNext(Node<E> cmp, Node<E> val) {
374	<	return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
373	>	/**
374	>	* Returns true if this node has been matched, including the
375	>	* case of artificial matches due to cancellation.
376	>	*/
377	>	final boolean isMatched() {
378	>	Object x = item;
379	>	return x == this || (x != null) != isData;
380		}
381
382	<	final void clearNext() {
383	<	UNSAFE.putOrderedObject(this, nextOffset, this);
382	>	/**
383	>	* Returns true if a node with the given mode cannot be
384	>	* appended to this node because this node is unmatched and
385	>	* has opposite data mode.
386	>	*/
387	>	final boolean cannotPrecede(boolean haveData) {
388	>	boolean d = isData;
389	>	Object x;
390	>	return d != haveData && (x = item) != this && (x != null) == d;
391		}
392
393		/**
394	<	* Returns a sun.misc.Unsafe.  Suitable for use in a 3rd party package.
135	<	* Replace with a simple call to Unsafe.getUnsafe when integrating
136	<	* into a jdk.
137	<	*
138	<	* @return a sun.misc.Unsafe
394	>	* Tries to artifically match a data node -- used by remove.
395		*/
396	<	private static sun.misc.Unsafe getUnsafe() {
397	<	try {
398	<	return sun.misc.Unsafe.getUnsafe();
399	<	} catch (SecurityException se) {
400	<	try {
145	<	return java.security.AccessController.doPrivileged
146	<	(new java.security
147	<	.PrivilegedExceptionAction<sun.misc.Unsafe>() {
148	<	public sun.misc.Unsafe run() throws Exception {
149	<	java.lang.reflect.Field f = sun.misc
150	<	.Unsafe.class.getDeclaredField("theUnsafe");
151	<	f.setAccessible(true);
152	<	return (sun.misc.Unsafe) f.get(null);
153	<	}});
154	<	} catch (java.security.PrivilegedActionException e) {
155	<	throw new RuntimeException("Could not initialize intrinsics",
156	<	e.getCause());
157	<	}
396	>	final boolean tryMatchData() {
397	>	Object x = item;
398	>	if (x != null && x != this && casItem(x, null)) {
399	>	LockSupport.unpark(waiter);
400	>	return true;
401		}
402	+	return false;
403		}
404
405	+	// Unsafe mechanics
406	+	private static final sun.misc.Unsafe UNSAFE = getUnsafe();
407	+	private static final long nextOffset =
408	+	objectFieldOffset(UNSAFE, "next", Node.class);
409	+	private static final long itemOffset =
410	+	objectFieldOffset(UNSAFE, "item", Node.class);
411	+	private static final long waiterOffset =
412	+	objectFieldOffset(UNSAFE, "waiter", Node.class);
413	+
414		private static final long serialVersionUID = -3375979862319811754L;
415		}
416
417	<	/**
418	<	* Padded version of AtomicReference used for head, tail and
166	<	* cleanMe, to alleviate contention across threads CASing one vs
167	<	* the other.
168	<	*/
169	<	static final class PaddedAtomicReference<T> extends AtomicReference<T> {
170	<	// enough padding for 64bytes with 4byte refs
171	<	Object p0, p1, p2, p3, p4, p5, p6, p7, p8, p9, pa, pb, pc, pd, pe;
172	<	PaddedAtomicReference(T r) { super(r); }
173	<	private static final long serialVersionUID = 8170090609809740854L;
174	<	}
417	>	/** head of the queue; null until first enqueue */
418	>	private transient volatile Node head;
419
420	+	/** predecessor of dangling unspliceable node */
421	+	private transient volatile Node cleanMe; // decl here to reduce contention
422
423	<	/** head of the queue */
424	<	private transient final PaddedAtomicReference<Node<E>> head;
423	>	/** tail of the queue; null until first append */
424	>	private transient volatile Node tail;
425
426	<	/** tail of the queue */
427	<	private transient final PaddedAtomicReference<Node<E>> tail;
426	>	// CAS methods for fields
427	>	private boolean casTail(Node cmp, Node val) {
428	>	return UNSAFE.compareAndSwapObject(this, tailOffset, cmp, val);
429	>	}
430
431	<	/**
432	<	* Reference to a cancelled node that might not yet have been
433	<	* unlinked from queue because it was the last inserted node
186	<	* when it cancelled.
187	<	*/
188	<	private transient final PaddedAtomicReference<Node<E>> cleanMe;
431	>	private boolean casHead(Node cmp, Node val) {
432	>	return UNSAFE.compareAndSwapObject(this, headOffset, cmp, val);
433	>	}
434
435	<	/**
436	<	* Tries to cas nh as new head; if successful, unlink
192	<	* old head's next node to avoid garbage retention.
193	<	*/
194	<	private boolean advanceHead(Node<E> h, Node<E> nh) {
195	<	if (h == head.get() && head.compareAndSet(h, nh)) {
196	<	h.clearNext(); // forget old next
197	<	return true;
198	<	}
199	<	return false;
435	>	private boolean casCleanMe(Node cmp, Node val) {
436	>	return UNSAFE.compareAndSwapObject(this, cleanMeOffset, cmp, val);
437		}
438
439	+	/*
440	+	* Possible values for "how" argument in xfer method. Beware that
441	+	* the order of assigned numerical values matters.
442	+	*/
443	+	private static final int NOW     = 0; // for untimed poll, tryTransfer
444	+	private static final int ASYNC   = 1; // for offer, put, add
445	+	private static final int SYNC    = 2; // for transfer, take
446	+	private static final int TIMEOUT = 3; // for timed poll, tryTransfer
447	+
448		/**
449	<	* Puts or takes an item. Used for most queue operations (except
204	<	* poll() and tryTransfer()). See the similar code in
205	<	* SynchronousQueue for detailed explanation.
449	>	* Implements all queuing methods. See above for explanation.
450		*
451	<	* @param e the item or if null, signifies that this is a take
452	<	* @param mode the wait mode: NOWAIT, TIMEOUT, WAIT
451	>	* @param e the item or null for take
452	>	* @param haveData true if this is a put else a take
453	>	* @param how NOW, ASYNC, SYNC, or TIMEOUT
454		* @param nanos timeout in nanosecs, used only if mode is TIMEOUT
455	<	* @return an item, or null on failure
455	>	* @return an item if matched, else e;
456	>	* @throws NullPointerException if haveData mode but e is null
457		*/
458	<	private E xfer(E e, int mode, long nanos) {
459	<	boolean isData = (e != null);
460	<	Node<E> s = null;
461	<	final PaddedAtomicReference<Node<E>> head = this.head;
216	<	final PaddedAtomicReference<Node<E>> tail = this.tail;
458	>	private Object xfer(Object e, boolean haveData, int how, long nanos) {
459	>	if (haveData && (e == null))
460	>	throw new NullPointerException();
461	>	Node s = null;                        // the node to append, if needed
462
463	<	for (;;) {
219	<	Node<E> t = tail.get();
220	<	Node<E> h = head.get();
463	>	retry: for (;;) {                     // restart on append race
464
465	<	if (t == h || t.isData == isData) {
466	<	if (s == null)
467	<	s = new Node<E>(e, isData);
468	<	Node<E> last = t.next;
469	<	if (last != null) {
470	<	if (t == tail.get())
471	<	tail.compareAndSet(t, last);
472	<	}
473	<	else if (t.casNext(null, s)) {
474	<	tail.compareAndSet(t, s);
475	<	return awaitFulfill(t, s, e, mode, nanos);
476	<	}
477	<	} else {
478	<	Node<E> first = h.next;
479	<	if (t == tail.get() && first != null &&
480	<	advanceHead(h, first)) {
481	<	Object x = first.get();
482	<	if (x != first && first.compareAndSet(x, e)) {
483	<	LockSupport.unpark(first.waiter);
484	<	return isData ? e : (E) x;
465	>	for (Node h = head, p = h; p != null;) { // find & match first node
466	>	boolean isData = p.isData;
467	>	Object item = p.item;
468	>	if (item != p && (item != null) == isData) { // unmatched
469	>	if (isData == haveData)   // can't match
470	>	break;
471	>	if (p.casItem(item, e)) { // match
472	>	Thread w = p.waiter;
473	>	while (p != h) {      // update head
474	>	Node n = p.next;  // by 2 unless singleton
475	>	if (n != null)
476	>	p = n;
477	>	if (head == h && casHead(h, p)) {
478	>	h.forgetNext();
479	>	break;
480	>	}                 // advance and retry
481	>	if ((h = head)   == null ||
482	>	(p = h.next) == null || !p.isMatched())
483	>	break;        // unless slack < 2
484	>	}
485	>	LockSupport.unpark(w);
486	>	return item;
487		}
488		}
489	+	Node n = p.next;
490	+	p = p != n ? n : (h = head);  // Use head if p offlist
491	+	}
492	+
493	+	if (how >= ASYNC) {               // No matches available
494	+	if (s == null)
495	+	s = new Node(e, haveData);
496	+	Node pred = tryAppend(s, haveData);
497	+	if (pred == null)
498	+	continue retry;           // lost race vs opposite mode
499	+	if (how >= SYNC)
500	+	return awaitMatch(pred, s, e, how, nanos);
501		}
502	+	return e; // not waiting
503		}
504		}
505
248	–
506		/**
507	<	* Version of xfer for poll() and tryTransfer, which
508	<	* simplifies control paths both here and in xfer.
507	>	* Tries to append node s as tail
508	>	* @param haveData true if appending in data mode
509	>	* @param s the node to append
510	>	* @return null on failure due to losing race with append in
511	>	* different mode, else s's predecessor, or s itself if no
512	>	* predecessor
513		*/
514	<	private E fulfill(E e) {
515	<	boolean isData = (e != null);
516	<	final PaddedAtomicReference<Node<E>> head = this.head;
517	<	final PaddedAtomicReference<Node<E>> tail = this.tail;
518	<
519	<	for (;;) {
520	<	Node<E> t = tail.get();
521	<	Node<E> h = head.get();
522	<
523	<	if (t == h || t.isData == isData) {
524	<	Node<E> last = t.next;
525	<	if (t == tail.get()) {
526	<	if (last != null)
527	<	tail.compareAndSet(t, last);
528	<	else
529	<	return null;
530	<	}
531	<	} else {
532	<	Node<E> first = h.next;
533	<	if (t == tail.get() &&
273	<	first != null &&
274	<	advanceHead(h, first)) {
275	<	Object x = first.get();
276	<	if (x != first && first.compareAndSet(x, e)) {
277	<	LockSupport.unpark(first.waiter);
278	<	return isData ? e : (E) x;
279	<	}
514	>	private Node tryAppend(Node s, boolean haveData) {
515	>	for (Node t = tail, p = t;;) { // move p to actual tail and append
516	>	Node n, u;                        // temps for reads of next & tail
517	>	if (p == null && (p = head) == null) {
518	>	if (casHead(null, s))
519	>	return s;                 // initialize
520	>	}
521	>	else if (p.cannotPrecede(haveData))
522	>	return null;                  // lost race vs opposite mode
523	>	else if ((n = p.next) != null)    // Not tail; keep traversing
524	>	p = p != t && t != (u = tail) ? (t = u) : // stale tail
525	>	p != n ? n : null;        // restart if off list
526	>	else if (!p.casNext(null, s))
527	>	p = p.next;                   // re-read on CAS failure
528	>	else {
529	>	if (p != t) {                 // Update if slack now >= 2
530	>	while ((tail != t || !casTail(t, s)) &&
531	>	(t = tail)   != null &&
532	>	(s = t.next) != null && // advance and retry
533	>	(s = s.next) != null && s != t);
534		}
535	+	return p;
536		}
537		}
538		}
539
540		/**
541	<	* Spins/blocks until node s is fulfilled or caller gives up,
287	<	* depending on wait mode.
541	>	* Spins/yields/blocks until node s is matched or caller gives up.
542		*
543	<	* @param pred the predecessor of waiting node
543	>	* @param pred the predecessor of s or s or null if none
544		* @param s the waiting node
545		* @param e the comparison value for checking match
546	<	* @param mode mode
546	>	* @param how either SYNC or TIMEOUT
547		* @param nanos timeout value
548	<	* @return matched item, or null if cancelled
548	>	* @return matched item, or e if unmatched on interrupt or timeout
549		*/
550	<	private E awaitFulfill(Node<E> pred, Node<E> s, E e,
551	<	int mode, long nanos) {
552	<	if (mode == NOWAIT)
299	<	return null;
300	<
301	<	long lastTime = (mode == TIMEOUT) ? System.nanoTime() : 0;
550	>	private Object awaitMatch(Node pred, Node s, Object e,
551	>	int how, long nanos) {
552	>	long lastTime = (how == TIMEOUT) ? System.nanoTime() : 0L;
553		Thread w = Thread.currentThread();
554	<	int spins = -1; // set to desired spin count below
554	>	int spins = -1; // initialized after first item and cancel checks
555	>	ThreadLocalRandom randomYields = null; // bound if needed
556	>
557		for (;;) {
558	<	if (w.isInterrupted())
559	<	s.compareAndSet(e, s);
560	<	Object x = s.get();
561	<	if (x != e) {                 // Node was matched or cancelled
309	<	advanceHead(pred, s);     // unlink if head
310	<	if (x == s) {             // was cancelled
311	<	clean(pred, s);
312	<	return null;
313	<	}
314	<	else if (x != null) {
315	<	s.set(s);             // avoid garbage retention
316	<	return (E) x;
317	<	}
318	<	else
319	<	return e;
558	>	Object item = s.item;
559	>	if (item != e) {                  // matched
560	>	s.forgetContents();           // avoid garbage
561	>	return item;
562		}
563	<	if (mode == TIMEOUT) {
564	<	long now = System.nanoTime();
565	<	nanos -= now - lastTime;
566	<	lastTime = now;
325	<	if (nanos <= 0) {
326	<	s.compareAndSet(e, s); // try to cancel
327	<	continue;
328	<	}
563	>	if ((w.isInterrupted() || (how == TIMEOUT && nanos <= 0)) &&
564	>	s.casItem(e, s)) {       // cancel
565	>	unsplice(pred, s);
566	>	return e;
567		}
568	<	if (spins < 0) {
569	<	Node<E> h = head.get(); // only spin if at head
570	<	spins = ((h.next != s) ? 0 :
571	<	(mode == TIMEOUT) ? maxTimedSpins :
334	<	maxUntimedSpins);
568	>
569	>	if (spins < 0) {                  // establish spins at/near front
570	>	if ((spins = spinsFor(pred, s.isData)) > 0)
571	>	randomYields = ThreadLocalRandom.current();
572		}
573	<	if (spins > 0)
573	>	else if (spins > 0) {             // spin, occasionally yield
574	>	if (randomYields.nextInt(FRONT_SPINS) == 0)
575	>	Thread.yield();
576		--spins;
338	–	else if (s.waiter == null)
339	–	s.waiter = w;
340	–	else if (mode != TIMEOUT) {
341	–	LockSupport.park(this);
342	–	s.waiter = null;
343	–	spins = -1;
577		}
578	<	else if (nanos > spinForTimeoutThreshold) {
579	<	LockSupport.parkNanos(this, nanos);
580	<	s.waiter = null;
581	<	spins = -1;
578	>	else if (s.waiter == null) {
579	>	shortenHeadPath();            // reduce slack before blocking
580	>	s.waiter = w;                 // request unpark
581	>	}
582	>	else if (how == TIMEOUT) {
583	>	long now = System.nanoTime();
584	>	if ((nanos -= now - lastTime) > 0)
585	>	LockSupport.parkNanos(this, nanos);
586	>	lastTime = now;
587	>	}
588	>	else {
589	>	LockSupport.park(this);
590	>	spins = -1;                   // spin if front upon wakeup
591		}
592		}
593		}
594
595		/**
596	<	* Returns validated tail for use in cleaning methods.
596	>	* Return spin/yield value for a node with given predecessor and
597	>	* data mode. See above for explanation.
598		*/
599	<	private Node<E> getValidatedTail() {
600	<	for (;;) {
601	<	Node<E> h = head.get();
602	<	Node<E> first = h.next;
603	<	if (first != null && first.get() == first) { // help advance
604	<	advanceHead(h, first);
605	<	continue;
606	<	}
607	<	Node<E> t = tail.get();
608	<	Node<E> last = t.next;
609	<	if (t == tail.get()) {
610	<	if (last != null)
611	<	tail.compareAndSet(t, last); // help advance
612	<	else
613	<	return t;
599	>	private static int spinsFor(Node pred, boolean haveData) {
600	>	if (MP && pred != null) {
601	>	boolean predData = pred.isData;
602	>	if (predData != haveData)         // front and phase change
603	>	return FRONT_SPINS + (FRONT_SPINS >>> 1);
604	>	if (predData != (pred.item != null)) // probably at front
605	>	return FRONT_SPINS;
606	>	if (pred.waiter == null)          // pred apparently spinning
607	>	return CHAINED_SPINS;
608	>	}
609	>	return 0;
610	>	}
611	>
612	>	/**
613	>	* Tries (once) to unsplice nodes between head and first unmatched
614	>	* or trailing node; failing on contention.
615	>	*/
616	>	private void shortenHeadPath() {
617	>	Node h, hn, p, q;
618	>	if ((p = h = head) != null && h.isMatched() &&
619	>	(q = hn = h.next) != null) {
620	>	Node n;
621	>	while ((n = q.next) != q) {
622	>	if (n == null || !q.isMatched()) {
623	>	if (hn != q && h.next == hn)
624	>	h.casNext(hn, q);
625	>	break;
626	>	}
627	>	p = q;
628	>	q = n;
629		}
630		}
631		}
632
633	+	/* -------------- Traversal methods -------------- */
634	+
635		/**
636	<	* Gets rid of cancelled node s with original predecessor pred.
637	<	*
378	<	* @param pred predecessor of cancelled node
379	<	* @param s the cancelled node
636	>	* Return the first unmatched node of the given mode, or null if
637	>	* none.  Used by methods isEmpty, hasWaitingConsumer.
638		*/
639	<	private void clean(Node<E> pred, Node<E> s) {
640	<	Thread w = s.waiter;
641	<	if (w != null) {             // Wake up thread
642	<	s.waiter = null;
643	<	if (w != Thread.currentThread())
644	<	LockSupport.unpark(w);
639	>	private Node firstOfMode(boolean data) {
640	>	for (Node p = head; p != null; ) {
641	>	if (!p.isMatched())
642	>	return p.isData == data? p : null;
643	>	Node n = p.next;
644	>	p = n != p ? n : head;
645		}
646	+	return null;
647	+	}
648	+
649	+	/**
650	+	* Returns the item in the first unmatched node with isData; or
651	+	* null if none. Used by peek.
652	+	*/
653	+	private Object firstDataItem() {
654	+	for (Node p = head; p != null; ) {
655	+	boolean isData = p.isData;
656	+	Object item = p.item;
657	+	if (item != p && (item != null) == isData)
658	+	return isData ? item : null;
659	+	Node n = p.next;
660	+	p = n != p ? n : head;
661	+	}
662	+	return null;
663	+	}
664	+
665	+	/**
666	+	* Traverse and count nodes of the given mode.
667	+	* Used by methds size and getWaitingConsumerCount.
668	+	*/
669	+	private int countOfMode(boolean data) {
670	+	int count = 0;
671	+	for (Node p = head; p != null; ) {
672	+	if (!p.isMatched()) {
673	+	if (p.isData != data)
674	+	return 0;
675	+	if (++count == Integer.MAX_VALUE) // saturated
676	+	break;
677	+	}
678	+	Node n = p.next;
679	+	if (n != p)
680	+	p = n;
681	+	else {
682	+	count = 0;
683	+	p = head;
684	+	}
685	+	}
686	+	return count;
687	+	}
688
689	<	if (pred == null)
690	<	return;
689	>	final class Itr implements Iterator<E> {
690	>	private Node nextNode;   // next node to return item for
691	>	private Object nextItem; // the corresponding item
692	>	private Node lastRet;    // last returned node, to support remove
693
694	+	/**
695	+	* Moves to next node after prev, or first node if prev null.
696	+	*/
697	+	private void advance(Node prev) {
698	+	lastRet = prev;
699	+	Node p;
700	+	if (prev == null || (p = prev.next) == prev)
701	+	p = head;
702	+	while (p != null) {
703	+	Object item = p.item;
704	+	if (p.isData) {
705	+	if (item != null && item != p) {
706	+	nextItem = item;
707	+	nextNode = p;
708	+	return;
709	+	}
710	+	}
711	+	else if (item == null)
712	+	break;
713	+	Node n = p.next;
714	+	p = n != p ? n : head;
715	+	}
716	+	nextNode = null;
717	+	}
718	+
719	+	Itr() {
720	+	advance(null);
721	+	}
722	+
723	+	public final boolean hasNext() {
724	+	return nextNode != null;
725	+	}
726	+
727	+	public final E next() {
728	+	Node p = nextNode;
729	+	if (p == null) throw new NoSuchElementException();
730	+	Object e = nextItem;
731	+	advance(p);
732	+	return (E) e;
733	+	}
734	+
735	+	public final void remove() {
736	+	Node p = lastRet;
737	+	if (p == null) throw new IllegalStateException();
738	+	lastRet = null;
739	+	findAndRemoveNode(p);
740	+	}
741	+	}
742	+
743	+	/* -------------- Removal methods -------------- */
744	+
745	+	/**
746	+	* Unsplices (now or later) the given deleted/cancelled node with
747	+	* the given predecessor.
748	+	*
749	+	* @param pred predecessor of node to be unspliced
750	+	* @param s the node to be unspliced
751	+	*/
752	+	private void unsplice(Node pred, Node s) {
753	+	s.forgetContents(); // clear unneeded fields
754		/*
755		* At any given time, exactly one node on list cannot be
756		* deleted -- the last inserted node. To accommodate this, if
757		* we cannot delete s, we save its predecessor as "cleanMe",
758	<	* processing the previously saved version first. At least one
759	<	* of node s or the node previously saved can always be
758	>	* processing the previously saved version first. Because only
759	>	* one node in the list can have a null next, at least one of
760	>	* node s or the node previously saved can always be
761		* processed, so this always terminates.
762		*/
763	<	while (pred.next == s) {
764	<	Node<E> oldpred = reclean();  // First, help get rid of cleanMe
765	<	Node<E> t = getValidatedTail();
766	<	if (s != t) {               // If not tail, try to unsplice
767	<	Node<E> sn = s.next;      // s.next == s means s already off list
768	<	if (sn == s || pred.casNext(s, sn))
763	>	if (pred != null && pred != s) {
764	>	while (pred.next == s) {
765	>	Node oldpred = cleanMe == null? null : reclean();
766	>	Node n = s.next;
767	>	if (n != null) {
768	>	if (n != s)
769	>	pred.casNext(s, n);
770		break;
771	+	}
772	+	if (oldpred == pred ||      // Already saved
773	+	(oldpred == null && casCleanMe(null, pred)))
774	+	break;                  // Postpone cleaning
775		}
408	–	else if (oldpred == pred || // Already saved
409	–	(oldpred == null && cleanMe.compareAndSet(null, pred)))
410	–	break;                  // Postpone cleaning
776		}
777		}
778
779		/**
780	<	* Tries to unsplice the cancelled node held in cleanMe that was
781	<	* previously uncleanable because it was at tail.
780	>	* Tries to unsplice the deleted/cancelled node held in cleanMe
781	>	* that was previously uncleanable because it was at tail.
782		*
783		* @return current cleanMe node (or null)
784		*/
785	<	private Node<E> reclean() {
785	>	private Node reclean() {
786		/*
787	<	* cleanMe is, or at one time was, predecessor of cancelled
788	<	* node s that was the tail so could not be unspliced.  If s
787	>	* cleanMe is, or at one time was, predecessor of a cancelled
788	>	* node s that was the tail so could not be unspliced.  If it
789		* is no longer the tail, try to unsplice if necessary and
790		* make cleanMe slot available.  This differs from similar
791	<	* code in clean() because we must check that pred still
792	<	* points to a cancelled node that must be unspliced -- if
793	<	* not, we can (must) clear cleanMe without unsplicing.
794	<	* This can loop only due to contention on casNext or
430	<	* clearing cleanMe.
791	>	* code in unsplice() because we must check that pred still
792	>	* points to a matched node that can be unspliced -- if not,
793	>	* we can (must) clear cleanMe without unsplicing.  This can
794	>	* loop only due to contention.
795		*/
796	<	Node<E> pred;
797	<	while ((pred = cleanMe.get()) != null) {
798	<	Node<E> t = getValidatedTail();
799	<	Node<E> s = pred.next;
800	<	if (s != t) {
801	<	Node<E> sn;
802	<	if (s == null || s == pred || s.get() != s ||
803	<	(sn = s.next) == s || pred.casNext(s, sn))
804	<	cleanMe.compareAndSet(pred, null);
796	>	Node pred;
797	>	while ((pred = cleanMe) != null) {
798	>	Node s = pred.next;
799	>	Node n;
800	>	if (s == null || s == pred || !s.isMatched())
801	>	casCleanMe(pred, null); // already gone
802	>	else if ((n = s.next) != null) {
803	>	if (n != s)
804	>	pred.casNext(s, n);
805	>	casCleanMe(pred, null);
806		}
807	<	else // s is still tail; cannot clean
807	>	else
808		break;
809		}
810		return pred;
811		}
812
813		/**
814	+	* Main implementation of Iterator.remove(). Find
815	+	* and unsplice the given node.
816	+	*/
817	+	final void findAndRemoveNode(Node s) {
818	+	if (s.tryMatchData()) {
819	+	Node pred = null;
820	+	Node p = head;
821	+	while (p != null) {
822	+	if (p == s) {
823	+	unsplice(pred, p);
824	+	break;
825	+	}
826	+	if (!p.isData && !p.isMatched())
827	+	break;
828	+	pred = p;
829	+	if ((p = p.next) == pred) { // stale
830	+	pred = null;
831	+	p = head;
832	+	}
833	+	}
834	+	}
835	+	}
836	+
837	+	/**
838	+	* Main implementation of remove(Object)
839	+	*/
840	+	private boolean findAndRemove(Object e) {
841	+	if (e != null) {
842	+	Node pred = null;
843	+	Node p = head;
844	+	while (p != null) {
845	+	Object item = p.item;
846	+	if (p.isData) {
847	+	if (item != null && item != p && e.equals(item) &&
848	+	p.tryMatchData()) {
849	+	unsplice(pred, p);
850	+	return true;
851	+	}
852	+	}
853	+	else if (item == null)
854	+	break;
855	+	pred = p;
856	+	if ((p = p.next) == pred) {
857	+	pred = null;
858	+	p = head;
859	+	}
860	+	}
861	+	}
862	+	return false;
863	+	}
864	+
865	+
866	+	/**
867		* Creates an initially empty {@code LinkedTransferQueue}.
868		*/
869		public LinkedTransferQueue() {
452	–	Node<E> dummy = new Node<E>(null, false);
453	–	head = new PaddedAtomicReference<Node<E>>(dummy);
454	–	tail = new PaddedAtomicReference<Node<E>>(dummy);
455	–	cleanMe = new PaddedAtomicReference<Node<E>>(null);
870		}
871
872		/**
#	Line 476 | Line 890 | public class LinkedTransferQueue<E> exte
890		* @throws NullPointerException if the specified element is null
891		*/
892		public void put(E e) {
893	<	offer(e);
893	>	xfer(e, true, ASYNC, 0);
894		}
895
896		/**
#	Line 489 | Line 903 | public class LinkedTransferQueue<E> exte
903		* @throws NullPointerException if the specified element is null
904		*/
905		public boolean offer(E e, long timeout, TimeUnit unit) {
906	<	return offer(e);
906	>	xfer(e, true, ASYNC, 0);
907	>	return true;
908		}
909
910		/**
#	Line 501 | Line 916 | public class LinkedTransferQueue<E> exte
916		* @throws NullPointerException if the specified element is null
917		*/
918		public boolean offer(E e) {
919	<	if (e == null) throw new NullPointerException();
505	<	xfer(e, NOWAIT, 0);
919	>	xfer(e, true, ASYNC, 0);
920		return true;
921		}
922
#	Line 515 | Line 929 | public class LinkedTransferQueue<E> exte
929		* @throws NullPointerException if the specified element is null
930		*/
931		public boolean add(E e) {
932	<	return offer(e);
932	>	xfer(e, true, ASYNC, 0);
933	>	return true;
934		}
935
936		/**
#	Line 529 | Line 944 | public class LinkedTransferQueue<E> exte
944		* @throws NullPointerException if the specified element is null
945		*/
946		public boolean tryTransfer(E e) {
947	<	if (e == null) throw new NullPointerException();
533	<	return fulfill(e) != null;
947	>	return xfer(e, true, NOW, 0) == null;
948		}
949
950		/**
#	Line 545 | Line 959 | public class LinkedTransferQueue<E> exte
959		* @throws NullPointerException if the specified element is null
960		*/
961		public void transfer(E e) throws InterruptedException {
962	<	if (e == null) throw new NullPointerException();
963	<	if (xfer(e, WAIT, 0) == null) {
550	<	Thread.interrupted();
962	>	if (xfer(e, true, SYNC, 0) != null) {
963	>	Thread.interrupted(); // failure possible only due to interrupt
964		throw new InterruptedException();
965		}
966		}
#	Line 568 | Line 981 | public class LinkedTransferQueue<E> exte
981		*/
982		public boolean tryTransfer(E e, long timeout, TimeUnit unit)
983		throws InterruptedException {
984	<	if (e == null) throw new NullPointerException();
572	<	if (xfer(e, TIMEOUT, unit.toNanos(timeout)) != null)
984	>	if (xfer(e, true, TIMEOUT, unit.toNanos(timeout)) == null)
985		return true;
986		if (!Thread.interrupted())
987		return false;
#	Line 577 | Line 989 | public class LinkedTransferQueue<E> exte
989		}
990
991		public E take() throws InterruptedException {
992	<	E e = xfer(null, WAIT, 0);
992	>	Object e = xfer(null, false, SYNC, 0);
993		if (e != null)
994	<	return e;
994	>	return (E)e;
995		Thread.interrupted();
996		throw new InterruptedException();
997		}
998
999		public E poll(long timeout, TimeUnit unit) throws InterruptedException {
1000	<	E e = xfer(null, TIMEOUT, unit.toNanos(timeout));
1000	>	Object e = xfer(null, false, TIMEOUT, unit.toNanos(timeout));
1001		if (e != null || !Thread.interrupted())
1002	<	return e;
1002	>	return (E)e;
1003		throw new InterruptedException();
1004		}
1005
1006		public E poll() {
1007	<	return fulfill(null);
1007	>	return (E)xfer(null, false, NOW, 0);
1008		}
1009
1010		/**
#	Line 631 | Line 1043 | public class LinkedTransferQueue<E> exte
1043		return n;
1044		}
1045
634	–	// Traversal-based methods
635	–
636	–	/**
637	–	* Returns head after performing any outstanding helping steps.
638	–	*/
639	–	private Node<E> traversalHead() {
640	–	for (;;) {
641	–	Node<E> t = tail.get();
642	–	Node<E> h = head.get();
643	–	Node<E> last = t.next;
644	–	Node<E> first = h.next;
645	–	if (t == tail.get()) {
646	–	if (last != null)
647	–	tail.compareAndSet(t, last);
648	–	else if (first != null) {
649	–	Object x = first.get();
650	–	if (x == first)
651	–	advanceHead(h, first);
652	–	else
653	–	return h;
654	–	}
655	–	else
656	–	return h;
657	–	}
658	–	reclean();
659	–	}
660	–	}
661	–
1046		/**
1047		* Returns an iterator over the elements in this queue in proper
1048		* sequence, from head to tail.
#	Line 676 | Line 1060 | public class LinkedTransferQueue<E> exte
1060		return new Itr();
1061		}
1062
679	–	/**
680	–	* Iterators. Basic strategy is to traverse list, treating
681	–	* non-data (i.e., request) nodes as terminating list.
682	–	* Once a valid data node is found, the item is cached
683	–	* so that the next call to next() will return it even
684	–	* if subsequently removed.
685	–	*/
686	–	class Itr implements Iterator<E> {
687	–	Node<E> next;        // node to return next
688	–	Node<E> pnext;       // predecessor of next
689	–	Node<E> curr;        // last returned node, for remove()
690	–	Node<E> pcurr;       // predecessor of curr, for remove()
691	–	E nextItem;          // Cache of next item, once committed to in next
692	–
693	–	Itr() {
694	–	advance();
695	–	}
696	–
697	–	/**
698	–	* Moves to next valid node and returns item to return for
699	–	* next(), or null if no such.
700	–	*/
701	–	private E advance() {
702	–	pcurr = pnext;
703	–	curr = next;
704	–	E item = nextItem;
705	–
706	–	for (;;) {
707	–	pnext = (next == null) ? traversalHead() : next;
708	–	next = pnext.next;
709	–	if (next == pnext) {
710	–	next = null;
711	–	continue;  // restart
712	–	}
713	–	if (next == null)
714	–	break;
715	–	Object x = next.get();
716	–	if (x != null && x != next) {
717	–	nextItem = (E) x;
718	–	break;
719	–	}
720	–	}
721	–	return item;
722	–	}
723	–
724	–	public boolean hasNext() {
725	–	return next != null;
726	–	}
727	–
728	–	public E next() {
729	–	if (next == null)
730	–	throw new NoSuchElementException();
731	–	return advance();
732	–	}
733	–
734	–	public void remove() {
735	–	Node<E> p = curr;
736	–	if (p == null)
737	–	throw new IllegalStateException();
738	–	Object x = p.get();
739	–	if (x != null && x != p && p.compareAndSet(x, p))
740	–	clean(pcurr, p);
741	–	}
742	–	}
743	–
1063		public E peek() {
1064	<	for (;;) {
746	<	Node<E> h = traversalHead();
747	<	Node<E> p = h.next;
748	<	if (p == null)
749	<	return null;
750	<	Object x = p.get();
751	<	if (p != x) {
752	<	if (!p.isData)
753	<	return null;
754	<	if (x != null)
755	<	return (E) x;
756	<	}
757	<	}
1064	>	return (E) firstDataItem();
1065		}
1066
1067		/**
#	Line 763 | Line 1070 | public class LinkedTransferQueue<E> exte
1070		* @return {@code true} if this queue contains no elements
1071		*/
1072		public boolean isEmpty() {
1073	<	for (;;) {
767	<	Node<E> h = traversalHead();
768	<	Node<E> p = h.next;
769	<	if (p == null)
770	<	return true;
771	<	Object x = p.get();
772	<	if (p != x) {
773	<	if (!p.isData)
774	<	return true;
775	<	if (x != null)
776	<	return false;
777	<	}
778	<	}
1073	>	return firstOfMode(true) == null;
1074		}
1075
1076		public boolean hasWaitingConsumer() {
1077	<	for (;;) {
783	<	Node<E> h = traversalHead();
784	<	Node<E> p = h.next;
785	<	if (p == null)
786	<	return false;
787	<	Object x = p.get();
788	<	if (p != x)
789	<	return !p.isData;
790	<	}
1077	>	return firstOfMode(false) != null;
1078		}
1079
1080		/**
#	Line 803 | Line 1090 | public class LinkedTransferQueue<E> exte
1090		* @return the number of elements in this queue
1091		*/
1092		public int size() {
1093	<	for (;;) {
807	<	int count = 0;
808	<	Node<E> pred = traversalHead();
809	<	for (;;) {
810	<	Node<E> q = pred.next;
811	<	if (q == pred) // restart
812	<	break;
813	<	if (q == null || !q.isData)
814	<	return count;
815	<	Object x = q.get();
816	<	if (x != null && x != q) {
817	<	if (++count == Integer.MAX_VALUE) // saturated
818	<	return count;
819	<	}
820	<	pred = q;
821	<	}
822	<	}
1093	>	return countOfMode(true);
1094		}
1095
1096		public int getWaitingConsumerCount() {
1097	<	// converse of size -- count valid non-data nodes
827	<	for (;;) {
828	<	int count = 0;
829	<	Node<E> pred = traversalHead();
830	<	for (;;) {
831	<	Node<E> q = pred.next;
832	<	if (q == pred) // restart
833	<	break;
834	<	if (q == null || q.isData)
835	<	return count;
836	<	Object x = q.get();
837	<	if (x == null) {
838	<	if (++count == Integer.MAX_VALUE) // saturated
839	<	return count;
840	<	}
841	<	pred = q;
842	<	}
843	<	}
1097	>	return countOfMode(false);
1098		}
1099
1100		/**
#	Line 855 | Line 1109 | public class LinkedTransferQueue<E> exte
1109		* @return {@code true} if this queue changed as a result of the call
1110		*/
1111		public boolean remove(Object o) {
1112	<	if (o == null)
859	<	return false;
860	<	for (;;) {
861	<	Node<E> pred = traversalHead();
862	<	for (;;) {
863	<	Node<E> q = pred.next;
864	<	if (q == pred) // restart
865	<	break;
866	<	if (q == null || !q.isData)
867	<	return false;
868	<	Object x = q.get();
869	<	if (x != null && x != q && o.equals(x) &&
870	<	q.compareAndSet(x, q)) {
871	<	clean(pred, q);
872	<	return true;
873	<	}
874	<	pred = q;
875	<	}
876	<	}
1112	>	return findAndRemove(o);
1113		}
1114
1115		/**
#	Line 912 | Line 1148 | public class LinkedTransferQueue<E> exte
1148		private void readObject(java.io.ObjectInputStream s)
1149		throws java.io.IOException, ClassNotFoundException {
1150		s.defaultReadObject();
915	–	resetHeadAndTail();
1151		for (;;) {
1152		@SuppressWarnings("unchecked") E item = (E) s.readObject();
1153		if (item == null)
#	Line 922 | Line 1157 | public class LinkedTransferQueue<E> exte
1157		}
1158		}
1159
925	–	// Support for resetting head/tail while deserializing
926	–	private void resetHeadAndTail() {
927	–	Node<E> dummy = new Node<E>(null, false);
928	–	UNSAFE.putObjectVolatile(this, headOffset,
929	–	new PaddedAtomicReference<Node<E>>(dummy));
930	–	UNSAFE.putObjectVolatile(this, tailOffset,
931	–	new PaddedAtomicReference<Node<E>>(dummy));
932	–	UNSAFE.putObjectVolatile(this, cleanMeOffset,
933	–	new PaddedAtomicReference<Node<E>>(null));
934	–	}
1160
1161		// Unsafe mechanics
1162
#	Line 943 | Line 1168 | public class LinkedTransferQueue<E> exte
1168		private static final long cleanMeOffset =
1169		objectFieldOffset(UNSAFE, "cleanMe", LinkedTransferQueue.class);
1170
946	–
1171		static long objectFieldOffset(sun.misc.Unsafe UNSAFE,
1172		String field, Class<?> klazz) {
1173		try {
#	Line 956 | Line 1180 | public class LinkedTransferQueue<E> exte
1180		}
1181		}
1182
959	–	/**
960	–	* Returns a sun.misc.Unsafe.  Suitable for use in a 3rd party package.
961	–	* Replace with a simple call to Unsafe.getUnsafe when integrating
962	–	* into a jdk.
963	–	*
964	–	* @return a sun.misc.Unsafe
965	–	*/
1183		private static sun.misc.Unsafe getUnsafe() {
1184		try {
1185		return sun.misc.Unsafe.getUnsafe();
#	Line 983 | Line 1200 | public class LinkedTransferQueue<E> exte
1200		}
1201		}
1202		}
1203	+
1204		}

Diff Legend

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