Fix missing check for SYS_membarrier in map header
[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
34 #ifdef __cplusplus
35 extern "C" {
36 #endif
37
38 struct cds_lfq_node_rcu_dummy {
39 struct cds_lfq_node_rcu parent;
40 struct rcu_head head;
41 struct cds_lfq_queue_rcu *q;
42 };
43
44 /*
45 * Lock-free RCU queue. Enqueue and dequeue operations hold a RCU read
46 * lock to deal with cmpxchg ABA problem. This queue is *not* circular:
47 * head points to the oldest node, tail points to the newest node.
48 * A dummy node is kept to ensure enqueue and dequeue can always proceed
49 * concurrently. Keeping a separate head and tail helps with large
50 * queues: enqueue and dequeue can proceed concurrently without
51 * wrestling for exclusive access to the same variables.
52 *
53 * Dequeue retry if it detects that it would be dequeueing the last node
54 * (it means a dummy node dequeue-requeue is in progress). This ensures
55 * that there is always at least one node in the queue.
56 *
57 * In the dequeue operation, we internally reallocate the dummy node
58 * upon dequeue/requeue and use call_rcu to free the old one after a
59 * grace period.
60 */
61
62 static inline
63 struct cds_lfq_node_rcu *make_dummy(struct cds_lfq_queue_rcu *q,
64 struct cds_lfq_node_rcu *next)
65 {
66 struct cds_lfq_node_rcu_dummy *dummy;
67
68 dummy = malloc(sizeof(struct cds_lfq_node_rcu_dummy));
69 assert(dummy);
70 dummy->parent.next = next;
71 dummy->parent.dummy = 1;
72 dummy->q = q;
73 return &dummy->parent;
74 }
75
76 static inline
77 void free_dummy_cb(struct rcu_head *head)
78 {
79 struct cds_lfq_node_rcu_dummy *dummy =
80 caa_container_of(head, struct cds_lfq_node_rcu_dummy, head);
81 free(dummy);
82 }
83
84 static inline
85 void rcu_free_dummy(struct cds_lfq_node_rcu *node)
86 {
87 struct cds_lfq_node_rcu_dummy *dummy;
88
89 assert(node->dummy);
90 dummy = caa_container_of(node, struct cds_lfq_node_rcu_dummy, parent);
91 dummy->q->queue_call_rcu(&dummy->head, free_dummy_cb);
92 }
93
94 static inline
95 void free_dummy(struct cds_lfq_node_rcu *node)
96 {
97 struct cds_lfq_node_rcu_dummy *dummy;
98
99 assert(node->dummy);
100 dummy = caa_container_of(node, struct cds_lfq_node_rcu_dummy, parent);
101 free(dummy);
102 }
103
104 static inline
105 void _cds_lfq_node_init_rcu(struct cds_lfq_node_rcu *node)
106 {
107 node->next = NULL;
108 node->dummy = 0;
109 }
110
111 static inline
112 void _cds_lfq_init_rcu(struct cds_lfq_queue_rcu *q,
113 void queue_call_rcu(struct rcu_head *head,
114 void (*func)(struct rcu_head *head)))
115 {
116 q->tail = make_dummy(q, NULL);
117 q->head = q->tail;
118 q->queue_call_rcu = queue_call_rcu;
119 }
120
121 /*
122 * The queue should be emptied before calling destroy.
123 *
124 * Return 0 on success, -EPERM if queue is not empty.
125 */
126 static inline
127 int _cds_lfq_destroy_rcu(struct cds_lfq_queue_rcu *q)
128 {
129 struct cds_lfq_node_rcu *head;
130
131 head = rcu_dereference(q->head);
132 if (!(head->dummy && head->next == NULL))
133 return -EPERM; /* not empty */
134 free_dummy(head);
135 return 0;
136 }
137
138 /*
139 * Should be called under rcu read lock critical section.
140 */
141 static inline
142 void _cds_lfq_enqueue_rcu(struct cds_lfq_queue_rcu *q,
143 struct cds_lfq_node_rcu *node)
144 {
145 /*
146 * uatomic_cmpxchg() implicit memory barrier orders earlier stores to
147 * node before publication.
148 */
149
150 for (;;) {
151 struct cds_lfq_node_rcu *tail, *next;
152
153 tail = rcu_dereference(q->tail);
154 next = uatomic_cmpxchg(&tail->next, NULL, node);
155 if (next == NULL) {
156 /*
157 * Tail was at the end of queue, we successfully
158 * appended to it. Now move tail (another
159 * enqueue might beat us to it, that's fine).
160 */
161 (void) uatomic_cmpxchg(&q->tail, tail, node);
162 return;
163 } else {
164 /*
165 * Failure to append to current tail.
166 * Help moving tail further and retry.
167 */
168 (void) uatomic_cmpxchg(&q->tail, tail, next);
169 continue;
170 }
171 }
172 }
173
174 static inline
175 void enqueue_dummy(struct cds_lfq_queue_rcu *q)
176 {
177 struct cds_lfq_node_rcu *node;
178
179 /* We need to reallocate to protect from ABA. */
180 node = make_dummy(q, NULL);
181 _cds_lfq_enqueue_rcu(q, node);
182 }
183
184 /*
185 * Should be called under rcu read lock critical section.
186 *
187 * The caller must wait for a grace period to pass before freeing the returned
188 * node or modifying the cds_lfq_node_rcu structure.
189 * Returns NULL if queue is empty.
190 */
191 static inline
192 struct cds_lfq_node_rcu *_cds_lfq_dequeue_rcu(struct cds_lfq_queue_rcu *q)
193 {
194 for (;;) {
195 struct cds_lfq_node_rcu *head, *next;
196
197 head = rcu_dereference(q->head);
198 next = rcu_dereference(head->next);
199 if (head->dummy && next == NULL)
200 return NULL; /* empty */
201 /*
202 * We never, ever allow dequeue to get to a state where
203 * the queue is empty (we need at least one node in the
204 * queue). This is ensured by checking if the head next
205 * is NULL, which means we need to enqueue a dummy node
206 * before we can hope dequeuing anything.
207 */
208 if (!next) {
209 enqueue_dummy(q);
210 next = rcu_dereference(head->next);
211 }
212 if (uatomic_cmpxchg(&q->head, head, next) != head)
213 continue; /* Concurrently pushed. */
214 if (head->dummy) {
215 /* Free dummy after grace period. */
216 rcu_free_dummy(head);
217 continue; /* try again */
218 }
219 return head;
220 }
221 }
222
223 #ifdef __cplusplus
224 }
225 #endif
226
227 #endif /* _URCU_RCULFQUEUE_STATIC_H */
This page took 0.034184 seconds and 4 git commands to generate.