[cvs] / jsr166 / src / main / java / util / concurrent / LinkedTransferQueue.java Repository:
ViewVC logotype

Annotation of /jsr166/src/main/java/util/concurrent/LinkedTransferQueue.java

Parent Directory Parent Directory | Revision Log Revision Log


Revision 1.68 - (view) (download)

1 : jsr166 1.1 /*
2 :     * Written by Doug Lea with assistance from members of JCP JSR-166
3 :     * Expert Group and released to the public domain, as explained at
4 : jsr166 1.39 * http://creativecommons.org/publicdomain/zero/1.0/
5 : jsr166 1.1 */
6 :    
7 :     package java.util.concurrent;
8 :    
9 :     import java.util.AbstractQueue;
10 : dl 1.52 import java.util.Arrays;
11 : jsr166 1.1 import java.util.Collection;
12 : dl 1.52 import java.util.Collections;
13 : jsr166 1.1 import java.util.Iterator;
14 :     import java.util.NoSuchElementException;
15 : jsr166 1.5 import java.util.Queue;
16 : dl 1.33 import java.util.concurrent.TimeUnit;
17 : jsr166 1.1 import java.util.concurrent.locks.LockSupport;
18 : dl 1.52 import java.util.Spliterator;
19 : dl 1.54 import java.util.Spliterators;
20 : dl 1.52 import java.util.stream.Stream;
21 :     import java.util.function.Consumer;
22 : dl 1.22
23 : jsr166 1.1 /**
24 : jsr166 1.6 * An unbounded {@link TransferQueue} based on linked nodes.
25 : jsr166 1.1 * This queue orders elements FIFO (first-in-first-out) with respect
26 :     * to any given producer. The <em>head</em> of the queue is that
27 :     * element that has been on the queue the longest time for some
28 :     * producer. The <em>tail</em> of the queue is that element that has
29 :     * been on the queue the shortest time for some producer.
30 :     *
31 : dl 1.40 * <p>Beware that, unlike in most collections, the {@code size} method
32 :     * is <em>NOT</em> a constant-time operation. Because of the
33 : jsr166 1.1 * asynchronous nature of these queues, determining the current number
34 : dl 1.40 * of elements requires a traversal of the elements, and so may report
35 :     * inaccurate results if this collection is modified during traversal.
36 : dl 1.41 * Additionally, the bulk operations {@code addAll},
37 :     * {@code removeAll}, {@code retainAll}, {@code containsAll},
38 :     * {@code equals}, and {@code toArray} are <em>not</em> guaranteed
39 : dl 1.40 * to be performed atomically. For example, an iterator operating
40 : dl 1.41 * concurrently with an {@code addAll} operation might view only some
41 : dl 1.40 * of the added elements.
42 : jsr166 1.1 *
43 :     * <p>This class and its iterator implement all of the
44 :     * <em>optional</em> methods of the {@link Collection} and {@link
45 :     * Iterator} interfaces.
46 :     *
47 :     * <p>Memory consistency effects: As with other concurrent
48 :     * collections, actions in a thread prior to placing an object into a
49 :     * {@code LinkedTransferQueue}
50 :     * <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a>
51 :     * actions subsequent to the access or removal of that element from
52 :     * the {@code LinkedTransferQueue} in another thread.
53 :     *
54 :     * <p>This class is a member of the
55 :     * <a href="{@docRoot}/../technotes/guides/collections/index.html">
56 :     * Java Collections Framework</a>.
57 :     *
58 :     * @since 1.7
59 :     * @author Doug Lea
60 :     * @param <E> the type of elements held in this collection
61 :     */
62 :     public class LinkedTransferQueue<E> extends AbstractQueue<E>
63 :     implements TransferQueue<E>, java.io.Serializable {
64 :     private static final long serialVersionUID = -3223113410248163686L;
65 :    
66 :     /*
67 : jsr166 1.8 * *** Overview of Dual Queues with Slack ***
68 : jsr166 1.1 *
69 : jsr166 1.8 * Dual Queues, introduced by Scherer and Scott
70 :     * (http://www.cs.rice.edu/~wns1/papers/2004-DISC-DDS.pdf) are
71 :     * (linked) queues in which nodes may represent either data or
72 :     * requests. When a thread tries to enqueue a data node, but
73 :     * encounters a request node, it instead "matches" and removes it;
74 :     * and vice versa for enqueuing requests. Blocking Dual Queues
75 :     * arrange that threads enqueuing unmatched requests block until
76 :     * other threads provide the match. Dual Synchronous Queues (see
77 :     * Scherer, Lea, & Scott
78 :     * http://www.cs.rochester.edu/u/scott/papers/2009_Scherer_CACM_SSQ.pdf)
79 :     * additionally arrange that threads enqueuing unmatched data also
80 :     * block. Dual Transfer Queues support all of these modes, as
81 :     * dictated by callers.
82 :     *
83 :     * A FIFO dual queue may be implemented using a variation of the
84 :     * Michael & Scott (M&S) lock-free queue algorithm
85 :     * (http://www.cs.rochester.edu/u/scott/papers/1996_PODC_queues.pdf).
86 :     * It maintains two pointer fields, "head", pointing to a
87 :     * (matched) node that in turn points to the first actual
88 :     * (unmatched) queue node (or null if empty); and "tail" that
89 :     * points to the last node on the queue (or again null if
90 :     * empty). For example, here is a possible queue with four data
91 :     * elements:
92 :     *
93 :     * head tail
94 :     * | |
95 :     * v v
96 :     * M -> U -> U -> U -> U
97 :     *
98 :     * The M&S queue algorithm is known to be prone to scalability and
99 :     * overhead limitations when maintaining (via CAS) these head and
100 :     * tail pointers. This has led to the development of
101 :     * contention-reducing variants such as elimination arrays (see
102 :     * Moir et al http://portal.acm.org/citation.cfm?id=1074013) and
103 :     * optimistic back pointers (see Ladan-Mozes & Shavit
104 :     * http://people.csail.mit.edu/edya/publications/OptimisticFIFOQueue-journal.pdf).
105 :     * However, the nature of dual queues enables a simpler tactic for
106 :     * improving M&S-style implementations when dual-ness is needed.
107 :     *
108 :     * In a dual queue, each node must atomically maintain its match
109 :     * status. While there are other possible variants, we implement
110 :     * this here as: for a data-mode node, matching entails CASing an
111 :     * "item" field from a non-null data value to null upon match, and
112 :     * vice-versa for request nodes, CASing from null to a data
113 :     * value. (Note that the linearization properties of this style of
114 :     * queue are easy to verify -- elements are made available by
115 :     * linking, and unavailable by matching.) Compared to plain M&S
116 :     * queues, this property of dual queues requires one additional
117 :     * successful atomic operation per enq/deq pair. But it also
118 :     * enables lower cost variants of queue maintenance mechanics. (A
119 :     * variation of this idea applies even for non-dual queues that
120 :     * support deletion of interior elements, such as
121 :     * j.u.c.ConcurrentLinkedQueue.)
122 :     *
123 :     * Once a node is matched, its match status can never again
124 :     * change. We may thus arrange that the linked list of them
125 :     * contain a prefix of zero or more matched nodes, followed by a
126 :     * suffix of zero or more unmatched nodes. (Note that we allow
127 :     * both the prefix and suffix to be zero length, which in turn
128 :     * means that we do not use a dummy header.) If we were not
129 :     * concerned with either time or space efficiency, we could
130 :     * correctly perform enqueue and dequeue operations by traversing
131 :     * from a pointer to the initial node; CASing the item of the
132 :     * first unmatched node on match and CASing the next field of the
133 :     * trailing node on appends. (Plus some special-casing when
134 :     * initially empty). While this would be a terrible idea in
135 :     * itself, it does have the benefit of not requiring ANY atomic
136 :     * updates on head/tail fields.
137 :     *
138 :     * We introduce here an approach that lies between the extremes of
139 :     * never versus always updating queue (head and tail) pointers.
140 :     * This offers a tradeoff between sometimes requiring extra
141 :     * traversal steps to locate the first and/or last unmatched
142 :     * nodes, versus the reduced overhead and contention of fewer
143 :     * updates to queue pointers. For example, a possible snapshot of
144 :     * a queue is:
145 :     *
146 :     * head tail
147 :     * | |
148 :     * v v
149 :     * M -> M -> U -> U -> U -> U
150 :     *
151 :     * The best value for this "slack" (the targeted maximum distance
152 :     * between the value of "head" and the first unmatched node, and
153 :     * similarly for "tail") is an empirical matter. We have found
154 :     * that using very small constants in the range of 1-3 work best
155 :     * over a range of platforms. Larger values introduce increasing
156 :     * costs of cache misses and risks of long traversal chains, while
157 :     * smaller values increase CAS contention and overhead.
158 :     *
159 :     * Dual queues with slack differ from plain M&S dual queues by
160 :     * virtue of only sometimes updating head or tail pointers when
161 :     * matching, appending, or even traversing nodes; in order to
162 :     * maintain a targeted slack. The idea of "sometimes" may be
163 :     * operationalized in several ways. The simplest is to use a
164 :     * per-operation counter incremented on each traversal step, and
165 :     * to try (via CAS) to update the associated queue pointer
166 :     * whenever the count exceeds a threshold. Another, that requires
167 :     * more overhead, is to use random number generators to update
168 :     * with a given probability per traversal step.
169 :     *
170 :     * In any strategy along these lines, because CASes updating
171 :     * fields may fail, the actual slack may exceed targeted
172 :     * slack. However, they may be retried at any time to maintain
173 :     * targets. Even when using very small slack values, this
174 :     * approach works well for dual queues because it allows all
175 :     * operations up to the point of matching or appending an item
176 :     * (hence potentially allowing progress by another thread) to be
177 :     * read-only, thus not introducing any further contention. As
178 :     * described below, we implement this by performing slack
179 :     * maintenance retries only after these points.
180 :     *
181 :     * As an accompaniment to such techniques, traversal overhead can
182 :     * be further reduced without increasing contention of head
183 :     * pointer updates: Threads may sometimes shortcut the "next" link
184 :     * path from the current "head" node to be closer to the currently
185 :     * known first unmatched node, and similarly for tail. Again, this
186 :     * may be triggered with using thresholds or randomization.
187 :     *
188 :     * These ideas must be further extended to avoid unbounded amounts
189 :     * of costly-to-reclaim garbage caused by the sequential "next"
190 :     * links of nodes starting at old forgotten head nodes: As first
191 :     * described in detail by Boehm
192 :     * (http://portal.acm.org/citation.cfm?doid=503272.503282) if a GC
193 :     * delays noticing that any arbitrarily old node has become
194 :     * garbage, all newer dead nodes will also be unreclaimed.
195 :     * (Similar issues arise in non-GC environments.) To cope with
196 :     * this in our implementation, upon CASing to advance the head
197 :     * pointer, we set the "next" link of the previous head to point
198 :     * only to itself; thus limiting the length of connected dead lists.
199 :     * (We also take similar care to wipe out possibly garbage
200 :     * retaining values held in other Node fields.) However, doing so
201 :     * adds some further complexity to traversal: If any "next"
202 :     * pointer links to itself, it indicates that the current thread
203 :     * has lagged behind a head-update, and so the traversal must
204 :     * continue from the "head". Traversals trying to find the
205 :     * current tail starting from "tail" may also encounter
206 :     * self-links, in which case they also continue at "head".
207 :     *
208 :     * It is tempting in slack-based scheme to not even use CAS for
209 :     * updates (similarly to Ladan-Mozes & Shavit). However, this
210 :     * cannot be done for head updates under the above link-forgetting
211 :     * mechanics because an update may leave head at a detached node.
212 :     * And while direct writes are possible for tail updates, they
213 :     * increase the risk of long retraversals, and hence long garbage
214 :     * chains, which can be much more costly than is worthwhile
215 :     * considering that the cost difference of performing a CAS vs
216 :     * write is smaller when they are not triggered on each operation
217 :     * (especially considering that writes and CASes equally require
218 :     * additional GC bookkeeping ("write barriers") that are sometimes
219 :     * more costly than the writes themselves because of contention).
220 :     *
221 :     * *** Overview of implementation ***
222 :     *
223 :     * We use a threshold-based approach to updates, with a slack
224 :     * threshold of two -- that is, we update head/tail when the
225 :     * current pointer appears to be two or more steps away from the
226 :     * first/last node. The slack value is hard-wired: a path greater
227 :     * than one is naturally implemented by checking equality of
228 :     * traversal pointers except when the list has only one element,
229 :     * in which case we keep slack threshold at one. Avoiding tracking
230 :     * explicit counts across method calls slightly simplifies an
231 :     * already-messy implementation. Using randomization would
232 :     * probably work better if there were a low-quality dirt-cheap
233 :     * per-thread one available, but even ThreadLocalRandom is too
234 :     * heavy for these purposes.
235 :     *
236 : dl 1.16 * With such a small slack threshold value, it is not worthwhile
237 :     * to augment this with path short-circuiting (i.e., unsplicing
238 :     * interior nodes) except in the case of cancellation/removal (see
239 :     * below).
240 : jsr166 1.8 *
241 :     * We allow both the head and tail fields to be null before any
242 :     * nodes are enqueued; initializing upon first append. This
243 :     * simplifies some other logic, as well as providing more
244 :     * efficient explicit control paths instead of letting JVMs insert
245 :     * implicit NullPointerExceptions when they are null. While not
246 :     * currently fully implemented, we also leave open the possibility
247 :     * of re-nulling these fields when empty (which is complicated to
248 :     * arrange, for little benefit.)
249 :     *
250 :     * All enqueue/dequeue operations are handled by the single method
251 :     * "xfer" with parameters indicating whether to act as some form
252 :     * of offer, put, poll, take, or transfer (each possibly with
253 :     * timeout). The relative complexity of using one monolithic
254 :     * method outweighs the code bulk and maintenance problems of
255 :     * using separate methods for each case.
256 :     *
257 :     * Operation consists of up to three phases. The first is
258 :     * implemented within method xfer, the second in tryAppend, and
259 :     * the third in method awaitMatch.
260 :     *
261 :     * 1. Try to match an existing node
262 :     *
263 :     * Starting at head, skip already-matched nodes until finding
264 :     * an unmatched node of opposite mode, if one exists, in which
265 :     * case matching it and returning, also if necessary updating
266 :     * head to one past the matched node (or the node itself if the
267 :     * list has no other unmatched nodes). If the CAS misses, then
268 :     * a loop retries advancing head by two steps until either
269 :     * success or the slack is at most two. By requiring that each
270 :     * attempt advances head by two (if applicable), we ensure that
271 :     * the slack does not grow without bound. Traversals also check
272 :     * if the initial head is now off-list, in which case they
273 :     * start at the new head.
274 :     *
275 :     * If no candidates are found and the call was untimed
276 :     * poll/offer, (argument "how" is NOW) return.
277 :     *
278 :     * 2. Try to append a new node (method tryAppend)
279 :     *
280 :     * Starting at current tail pointer, find the actual last node
281 :     * and try to append a new node (or if head was null, establish
282 :     * the first node). Nodes can be appended only if their
283 :     * predecessors are either already matched or are of the same
284 :     * mode. If we detect otherwise, then a new node with opposite
285 :     * mode must have been appended during traversal, so we must
286 :     * restart at phase 1. The traversal and update steps are
287 :     * otherwise similar to phase 1: Retrying upon CAS misses and
288 :     * checking for staleness. In particular, if a self-link is
289 :     * encountered, then we can safely jump to a node on the list
290 :     * by continuing the traversal at current head.
291 :     *
292 :     * On successful append, if the call was ASYNC, return.
293 :     *
294 :     * 3. Await match or cancellation (method awaitMatch)
295 :     *
296 :     * Wait for another thread to match node; instead cancelling if
297 :     * the current thread was interrupted or the wait timed out. On
298 :     * multiprocessors, we use front-of-queue spinning: If a node
299 :     * appears to be the first unmatched node in the queue, it
300 :     * spins a bit before blocking. In either case, before blocking
301 :     * it tries to unsplice any nodes between the current "head"
302 :     * and the first unmatched node.
303 :     *
304 :     * Front-of-queue spinning vastly improves performance of
305 :     * heavily contended queues. And so long as it is relatively
306 :     * brief and "quiet", spinning does not much impact performance
307 :     * of less-contended queues. During spins threads check their
308 :     * interrupt status and generate a thread-local random number
309 :     * to decide to occasionally perform a Thread.yield. While
310 : jsr166 1.44 * yield has underdefined specs, we assume that it might help,
311 : jsr166 1.45 * and will not hurt, in limiting impact of spinning on busy
312 : jsr166 1.8 * systems. We also use smaller (1/2) spins for nodes that are
313 :     * not known to be front but whose predecessors have not
314 :     * blocked -- these "chained" spins avoid artifacts of
315 :     * front-of-queue rules which otherwise lead to alternating
316 :     * nodes spinning vs blocking. Further, front threads that
317 :     * represent phase changes (from data to request node or vice
318 :     * versa) compared to their predecessors receive additional
319 :     * chained spins, reflecting longer paths typically required to
320 :     * unblock threads during phase changes.
321 : dl 1.16 *
322 :     *
323 :     * ** Unlinking removed interior nodes **
324 :     *
325 :     * In addition to minimizing garbage retention via self-linking
326 :     * described above, we also unlink removed interior nodes. These
327 :     * may arise due to timed out or interrupted waits, or calls to
328 :     * remove(x) or Iterator.remove. Normally, given a node that was
329 :     * at one time known to be the predecessor of some node s that is
330 :     * to be removed, we can unsplice s by CASing the next field of
331 :     * its predecessor if it still points to s (otherwise s must
332 :     * already have been removed or is now offlist). But there are two
333 :     * situations in which we cannot guarantee to make node s
334 :     * unreachable in this way: (1) If s is the trailing node of list
335 :     * (i.e., with null next), then it is pinned as the target node
336 : jsr166 1.23 * for appends, so can only be removed later after other nodes are
337 : dl 1.16 * appended. (2) We cannot necessarily unlink s given a
338 :     * predecessor node that is matched (including the case of being
339 : jsr166 1.17 * cancelled): the predecessor may already be unspliced, in which
340 :     * case some previous reachable node may still point to s.
341 :     * (For further explanation see Herlihy & Shavit "The Art of
342 : dl 1.16 * Multiprocessor Programming" chapter 9). Although, in both
343 :     * cases, we can rule out the need for further action if either s
344 :     * or its predecessor are (or can be made to be) at, or fall off
345 :     * from, the head of list.
346 :     *
347 :     * Without taking these into account, it would be possible for an
348 :     * unbounded number of supposedly removed nodes to remain
349 :     * reachable. Situations leading to such buildup are uncommon but
350 :     * can occur in practice; for example when a series of short timed
351 :     * calls to poll repeatedly time out but never otherwise fall off
352 :     * the list because of an untimed call to take at the front of the
353 :     * queue.
354 :     *
355 :     * When these cases arise, rather than always retraversing the
356 :     * entire list to find an actual predecessor to unlink (which
357 :     * won't help for case (1) anyway), we record a conservative
358 : jsr166 1.24 * estimate of possible unsplice failures (in "sweepVotes").
359 :     * We trigger a full sweep when the estimate exceeds a threshold
360 :     * ("SWEEP_THRESHOLD") indicating the maximum number of estimated
361 :     * removal failures to tolerate before sweeping through, unlinking
362 :     * cancelled nodes that were not unlinked upon initial removal.
363 :     * We perform sweeps by the thread hitting threshold (rather than
364 :     * background threads or by spreading work to other threads)
365 :     * because in the main contexts in which removal occurs, the
366 :     * caller is already timed-out, cancelled, or performing a
367 :     * potentially O(n) operation (e.g. remove(x)), none of which are
368 :     * time-critical enough to warrant the overhead that alternatives
369 :     * would impose on other threads.
370 : dl 1.16 *
371 :     * Because the sweepVotes estimate is conservative, and because
372 :     * nodes become unlinked "naturally" as they fall off the head of
373 :     * the queue, and because we allow votes to accumulate even while
374 : jsr166 1.17 * sweeps are in progress, there are typically significantly fewer
375 : dl 1.16 * such nodes than estimated. Choice of a threshold value
376 :     * balances the likelihood of wasted effort and contention, versus
377 :     * providing a worst-case bound on retention of interior nodes in
378 :     * quiescent queues. The value defined below was chosen
379 :     * empirically to balance these under various timeout scenarios.
380 :     *
381 :     * Note that we cannot self-link unlinked interior nodes during
382 :     * sweeps. However, the associated garbage chains terminate when
383 :     * some successor ultimately falls off the head of the list and is
384 :     * self-linked.
385 : jsr166 1.8 */
386 :    
387 :     /** True if on multiprocessor */
388 :     private static final boolean MP =
389 :     Runtime.getRuntime().availableProcessors() > 1;
390 :    
391 :     /**
392 :     * The number of times to spin (with randomly interspersed calls
393 :     * to Thread.yield) on multiprocessor before blocking when a node
394 :     * is apparently the first waiter in the queue. See above for
395 :     * explanation. Must be a power of two. The value is empirically
396 :     * derived -- it works pretty well across a variety of processors,
397 :     * numbers of CPUs, and OSes.
398 :     */
399 :     private static final int FRONT_SPINS = 1 << 7;
400 :    
401 :     /**
402 :     * The number of times to spin before blocking when a node is
403 :     * preceded by another node that is apparently spinning. Also
404 :     * serves as an increment to FRONT_SPINS on phase changes, and as
405 :     * base average frequency for yielding during spins. Must be a
406 :     * power of two.
407 :     */
408 :     private static final int CHAINED_SPINS = FRONT_SPINS >>> 1;
409 :    
410 :     /**
411 : dl 1.16 * The maximum number of estimated removal failures (sweepVotes)
412 :     * to tolerate before sweeping through the queue unlinking
413 :     * cancelled nodes that were not unlinked upon initial
414 :     * removal. See above for explanation. The value must be at least
415 :     * two to avoid useless sweeps when removing trailing nodes.
416 :     */
417 :     static final int SWEEP_THRESHOLD = 32;
418 :    
419 :     /**
420 : jsr166 1.8 * Queue nodes. Uses Object, not E, for items to allow forgetting
421 :     * them after use. Relies heavily on Unsafe mechanics to minimize
422 : dl 1.16 * unnecessary ordering constraints: Writes that are intrinsically
423 :     * ordered wrt other accesses or CASes use simple relaxed forms.
424 : jsr166 1.8 */
425 : jsr166 1.14 static final class Node {
426 : jsr166 1.8 final boolean isData; // false if this is a request node
427 :     volatile Object item; // initially non-null if isData; CASed to match
428 : jsr166 1.14 volatile Node next;
429 : jsr166 1.8 volatile Thread waiter; // null until waiting
430 :    
431 :     // CAS methods for fields
432 : jsr166 1.14 final boolean casNext(Node cmp, Node val) {
433 : jsr166 1.8 return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
434 :     }
435 : jsr166 1.1
436 : jsr166 1.8 final boolean casItem(Object cmp, Object val) {
437 : dl 1.33 // assert cmp == null || cmp.getClass() != Node.class;
438 : jsr166 1.8 return UNSAFE.compareAndSwapObject(this, itemOffset, cmp, val);
439 :     }
440 : jsr166 1.1
441 : jsr166 1.8 /**
442 : jsr166 1.25 * Constructs a new node. Uses relaxed write because item can
443 :     * only be seen after publication via casNext.
444 : jsr166 1.8 */
445 : jsr166 1.14 Node(Object item, boolean isData) {
446 : jsr166 1.8 UNSAFE.putObject(this, itemOffset, item); // relaxed write
447 :     this.isData = isData;
448 :     }
449 : jsr166 1.1
450 : jsr166 1.8 /**
451 :     * Links node to itself to avoid garbage retention. Called
452 :     * only after CASing head field, so uses relaxed write.
453 :     */
454 :     final void forgetNext() {
455 :     UNSAFE.putObject(this, nextOffset, this);
456 :     }
457 : jsr166 1.1
458 : jsr166 1.8 /**
459 : dl 1.16 * Sets item to self and waiter to null, to avoid garbage
460 :     * retention after matching or cancelling. Uses relaxed writes
461 : dl 1.22 * because order is already constrained in the only calling
462 : dl 1.16 * contexts: item is forgotten only after volatile/atomic
463 :     * mechanics that extract items. Similarly, clearing waiter
464 :     * follows either CAS or return from park (if ever parked;
465 :     * else we don't care).
466 : jsr166 1.8 */
467 :     final void forgetContents() {
468 : dl 1.16 UNSAFE.putObject(this, itemOffset, this);
469 :     UNSAFE.putObject(this, waiterOffset, null);
470 : jsr166 1.8 }
471 : jsr166 1.1
472 : jsr166 1.8 /**
473 :     * Returns true if this node has been matched, including the
474 :     * case of artificial matches due to cancellation.
475 :     */
476 :     final boolean isMatched() {
477 :     Object x = item;
478 : jsr166 1.11 return (x == this) || ((x == null) == isData);
479 :     }
480 :    
481 :     /**
482 :     * Returns true if this is an unmatched request node.
483 :     */
484 :     final boolean isUnmatchedRequest() {
485 :     return !isData && item == null;
486 : jsr166 1.8 }
487 : jsr166 1.1
488 : jsr166 1.8 /**
489 :     * Returns true if a node with the given mode cannot be
490 :     * appended to this node because this node is unmatched and
491 :     * has opposite data mode.
492 :     */
493 :     final boolean cannotPrecede(boolean haveData) {
494 :     boolean d = isData;
495 :     Object x;
496 :     return d != haveData && (x = item) != this && (x != null) == d;
497 :     }
498 : jsr166 1.1
499 : jsr166 1.8 /**
500 :     * Tries to artificially match a data node -- used by remove.
501 :     */
502 :     final boolean tryMatchData() {
503 : dl 1.33 // assert isData;
504 : jsr166 1.8 Object x = item;
505 :     if (x != null && x != this && casItem(x, null)) {
506 :     LockSupport.unpark(waiter);
507 :     return true;
508 :     }
509 :     return false;
510 : jsr166 1.1 }
511 :    
512 : dl 1.38 private static final long serialVersionUID = -3375979862319811754L;
513 :    
514 : jsr166 1.4 // Unsafe mechanics
515 : dl 1.38 private static final sun.misc.Unsafe UNSAFE;
516 :     private static final long itemOffset;
517 :     private static final long nextOffset;
518 :     private static final long waiterOffset;
519 :     static {
520 :     try {
521 :     UNSAFE = sun.misc.Unsafe.getUnsafe();
522 : jsr166 1.43 Class<?> k = Node.class;
523 : dl 1.38 itemOffset = UNSAFE.objectFieldOffset
524 :     (k.getDeclaredField("item"));
525 :     nextOffset = UNSAFE.objectFieldOffset
526 :     (k.getDeclaredField("next"));
527 :     waiterOffset = UNSAFE.objectFieldOffset
528 :     (k.getDeclaredField("waiter"));
529 :     } catch (Exception e) {
530 :     throw new Error(e);
531 :     }
532 :     }
533 : jsr166 1.1 }
534 :    
535 : jsr166 1.8 /** head of the queue; null until first enqueue */
536 : jsr166 1.14 transient volatile Node head;
537 : jsr166 1.8
538 :     /** tail of the queue; null until first append */
539 : jsr166 1.14 private transient volatile Node tail;
540 : jsr166 1.1
541 : dl 1.16 /** The number of apparent failures to unsplice removed nodes */
542 :     private transient volatile int sweepVotes;
543 :    
544 : jsr166 1.8 // CAS methods for fields
545 : jsr166 1.14 private boolean casTail(Node cmp, Node val) {
546 : jsr166 1.8 return UNSAFE.compareAndSwapObject(this, tailOffset, cmp, val);
547 :     }
548 : jsr166 1.1
549 : jsr166 1.14 private boolean casHead(Node cmp, Node val) {
550 : jsr166 1.8 return UNSAFE.compareAndSwapObject(this, headOffset, cmp, val);
551 :     }
552 : jsr166 1.1
553 : dl 1.16 private boolean casSweepVotes(int cmp, int val) {
554 :     return UNSAFE.compareAndSwapInt(this, sweepVotesOffset, cmp, val);
555 : jsr166 1.8 }
556 : jsr166 1.1
557 : jsr166 1.8 /*
558 : jsr166 1.14 * Possible values for "how" argument in xfer method.
559 : jsr166 1.1 */
560 : jsr166 1.14 private static final int NOW = 0; // for untimed poll, tryTransfer
561 :     private static final int ASYNC = 1; // for offer, put, add
562 :     private static final int SYNC = 2; // for transfer, take
563 :     private static final int TIMED = 3; // for timed poll, tryTransfer
564 : jsr166 1.1
565 : jsr166 1.10 @SuppressWarnings("unchecked")
566 :     static <E> E cast(Object item) {
567 : dl 1.33 // assert item == null || item.getClass() != Node.class;
568 : jsr166 1.10 return (E) item;
569 :     }
570 :    
571 : jsr166 1.1 /**
572 : jsr166 1.8 * Implements all queuing methods. See above for explanation.
573 : jsr166 1.1 *
574 : jsr166 1.8 * @param e the item or null for take
575 :     * @param haveData true if this is a put, else a take
576 : jsr166 1.14 * @param how NOW, ASYNC, SYNC, or TIMED
577 :     * @param nanos timeout in nanosecs, used only if mode is TIMED
578 : jsr166 1.8 * @return an item if matched, else e
579 :     * @throws NullPointerException if haveData mode but e is null
580 : jsr166 1.1 */
581 : jsr166 1.8 private E xfer(E e, boolean haveData, int how, long nanos) {
582 :     if (haveData && (e == null))
583 :     throw new NullPointerException();
584 : jsr166 1.14 Node s = null; // the node to append, if needed
585 : jsr166 1.1
586 : jsr166 1.29 retry:
587 :     for (;;) { // restart on append race
588 : jsr166 1.1
589 : jsr166 1.14 for (Node h = head, p = h; p != null;) { // find & match first node
590 : jsr166 1.8 boolean isData = p.isData;
591 :     Object item = p.item;
592 :     if (item != p && (item != null) == isData) { // unmatched
593 :     if (isData == haveData) // can't match
594 :     break;
595 :     if (p.casItem(item, e)) { // match
596 : jsr166 1.14 for (Node q = p; q != h;) {
597 : dl 1.16 Node n = q.next; // update by 2 unless singleton
598 : jsr166 1.37 if (head == h && casHead(h, n == null ? q : n)) {
599 : jsr166 1.8 h.forgetNext();
600 :     break;
601 :     } // advance and retry
602 :     if ((h = head) == null ||
603 :     (q = h.next) == null || !q.isMatched())
604 :     break; // unless slack < 2
605 :     }
606 :     LockSupport.unpark(p.waiter);
607 : jsr166 1.46 return LinkedTransferQueue.<E>cast(item);
608 : jsr166 1.1 }
609 :     }
610 : jsr166 1.14 Node n = p.next;
611 : jsr166 1.8 p = (p != n) ? n : (h = head); // Use head if p offlist
612 :     }
613 :    
614 : jsr166 1.14 if (how != NOW) { // No matches available
615 : jsr166 1.8 if (s == null)
616 : jsr166 1.14 s = new Node(e, haveData);
617 :     Node pred = tryAppend(s, haveData);
618 : jsr166 1.8 if (pred == null)
619 :     continue retry; // lost race vs opposite mode
620 : jsr166 1.14 if (how != ASYNC)
621 :     return awaitMatch(s, pred, e, (how == TIMED), nanos);
622 : jsr166 1.1 }
623 : jsr166 1.8 return e; // not waiting
624 : jsr166 1.1 }
625 :     }
626 :    
627 :     /**
628 : jsr166 1.8 * Tries to append node s as tail.
629 :     *
630 :     * @param s the node to append
631 :     * @param haveData true if appending in data mode
632 :     * @return null on failure due to losing race with append in
633 :     * different mode, else s's predecessor, or s itself if no
634 :     * predecessor
635 : jsr166 1.1 */
636 : jsr166 1.14 private Node tryAppend(Node s, boolean haveData) {
637 :     for (Node t = tail, p = t;;) { // move p to last node and append
638 :     Node n, u; // temps for reads of next & tail
639 : jsr166 1.8 if (p == null && (p = head) == null) {
640 :     if (casHead(null, s))
641 :     return s; // initialize
642 :     }
643 :     else if (p.cannotPrecede(haveData))
644 :     return null; // lost race vs opposite mode
645 :     else if ((n = p.next) != null) // not last; keep traversing
646 :     p = p != t && t != (u = tail) ? (t = u) : // stale tail
647 :     (p != n) ? n : null; // restart if off list
648 :     else if (!p.casNext(null, s))
649 :     p = p.next; // re-read on CAS failure
650 :     else {
651 :     if (p != t) { // update if slack now >= 2
652 :     while ((tail != t || !casTail(t, s)) &&
653 :     (t = tail) != null &&
654 :     (s = t.next) != null && // advance and retry
655 :     (s = s.next) != null && s != t);
656 : jsr166 1.1 }
657 : jsr166 1.8 return p;
658 : jsr166 1.1 }
659 :     }
660 :     }
661 :    
662 :     /**
663 : jsr166 1.8 * Spins/yields/blocks until node s is matched or caller gives up.
664 : jsr166 1.1 *
665 :     * @param s the waiting node
666 : jsr166 1.8 * @param pred the predecessor of s, or s itself if it has no
667 :     * predecessor, or null if unknown (the null case does not occur
668 :     * in any current calls but may in possible future extensions)
669 : jsr166 1.1 * @param e the comparison value for checking match
670 : jsr166 1.14 * @param timed if true, wait only until timeout elapses
671 :     * @param nanos timeout in nanosecs, used only if timed is true
672 : jsr166 1.8 * @return matched item, or e if unmatched on interrupt or timeout
673 : jsr166 1.1 */
674 : jsr166 1.14 private E awaitMatch(Node s, Node pred, E e, boolean timed, long nanos) {
675 : jsr166 1.51 final long deadline = timed ? System.nanoTime() + nanos : 0L;
676 : jsr166 1.8 Thread w = Thread.currentThread();
677 :     int spins = -1; // initialized after first item and cancel checks
678 :     ThreadLocalRandom randomYields = null; // bound if needed
679 : jsr166 1.1
680 :     for (;;) {
681 : jsr166 1.8 Object item = s.item;
682 :     if (item != e) { // matched
683 : dl 1.33 // assert item != s;
684 : jsr166 1.8 s.forgetContents(); // avoid garbage
685 : jsr166 1.46 return LinkedTransferQueue.<E>cast(item);
686 : jsr166 1.8 }
687 : jsr166 1.14 if ((w.isInterrupted() || (timed && nanos <= 0)) &&
688 : dl 1.16 s.casItem(e, s)) { // cancel
689 : jsr166 1.8 unsplice(pred, s);
690 :     return e;
691 :     }
692 :    
693 :     if (spins < 0) { // establish spins at/near front
694 :     if ((spins = spinsFor(pred, s.isData)) > 0)
695 :     randomYields = ThreadLocalRandom.current();
696 :     }
697 :     else if (spins > 0) { // spin
698 : dl 1.16 --spins;
699 :     if (randomYields.nextInt(CHAINED_SPINS) == 0)
700 : jsr166 1.8 Thread.yield(); // occasionally yield
701 :     }
702 :     else if (s.waiter == null) {
703 :     s.waiter = w; // request unpark then recheck
704 : jsr166 1.1 }
705 : jsr166 1.14 else if (timed) {
706 : jsr166 1.51 nanos = deadline - System.nanoTime();
707 :     if (nanos > 0L)
708 : jsr166 1.8 LockSupport.parkNanos(this, nanos);
709 : jsr166 1.1 }
710 : jsr166 1.8 else {
711 : jsr166 1.1 LockSupport.park(this);
712 :     }
713 : jsr166 1.8 }
714 :     }
715 :    
716 :     /**
717 :     * Returns spin/yield value for a node with given predecessor and
718 :     * data mode. See above for explanation.
719 :     */
720 : jsr166 1.14 private static int spinsFor(Node pred, boolean haveData) {
721 : jsr166 1.8 if (MP && pred != null) {
722 :     if (pred.isData != haveData) // phase change
723 :     return FRONT_SPINS + CHAINED_SPINS;
724 :     if (pred.isMatched()) // probably at front
725 :     return FRONT_SPINS;
726 :     if (pred.waiter == null) // pred apparently spinning
727 :     return CHAINED_SPINS;
728 :     }
729 :     return 0;
730 :     }
731 :    
732 :     /* -------------- Traversal methods -------------- */
733 :    
734 :     /**
735 : jsr166 1.14 * Returns the successor of p, or the head node if p.next has been
736 :     * linked to self, which will only be true if traversing with a
737 :     * stale pointer that is now off the list.
738 :     */
739 :     final Node succ(Node p) {
740 :     Node next = p.next;
741 :     return (p == next) ? head : next;
742 :     }
743 :    
744 :     /**
745 : jsr166 1.8 * Returns the first unmatched node of the given mode, or null if
746 :     * none. Used by methods isEmpty, hasWaitingConsumer.
747 :     */
748 : jsr166 1.14 private Node firstOfMode(boolean isData) {
749 :     for (Node p = head; p != null; p = succ(p)) {
750 : jsr166 1.8 if (!p.isMatched())
751 : jsr166 1.14 return (p.isData == isData) ? p : null;
752 : jsr166 1.8 }
753 :     return null;
754 :     }
755 :    
756 :     /**
757 : dl 1.52 * Version of firstOfMode used by Spliterator
758 :     */
759 :     final Node firstDataNode() {
760 :     for (Node p = head; p != null;) {
761 :     Object item = p.item;
762 :     if (p.isData) {
763 :     if (item != null && item != p)
764 :     return p;
765 :     }
766 :     else if (item == null)
767 :     break;
768 :     if (p == (p = p.next))
769 :     p = head;
770 :     }
771 :     return null;
772 :     }
773 :    
774 :     /**
775 : jsr166 1.8 * Returns the item in the first unmatched node with isData; or
776 :     * null if none. Used by peek.
777 :     */
778 :     private E firstDataItem() {
779 : jsr166 1.14 for (Node p = head; p != null; p = succ(p)) {
780 : jsr166 1.8 Object item = p.item;
781 : jsr166 1.14 if (p.isData) {
782 :     if (item != null && item != p)
783 : jsr166 1.46 return LinkedTransferQueue.<E>cast(item);
784 : jsr166 1.14 }
785 :     else if (item == null)
786 :     return null;
787 : jsr166 1.8 }
788 :     return null;
789 :     }
790 :    
791 : jsr166 1.1 /**
792 : jsr166 1.8 * Traverses and counts unmatched nodes of the given mode.
793 :     * Used by methods size and getWaitingConsumerCount.
794 : jsr166 1.1 */
795 : jsr166 1.8 private int countOfMode(boolean data) {
796 :     int count = 0;
797 : jsr166 1.14 for (Node p = head; p != null; ) {
798 : jsr166 1.8 if (!p.isMatched()) {
799 :     if (p.isData != data)
800 :     return 0;
801 :     if (++count == Integer.MAX_VALUE) // saturated
802 :     break;
803 :     }
804 : jsr166 1.14 Node n = p.next;
805 : jsr166 1.8 if (n != p)
806 :     p = n;
807 :     else {
808 :     count = 0;
809 :     p = head;
810 : jsr166 1.1 }
811 : jsr166 1.8 }
812 :     return count;
813 :     }
814 :    
815 :     final class Itr implements Iterator<E> {
816 : jsr166 1.14 private Node nextNode; // next node to return item for
817 :     private E nextItem; // the corresponding item
818 :     private Node lastRet; // last returned node, to support remove
819 :     private Node lastPred; // predecessor to unlink lastRet
820 : jsr166 1.8
821 :     /**
822 :     * Moves to next node after prev, or first node if prev null.
823 :     */
824 : jsr166 1.14 private void advance(Node prev) {
825 : dl 1.33 /*
826 :     * To track and avoid buildup of deleted nodes in the face
827 :     * of calls to both Queue.remove and Itr.remove, we must
828 :     * include variants of unsplice and sweep upon each
829 :     * advance: Upon Itr.remove, we may need to catch up links
830 :     * from lastPred, and upon other removes, we might need to
831 :     * skip ahead from stale nodes and unsplice deleted ones
832 :     * found while advancing.
833 :     */
834 :    
835 :     Node r, b; // reset lastPred upon possible deletion of lastRet
836 :     if ((r = lastRet) != null && !r.isMatched())
837 :     lastPred = r; // next lastPred is old lastRet
838 :     else if ((b = lastPred) == null || b.isMatched())
839 :     lastPred = null; // at start of list
840 : jsr166 1.34 else {
841 : dl 1.33 Node s, n; // help with removal of lastPred.next
842 :     while ((s = b.next) != null &&
843 :     s != b && s.isMatched() &&
844 :     (n = s.next) != null && n != s)
845 :     b.casNext(s, n);
846 :     }
847 :    
848 :     this.lastRet = prev;
849 : jsr166 1.35
850 : dl 1.33 for (Node p = prev, s, n;;) {
851 :     s = (p == null) ? head : p.next;
852 :     if (s == null)
853 :     break;
854 :     else if (s == p) {
855 :     p = null;
856 :     continue;
857 :     }
858 :     Object item = s.item;
859 :     if (s.isData) {
860 :     if (item != null && item != s) {
861 : jsr166 1.31 nextItem = LinkedTransferQueue.<E>cast(item);
862 : dl 1.33 nextNode = s;
863 : jsr166 1.8 return;
864 :     }
865 : jsr166 1.34 }
866 : jsr166 1.8 else if (item == null)
867 :     break;
868 : dl 1.33 // assert s.isMatched();
869 :     if (p == null)
870 :     p = s;
871 :     else if ((n = s.next) == null)
872 :     break;
873 :     else if (s == n)
874 :     p = null;
875 :     else
876 :     p.casNext(s, n);
877 : jsr166 1.1 }
878 : jsr166 1.8 nextNode = null;
879 : dl 1.33 nextItem = null;
880 : jsr166 1.8 }
881 :    
882 :     Itr() {
883 :     advance(null);
884 :     }
885 :    
886 :     public final boolean hasNext() {
887 :     return nextNode != null;
888 :     }
889 :    
890 :     public final E next() {
891 : jsr166 1.14 Node p = nextNode;
892 : jsr166 1.8 if (p == null) throw new NoSuchElementException();
893 :     E e = nextItem;
894 :     advance(p);
895 :     return e;
896 :     }
897 :    
898 :     public final void remove() {
899 : dl 1.33 final Node lastRet = this.lastRet;
900 :     if (lastRet == null)
901 :     throw new IllegalStateException();
902 :     this.lastRet = null;
903 :     if (lastRet.tryMatchData())
904 :     unsplice(lastPred, lastRet);
905 : jsr166 1.1 }
906 :     }
907 : jsr166 1.53
908 : dl 1.57 /** A customized variant of Spliterators.IteratorSpliterator */
909 : dl 1.52 static final class LTQSpliterator<E> implements Spliterator<E> {
910 : dl 1.60 static final int MAX_BATCH = 1 << 25; // max batch array size;
911 : dl 1.52 final LinkedTransferQueue<E> queue;
912 :     Node current; // current node; null until initialized
913 :     int batch; // batch size for splits
914 :     boolean exhausted; // true when no more nodes
915 : jsr166 1.53 LTQSpliterator(LinkedTransferQueue<E> queue) {
916 : dl 1.52 this.queue = queue;
917 :     }
918 :    
919 :     public Spliterator<E> trySplit() {
920 : dl 1.60 Node p;
921 : dl 1.52 final LinkedTransferQueue<E> q = this.queue;
922 : dl 1.60 int b = batch;
923 :     int n = (b <= 0) ? 1 : (b >= MAX_BATCH) ? MAX_BATCH : b + 1;
924 : jsr166 1.58 if (!exhausted &&
925 : dl 1.54 ((p = current) != null || (p = q.firstDataNode()) != null) &&
926 :     p.next != null) {
927 : dl 1.63 Object[] a = new Object[n];
928 : dl 1.52 int i = 0;
929 :     do {
930 :     if ((a[i] = p.item) != null)
931 :     ++i;
932 : jsr166 1.53 if (p == (p = p.next))
933 : dl 1.52 p = q.firstDataNode();
934 :     } while (p != null && i < n);
935 :     if ((current = p) == null)
936 :     exhausted = true;
937 : dl 1.60 if (i > 0) {
938 :     batch = i;
939 :     return Spliterators.spliterator
940 :     (a, 0, i, Spliterator.ORDERED | Spliterator.NONNULL |
941 :     Spliterator.CONCURRENT);
942 :     }
943 : dl 1.52 }
944 :     return null;
945 :     }
946 :    
947 :     @SuppressWarnings("unchecked")
948 : dl 1.61 public void forEachRemaining(Consumer<? super E> action) {
949 : dl 1.52 Node p;
950 :     if (action == null) throw new NullPointerException();
951 :     final LinkedTransferQueue<E> q = this.queue;
952 :     if (!exhausted &&
953 :     ((p = current) != null || (p = q.firstDataNode()) != null)) {
954 :     exhausted = true;
955 :     do {
956 :     Object e = p.item;
957 : jsr166 1.53 if (p == (p = p.next))
958 : dl 1.52 p = q.firstDataNode();
959 :     if (e != null)
960 :     action.accept((E)e);
961 :     } while (p != null);
962 :     }
963 :     }
964 :    
965 :     @SuppressWarnings("unchecked")
966 :     public boolean tryAdvance(Consumer<? super E> action) {
967 :     Node p;
968 :     if (action == null) throw new NullPointerException();
969 :     final LinkedTransferQueue<E> q = this.queue;
970 :     if (!exhausted &&
971 :     ((p = current) != null || (p = q.firstDataNode()) != null)) {
972 :     Object e;
973 :     do {
974 :     e = p.item;
975 : jsr166 1.53 if (p == (p = p.next))
976 : dl 1.52 p = q.firstDataNode();
977 :     } while (e == null && p != null);
978 :     if ((current = p) == null)
979 :     exhausted = true;
980 :     if (e != null) {
981 :     action.accept((E)e);
982 :     return true;
983 :     }
984 :     }
985 :     return false;
986 :     }
987 :    
988 : dl 1.54 public long estimateSize() { return Long.MAX_VALUE; }
989 :    
990 : dl 1.52 public int characteristics() {
991 :     return Spliterator.ORDERED | Spliterator.NONNULL |
992 :     Spliterator.CONCURRENT;
993 :     }
994 :     }
995 :    
996 : jsr166 1.67 /**
997 :     * Returns a {@link Spliterator} over the elements in this queue.
998 :     *
999 : jsr166 1.68 * <p>The returned spliterator is
1000 :     * <a href="package-summary.html#Weakly"><i>weakly consistent</i></a>.
1001 :     *
1002 : jsr166 1.67 * <p>The {@code Spliterator} reports {@link Spliterator#CONCURRENT},
1003 :     * {@link Spliterator#ORDERED}, and {@link Spliterator#NONNULL}.
1004 :     *
1005 :     * @implNote
1006 :     * The {@code Spliterator} implements {@code trySplit} to permit limited
1007 :     * parallelism.
1008 :     *
1009 :     * @return a {@code Spliterator} over the elements in this queue
1010 :     * @since 1.8
1011 :     */
1012 : dl 1.56 public Spliterator<E> spliterator() {
1013 : dl 1.52 return new LTQSpliterator<E>(this);
1014 :     }
1015 :    
1016 : jsr166 1.8 /* -------------- Removal methods -------------- */
1017 :    
1018 : jsr166 1.1 /**
1019 : jsr166 1.8 * Unsplices (now or later) the given deleted/cancelled node with
1020 :     * the given predecessor.
1021 : jsr166 1.1 *
1022 : dl 1.16 * @param pred a node that was at one time known to be the
1023 :     * predecessor of s, or null or s itself if s is/was at head
1024 : jsr166 1.8 * @param s the node to be unspliced
1025 : jsr166 1.1 */
1026 : dl 1.16 final void unsplice(Node pred, Node s) {
1027 :     s.forgetContents(); // forget unneeded fields
1028 : jsr166 1.1 /*
1029 : dl 1.16 * See above for rationale. Briefly: if pred still points to
1030 :     * s, try to unlink s. If s cannot be unlinked, because it is
1031 :     * trailing node or pred might be unlinked, and neither pred
1032 :     * nor s are head or offlist, add to sweepVotes, and if enough
1033 :     * votes have accumulated, sweep.
1034 : jsr166 1.1 */
1035 : dl 1.16 if (pred != null && pred != s && pred.next == s) {
1036 :     Node n = s.next;
1037 :     if (n == null ||
1038 :     (n != s && pred.casNext(s, n) && pred.isMatched())) {
1039 :     for (;;) { // check if at, or could be, head
1040 :     Node h = head;
1041 :     if (h == pred || h == s || h == null)
1042 :     return; // at head or list empty
1043 :     if (!h.isMatched())
1044 :     break;
1045 :     Node hn = h.next;
1046 :     if (hn == null)
1047 :     return; // now empty
1048 :     if (hn != h && casHead(h, hn))
1049 :     h.forgetNext(); // advance head
1050 : jsr166 1.8 }
1051 : dl 1.16 if (pred.next != pred && s.next != s) { // recheck if offlist
1052 :     for (;;) { // sweep now if enough votes
1053 :     int v = sweepVotes;
1054 :     if (v < SWEEP_THRESHOLD) {
1055 :     if (casSweepVotes(v, v + 1))
1056 :     break;
1057 :     }
1058 :     else if (casSweepVotes(v, 0)) {
1059 :     sweep();
1060 :     break;
1061 :     }
1062 :     }
1063 : jsr166 1.12 }
1064 : jsr166 1.1 }
1065 :     }
1066 :     }
1067 :    
1068 :     /**
1069 : jsr166 1.26 * Unlinks matched (typically cancelled) nodes encountered in a
1070 :     * traversal from head.
1071 : jsr166 1.1 */
1072 : dl 1.16 private void sweep() {
1073 : jsr166 1.20 for (Node p = head, s, n; p != null && (s = p.next) != null; ) {
1074 : jsr166 1.28 if (!s.isMatched())
1075 :     // Unmatched nodes are never self-linked
1076 : jsr166 1.20 p = s;
1077 : jsr166 1.28 else if ((n = s.next) == null) // trailing node is pinned
1078 : jsr166 1.20 break;
1079 : jsr166 1.28 else if (s == n) // stale
1080 :     // No need to also check for p == s, since that implies s == n
1081 :     p = head;
1082 : jsr166 1.20 else
1083 : dl 1.16 p.casNext(s, n);
1084 : jsr166 1.8 }
1085 :     }
1086 :    
1087 :     /**
1088 :     * Main implementation of remove(Object)
1089 :     */
1090 :     private boolean findAndRemove(Object e) {
1091 :     if (e != null) {
1092 : jsr166 1.14 for (Node pred = null, p = head; p != null; ) {
1093 : jsr166 1.8 Object item = p.item;
1094 :     if (p.isData) {
1095 :     if (item != null && item != p && e.equals(item) &&
1096 :     p.tryMatchData()) {
1097 :     unsplice(pred, p);
1098 :     return true;
1099 :     }
1100 :     }
1101 :     else if (item == null)
1102 :     break;
1103 :     pred = p;
1104 : jsr166 1.11 if ((p = p.next) == pred) { // stale
1105 : jsr166 1.8 pred = null;
1106 :     p = head;
1107 :     }
1108 :     }
1109 :     }
1110 :     return false;
1111 :     }
1112 :    
1113 :     /**
1114 : jsr166 1.1 * Creates an initially empty {@code LinkedTransferQueue}.
1115 :     */
1116 :     public LinkedTransferQueue() {
1117 :     }
1118 :    
1119 :     /**
1120 :     * Creates a {@code LinkedTransferQueue}
1121 :     * initially containing the elements of the given collection,
1122 :     * added in traversal order of the collection's iterator.
1123 :     *
1124 :     * @param c the collection of elements to initially contain
1125 :     * @throws NullPointerException if the specified collection or any
1126 :     * of its elements are null
1127 :     */
1128 :     public LinkedTransferQueue(Collection<? extends E> c) {
1129 :     this();
1130 :     addAll(c);
1131 :     }
1132 :    
1133 : jsr166 1.4 /**
1134 : jsr166 1.5 * Inserts the specified element at the tail of this queue.
1135 :     * As the queue is unbounded, this method will never block.
1136 :     *
1137 :     * @throws NullPointerException if the specified element is null
1138 : jsr166 1.4 */
1139 : jsr166 1.5 public void put(E e) {
1140 : jsr166 1.8 xfer(e, true, ASYNC, 0);
1141 : jsr166 1.1 }
1142 :    
1143 : jsr166 1.4 /**
1144 : jsr166 1.5 * Inserts the specified element at the tail of this queue.
1145 :     * As the queue is unbounded, this method will never block or
1146 :     * return {@code false}.
1147 :     *
1148 :     * @return {@code true} (as specified by
1149 : jsr166 1.42 * {@link java.util.concurrent.BlockingQueue#offer(Object,long,TimeUnit)
1150 :     * BlockingQueue.offer})
1151 : jsr166 1.5 * @throws NullPointerException if the specified element is null
1152 : jsr166 1.4 */
1153 : jsr166 1.5 public boolean offer(E e, long timeout, TimeUnit unit) {
1154 : jsr166 1.8 xfer(e, true, ASYNC, 0);
1155 :     return true;
1156 : jsr166 1.1 }
1157 :    
1158 : jsr166 1.4 /**
1159 : jsr166 1.5 * Inserts the specified element at the tail of this queue.
1160 :     * As the queue is unbounded, this method will never return {@code false}.
1161 :     *
1162 : jsr166 1.32 * @return {@code true} (as specified by {@link Queue#offer})
1163 : jsr166 1.5 * @throws NullPointerException if the specified element is null
1164 : jsr166 1.4 */
1165 : jsr166 1.1 public boolean offer(E e) {
1166 : jsr166 1.8 xfer(e, true, ASYNC, 0);
1167 : jsr166 1.1 return true;
1168 :     }
1169 :    
1170 : jsr166 1.4 /**
1171 : jsr166 1.5 * Inserts the specified element at the tail of this queue.
1172 :     * As the queue is unbounded, this method will never throw
1173 :     * {@link IllegalStateException} or return {@code false}.
1174 :     *
1175 :     * @return {@code true} (as specified by {@link Collection#add})
1176 :     * @throws NullPointerException if the specified element is null
1177 : jsr166 1.4 */
1178 : jsr166 1.1 public boolean add(E e) {
1179 : jsr166 1.8 xfer(e, true, ASYNC, 0);
1180 :     return true;
1181 : jsr166 1.5 }
1182 :    
1183 :     /**
1184 : jsr166 1.6 * Transfers the element to a waiting consumer immediately, if possible.
1185 :     *
1186 :     * <p>More precisely, transfers the specified element immediately
1187 :     * if there exists a consumer already waiting to receive it (in
1188 :     * {@link #take} or timed {@link #poll(long,TimeUnit) poll}),
1189 :     * otherwise returning {@code false} without enqueuing the element.
1190 : jsr166 1.5 *
1191 :     * @throws NullPointerException if the specified element is null
1192 :     */
1193 :     public boolean tryTransfer(E e) {
1194 : jsr166 1.8 return xfer(e, true, NOW, 0) == null;
1195 : jsr166 1.1 }
1196 :    
1197 : jsr166 1.4 /**
1198 : jsr166 1.6 * Transfers the element to a consumer, waiting if necessary to do so.
1199 :     *
1200 :     * <p>More precisely, transfers the specified element immediately
1201 :     * if there exists a consumer already waiting to receive it (in
1202 :     * {@link #take} or timed {@link #poll(long,TimeUnit) poll}),
1203 :     * else inserts the specified element at the tail of this queue
1204 :     * and waits until the element is received by a consumer.
1205 : jsr166 1.5 *
1206 :     * @throws NullPointerException if the specified element is null
1207 : jsr166 1.4 */
1208 : jsr166 1.1 public void transfer(E e) throws InterruptedException {
1209 : jsr166 1.8 if (xfer(e, true, SYNC, 0) != null) {
1210 :     Thread.interrupted(); // failure possible only due to interrupt
1211 : jsr166 1.1 throw new InterruptedException();
1212 :     }
1213 :     }
1214 :    
1215 : jsr166 1.4 /**
1216 : jsr166 1.6 * Transfers the element to a consumer if it is possible to do so
1217 :     * before the timeout elapses.
1218 :     *
1219 :     * <p>More precisely, transfers the specified element immediately
1220 :     * if there exists a consumer already waiting to receive it (in
1221 :     * {@link #take} or timed {@link #poll(long,TimeUnit) poll}),
1222 :     * else inserts the specified element at the tail of this queue
1223 :     * and waits until the element is received by a consumer,
1224 :     * returning {@code false} if the specified wait time elapses
1225 :     * before the element can be transferred.
1226 : jsr166 1.5 *
1227 :     * @throws NullPointerException if the specified element is null
1228 : jsr166 1.4 */
1229 : jsr166 1.1 public boolean tryTransfer(E e, long timeout, TimeUnit unit)
1230 :     throws InterruptedException {
1231 : jsr166 1.14 if (xfer(e, true, TIMED, unit.toNanos(timeout)) == null)
1232 : jsr166 1.1 return true;
1233 :     if (!Thread.interrupted())
1234 :     return false;
1235 :     throw new InterruptedException();
1236 :     }
1237 :    
1238 :     public E take() throws InterruptedException {
1239 : jsr166 1.8 E e = xfer(null, false, SYNC, 0);
1240 : jsr166 1.1 if (e != null)
1241 : jsr166 1.5 return e;
1242 : jsr166 1.1 Thread.interrupted();
1243 :     throw new InterruptedException();
1244 :     }
1245 :    
1246 :     public E poll(long timeout, TimeUnit unit) throws InterruptedException {
1247 : jsr166 1.14 E e = xfer(null, false, TIMED, unit.toNanos(timeout));
1248 : jsr166 1.1 if (e != null || !Thread.interrupted())
1249 : jsr166 1.5 return e;
1250 : jsr166 1.1 throw new InterruptedException();
1251 :     }
1252 :    
1253 :     public E poll() {
1254 : jsr166 1.8 return xfer(null, false, NOW, 0);
1255 : jsr166 1.1 }
1256 :    
1257 : jsr166 1.4 /**
1258 :     * @throws NullPointerException {@inheritDoc}
1259 :     * @throws IllegalArgumentException {@inheritDoc}
1260 :     */
1261 : jsr166 1.1 public int drainTo(Collection<? super E> c) {
1262 :     if (c == null)
1263 :     throw new NullPointerException();
1264 :     if (c == this)
1265 :     throw new IllegalArgumentException();
1266 :     int n = 0;
1267 : jsr166 1.47 for (E e; (e = poll()) != null;) {
1268 : jsr166 1.1 c.add(e);
1269 :     ++n;
1270 :     }
1271 :     return n;
1272 :     }
1273 :    
1274 : jsr166 1.4 /**
1275 :     * @throws NullPointerException {@inheritDoc}
1276 :     * @throws IllegalArgumentException {@inheritDoc}
1277 :     */
1278 : jsr166 1.1 public int drainTo(Collection<? super E> c, int maxElements) {
1279 :     if (c == null)
1280 :     throw new NullPointerException();
1281 :     if (c == this)
1282 :     throw new IllegalArgumentException();
1283 :     int n = 0;
1284 : jsr166 1.47 for (E e; n < maxElements && (e = poll()) != null;) {
1285 : jsr166 1.1 c.add(e);
1286 :     ++n;
1287 :     }
1288 :     return n;
1289 :     }
1290 :    
1291 : jsr166 1.5 /**
1292 : jsr166 1.36 * Returns an iterator over the elements in this queue in proper sequence.
1293 :     * The elements will be returned in order from first (head) to last (tail).
1294 : jsr166 1.5 *
1295 : jsr166 1.68 * <p>The returned iterator is
1296 :     * <a href="package-summary.html#Weakly"><i>weakly consistent</i></a>.
1297 : jsr166 1.5 *
1298 :     * @return an iterator over the elements in this queue in proper sequence
1299 :     */
1300 : jsr166 1.1 public Iterator<E> iterator() {
1301 :     return new Itr();
1302 :     }
1303 :    
1304 :     public E peek() {
1305 : jsr166 1.8 return firstDataItem();
1306 : jsr166 1.1 }
1307 :    
1308 : jsr166 1.6 /**
1309 :     * Returns {@code true} if this queue contains no elements.
1310 :     *
1311 :     * @return {@code true} if this queue contains no elements
1312 :     */
1313 : jsr166 1.1 public boolean isEmpty() {
1314 : dl 1.21 for (Node p = head; p != null; p = succ(p)) {
1315 :     if (!p.isMatched())
1316 :     return !p.isData;
1317 :     }
1318 :     return true;
1319 : jsr166 1.1 }
1320 :    
1321 :     public boolean hasWaitingConsumer() {
1322 : jsr166 1.8 return firstOfMode(false) != null;
1323 : jsr166 1.1 }
1324 :    
1325 :     /**
1326 :     * Returns the number of elements in this queue. If this queue
1327 :     * contains more than {@code Integer.MAX_VALUE} elements, returns
1328 :     * {@code Integer.MAX_VALUE}.
1329 :     *
1330 :     * <p>Beware that, unlike in most collections, this method is
1331 :     * <em>NOT</em> a constant-time operation. Because of the
1332 :     * asynchronous nature of these queues, determining the current
1333 :     * number of elements requires an O(n) traversal.
1334 :     *
1335 :     * @return the number of elements in this queue
1336 :     */
1337 :     public int size() {
1338 : jsr166 1.8 return countOfMode(true);
1339 : jsr166 1.1 }
1340 :    
1341 :     public int getWaitingConsumerCount() {
1342 : jsr166 1.8 return countOfMode(false);
1343 : jsr166 1.1 }
1344 :    
1345 : jsr166 1.6 /**
1346 :     * Removes a single instance of the specified element from this queue,
1347 :     * if it is present. More formally, removes an element {@code e} such
1348 :     * that {@code o.equals(e)}, if this queue contains one or more such
1349 :     * elements.
1350 :     * Returns {@code true} if this queue contained the specified element
1351 :     * (or equivalently, if this queue changed as a result of the call).
1352 :     *
1353 :     * @param o element to be removed from this queue, if present
1354 :     * @return {@code true} if this queue changed as a result of the call
1355 :     */
1356 : jsr166 1.1 public boolean remove(Object o) {
1357 : jsr166 1.8 return findAndRemove(o);
1358 : jsr166 1.1 }
1359 :    
1360 :     /**
1361 : jsr166 1.30 * Returns {@code true} if this queue contains the specified element.
1362 :     * More formally, returns {@code true} if and only if this queue contains
1363 :     * at least one element {@code e} such that {@code o.equals(e)}.
1364 :     *
1365 :     * @param o object to be checked for containment in this queue
1366 :     * @return {@code true} if this queue contains the specified element
1367 :     */
1368 :     public boolean contains(Object o) {
1369 :     if (o == null) return false;
1370 :     for (Node p = head; p != null; p = succ(p)) {
1371 :     Object item = p.item;
1372 :     if (p.isData) {
1373 :     if (item != null && item != p && o.equals(item))
1374 :     return true;
1375 :     }
1376 :     else if (item == null)
1377 :     break;
1378 :     }
1379 :     return false;
1380 :     }
1381 :    
1382 :     /**
1383 : jsr166 1.5 * Always returns {@code Integer.MAX_VALUE} because a
1384 :     * {@code LinkedTransferQueue} is not capacity constrained.
1385 :     *
1386 :     * @return {@code Integer.MAX_VALUE} (as specified by
1387 : jsr166 1.42 * {@link java.util.concurrent.BlockingQueue#remainingCapacity()
1388 :     * BlockingQueue.remainingCapacity})
1389 : jsr166 1.5 */
1390 :     public int remainingCapacity() {
1391 :     return Integer.MAX_VALUE;
1392 :     }
1393 :    
1394 :     /**
1395 : jsr166 1.50 * Saves this queue to a stream (that is, serializes it).
1396 : jsr166 1.1 *
1397 : jsr166 1.65 * @param s the stream
1398 : jsr166 1.66 * @throws java.io.IOException if an I/O error occurs
1399 : jsr166 1.1 * @serialData All of the elements (each an {@code E}) in
1400 :     * the proper order, followed by a null
1401 :     */
1402 :     private void writeObject(java.io.ObjectOutputStream s)
1403 :     throws java.io.IOException {
1404 :     s.defaultWriteObject();
1405 :     for (E e : this)
1406 :     s.writeObject(e);
1407 :     // Use trailing null as sentinel
1408 :     s.writeObject(null);
1409 :     }
1410 :    
1411 :     /**
1412 : jsr166 1.50 * Reconstitutes this queue from a stream (that is, deserializes it).
1413 : jsr166 1.65 * @param s the stream
1414 : jsr166 1.66 * @throws ClassNotFoundException if the class of a serialized object
1415 :     * could not be found
1416 :     * @throws java.io.IOException if an I/O error occurs
1417 : jsr166 1.1 */
1418 :     private void readObject(java.io.ObjectInputStream s)
1419 :     throws java.io.IOException, ClassNotFoundException {
1420 :     s.defaultReadObject();
1421 :     for (;;) {
1422 : jsr166 1.49 @SuppressWarnings("unchecked")
1423 :     E item = (E) s.readObject();
1424 : jsr166 1.1 if (item == null)
1425 :     break;
1426 :     else
1427 :     offer(item);
1428 :     }
1429 :     }
1430 :    
1431 : jsr166 1.3 // Unsafe mechanics
1432 : jsr166 1.1
1433 : dl 1.38 private static final sun.misc.Unsafe UNSAFE;
1434 :     private static final long headOffset;
1435 :     private static final long tailOffset;
1436 :     private static final long sweepVotesOffset;
1437 :     static {
1438 : jsr166 1.1 try {
1439 : dl 1.38 UNSAFE = sun.misc.Unsafe.getUnsafe();
1440 : jsr166 1.43 Class<?> k = LinkedTransferQueue.class;
1441 : dl 1.38 headOffset = UNSAFE.objectFieldOffset
1442 :     (k.getDeclaredField("head"));
1443 :     tailOffset = UNSAFE.objectFieldOffset
1444 :     (k.getDeclaredField("tail"));
1445 :     sweepVotesOffset = UNSAFE.objectFieldOffset
1446 :     (k.getDeclaredField("sweepVotes"));
1447 :     } catch (Exception e) {
1448 :     throw new Error(e);
1449 : jsr166 1.1 }
1450 :     }
1451 :     }

Doug Lea
ViewVC Help
Powered by ViewVC 1.0.8