b83862f26c21c400484aa8f43f4c39c0516ba709
[urcu.git] / urcu / static / rculfqueue.h
1 #ifndef _URCU_RCULFQUEUE_STATIC_H
2 #define _URCU_RCULFQUEUE_STATIC_H
3
4 /*
5 * rculfqueue-static.h
6 *
7 * Userspace RCU library - Lock-Free RCU Queue
8 *
9 * Copyright 2010 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
10 *
11 * TO BE INCLUDED ONLY IN LGPL-COMPATIBLE CODE. See rculfqueue.h for linking
12 * dynamically with the userspace rcu library.
13 *
14 * This library is free software; you can redistribute it and/or
15 * modify it under the terms of the GNU Lesser General Public
16 * License as published by the Free Software Foundation; either
17 * version 2.1 of the License, or (at your option) any later version.
18 *
19 * This library is distributed in the hope that it will be useful,
20 * but WITHOUT ANY WARRANTY; without even the implied warranty of
21 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
22 * Lesser General Public License for more details.
23 *
24 * You should have received a copy of the GNU Lesser General Public
25 * License along with this library; if not, write to the Free Software
26 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
27 */
28
29 #include <urcu-call-rcu.h>
30 #include <urcu/uatomic.h>
31 #include <assert.h>
32 #include <errno.h>
33 /* A urcu implementation header should be already included. */
34
35 #ifdef __cplusplus
36 extern "C" {
37 #endif
38
39 struct cds_lfq_node_rcu_dummy {
40 struct cds_lfq_node_rcu parent;
41 struct rcu_head head;
42 struct cds_lfq_queue_rcu *q;
43 };
44
45 /*
46 * Lock-free RCU queue. Enqueue and dequeue operations hold a RCU read
47 * lock to deal with cmpxchg ABA problem. This queue is *not* circular:
48 * head points to the oldest node, tail points to the newest node.
49 * Dummy nodes are kept to ensure enqueue and dequeue can always proceed
50 * concurrently. Keeping a separate head and tail helps with large
51 * queues: enqueue and dequeue can proceed concurrently without
52 * wrestling for exclusive access to the same variables.
53 *
54 * We keep two dummy nodes in the queue to distinguish between empty queue
55 * state and intermediate state while a dummy node dequeue/requeue is
56 * being performed. Dequeue retry if it detects that it would be
57 * dequeueing the last node (it means a dummy node dequeue-requeue is in
58 * progress). This ensures that there is always at least one node in
59 * the queue. In a situation where the two dummy nodes are being
60 * requeued (they therefore don't appear in the queue at a given
61 * moment), we are certain that there is at least one non-dummy node in
62 * the queue (ensured by the test for NULL next node upon dequeue).
63 *
64 * In the dequeue operation, we internally reallocate the dummy nodes
65 * upon dequeue/requeue and use call_rcu to free them after a grace
66 * period.
67 */
68
69 static inline
70 int is_dummy(struct cds_lfq_node_rcu *node)
71 {
72 return ((unsigned long) node) & 0x1UL;
73 }
74
75 static inline
76 struct cds_lfq_node_rcu *make_dummy(struct cds_lfq_queue_rcu *q,
77 struct cds_lfq_node_rcu *next)
78 {
79 struct cds_lfq_node_rcu_dummy *dummy;
80
81 dummy = malloc(sizeof(struct cds_lfq_node_rcu_dummy));
82 dummy->parent.next = next;
83 dummy->q = q;
84 return (struct cds_lfq_node_rcu *) (((unsigned long) &dummy->parent) | 0x1UL);
85 }
86
87 static inline
88 struct cds_lfq_node_rcu *get_node(struct cds_lfq_node_rcu *node)
89 {
90 return (struct cds_lfq_node_rcu *) (((unsigned long )node) & ~0x1UL);
91 }
92
93 static inline
94 void free_dummy(struct rcu_head *head)
95 {
96 struct cds_lfq_node_rcu_dummy *dummy =
97 caa_container_of(head, struct cds_lfq_node_rcu_dummy, head);
98 free(dummy);
99 }
100
101 static inline
102 void rcu_free_dummy(struct cds_lfq_node_rcu *node)
103 {
104 struct cds_lfq_node_rcu_dummy *dummy;
105
106 dummy = caa_container_of(get_node(node), struct cds_lfq_node_rcu_dummy,
107 parent);
108 dummy->q->queue_call_rcu(&dummy->head, free_dummy);
109 }
110
111 static inline
112 void _cds_lfq_node_init_rcu(struct cds_lfq_node_rcu *node)
113 {
114 node->next = NULL;
115 }
116
117 static inline
118 void _cds_lfq_init_rcu(struct cds_lfq_queue_rcu *q,
119 void queue_call_rcu(struct rcu_head *head,
120 void (*func)(struct rcu_head *head)))
121 {
122 q->tail = make_dummy(q, NULL);
123 q->head = make_dummy(q, q->tail);
124 q->queue_call_rcu = queue_call_rcu;
125 }
126
127 /*
128 * The queue should be emptied before calling destroy.
129 *
130 * Return 0 on success, -EPERM if queue is not empty.
131 */
132 static inline
133 int _cds_lfq_destroy_rcu(struct cds_lfq_queue_rcu *q)
134 {
135 struct cds_lfq_node_rcu *head, *next;
136
137 head = rcu_dereference(q->head);
138 next = rcu_dereference(get_node(head)->next);
139 if (!(is_dummy(head) && is_dummy(next) && get_node(next)->next == NULL))
140 return -EPERM; /* not empty */
141 rcu_free_dummy(head);
142 rcu_free_dummy(next);
143 return 0;
144 }
145
146 /*
147 * Should be called under rcu read lock critical section.
148 */
149 static inline
150 void _cds_lfq_enqueue_rcu(struct cds_lfq_queue_rcu *q,
151 struct cds_lfq_node_rcu *node)
152 {
153 /*
154 * uatomic_cmpxchg() implicit memory barrier orders earlier stores to
155 * node before publication.
156 */
157
158 for (;;) {
159 struct cds_lfq_node_rcu *tail, *next;
160
161 tail = rcu_dereference(q->tail);
162 next = uatomic_cmpxchg(&get_node(tail)->next, NULL, node);
163 if (next == NULL) {
164 /*
165 * Tail was at the end of queue, we successfully
166 * appended to it. Now move tail (another
167 * enqueue might beat us to it, that's fine).
168 */
169 (void) uatomic_cmpxchg(&q->tail, tail, node);
170 return;
171 } else {
172 /*
173 * Failure to append to current tail.
174 * Help moving tail further and retry.
175 */
176 (void) uatomic_cmpxchg(&q->tail, tail, next);
177 continue;
178 }
179 }
180 }
181
182 /*
183 * Should be called under rcu read lock critical section.
184 *
185 * The caller must wait for a grace period to pass before freeing the returned
186 * node or modifying the cds_lfq_node_rcu structure.
187 * Returns NULL if queue is empty.
188 */
189 static inline
190 struct cds_lfq_node_rcu *_cds_lfq_dequeue_rcu(struct cds_lfq_queue_rcu *q)
191 {
192 for (;;) {
193 struct cds_lfq_node_rcu *head, *next;
194
195 head = rcu_dereference(q->head);
196 next = rcu_dereference(get_node(head)->next);
197 if (is_dummy(head) && is_dummy(next) && get_node(next)->next == NULL)
198 return NULL; /* empty */
199 /*
200 * We never, ever allow dequeue to get to a state where
201 * the queue is empty (we need at least one node in the
202 * queue). This is ensured by checking if the head next
203 * is NULL and retry in that case (this means a
204 * concurrent dummy node re-enqueue is in progress).
205 */
206 if (next) {
207 if (uatomic_cmpxchg(&q->head, head, next) == head) {
208 if (is_dummy(head)) {
209 struct cds_lfq_node_rcu *node;
210 /*
211 * Requeue dummy. We need to
212 * reallocate to protect from
213 * ABA.
214 */
215 rcu_free_dummy(head);
216 node = make_dummy(q, NULL);
217 _cds_lfq_enqueue_rcu(q, node);
218 continue; /* try again */
219 }
220 return head;
221 } else {
222 /* Concurrently pushed, retry */
223 continue;
224 }
225 } else {
226 /* Dummy node re-enqueue is in progress, retry. */
227 continue;
228 }
229 }
230 }
231
232 #ifdef __cplusplus
233 }
234 #endif
235
236 #endif /* _URCU_RCULFQUEUE_STATIC_H */
This page took 0.034257 seconds and 4 git commands to generate.