2 * lock-free-queue.c: Lock free queue.
4 * (C) Copyright 2011 Novell, Inc
6 * Permission is hereby granted, free of charge, to any person obtaining
7 * a copy of this software and associated documentation files (the
8 * "Software"), to deal in the Software without restriction, including
9 * without limitation the rights to use, copy, modify, merge, publish,
10 * distribute, sublicense, and/or sell copies of the Software, and to
11 * permit persons to whom the Software is furnished to do so, subject to
12 * the following conditions:
14 * The above copyright notice and this permission notice shall be
15 * included in all copies or substantial portions of the Software.
17 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
18 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
19 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
20 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
21 * LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
22 * OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
23 * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
27 * This is an implementation of a lock-free queue, as described in
29 * Simple, Fast, and Practical Non-Blocking and Blocking
30 * Concurrent Queue Algorithms
31 * Maged M. Michael, Michael L. Scott
34 * A few slight modifications have been made:
36 * We use hazard pointers to rule out the ABA problem, instead of the
37 * counter as in the paper.
39 * Memory management of the queue entries is done by the caller, not
40 * by the queue implementation. This implies that the dequeue
41 * function must return the queue entry, not just the data.
43 * Therefore, the dummy entry must never be returned. We do this by
44 * re-enqueuing a new dummy entry after we dequeue one and then
45 * retrying the dequeue. We need more than one dummy because they
46 * must be hazardly freed.
54 #include <mono/utils/mono-membar.h>
55 #include <mono/utils/hazard-pointer.h>
56 #include <mono/utils/atomic.h>
58 #include <mono/utils/lock-free-queue.h>
60 #define INVALID_NEXT ((MonoLockFreeQueueNode *volatile)-1)
61 #define END_MARKER ((MonoLockFreeQueueNode *volatile)-2)
62 #define FREE_NEXT ((MonoLockFreeQueueNode *volatile)-3)
65 * Initialize a lock-free queue in-place at @q.
68 mono_lock_free_queue_init (MonoLockFreeQueue
*q
)
71 for (i
= 0; i
< MONO_LOCK_FREE_QUEUE_NUM_DUMMIES
; ++i
) {
72 q
->dummies
[i
].node
.next
= (i
== 0) ? END_MARKER
: FREE_NEXT
;
73 q
->dummies
[i
].in_use
= i
== 0 ? 1 : 0;
75 q
->dummies
[i
].node
.in_queue
= i
== 0 ? TRUE
: FALSE
;
79 q
->head
= q
->tail
= &q
->dummies
[0].node
;
84 * Initialize @node's state. If @poison is TRUE, @node may not be enqueued to a
85 * queue - @mono_lock_free_queue_node_unpoison must be called first; otherwise,
86 * the node can be enqueued right away.
88 * The poisoning feature is mainly intended for ensuring correctness in complex
89 * lock-free code that uses the queue. For example, in some code that reuses
90 * nodes, nodes can be poisoned when they're dequeued, and then unpoisoned and
91 * enqueued in their hazard free callback.
94 mono_lock_free_queue_node_init (MonoLockFreeQueueNode
*node
, gboolean poison
)
96 node
->next
= poison
? INVALID_NEXT
: FREE_NEXT
;
98 node
->in_queue
= FALSE
;
103 * Unpoisons @node so that it may be enqueued.
106 mono_lock_free_queue_node_unpoison (MonoLockFreeQueueNode
*node
)
108 g_assert (node
->next
== INVALID_NEXT
);
110 g_assert (!node
->in_queue
);
112 node
->next
= FREE_NEXT
;
116 * Enqueue @node to @q. @node must have been initialized by a prior call to
117 * @mono_lock_free_queue_node_init, and must not be in a poisoned state.
120 mono_lock_free_queue_enqueue (MonoLockFreeQueue
*q
, MonoLockFreeQueueNode
*node
)
122 MonoThreadHazardPointers
*hp
= mono_hazard_pointer_get ();
123 MonoLockFreeQueueNode
*tail
;
126 g_assert (!node
->in_queue
);
127 node
->in_queue
= TRUE
;
128 mono_memory_write_barrier ();
131 g_assert (node
->next
== FREE_NEXT
);
132 node
->next
= END_MARKER
;
134 MonoLockFreeQueueNode
*next
;
136 tail
= (MonoLockFreeQueueNode
*) mono_get_hazardous_pointer ((gpointer
volatile*)&q
->tail
, hp
, 0);
137 mono_memory_read_barrier ();
139 * We never dereference next so we don't need a
143 mono_memory_read_barrier ();
145 /* Are tail and next consistent? */
146 if (tail
== q
->tail
) {
147 g_assert (next
!= INVALID_NEXT
&& next
!= FREE_NEXT
);
148 g_assert (next
!= tail
);
150 if (next
== END_MARKER
) {
152 * Here we require that nodes that
153 * have been dequeued don't have
154 * next==END_MARKER. If they did, we
155 * might append to a node that isn't
156 * in the queue anymore here.
158 if (InterlockedCompareExchangePointer ((gpointer
volatile*)&tail
->next
, node
, END_MARKER
) == END_MARKER
)
161 /* Try to advance tail */
162 InterlockedCompareExchangePointer ((gpointer
volatile*)&q
->tail
, next
, tail
);
166 mono_memory_write_barrier ();
167 mono_hazard_pointer_clear (hp
, 0);
170 /* Try to advance tail */
171 InterlockedCompareExchangePointer ((gpointer
volatile*)&q
->tail
, node
, tail
);
173 mono_memory_write_barrier ();
174 mono_hazard_pointer_clear (hp
, 0);
178 free_dummy (gpointer _dummy
)
180 MonoLockFreeQueueDummy
*dummy
= (MonoLockFreeQueueDummy
*) _dummy
;
181 mono_lock_free_queue_node_unpoison (&dummy
->node
);
182 g_assert (dummy
->in_use
);
183 mono_memory_write_barrier ();
187 static MonoLockFreeQueueDummy
*
188 get_dummy (MonoLockFreeQueue
*q
)
191 for (i
= 0; i
< MONO_LOCK_FREE_QUEUE_NUM_DUMMIES
; ++i
) {
192 MonoLockFreeQueueDummy
*dummy
= &q
->dummies
[i
];
197 if (InterlockedCompareExchange (&dummy
->in_use
, 1, 0) == 0)
204 is_dummy (MonoLockFreeQueue
*q
, MonoLockFreeQueueNode
*n
)
206 return n
>= &q
->dummies
[0].node
&& n
<= &q
->dummies
[MONO_LOCK_FREE_QUEUE_NUM_DUMMIES
-1].node
;
210 try_reenqueue_dummy (MonoLockFreeQueue
*q
)
212 MonoLockFreeQueueDummy
*dummy
;
217 dummy
= get_dummy (q
);
221 if (InterlockedCompareExchange (&q
->has_dummy
, 1, 0) != 0) {
226 mono_lock_free_queue_enqueue (q
, &dummy
->node
);
232 * Dequeues a node from @q. Returns NULL if no nodes are available. The returned
233 * node is hazardous and must be freed with @mono_thread_hazardous_try_free or
234 * @mono_thread_hazardous_queue_free - it must not be freed directly.
236 MonoLockFreeQueueNode
*
237 mono_lock_free_queue_dequeue (MonoLockFreeQueue
*q
)
239 MonoThreadHazardPointers
*hp
= mono_hazard_pointer_get ();
240 MonoLockFreeQueueNode
*head
;
244 MonoLockFreeQueueNode
*tail
, *next
;
246 head
= (MonoLockFreeQueueNode
*) mono_get_hazardous_pointer ((gpointer
volatile*)&q
->head
, hp
, 0);
247 tail
= (MonoLockFreeQueueNode
*)q
->tail
;
248 mono_memory_read_barrier ();
250 mono_memory_read_barrier ();
252 /* Are head, tail and next consistent? */
253 if (head
== q
->head
) {
254 g_assert (next
!= INVALID_NEXT
&& next
!= FREE_NEXT
);
255 g_assert (next
!= head
);
257 /* Is queue empty or tail behind? */
259 if (next
== END_MARKER
) {
261 mono_hazard_pointer_clear (hp
, 0);
264 * We only continue if we
265 * reenqueue the dummy
266 * ourselves, so as not to
267 * wait for threads that might
270 if (!is_dummy (q
, head
) && try_reenqueue_dummy (q
))
276 /* Try to advance tail */
277 InterlockedCompareExchangePointer ((gpointer
volatile*)&q
->tail
, next
, tail
);
279 g_assert (next
!= END_MARKER
);
280 /* Try to dequeue head */
281 if (InterlockedCompareExchangePointer ((gpointer
volatile*)&q
->head
, next
, head
) == head
)
286 mono_memory_write_barrier ();
287 mono_hazard_pointer_clear (hp
, 0);
291 * The head is dequeued now, so we know it's this thread's
292 * responsibility to free it - no other thread can.
294 mono_memory_write_barrier ();
295 mono_hazard_pointer_clear (hp
, 0);
297 g_assert (head
->next
);
299 * Setting next here isn't necessary for correctness, but we
300 * do it to make sure that we catch dereferencing next in a
301 * node that's not in the queue anymore.
303 head
->next
= INVALID_NEXT
;
305 g_assert (head
->in_queue
);
306 head
->in_queue
= FALSE
;
307 mono_memory_write_barrier ();
310 if (is_dummy (q
, head
)) {
311 g_assert (q
->has_dummy
);
313 mono_memory_write_barrier ();
314 mono_thread_hazardous_try_free (head
, free_dummy
);
315 if (try_reenqueue_dummy (q
))
320 /* The caller must hazardously free the node. */