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.25 by jsr166, Fri Jul 24 23:48:26 2009 UTC vs.
Revision 1.62 by jsr166, Mon Nov 2 03:01:10 2009 UTC

#	Line 5 | Line 5
5		*/
6
7		package jsr166y;
8	+
9		import java.util.concurrent.*;
9	–	import java.util.concurrent.locks.*;
10	–	import java.util.concurrent.atomic.*;
11	–	import java.util.*;
12	–	import java.io.*;
10
11	+	import java.util.AbstractQueue;
12	+	import java.util.Collection;
13	+	import java.util.ConcurrentModificationException;
14	+	import java.util.Iterator;
15	+	import java.util.NoSuchElementException;
16	+	import java.util.Queue;
17	+	import java.util.concurrent.locks.LockSupport;
18		/**
19	<	* An unbounded {@linkplain TransferQueue} based on linked nodes.
19	>	* An unbounded {@link TransferQueue} based on linked nodes.
20		* This queue orders elements FIFO (first-in-first-out) with respect
21		* to any given producer.  The <em>head</em> of the queue is that
22		* element that has been on the queue the longest time for some
#	Line 48 | 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
52	<	* SynchronousQueues. See the internal documentation, as well as
53	<	* the PPoPP 2006 paper "Scalable Synchronous Queues" by Scherer,
54	<	* Lea & Scott
55	<	* (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 interior elements, such as
109	>	* j.u.c.ConcurrentLinkedQueue.)
110	>	*
111	>	* Once a node is matched, its match status can never again
112	>	* change.  We may thus arrange that the linked list of them
113	>	* contain a prefix of zero or more matched nodes, followed by a
114	>	* suffix of zero or more unmatched nodes. (Note that we allow
115	>	* both the prefix and suffix to be zero length, which in turn
116	>	* means that we do not use a dummy header.)  If we were not
117	>	* concerned with either time or space efficiency, we could
118	>	* correctly perform enqueue and dequeue operations by traversing
119	>	* from a pointer to the initial node; CASing the item of the
120	>	* first unmatched node on match and CASing the next field of the
121	>	* trailing node on appends. (Plus some special-casing when
122	>	* initially empty).  While this would be a terrible idea in
123	>	* itself, it does have the benefit of not requiring ANY atomic
124	>	* updates on head/tail fields.
125	>	*
126	>	* We introduce here an approach that lies between the extremes of
127	>	* never versus always updating queue (head and tail) pointers.
128	>	* This offers a tradeoff between sometimes requiring extra
129	>	* traversal steps to locate the first and/or last unmatched
130	>	* nodes, versus the reduced overhead and contention of fewer
131	>	* updates to queue pointers. For example, a possible snapshot of
132	>	* a queue is:
133	>	*
134	>	*  head           tail
135	>	*    |              |
136	>	*    v              v
137	>	*    M -> M -> U -> U -> U -> U
138	>	*
139	>	* The best value for this "slack" (the targeted maximum distance
140	>	* between the value of "head" and the first unmatched node, and
141	>	* similarly for "tail") is an empirical matter. We have found
142	>	* that using very small constants in the range of 1-3 work best
143	>	* over a range of platforms. Larger values introduce increasing
144	>	* costs of cache misses and risks of long traversal chains, while
145	>	* smaller values increase CAS contention and overhead.
146	>	*
147	>	* Dual queues with slack differ from plain M&S dual queues by
148	>	* virtue of only sometimes updating head or tail pointers when
149	>	* matching, appending, or even traversing nodes; in order to
150	>	* maintain a targeted slack.  The idea of "sometimes" may be
151	>	* operationalized in several ways. The simplest is to use a
152	>	* per-operation counter incremented on each traversal step, and
153	>	* to try (via CAS) to update the associated queue pointer
154	>	* whenever the count exceeds a threshold. Another, that requires
155	>	* more overhead, is to use random number generators to update
156	>	* with a given probability per traversal step.
157	>	*
158	>	* In any strategy along these lines, because CASes updating
159	>	* fields may fail, the actual slack may exceed targeted
160	>	* slack. However, they may be retried at any time to maintain
161	>	* targets.  Even when using very small slack values, this
162	>	* approach works well for dual queues because it allows all
163	>	* operations up to the point of matching or appending an item
164	>	* (hence potentially allowing progress by another thread) to be
165	>	* read-only, thus not introducing any further contention. As
166	>	* described below, we implement this by performing slack
167	>	* maintenance retries only after these points.
168	>	*
169	>	* As an accompaniment to such techniques, traversal overhead can
170	>	* be further reduced without increasing contention of head
171	>	* pointer updates: Threads may sometimes shortcut the "next" link
172	>	* path from the current "head" node to be closer to the currently
173	>	* known first unmatched node, and similarly for tail. Again, this
174	>	* may be triggered with using thresholds or randomization.
175	>	*
176	>	* These ideas must be further extended to avoid unbounded amounts
177	>	* of costly-to-reclaim garbage caused by the sequential "next"
178	>	* links of nodes starting at old forgotten head nodes: As first
179	>	* described in detail by Boehm
180	>	* (http://portal.acm.org/citation.cfm?doid=503272.503282) if a GC
181	>	* delays noticing that any arbitrarily old node has become
182	>	* garbage, all newer dead nodes will also be unreclaimed.
183	>	* (Similar issues arise in non-GC environments.)  To cope with
184	>	* this in our implementation, upon CASing to advance the head
185	>	* pointer, we set the "next" link of the previous head to point
186	>	* only to itself; thus limiting the length of connected dead lists.
187	>	* (We also take similar care to wipe out possibly garbage
188	>	* retaining values held in other Node fields.)  However, doing so
189	>	* adds some further complexity to traversal: If any "next"
190	>	* pointer links to itself, it indicates that the current thread
191	>	* has lagged behind a head-update, and so the traversal must
192	>	* continue from the "head".  Traversals trying to find the
193	>	* current tail starting from "tail" may also encounter
194	>	* self-links, in which case they also continue at "head".
195	>	*
196	>	* It is tempting in slack-based scheme to not even use CAS for
197	>	* updates (similarly to Ladan-Mozes & Shavit). However, this
198	>	* cannot be done for head updates under the above link-forgetting
199	>	* mechanics because an update may leave head at a detached node.
200	>	* And while direct writes are possible for tail updates, they
201	>	* increase the risk of long retraversals, and hence long garbage
202	>	* chains, which can be much more costly than is worthwhile
203	>	* considering that the cost difference of performing a CAS vs
204	>	* write is smaller when they are not triggered on each operation
205	>	* (especially considering that writes and CASes equally require
206	>	* additional GC bookkeeping ("write barriers") that are sometimes
207	>	* more costly than the writes themselves because of contention).
208	>	*
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	>	*
227	>	* *** Overview of implementation ***
228	>	*
229	>	* We use a threshold-based approach to updates, with a slack
230	>	* threshold of two -- that is, we update head/tail when the
231	>	* current pointer appears to be two or more steps away from the
232	>	* first/last node. The slack value is hard-wired: a path greater
233	>	* than one is naturally implemented by checking equality of
234	>	* traversal pointers except when the list has only one element,
235	>	* in which case we keep slack threshold at one. Avoiding tracking
236	>	* explicit counts across method calls slightly simplifies an
237	>	* already-messy implementation. Using randomization would
238	>	* probably work better if there were a low-quality dirt-cheap
239	>	* per-thread one available, but even ThreadLocalRandom is too
240	>	* heavy for these purposes.
241	>	*
242	>	* With such a small slack threshold value, it is rarely
243	>	* worthwhile to augment this with path short-circuiting; i.e.,
244	>	* unsplicing nodes between head and the first unmatched node, or
245	>	* 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).
251	>	*
252	>	* We allow both the head and tail fields to be null before any
253	>	* nodes are enqueued; initializing upon first append.  This
254	>	* simplifies some other logic, as well as providing more
255	>	* efficient explicit control paths instead of letting JVMs insert
256	>	* implicit NullPointerExceptions when they are null.  While not
257	>	* currently fully implemented, we also leave open the possibility
258	>	* of re-nulling these fields when empty (which is complicated to
259	>	* arrange, for little benefit.)
260	>	*
261	>	* All enqueue/dequeue operations are handled by the single method
262	>	* "xfer" with parameters indicating whether to act as some form
263	>	* of offer, put, poll, take, or transfer (each possibly with
264	>	* timeout). The relative complexity of using one monolithic
265	>	* method outweighs the code bulk and maintenance problems of
266	>	* using separate methods for each case.
267	>	*
268	>	* Operation consists of up to three phases. The first is
269	>	* implemented within method xfer, the second in tryAppend, and
270	>	* the third in method awaitMatch.
271	>	*
272	>	* 1. Try to match an existing node
273	>	*
274	>	*    Starting at head, skip already-matched nodes until finding
275	>	*    an unmatched node of opposite mode, if one exists, in which
276	>	*    case matching it and returning, also if necessary updating
277	>	*    head to one past the matched node (or the node itself if the
278	>	*    list has no other unmatched nodes). If the CAS misses, then
279	>	*    a loop retries advancing head by two steps until either
280	>	*    success or the slack is at most two. By requiring that each
281	>	*    attempt advances head by two (if applicable), we ensure that
282	>	*    the slack does not grow without bound. Traversals also check
283	>	*    if the initial head is now off-list, in which case they
284	>	*    start at the new head.
285	>	*
286	>	*    If no candidates are found and the call was untimed
287	>	*    poll/offer, (argument "how" is NOW) return.
288	>	*
289	>	* 2. Try to append a new node (method tryAppend)
290	>	*
291	>	*    Starting at current tail pointer, find the actual last node
292	>	*    and try to append a new node (or if head was null, establish
293	>	*    the first node). Nodes can be appended only if their
294	>	*    predecessors are either already matched or are of the same
295	>	*    mode. If we detect otherwise, then a new node with opposite
296	>	*    mode must have been appended during traversal, so we must
297	>	*    restart at phase 1. The traversal and update steps are
298	>	*    otherwise similar to phase 1: Retrying upon CAS misses and
299	>	*    checking for staleness.  In particular, if a self-link is
300	>	*    encountered, then we can safely jump to a node on the list
301	>	*    by continuing the traversal at current head.
302	>	*
303	>	*    On successful append, if the call was ASYNC, return.
304	>	*
305	>	* 3. Await match or cancellation (method awaitMatch)
306	>	*
307	>	*    Wait for another thread to match node; instead cancelling if
308	>	*    the current thread was interrupted or the wait timed out. On
309	>	*    multiprocessors, we use front-of-queue spinning: If a node
310	>	*    appears to be the first unmatched node in the queue, it
311	>	*    spins a bit before blocking. In either case, before blocking
312	>	*    it tries to unsplice any nodes between the current "head"
313	>	*    and the first unmatched node.
314	>	*
315	>	*    Front-of-queue spinning vastly improves performance of
316	>	*    heavily contended queues. And so long as it is relatively
317	>	*    brief and "quiet", spinning does not much impact performance
318	>	*    of less-contended queues.  During spins threads check their
319	>	*    interrupt status and generate a thread-local random number
320	>	*    to decide to occasionally perform a Thread.yield. While
321	>	*    yield has underdefined specs, we assume that might it help,
322	>	*    and will not hurt in limiting impact of spinning on busy
323	>	*    systems.  We also use smaller (1/2) spins for nodes that are
324	>	*    not known to be front but whose predecessors have not
325	>	*    blocked -- these "chained" spins avoid artifacts of
326	>	*    front-of-queue rules which otherwise lead to alternating
327	>	*    nodes spinning vs blocking. Further, front threads that
328	>	*    represent phase changes (from data to request node or vice
329	>	*    versa) compared to their predecessors receive additional
330	>	*    chained spins, reflecting longer paths typically required to
331	>	*    unblock threads during phase changes.
332	>	*/
333	>
334	>	/** True if on multiprocessor */
335	>	private static final boolean MP =
336	>	Runtime.getRuntime().availableProcessors() > 1;
337
338		/**
339	<	* The number of times to spin before blocking in untimed waits.
340	<	* This is greater than timed value because untimed waits spin
341	<	* faster since they don't need to check times on each spin.
339	>	* The number of times to spin (with randomly interspersed calls
340	>	* to Thread.yield) on multiprocessor before blocking when a node
341	>	* is apparently the first waiter in the queue.  See above for
342	>	* explanation. Must be a power of two. The value is empirically
343	>	* derived -- it works pretty well across a variety of processors,
344	>	* numbers of CPUs, and OSes.
345		*/
346	<	static final int maxUntimedSpins = maxTimedSpins * 16;
346	>	private static final int FRONT_SPINS   = 1 << 7;
347
348		/**
349	<	* The number of nanoseconds for which it is faster to spin
350	<	* rather than to use timed park. A rough estimate suffices.
349	>	* The number of times to spin before blocking when a node is
350	>	* preceded by another node that is apparently spinning.  Also
351	>	* serves as an increment to FRONT_SPINS on phase changes, and as
352	>	* base average frequency for yielding during spins. Must be a
353	>	* power of two.
354		*/
355	<	static final long spinForTimeoutThreshold = 1000L;
355	>	private static final int CHAINED_SPINS = FRONT_SPINS >>> 1;
356
357		/**
358	<	* Node class for LinkedTransferQueue. Opportunistically
359	<	* subclasses from AtomicReference to represent item. Uses Object,
360	<	* not E, to allow setting item to "this" after use, to avoid
361	<	* garbage retention. Similarly, setting the next field to this is
362	<	* used as sentinel that node is off list.
363	<	*/
364	<	static final class Node<E> extends AtomicReference<Object> {
365	<	volatile Node<E> next;
366	<	volatile Thread waiter;       // to control park/unpark
367	<	final boolean isData;
358	>	* Queue nodes. Uses Object, not E, for items to allow forgetting
359	>	* them after use.  Relies heavily on Unsafe mechanics to minimize
360	>	* unnecessary ordering constraints: Writes that intrinsically
361	>	* precede or follow CASes use simple relaxed forms.  Other
362	>	* cleanups use releasing/lazy writes.
363	>	*/
364	>	static final class Node {
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;
368	>	volatile Thread waiter; // null until waiting
369	>
370	>	// CAS methods for fields
371	>	final boolean casNext(Node cmp, Node 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
380	<	Node(E item, boolean isData) {
381	<	super(item);
380	>	/**
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) {
385	>	UNSAFE.putObject(this, itemOffset, item); // relaxed write
386		this.isData = isData;
387		}
388
389	<	@SuppressWarnings("rawtypes")
390	<	static final AtomicReferenceFieldUpdater<Node, Node>
391	<	nextUpdater = AtomicReferenceFieldUpdater.newUpdater
392	<	(Node.class, Node.class, "next");
389	>	/**
390	>	* Links node to itself to avoid garbage retention.  Called
391	>	* only after CASing head field, so uses relaxed write.
392	>	*/
393	>	final void forgetNext() {
394	>	UNSAFE.putObject(this, nextOffset, this);
395	>	}
396	>
397	>	/**
398	>	* Sets item to self (using a releasing/lazy write) and waiter
399	>	* to null, to avoid garbage retention after extracting or
400	>	* cancelling.
401	>	*/
402	>	final void forgetContents() {
403	>	UNSAFE.putOrderedObject(this, itemOffset, this);
404	>	UNSAFE.putOrderedObject(this, waiterOffset, null);
405	>	}
406	>
407	>	/**
408	>	* Returns true if this node has been matched, including the
409	>	* case of artificial matches due to cancellation.
410	>	*/
411	>	final boolean isMatched() {
412	>	Object x = item;
413	>	return (x == this) || ((x == null) == isData);
414	>	}
415
416	<	final boolean casNext(Node<E> cmp, Node<E> val) {
417	<	return nextUpdater.compareAndSet(this, cmp, val);
416	>	/**
417	>	* Returns true if this is an unmatched request node.
418	>	*/
419	>	final boolean isUnmatchedRequest() {
420	>	return !isData && item == null;
421		}
422
423	<	final void clearNext() {
424	<	nextUpdater.lazySet(this, this);
423	>	/**
424	>	* Returns true if a node with the given mode cannot be
425	>	* appended to this node because this node is unmatched and
426	>	* has opposite data mode.
427	>	*/
428	>	final boolean cannotPrecede(boolean haveData) {
429	>	boolean d = isData;
430	>	Object x;
431	>	return d != haveData && (x = item) != this && (x != null) == d;
432		}
433
434	+	/**
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);
442	+	return true;
443	+	}
444	+	return false;
445	+	}
446	+
447	+	// Unsafe mechanics
448	+	private static final sun.misc.Unsafe UNSAFE = getUnsafe();
449	+	private static final long nextOffset =
450	+	objectFieldOffset(UNSAFE, "next", Node.class);
451	+	private static final long itemOffset =
452	+	objectFieldOffset(UNSAFE, "item", Node.class);
453	+	private static final long waiterOffset =
454	+	objectFieldOffset(UNSAFE, "waiter", Node.class);
455	+
456		private static final long serialVersionUID = -3375979862319811754L;
457		}
458
459	<	/**
460	<	* Padded version of AtomicReference used for head, tail and
131	<	* cleanMe, to alleviate contention across threads CASing one vs
132	<	* the other.
133	<	*/
134	<	static final class PaddedAtomicReference<T> extends AtomicReference<T> {
135	<	// enough padding for 64bytes with 4byte refs
136	<	Object p0, p1, p2, p3, p4, p5, p6, p7, p8, p9, pa, pb, pc, pd, pe;
137	<	PaddedAtomicReference(T r) { super(r); }
138	<	private static final long serialVersionUID = 8170090609809740854L;
139	<	}
459	>	/** head of the queue; null until first enqueue */
460	>	transient volatile Node head;
461
462	+	/** predecessor of dangling unspliceable node */
463	+	private transient volatile Node cleanMe; // decl here reduces contention
464
465	<	/** head of the queue */
466	<	private transient final PaddedAtomicReference<Node<E>> head;
465	>	/** tail of the queue; null until first append */
466	>	private transient volatile Node tail;
467
468	<	/** tail of the queue */
469	<	private transient final PaddedAtomicReference<Node<E>> tail;
468	>	// CAS methods for fields
469	>	private boolean casTail(Node cmp, Node val) {
470	>	return UNSAFE.compareAndSwapObject(this, tailOffset, cmp, val);
471	>	}
472
473	<	/**
474	<	* Reference to a cancelled node that might not yet have been
475	<	* unlinked from queue because it was the last inserted node
151	<	* when it cancelled.
152	<	*/
153	<	private transient final PaddedAtomicReference<Node<E>> cleanMe;
473	>	private boolean casHead(Node cmp, Node val) {
474	>	return UNSAFE.compareAndSwapObject(this, headOffset, cmp, val);
475	>	}
476
477	<	/**
478	<	* Tries to cas nh as new head; if successful, unlink
479	<	* old head's next node to avoid garbage retention.
477	>	private boolean casCleanMe(Node cmp, Node val) {
478	>	return UNSAFE.compareAndSwapObject(this, cleanMeOffset, cmp, val);
479	>	}
480	>
481	>	/*
482	>	* Possible values for "how" argument in xfer method. Beware that
483	>	* the order of assigned numerical values matters.
484		*/
485	<	private boolean advanceHead(Node<E> h, Node<E> nh) {
486	<	if (h == head.get() && head.compareAndSet(h, nh)) {
487	<	h.clearNext(); // forget old next
488	<	return true;
489	<	}
490	<	return false;
485	>	private static final int NOW     = 0; // for untimed poll, tryTransfer
486	>	private static final int ASYNC   = 1; // for offer, put, add
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	<	* Puts or takes an item. Used for most queue operations (except
169	<	* poll() and tryTransfer()). See the similar code in
170	<	* SynchronousQueue for detailed explanation.
497	>	* Implements all queuing methods. See above for explanation.
498		*
499	<	* @param e the item or if null, signifies that this is a take
500	<	* @param mode the wait mode: NOWAIT, TIMEOUT, WAIT
499	>	* @param e the item or null for take
500	>	* @param haveData true if this is a put, else a take
501	>	* @param how NOW, ASYNC, SYNC, or TIMEOUT
502		* @param nanos timeout in nanosecs, used only if mode is TIMEOUT
503	<	* @return an item, or null on failure
503	>	* @return an item if matched, else e
504	>	* @throws NullPointerException if haveData mode but e is null
505		*/
506	<	private E xfer(E e, int mode, long nanos) {
507	<	boolean isData = (e != null);
508	<	Node<E> s = null;
509	<	final PaddedAtomicReference<Node<E>> head = this.head;
181	<	final PaddedAtomicReference<Node<E>> tail = this.tail;
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
510
511	<	for (;;) {
184	<	Node<E> t = tail.get();
185	<	Node<E> h = head.get();
511	>	retry: for (;;) {                     // restart on append race
512
513	<	if (t != null && (t == h || t.isData == isData)) {
514	<	if (s == null)
515	<	s = new Node<E>(e, isData);
516	<	Node<E> last = t.next;
517	<	if (last != null) {
518	<	if (t == tail.get())
519	<	tail.compareAndSet(t, last);
520	<	}
521	<	else if (t.casNext(null, s)) {
522	<	tail.compareAndSet(t, s);
523	<	return awaitFulfill(t, s, e, mode, nanos);
513	>	for (Node h = head, p = h; p != null;) { // find & match first node
514	>	boolean isData = p.isData;
515	>	Object item = p.item;
516	>	if (item != p && (item != null) == isData) { // unmatched
517	>	if (isData == haveData)   // can't match
518	>	break;
519	>	if (p.casItem(item, e)) { // match
520	>	for (Node q = p; q != h;) {
521	>	Node n = q.next;  // update head by 2
522	>	if (n != null)    // unless singleton
523	>	q = n;
524	>	if (head == h && casHead(h, q)) {
525	>	h.forgetNext();
526	>	break;
527	>	}                 // advance and retry
528	>	if ((h = head)   == null ||
529	>	(q = h.next) == null || !q.isMatched())
530	>	break;        // unless slack < 2
531	>	}
532	>	LockSupport.unpark(p.waiter);
533	>	return this.<E>cast(item);
534	>	}
535		}
536	+	Node n = p.next;
537	+	p = (p != n) ? n : (h = head); // Use head if p offlist
538		}
539
540	<	else if (h != null) {
541	<	Node<E> first = h.next;
542	<	if (t == tail.get() && first != null &&
543	<	advanceHead(h, first)) {
544	<	Object x = first.get();
545	<	if (x != first && first.compareAndSet(x, e)) {
546	<	LockSupport.unpark(first.waiter);
547	<	return isData ? e : (E) x;
209	<	}
210	<	}
540	>	if (how >= ASYNC) {               // No matches available
541	>	if (s == null)
542	>	s = new Node(e, haveData);
543	>	Node pred = tryAppend(s, haveData);
544	>	if (pred == null)
545	>	continue retry;           // lost race vs opposite mode
546	>	if (how >= SYNC)
547	>	return awaitMatch(s, pred, e, how, nanos);
548		}
549	+	return e; // not waiting
550		}
551		}
552
215	–
553		/**
554	<	* Version of xfer for poll() and tryTransfer, which
555	<	* simplifies control paths both here and in xfer.
556	<	*/
557	<	private E fulfill(E e) {
558	<	boolean isData = (e != null);
559	<	final PaddedAtomicReference<Node<E>> head = this.head;
560	<	final PaddedAtomicReference<Node<E>> tail = this.tail;
561	<
562	<	for (;;) {
563	<	Node<E> t = tail.get();
564	<	Node<E> h = head.get();
565	<
566	<	if (t != null && (t == h || t.isData == isData)) {
567	<	Node<E> last = t.next;
568	<	if (t == tail.get()) {
569	<	if (last != null)
570	<	tail.compareAndSet(t, last);
571	<	else
572	<	return null;
573	<	}
574	<	}
575	<	else if (h != null) {
576	<	Node<E> first = h.next;
577	<	if (t == tail.get() &&
578	<	first != null &&
579	<	advanceHead(h, first)) {
580	<	Object x = first.get();
581	<	if (x != first && first.compareAndSet(x, e)) {
245	<	LockSupport.unpark(first.waiter);
246	<	return isData ? e : (E) x;
247	<	}
554	>	* Tries to append node s as tail.
555	>	*
556	>	* @param s the node to append
557	>	* @param haveData true if appending in data mode
558	>	* @return null on failure due to losing race with append in
559	>	* different mode, else s's predecessor, or s itself if no
560	>	* predecessor
561	>	*/
562	>	private Node tryAppend(Node s, boolean haveData) {
563	>	for (Node t = tail, p = t;;) {        // move p to last node and append
564	>	Node n, u;                        // temps for reads of next & tail
565	>	if (p == null && (p = head) == null) {
566	>	if (casHead(null, s))
567	>	return s;                 // initialize
568	>	}
569	>	else if (p.cannotPrecede(haveData))
570	>	return null;                  // lost race vs opposite mode
571	>	else if ((n = p.next) != null)    // not last; keep traversing
572	>	p = p != t && t != (u = tail) ? (t = u) : // stale tail
573	>	(p != n) ? n : null;      // restart if off list
574	>	else if (!p.casNext(null, s))
575	>	p = p.next;                   // re-read on CAS failure
576	>	else {
577	>	if (p != t) {                 // update if slack now >= 2
578	>	while ((tail != t || !casTail(t, s)) &&
579	>	(t = tail)   != null &&
580	>	(s = t.next) != null && // advance and retry
581	>	(s = s.next) != null && s != t);
582		}
583	+	return p;
584		}
585		}
586		}
587
588		/**
589	<	* Spins/blocks until node s is fulfilled or caller gives up,
255	<	* depending on wait mode.
589	>	* Spins/yields/blocks until node s is matched or caller gives up.
590		*
257	–	* @param pred the predecessor of waiting node
591		* @param s the waiting node
592	+	* @param pred the predecessor of s, or s itself if it has no
593	+	* predecessor, or null if unknown (the null case does not occur
594	+	* in any current calls but may in possible future extensions)
595		* @param e the comparison value for checking match
596	<	* @param mode mode
596	>	* @param how either SYNC or TIMEOUT
597		* @param nanos timeout value
598	<	* @return matched item, or s if cancelled
598	>	* @return matched item, or e if unmatched on interrupt or timeout
599		*/
600	<	private E awaitFulfill(Node<E> pred, Node<E> s, E e,
601	<	int mode, long nanos) {
266	<	if (mode == NOWAIT)
267	<	return null;
268	<
269	<	long lastTime = (mode == TIMEOUT) ? System.nanoTime() : 0;
600	>	private E awaitMatch(Node s, Node pred, E e, int how, long nanos) {
601	>	long lastTime = (how == TIMEOUT) ? System.nanoTime() : 0L;
602		Thread w = Thread.currentThread();
603	<	int spins = -1; // set to desired spin count below
603	>	int spins = -1; // initialized after first item and cancel checks
604	>	ThreadLocalRandom randomYields = null; // bound if needed
605	>
606		for (;;) {
607	<	if (w.isInterrupted())
608	<	s.compareAndSet(e, s);
609	<	Object x = s.get();
610	<	if (x != e) {                 // Node was matched or cancelled
611	<	advanceHead(pred, s);     // unlink if head
612	<	if (x == s) {             // was cancelled
613	<	clean(pred, s);
614	<	return null;
615	<	}
616	<	else if (x != null) {
617	<	s.set(s);             // avoid garbage retention
618	<	return (E) x;
619	<	}
620	<	else
621	<	return e;
607	>	Object item = s.item;
608	>	if (item != e) {                  // matched
609	>	assert item != s;
610	>	s.forgetContents();           // avoid garbage
611	>	return this.<E>cast(item);
612	>	}
613	>	if ((w.isInterrupted() || (how == TIMEOUT && nanos <= 0)) &&
614	>	s.casItem(e, s)) {       // cancel
615	>	unsplice(pred, s);
616	>	return e;
617	>	}
618	>
619	>	if (spins < 0) {                  // establish spins at/near front
620	>	if ((spins = spinsFor(pred, s.isData)) > 0)
621	>	randomYields = ThreadLocalRandom.current();
622	>	}
623	>	else if (spins > 0) {             // spin
624	>	if (--spins == 0)
625	>	shortenHeadPath();        // reduce slack before blocking
626	>	else if (randomYields.nextInt(CHAINED_SPINS) == 0)
627	>	Thread.yield();           // occasionally yield
628		}
629	<	if (mode == TIMEOUT) {
629	>	else if (s.waiter == null) {
630	>	s.waiter = w;                 // request unpark then recheck
631	>	}
632	>	else if (how == TIMEOUT) {
633		long now = System.nanoTime();
634	<	nanos -= now - lastTime;
634	>	if ((nanos -= now - lastTime) > 0)
635	>	LockSupport.parkNanos(this, nanos);
636		lastTime = now;
293	–	if (nanos <= 0) {
294	–	s.compareAndSet(e, s); // try to cancel
295	–	continue;
296	–	}
637		}
638	<	if (spins < 0) {
299	<	Node<E> h = head.get(); // only spin if at head
300	<	spins = ((h != null && h.next == s) ?
301	<	((mode == TIMEOUT) ?
302	<	maxTimedSpins : maxUntimedSpins) : 0);
303	<	}
304	<	if (spins > 0)
305	<	--spins;
306	<	else if (s.waiter == null)
307	<	s.waiter = w;
308	<	else if (mode != TIMEOUT) {
638	>	else {
639		LockSupport.park(this);
640		s.waiter = null;
641	<	spins = -1;
312	<	}
313	<	else if (nanos > spinForTimeoutThreshold) {
314	<	LockSupport.parkNanos(this, nanos);
315	<	s.waiter = null;
316	<	spins = -1;
641	>	spins = -1;                   // spin if front upon wakeup
642		}
643		}
644		}
645
646		/**
647	<	* Returns validated tail for use in cleaning methods.
647	>	* Returns spin/yield value for a node with given predecessor and
648	>	* data mode. See above for explanation.
649		*/
650	<	private Node<E> getValidatedTail() {
651	<	for (;;) {
652	<	Node<E> h = head.get();
653	<	Node<E> first = h.next;
654	<	if (first != null && first.next == first) { // help advance
655	<	advanceHead(h, first);
656	<	continue;
650	>	private static int spinsFor(Node pred, boolean haveData) {
651	>	if (MP && pred != null) {
652	>	if (pred.isData != haveData)      // phase change
653	>	return FRONT_SPINS + CHAINED_SPINS;
654	>	if (pred.isMatched())             // probably at front
655	>	return FRONT_SPINS;
656	>	if (pred.waiter == null)          // pred apparently spinning
657	>	return CHAINED_SPINS;
658	>	}
659	>	return 0;
660	>	}
661	>
662	>	/**
663	>	* Tries (once) to unsplice nodes between head and first unmatched
664	>	* or trailing node; failing on contention.
665	>	*/
666	>	private void shortenHeadPath() {
667	>	Node h, hn, p, q;
668	>	if ((p = h = head) != null && h.isMatched() &&
669	>	(q = hn = h.next) != null) {
670	>	Node n;
671	>	while ((n = q.next) != q) {
672	>	if (n == null || !q.isMatched()) {
673	>	if (hn != q && h.next == hn)
674	>	h.casNext(hn, q);
675	>	break;
676	>	}
677	>	p = q;
678	>	q = n;
679		}
680	<	Node<E> t = tail.get();
681	<	Node<E> last = t.next;
682	<	if (t == tail.get()) {
683	<	if (last != null)
684	<	tail.compareAndSet(t, last); // help advance
685	<	else
686	<	return t;
680	>	}
681	>	}
682	>
683	>	/* -------------- Traversal methods -------------- */
684	>
685	>	/**
686	>	* Returns the successor of p, or the head node if p.next has been
687	>	* linked to self, which will only be true if traversing with a
688	>	* stale pointer that is now off the list.
689	>	*/
690	>	final Node succ(Node p) {
691	>	Node next = p.next;
692	>	return (p == next) ? head : next;
693	>	}
694	>
695	>	/**
696	>	* Returns the first unmatched node of the given mode, or null if
697	>	* none.  Used by methods isEmpty, hasWaitingConsumer.
698	>	*/
699	>	private Node firstOfMode(boolean isData) {
700	>	for (Node p = head; p != null; p = succ(p)) {
701	>	if (!p.isMatched())
702	>	return (p.isData == isData) ? p : null;
703	>	}
704	>	return null;
705	>	}
706	>
707	>	/**
708	>	* Returns the item in the first unmatched node with isData; or
709	>	* null if none.  Used by peek.
710	>	*/
711	>	private E firstDataItem() {
712	>	for (Node p = head; p != null; p = succ(p)) {
713	>	Object item = p.item;
714	>	if (p.isData) {
715	>	if (item != null && item != p)
716	>	return this.<E>cast(item);
717		}
718	+	else if (item == null)
719	+	return null;
720		}
721	+	return null;
722		}
723
724		/**
725	<	* Gets rid of cancelled node s with original predecessor pred.
726	<	*
346	<	* @param pred predecessor of cancelled node
347	<	* @param s the cancelled node
725	>	* Traverses and counts unmatched nodes of the given mode.
726	>	* Used by methods size and getWaitingConsumerCount.
727		*/
728	<	private void clean(Node<E> pred, Node<E> s) {
729	<	Thread w = s.waiter;
730	<	if (w != null) {             // Wake up thread
731	<	s.waiter = null;
732	<	if (w != Thread.currentThread())
733	<	LockSupport.unpark(w);
728	>	private int countOfMode(boolean data) {
729	>	int count = 0;
730	>	for (Node p = head; p != null; ) {
731	>	if (!p.isMatched()) {
732	>	if (p.isData != data)
733	>	return 0;
734	>	if (++count == Integer.MAX_VALUE) // saturated
735	>	break;
736	>	}
737	>	Node n = p.next;
738	>	if (n != p)
739	>	p = n;
740	>	else {
741	>	count = 0;
742	>	p = head;
743	>	}
744	>	}
745	>	return count;
746	>	}
747	>
748	>	final class Itr implements Iterator<E> {
749	>	private Node nextNode;   // next node to return item for
750	>	private E nextItem;      // the corresponding item
751	>	private Node lastRet;    // last returned node, to support remove
752	>	private Node lastPred;   // predecessor to unlink lastRet
753	>
754	>	/**
755	>	* Moves to next node after prev, or first node if prev null.
756	>	*/
757	>	private void advance(Node prev) {
758	>	lastPred = lastRet;
759	>	lastRet = prev;
760	>	for (Node p = (prev == null) ? head : succ(prev);
761	>	p != null; p = succ(p)) {
762	>	Object item = p.item;
763	>	if (p.isData) {
764	>	if (item != null && item != p) {
765	>	nextItem = LinkedTransferQueue.this.<E>cast(item);
766	>	nextNode = p;
767	>	return;
768	>	}
769	>	}
770	>	else if (item == null)
771	>	break;
772	>	}
773	>	nextNode = null;
774		}
775
776	<	if (pred == null)
777	<	return;
776	>	Itr() {
777	>	advance(null);
778	>	}
779	>
780	>	public final boolean hasNext() {
781	>	return nextNode != null;
782	>	}
783
784	+	public final E next() {
785	+	Node p = nextNode;
786	+	if (p == null) throw new NoSuchElementException();
787	+	E e = nextItem;
788	+	advance(p);
789	+	return e;
790	+	}
791	+
792	+	public final void remove() {
793	+	Node p = lastRet;
794	+	if (p == null) throw new IllegalStateException();
795	+	findAndRemoveDataNode(lastPred, p);
796	+	}
797	+	}
798	+
799	+	/* -------------- Removal methods -------------- */
800	+
801	+	/**
802	+	* Unsplices (now or later) the given deleted/cancelled node with
803	+	* the given predecessor.
804	+	*
805	+	* @param pred predecessor of node to be unspliced
806	+	* @param s the node to be unspliced
807	+	*/
808	+	private void unsplice(Node pred, Node s) {
809	+	s.forgetContents(); // clear unneeded fields
810		/*
811		* At any given time, exactly one node on list cannot be
812	<	* deleted -- the last inserted node. To accommodate this, if
813	<	* we cannot delete s, we save its predecessor as "cleanMe",
814	<	* processing the previously saved version first. At least one
815	<	* of node s or the node previously saved can always be
812	>	* unlinked -- the last inserted node. To accommodate this, if
813	>	* we cannot unlink s, we save its predecessor as "cleanMe",
814	>	* processing the previously saved version first. Because only
815	>	* one node in the list can have a null next, at least one of
816	>	* node s or the node previously saved can always be
817		* processed, so this always terminates.
818		*/
819	<	while (pred.next == s) {
820	<	Node<E> oldpred = reclean();  // First, help get rid of cleanMe
821	<	Node<E> t = getValidatedTail();
822	<	if (s != t) {               // If not tail, try to unsplice
823	<	Node<E> sn = s.next;      // s.next == s means s already off list
824	<	if (sn == s || pred.casNext(s, sn))
819	>	if (pred != null && pred != s) {
820	>	while (pred.next == s) {
821	>	Node oldpred = (cleanMe == null) ? null : reclean();
822	>	Node n = s.next;
823	>	if (n != null) {
824	>	if (n != s)
825	>	pred.casNext(s, n);
826		break;
827	+	}
828	+	if (oldpred == pred ||      // Already saved
829	+	((oldpred == null || oldpred.next == s) &&
830	+	casCleanMe(oldpred, pred))) {
831	+	break;
832	+	}
833		}
376	–	else if (oldpred == pred || // Already saved
377	–	(oldpred == null && cleanMe.compareAndSet(null, pred)))
378	–	break;                  // Postpone cleaning
834		}
835		}
836
837		/**
838	<	* Tries to unsplice the cancelled node held in cleanMe that was
839	<	* previously uncleanable because it was at tail.
838	>	* Tries to unsplice the deleted/cancelled node held in cleanMe
839	>	* that was previously uncleanable because it was at tail.
840		*
841		* @return current cleanMe node (or null)
842		*/
843	<	private Node<E> reclean() {
843	>	private Node reclean() {
844		/*
845	<	* cleanMe is, or at one time was, predecessor of cancelled
846	<	* node s that was the tail so could not be unspliced.  If s
845	>	* cleanMe is, or at one time was, predecessor of a cancelled
846	>	* node s that was the tail so could not be unspliced.  If it
847		* is no longer the tail, try to unsplice if necessary and
848		* make cleanMe slot available.  This differs from similar
849	<	* code in clean() because we must check that pred still
850	<	* points to a cancelled node that must be unspliced -- if
851	<	* not, we can (must) clear cleanMe without unsplicing.
852	<	* This can loop only due to contention on casNext or
398	<	* clearing cleanMe.
849	>	* code in unsplice() because we must check that pred still
850	>	* points to a matched node that can be unspliced -- if not,
851	>	* we can (must) clear cleanMe without unsplicing.  This can
852	>	* loop only due to contention.
853		*/
854	<	Node<E> pred;
855	<	while ((pred = cleanMe.get()) != null) {
856	<	Node<E> t = getValidatedTail();
857	<	Node<E> s = pred.next;
858	<	if (s != t) {
859	<	Node<E> sn;
860	<	if (s == null || s == pred || s.get() != s ||
861	<	(sn = s.next) == s || pred.casNext(s, sn))
862	<	cleanMe.compareAndSet(pred, null);
854	>	Node pred;
855	>	while ((pred = cleanMe) != null) {
856	>	Node s = pred.next;
857	>	Node n;
858	>	if (s == null || s == pred || !s.isMatched())
859	>	casCleanMe(pred, null); // already gone
860	>	else if ((n = s.next) != null) {
861	>	if (n != s)
862	>	pred.casNext(s, n);
863	>	casCleanMe(pred, null);
864		}
865	<	else // s is still tail; cannot clean
865	>	else
866		break;
867		}
868		return pred;
869		}
870
871		/**
872	+	* Main implementation of Iterator.remove(). Find
873	+	* and unsplice the given data node.
874	+	* @param possiblePred possible predecessor of s
875	+	* @param s the node to remove
876	+	*/
877	+	final void findAndRemoveDataNode(Node possiblePred, Node s) {
878	+	assert s.isData;
879	+	if (s.tryMatchData()) {
880	+	if (possiblePred != null && possiblePred.next == s)
881	+	unsplice(possiblePred, s); // was actual predecessor
882	+	else {
883	+	for (Node pred = null, p = head; p != null; ) {
884	+	if (p == s) {
885	+	unsplice(pred, p);
886	+	break;
887	+	}
888	+	if (p.isUnmatchedRequest())
889	+	break;
890	+	pred = p;
891	+	if ((p = p.next) == pred) { // stale
892	+	pred = null;
893	+	p = head;
894	+	}
895	+	}
896	+	}
897	+	}
898	+	}
899	+
900	+	/**
901	+	* Main implementation of remove(Object)
902	+	*/
903	+	private boolean findAndRemove(Object e) {
904	+	if (e != null) {
905	+	for (Node pred = null, p = head; p != null; ) {
906	+	Object item = p.item;
907	+	if (p.isData) {
908	+	if (item != null && item != p && e.equals(item) &&
909	+	p.tryMatchData()) {
910	+	unsplice(pred, p);
911	+	return true;
912	+	}
913	+	}
914	+	else if (item == null)
915	+	break;
916	+	pred = p;
917	+	if ((p = p.next) == pred) { // stale
918	+	pred = null;
919	+	p = head;
920	+	}
921	+	}
922	+	}
923	+	return false;
924	+	}
925	+
926	+
927	+	/**
928		* Creates an initially empty {@code LinkedTransferQueue}.
929		*/
930		public LinkedTransferQueue() {
420	–	Node<E> dummy = new Node<E>(null, false);
421	–	head = new PaddedAtomicReference<Node<E>>(dummy);
422	–	tail = new PaddedAtomicReference<Node<E>>(dummy);
423	–	cleanMe = new PaddedAtomicReference<Node<E>>(null);
931		}
932
933		/**
#	Line 437 | Line 944 | public class LinkedTransferQueue<E> exte
944		addAll(c);
945		}
946
947	<	public void put(E e) throws InterruptedException {
948	<	if (e == null) throw new NullPointerException();
949	<	if (Thread.interrupted()) throw new InterruptedException();
950	<	xfer(e, NOWAIT, 0);
947	>	/**
948	>	* Inserts the specified element at the tail of this queue.
949	>	* As the queue is unbounded, this method will never block.
950	>	*
951	>	* @throws NullPointerException if the specified element is null
952	>	*/
953	>	public void put(E e) {
954	>	xfer(e, true, ASYNC, 0);
955		}
956
957	<	public boolean offer(E e, long timeout, TimeUnit unit)
958	<	throws InterruptedException {
959	<	if (e == null) throw new NullPointerException();
960	<	if (Thread.interrupted()) throw new InterruptedException();
961	<	xfer(e, NOWAIT, 0);
957	>	/**
958	>	* Inserts the specified element at the tail of this queue.
959	>	* As the queue is unbounded, this method will never block or
960	>	* return {@code false}.
961	>	*
962	>	* @return {@code true} (as specified by
963	>	*  {@link BlockingQueue#offer(Object,long,TimeUnit) BlockingQueue.offer})
964	>	* @throws NullPointerException if the specified element is null
965	>	*/
966	>	public boolean offer(E e, long timeout, TimeUnit unit) {
967	>	xfer(e, true, ASYNC, 0);
968		return true;
969		}
970
971	+	/**
972	+	* Inserts the specified element at the tail of this queue.
973	+	* As the queue is unbounded, this method will never return {@code false}.
974	+	*
975	+	* @return {@code true} (as specified by
976	+	*         {@link BlockingQueue#offer(Object) BlockingQueue.offer})
977	+	* @throws NullPointerException if the specified element is null
978	+	*/
979		public boolean offer(E e) {
980	<	if (e == null) throw new NullPointerException();
456	<	xfer(e, NOWAIT, 0);
980	>	xfer(e, true, ASYNC, 0);
981		return true;
982		}
983
984	+	/**
985	+	* Inserts the specified element at the tail of this queue.
986	+	* As the queue is unbounded, this method will never throw
987	+	* {@link IllegalStateException} or return {@code false}.
988	+	*
989	+	* @return {@code true} (as specified by {@link Collection#add})
990	+	* @throws NullPointerException if the specified element is null
991	+	*/
992		public boolean add(E e) {
993	<	if (e == null) throw new NullPointerException();
462	<	xfer(e, NOWAIT, 0);
993	>	xfer(e, true, ASYNC, 0);
994		return true;
995		}
996
997	+	/**
998	+	* Transfers the element to a waiting consumer immediately, if possible.
999	+	*
1000	+	* <p>More precisely, transfers the specified element immediately
1001	+	* if there exists a consumer already waiting to receive it (in
1002	+	* {@link #take} or timed {@link #poll(long,TimeUnit) poll}),
1003	+	* otherwise returning {@code false} without enqueuing the element.
1004	+	*
1005	+	* @throws NullPointerException if the specified element is null
1006	+	*/
1007	+	public boolean tryTransfer(E e) {
1008	+	return xfer(e, true, NOW, 0) == null;
1009	+	}
1010	+
1011	+	/**
1012	+	* Transfers the element to a consumer, waiting if necessary to do so.
1013	+	*
1014	+	* <p>More precisely, transfers the specified element immediately
1015	+	* if there exists a consumer already waiting to receive it (in
1016	+	* {@link #take} or timed {@link #poll(long,TimeUnit) poll}),
1017	+	* else inserts the specified element at the tail of this queue
1018	+	* and waits until the element is received by a consumer.
1019	+	*
1020	+	* @throws NullPointerException if the specified element is null
1021	+	*/
1022		public void transfer(E e) throws InterruptedException {
1023	<	if (e == null) throw new NullPointerException();
1024	<	if (xfer(e, WAIT, 0) == null) {
469	<	Thread.interrupted();
1023	>	if (xfer(e, true, SYNC, 0) != null) {
1024	>	Thread.interrupted(); // failure possible only due to interrupt
1025		throw new InterruptedException();
1026		}
1027		}
1028
1029	+	/**
1030	+	* Transfers the element to a consumer if it is possible to do so
1031	+	* before the timeout elapses.
1032	+	*
1033	+	* <p>More precisely, transfers the specified element immediately
1034	+	* if there exists a consumer already waiting to receive it (in
1035	+	* {@link #take} or timed {@link #poll(long,TimeUnit) poll}),
1036	+	* else inserts the specified element at the tail of this queue
1037	+	* and waits until the element is received by a consumer,
1038	+	* returning {@code false} if the specified wait time elapses
1039	+	* before the element can be transferred.
1040	+	*
1041	+	* @throws NullPointerException if the specified element is null
1042	+	*/
1043		public boolean tryTransfer(E e, long timeout, TimeUnit unit)
1044		throws InterruptedException {
1045	<	if (e == null) throw new NullPointerException();
477	<	if (xfer(e, TIMEOUT, unit.toNanos(timeout)) != null)
1045	>	if (xfer(e, true, TIMEOUT, unit.toNanos(timeout)) == null)
1046		return true;
1047		if (!Thread.interrupted())
1048		return false;
1049		throw new InterruptedException();
1050		}
1051
484	–	public boolean tryTransfer(E e) {
485	–	if (e == null) throw new NullPointerException();
486	–	return fulfill(e) != null;
487	–	}
488	–
1052		public E take() throws InterruptedException {
1053	<	Object e = xfer(null, WAIT, 0);
1053	>	E e = xfer(null, false, SYNC, 0);
1054		if (e != null)
1055	<	return (E) e;
1055	>	return e;
1056		Thread.interrupted();
1057		throw new InterruptedException();
1058		}
1059
1060		public E poll(long timeout, TimeUnit unit) throws InterruptedException {
1061	<	Object e = xfer(null, TIMEOUT, unit.toNanos(timeout));
1061	>	E e = xfer(null, false, TIMEOUT, unit.toNanos(timeout));
1062		if (e != null || !Thread.interrupted())
1063	<	return (E) e;
1063	>	return e;
1064		throw new InterruptedException();
1065		}
1066
1067		public E poll() {
1068	<	return fulfill(null);
1068	>	return xfer(null, false, NOW, 0);
1069		}
1070
1071	+	/**
1072	+	* @throws NullPointerException     {@inheritDoc}
1073	+	* @throws IllegalArgumentException {@inheritDoc}
1074	+	*/
1075		public int drainTo(Collection<? super E> c) {
1076		if (c == null)
1077		throw new NullPointerException();
#	Line 519 | Line 1086 | public class LinkedTransferQueue<E> exte
1086		return n;
1087		}
1088
1089	+	/**
1090	+	* @throws NullPointerException     {@inheritDoc}
1091	+	* @throws IllegalArgumentException {@inheritDoc}
1092	+	*/
1093		public int drainTo(Collection<? super E> c, int maxElements) {
1094		if (c == null)
1095		throw new NullPointerException();
#	Line 533 | Line 1104 | public class LinkedTransferQueue<E> exte
1104		return n;
1105		}
1106
536	–	// Traversal-based methods
537	–
1107		/**
1108	<	* Returns head after performing any outstanding helping steps.
1108	>	* Returns an iterator over the elements in this queue in proper
1109	>	* sequence, from head to tail.
1110	>	*
1111	>	* <p>The returned iterator is a "weakly consistent" iterator that
1112	>	* will never throw
1113	>	* {@link ConcurrentModificationException ConcurrentModificationException},
1114	>	* and guarantees to traverse elements as they existed upon
1115	>	* construction of the iterator, and may (but is not guaranteed
1116	>	* to) reflect any modifications subsequent to construction.
1117	>	*
1118	>	* @return an iterator over the elements in this queue in proper sequence
1119		*/
541	–	private Node<E> traversalHead() {
542	–	for (;;) {
543	–	Node<E> t = tail.get();
544	–	Node<E> h = head.get();
545	–	if (h != null && t != null) {
546	–	Node<E> last = t.next;
547	–	Node<E> first = h.next;
548	–	if (t == tail.get()) {
549	–	if (last != null)
550	–	tail.compareAndSet(t, last);
551	–	else if (first != null) {
552	–	Object x = first.get();
553	–	if (x == first)
554	–	advanceHead(h, first);
555	–	else
556	–	return h;
557	–	}
558	–	else
559	–	return h;
560	–	}
561	–	}
562	–	reclean();
563	–	}
564	–	}
565	–
566	–
1120		public Iterator<E> iterator() {
1121		return new Itr();
1122		}
1123
571	–	/**
572	–	* Iterators. Basic strategy is to traverse list, treating
573	–	* non-data (i.e., request) nodes as terminating list.
574	–	* Once a valid data node is found, the item is cached
575	–	* so that the next call to next() will return it even
576	–	* if subsequently removed.
577	–	*/
578	–	class Itr implements Iterator<E> {
579	–	Node<E> next;        // node to return next
580	–	Node<E> pnext;       // predecessor of next
581	–	Node<E> snext;       // successor of next
582	–	Node<E> curr;        // last returned node, for remove()
583	–	Node<E> pcurr;       // predecessor of curr, for remove()
584	–	E nextItem;        // Cache of next item, once committed to in next
585	–
586	–	Itr() {
587	–	findNext();
588	–	}
589	–
590	–	/**
591	–	* Ensures next points to next valid node, or null if none.
592	–	*/
593	–	void findNext() {
594	–	for (;;) {
595	–	Node<E> pred = pnext;
596	–	Node<E> q = next;
597	–	if (pred == null || pred == q) {
598	–	pred = traversalHead();
599	–	q = pred.next;
600	–	}
601	–	if (q == null || !q.isData) {
602	–	next = null;
603	–	return;
604	–	}
605	–	Object x = q.get();
606	–	Node<E> s = q.next;
607	–	if (x != null && q != x && q != s) {
608	–	nextItem = (E) x;
609	–	snext = s;
610	–	pnext = pred;
611	–	next = q;
612	–	return;
613	–	}
614	–	pnext = q;
615	–	next = s;
616	–	}
617	–	}
618	–
619	–	public boolean hasNext() {
620	–	return next != null;
621	–	}
622	–
623	–	public E next() {
624	–	if (next == null) throw new NoSuchElementException();
625	–	pcurr = pnext;
626	–	curr = next;
627	–	pnext = next;
628	–	next = snext;
629	–	E x = nextItem;
630	–	findNext();
631	–	return x;
632	–	}
633	–
634	–	public void remove() {
635	–	Node<E> p = curr;
636	–	if (p == null)
637	–	throw new IllegalStateException();
638	–	Object x = p.get();
639	–	if (x != null && x != p && p.compareAndSet(x, p))
640	–	clean(pcurr, p);
641	–	}
642	–	}
643	–
1124		public E peek() {
1125	<	for (;;) {
646	<	Node<E> h = traversalHead();
647	<	Node<E> p = h.next;
648	<	if (p == null)
649	<	return null;
650	<	Object x = p.get();
651	<	if (p != x) {
652	<	if (!p.isData)
653	<	return null;
654	<	if (x != null)
655	<	return (E) x;
656	<	}
657	<	}
1125	>	return firstDataItem();
1126		}
1127
1128	+	/**
1129	+	* Returns {@code true} if this queue contains no elements.
1130	+	*
1131	+	* @return {@code true} if this queue contains no elements
1132	+	*/
1133		public boolean isEmpty() {
1134	<	for (;;) {
662	<	Node<E> h = traversalHead();
663	<	Node<E> p = h.next;
664	<	if (p == null)
665	<	return true;
666	<	Object x = p.get();
667	<	if (p != x) {
668	<	if (!p.isData)
669	<	return true;
670	<	if (x != null)
671	<	return false;
672	<	}
673	<	}
1134	>	return firstOfMode(true) == null;
1135		}
1136
1137		public boolean hasWaitingConsumer() {
1138	<	for (;;) {
678	<	Node<E> h = traversalHead();
679	<	Node<E> p = h.next;
680	<	if (p == null)
681	<	return false;
682	<	Object x = p.get();
683	<	if (p != x)
684	<	return !p.isData;
685	<	}
1138	>	return firstOfMode(false) != null;
1139		}
1140
1141		/**
#	Line 698 | Line 1151 | public class LinkedTransferQueue<E> exte
1151		* @return the number of elements in this queue
1152		*/
1153		public int size() {
1154	<	int count = 0;
702	<	Node<E> h = traversalHead();
703	<	for (Node<E> p = h.next; p != null && p.isData; p = p.next) {
704	<	Object x = p.get();
705	<	if (x != null && x != p) {
706	<	if (++count == Integer.MAX_VALUE) // saturated
707	<	break;
708	<	}
709	<	}
710	<	return count;
1154	>	return countOfMode(true);
1155		}
1156
1157		public int getWaitingConsumerCount() {
1158	<	int count = 0;
715	<	Node<E> h = traversalHead();
716	<	for (Node<E> p = h.next; p != null && !p.isData; p = p.next) {
717	<	if (p.get() == null) {
718	<	if (++count == Integer.MAX_VALUE)
719	<	break;
720	<	}
721	<	}
722	<	return count;
1158	>	return countOfMode(false);
1159		}
1160
1161	<	public int remainingCapacity() {
1162	<	return Integer.MAX_VALUE;
1161	>	/**
1162	>	* Removes a single instance of the specified element from this queue,
1163	>	* if it is present.  More formally, removes an element {@code e} such
1164	>	* that {@code o.equals(e)}, if this queue contains one or more such
1165	>	* elements.
1166	>	* Returns {@code true} if this queue contained the specified element
1167	>	* (or equivalently, if this queue changed as a result of the call).
1168	>	*
1169	>	* @param o element to be removed from this queue, if present
1170	>	* @return {@code true} if this queue changed as a result of the call
1171	>	*/
1172	>	public boolean remove(Object o) {
1173	>	return findAndRemove(o);
1174		}
1175
1176	<	public boolean remove(Object o) {
1177	<	if (o == null)
1178	<	return false;
1179	<	for (;;) {
1180	<	Node<E> pred = traversalHead();
1181	<	for (;;) {
1182	<	Node<E> q = pred.next;
1183	<	if (q == null || !q.isData)
1184	<	return false;
738	<	if (q == pred) // restart
739	<	break;
740	<	Object x = q.get();
741	<	if (x != null && x != q && o.equals(x) &&
742	<	q.compareAndSet(x, q)) {
743	<	clean(pred, q);
744	<	return true;
745	<	}
746	<	pred = q;
747	<	}
748	<	}
1176	>	/**
1177	>	* Always returns {@code Integer.MAX_VALUE} because a
1178	>	* {@code LinkedTransferQueue} is not capacity constrained.
1179	>	*
1180	>	* @return {@code Integer.MAX_VALUE} (as specified by
1181	>	*         {@link BlockingQueue#remainingCapacity()})
1182	>	*/
1183	>	public int remainingCapacity() {
1184	>	return Integer.MAX_VALUE;
1185		}
1186
1187		/**
1188	<	* Save the state to a stream (that is, serialize it).
1188	>	* Saves the state to a stream (that is, serializes it).
1189		*
1190		* @serialData All of the elements (each an {@code E}) in
1191		* the proper order, followed by a null
#	Line 765 | Line 1201 | public class LinkedTransferQueue<E> exte
1201		}
1202
1203		/**
1204	<	* Reconstitute the Queue instance from a stream (that is,
1205	<	* deserialize it).
1204	>	* Reconstitutes the Queue instance from a stream (that is,
1205	>	* deserializes it).
1206		*
1207		* @param s the stream
1208		*/
1209		private void readObject(java.io.ObjectInputStream s)
1210		throws java.io.IOException, ClassNotFoundException {
1211		s.defaultReadObject();
776	–	resetHeadAndTail();
1212		for (;;) {
1213		@SuppressWarnings("unchecked") E item = (E) s.readObject();
1214		if (item == null)
#	Line 783 | Line 1218 | public class LinkedTransferQueue<E> exte
1218		}
1219		}
1220
1221	<	// Support for resetting head/tail while deserializing
1222	<	private void resetHeadAndTail() {
1223	<	Node<E> dummy = new Node<E>(null, false);
1224	<	UNSAFE.putObjectVolatile(this, headOffset,
1225	<	new PaddedAtomicReference<Node<E>>(dummy));
1226	<	UNSAFE.putObjectVolatile(this, tailOffset,
1227	<	new PaddedAtomicReference<Node<E>>(dummy));
1228	<	UNSAFE.putObjectVolatile(this, cleanMeOffset,
1229	<	new PaddedAtomicReference<Node<E>>(null));
1221	>	// Unsafe mechanics
1222	>
1223	>	private static final sun.misc.Unsafe UNSAFE = getUnsafe();
1224	>	private static final long headOffset =
1225	>	objectFieldOffset(UNSAFE, "head", LinkedTransferQueue.class);
1226	>	private static final long tailOffset =
1227	>	objectFieldOffset(UNSAFE, "tail", LinkedTransferQueue.class);
1228	>	private static final long cleanMeOffset =
1229	>	objectFieldOffset(UNSAFE, "cleanMe", LinkedTransferQueue.class);
1230	>
1231	>	static long objectFieldOffset(sun.misc.Unsafe UNSAFE,
1232	>	String field, Class<?> klazz) {
1233	>	try {
1234	>	return UNSAFE.objectFieldOffset(klazz.getDeclaredField(field));
1235	>	} catch (NoSuchFieldException e) {
1236	>	// Convert Exception to corresponding Error
1237	>	NoSuchFieldError error = new NoSuchFieldError(field);
1238	>	error.initCause(e);
1239	>	throw error;
1240	>	}
1241		}
1242
1243	<	// Unsafe mechanics for jsr166y 3rd party package.
1244	<	private static sun.misc.Unsafe getUnsafe() {
1243	>	/**
1244	>	* Returns a sun.misc.Unsafe.  Suitable for use in a 3rd party package.
1245	>	* Replace with a simple call to Unsafe.getUnsafe when integrating
1246	>	* into a jdk.
1247	>	*
1248	>	* @return a sun.misc.Unsafe
1249	>	*/
1250	>	static sun.misc.Unsafe getUnsafe() {
1251		try {
1252		return sun.misc.Unsafe.getUnsafe();
1253		} catch (SecurityException se) {
1254		try {
1255		return java.security.AccessController.doPrivileged
1256	<	(new java.security.PrivilegedExceptionAction<sun.misc.Unsafe>() {
1256	>	(new java.security
1257	>	.PrivilegedExceptionAction<sun.misc.Unsafe>() {
1258		public sun.misc.Unsafe run() throws Exception {
1259	<	return getUnsafeByReflection();
1259	>	java.lang.reflect.Field f = sun.misc
1260	>	.Unsafe.class.getDeclaredField("theUnsafe");
1261	>	f.setAccessible(true);
1262	>	return (sun.misc.Unsafe) f.get(null);
1263		}});
1264		} catch (java.security.PrivilegedActionException e) {
1265		throw new RuntimeException("Could not initialize intrinsics",
#	Line 812 | Line 1268 | public class LinkedTransferQueue<E> exte
1268		}
1269		}
1270
815	–	private static sun.misc.Unsafe getUnsafeByReflection()
816	–	throws NoSuchFieldException, IllegalAccessException {
817	–	java.lang.reflect.Field f =
818	–	sun.misc.Unsafe.class.getDeclaredField("theUnsafe");
819	–	f.setAccessible(true);
820	–	return (sun.misc.Unsafe) f.get(null);
821	–	}
822	–
823	–	private static long fieldOffset(String fieldName, Class<?> klazz) {
824	–	try {
825	–	return UNSAFE.objectFieldOffset(klazz.getDeclaredField(fieldName));
826	–	} catch (NoSuchFieldException e) {
827	–	// Convert Exception to Error
828	–	NoSuchFieldError error = new NoSuchFieldError(fieldName);
829	–	error.initCause(e);
830	–	throw error;
831	–	}
832	–	}
833	–
834	–	private static final sun.misc.Unsafe UNSAFE = getUnsafe();
835	–	static final long headOffset =
836	–	fieldOffset("head", LinkedTransferQueue.class);
837	–	static final long tailOffset =
838	–	fieldOffset("tail", LinkedTransferQueue.class);
839	–	static final long cleanMeOffset =
840	–	fieldOffset("cleanMe", LinkedTransferQueue.class);
841	–
1271		}

Diff Legend

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