Add rcu_flavor
[urcu.git] / urcu-qsbr.c
1 /*
2 * urcu-qsbr.c
3 *
4 * Userspace RCU QSBR library
5 *
6 * Copyright (c) 2009 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
7 * Copyright (c) 2009 Paul E. McKenney, IBM Corporation.
8 *
9 * This library is free software; you can redistribute it and/or
10 * modify it under the terms of the GNU Lesser General Public
11 * License as published by the Free Software Foundation; either
12 * version 2.1 of the License, or (at your option) any later version.
13 *
14 * This library is distributed in the hope that it will be useful,
15 * but WITHOUT ANY WARRANTY; without even the implied warranty of
16 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
17 * Lesser General Public License for more details.
18 *
19 * You should have received a copy of the GNU Lesser General Public
20 * License along with this library; if not, write to the Free Software
21 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
22 *
23 * IBM's contributions to this file may be relicensed under LGPLv2 or later.
24 */
25
26 #define _GNU_SOURCE
27 #define _LGPL_SOURCE
28 #include <stdio.h>
29 #include <pthread.h>
30 #include <signal.h>
31 #include <assert.h>
32 #include <stdlib.h>
33 #include <stdint.h>
34 #include <string.h>
35 #include <errno.h>
36 #include <poll.h>
37
38 #include "urcu/wfqueue.h"
39 #include "urcu/map/urcu-qsbr.h"
40 #define BUILD_QSBR_LIB
41 #include "urcu/static/urcu-qsbr.h"
42 #include "urcu-pointer.h"
43
44 /* Do not #define _LGPL_SOURCE to ensure we can emit the wrapper symbols */
45 #undef _LGPL_SOURCE
46 #include "urcu-qsbr.h"
47 #define _LGPL_SOURCE
48
49 void __attribute__((destructor)) rcu_exit(void);
50
51 static pthread_mutex_t rcu_gp_lock = PTHREAD_MUTEX_INITIALIZER;
52
53 int32_t gp_futex;
54
55 /*
56 * Global grace period counter.
57 */
58 unsigned long rcu_gp_ctr = RCU_GP_ONLINE;
59
60 /*
61 * Active attempts to check for reader Q.S. before calling futex().
62 */
63 #define RCU_QS_ACTIVE_ATTEMPTS 100
64
65 /*
66 * Written to only by each individual reader. Read by both the reader and the
67 * writers.
68 */
69 struct rcu_reader __thread rcu_reader;
70
71 #ifdef DEBUG_YIELD
72 unsigned int yield_active;
73 unsigned int __thread rand_yield;
74 #endif
75
76 static CDS_LIST_HEAD(registry);
77
78 static void mutex_lock(pthread_mutex_t *mutex)
79 {
80 int ret;
81
82 #ifndef DISTRUST_SIGNALS_EXTREME
83 ret = pthread_mutex_lock(mutex);
84 if (ret) {
85 perror("Error in pthread mutex lock");
86 exit(-1);
87 }
88 #else /* #ifndef DISTRUST_SIGNALS_EXTREME */
89 while ((ret = pthread_mutex_trylock(mutex)) != 0) {
90 if (ret != EBUSY && ret != EINTR) {
91 printf("ret = %d, errno = %d\n", ret, errno);
92 perror("Error in pthread mutex lock");
93 exit(-1);
94 }
95 poll(NULL,0,10);
96 }
97 #endif /* #else #ifndef DISTRUST_SIGNALS_EXTREME */
98 }
99
100 static void mutex_unlock(pthread_mutex_t *mutex)
101 {
102 int ret;
103
104 ret = pthread_mutex_unlock(mutex);
105 if (ret) {
106 perror("Error in pthread mutex unlock");
107 exit(-1);
108 }
109 }
110
111 /*
112 * synchronize_rcu() waiting. Single thread.
113 */
114 static void wait_gp(void)
115 {
116 /* Read reader_gp before read futex */
117 cmm_smp_rmb();
118 if (uatomic_read(&gp_futex) == -1)
119 futex_noasync(&gp_futex, FUTEX_WAIT, -1,
120 NULL, NULL, 0);
121 }
122
123 static void update_counter_and_wait(void)
124 {
125 CDS_LIST_HEAD(qsreaders);
126 int wait_loops = 0;
127 struct rcu_reader *index, *tmp;
128
129 #if (CAA_BITS_PER_LONG < 64)
130 /* Switch parity: 0 -> 1, 1 -> 0 */
131 CMM_STORE_SHARED(rcu_gp_ctr, rcu_gp_ctr ^ RCU_GP_CTR);
132 #else /* !(CAA_BITS_PER_LONG < 64) */
133 /* Increment current G.P. */
134 CMM_STORE_SHARED(rcu_gp_ctr, rcu_gp_ctr + RCU_GP_CTR);
135 #endif /* !(CAA_BITS_PER_LONG < 64) */
136
137 /*
138 * Must commit rcu_gp_ctr update to memory before waiting for
139 * quiescent state. Failure to do so could result in the writer
140 * waiting forever while new readers are always accessing data
141 * (no progress). Enforce compiler-order of store to rcu_gp_ctr
142 * before load rcu_reader ctr.
143 */
144 cmm_barrier();
145
146 /*
147 * Adding a cmm_smp_mb() which is _not_ formally required, but makes the
148 * model easier to understand. It does not have a big performance impact
149 * anyway, given this is the write-side.
150 */
151 cmm_smp_mb();
152
153 /*
154 * Wait for each thread rcu_reader_qs_gp count to become 0.
155 */
156 for (;;) {
157 wait_loops++;
158 if (wait_loops >= RCU_QS_ACTIVE_ATTEMPTS) {
159 uatomic_set(&gp_futex, -1);
160 /*
161 * Write futex before write waiting (the other side
162 * reads them in the opposite order).
163 */
164 cmm_smp_wmb();
165 cds_list_for_each_entry(index, &registry, node) {
166 _CMM_STORE_SHARED(index->waiting, 1);
167 }
168 /* Write futex before read reader_gp */
169 cmm_smp_mb();
170 }
171 cds_list_for_each_entry_safe(index, tmp, &registry, node) {
172 if (!rcu_gp_ongoing(&index->ctr))
173 cds_list_move(&index->node, &qsreaders);
174 }
175
176 if (cds_list_empty(&registry)) {
177 if (wait_loops >= RCU_QS_ACTIVE_ATTEMPTS) {
178 /* Read reader_gp before write futex */
179 cmm_smp_mb();
180 uatomic_set(&gp_futex, 0);
181 }
182 break;
183 } else {
184 if (wait_loops >= RCU_QS_ACTIVE_ATTEMPTS) {
185 wait_gp();
186 } else {
187 #ifndef HAS_INCOHERENT_CACHES
188 caa_cpu_relax();
189 #else /* #ifndef HAS_INCOHERENT_CACHES */
190 cmm_smp_mb();
191 #endif /* #else #ifndef HAS_INCOHERENT_CACHES */
192 }
193 }
194 }
195 /* put back the reader list in the registry */
196 cds_list_splice(&qsreaders, &registry);
197 }
198
199 /*
200 * Using a two-subphases algorithm for architectures with smaller than 64-bit
201 * long-size to ensure we do not encounter an overflow bug.
202 */
203
204 #if (CAA_BITS_PER_LONG < 64)
205 void synchronize_rcu(void)
206 {
207 unsigned long was_online;
208
209 was_online = rcu_reader.ctr;
210
211 /* All threads should read qparity before accessing data structure
212 * where new ptr points to. In the "then" case, rcu_thread_offline
213 * includes a memory barrier.
214 *
215 * Mark the writer thread offline to make sure we don't wait for
216 * our own quiescent state. This allows using synchronize_rcu()
217 * in threads registered as readers.
218 */
219 if (was_online)
220 rcu_thread_offline();
221 else
222 cmm_smp_mb();
223
224 mutex_lock(&rcu_gp_lock);
225
226 if (cds_list_empty(&registry))
227 goto out;
228
229 /*
230 * Wait for previous parity to be empty of readers.
231 */
232 update_counter_and_wait(); /* 0 -> 1, wait readers in parity 0 */
233
234 /*
235 * Must finish waiting for quiescent state for parity 0 before
236 * committing next rcu_gp_ctr update to memory. Failure to
237 * do so could result in the writer waiting forever while new
238 * readers are always accessing data (no progress). Enforce
239 * compiler-order of load rcu_reader ctr before store to
240 * rcu_gp_ctr.
241 */
242 cmm_barrier();
243
244 /*
245 * Adding a cmm_smp_mb() which is _not_ formally required, but makes the
246 * model easier to understand. It does not have a big performance impact
247 * anyway, given this is the write-side.
248 */
249 cmm_smp_mb();
250
251 /*
252 * Wait for previous parity to be empty of readers.
253 */
254 update_counter_and_wait(); /* 1 -> 0, wait readers in parity 1 */
255 out:
256 mutex_unlock(&rcu_gp_lock);
257
258 /*
259 * Finish waiting for reader threads before letting the old ptr being
260 * freed.
261 */
262 if (was_online)
263 rcu_thread_online();
264 else
265 cmm_smp_mb();
266 }
267 #else /* !(CAA_BITS_PER_LONG < 64) */
268 void synchronize_rcu(void)
269 {
270 unsigned long was_online;
271
272 was_online = rcu_reader.ctr;
273
274 /*
275 * Mark the writer thread offline to make sure we don't wait for
276 * our own quiescent state. This allows using synchronize_rcu()
277 * in threads registered as readers.
278 */
279 if (was_online)
280 rcu_thread_offline();
281 else
282 cmm_smp_mb();
283
284 mutex_lock(&rcu_gp_lock);
285 if (cds_list_empty(&registry))
286 goto out;
287 update_counter_and_wait();
288 out:
289 mutex_unlock(&rcu_gp_lock);
290
291 if (was_online)
292 rcu_thread_online();
293 else
294 cmm_smp_mb();
295 }
296 #endif /* !(CAA_BITS_PER_LONG < 64) */
297
298 /*
299 * library wrappers to be used by non-LGPL compatible source code.
300 */
301
302 void rcu_read_lock(void)
303 {
304 _rcu_read_lock();
305 }
306
307 void rcu_read_unlock(void)
308 {
309 _rcu_read_unlock();
310 }
311
312 void rcu_quiescent_state(void)
313 {
314 _rcu_quiescent_state();
315 }
316
317 void rcu_thread_offline(void)
318 {
319 _rcu_thread_offline();
320 }
321
322 void rcu_thread_online(void)
323 {
324 _rcu_thread_online();
325 }
326
327 void rcu_register_thread(void)
328 {
329 rcu_reader.tid = pthread_self();
330 assert(rcu_reader.ctr == 0);
331
332 mutex_lock(&rcu_gp_lock);
333 cds_list_add(&rcu_reader.node, &registry);
334 mutex_unlock(&rcu_gp_lock);
335 _rcu_thread_online();
336 }
337
338 void rcu_unregister_thread(void)
339 {
340 /*
341 * We have to make the thread offline otherwise we end up dealocking
342 * with a waiting writer.
343 */
344 _rcu_thread_offline();
345 mutex_lock(&rcu_gp_lock);
346 cds_list_del(&rcu_reader.node);
347 mutex_unlock(&rcu_gp_lock);
348 }
349
350 void rcu_exit(void)
351 {
352 /*
353 * Assertion disabled because call_rcu threads are now rcu
354 * readers, and left running at exit.
355 * assert(cds_list_empty(&registry));
356 */
357 }
358
359 DEFINE_RCU_FLAVOR()
360
361 #include "urcu-call-rcu-impl.h"
362 #include "urcu-defer-impl.h"
This page took 0.045244 seconds and 4 git commands to generate.