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