a35dcfc5aea3674e9747ef69daea0f49def7a972
[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/wfcqueue.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 #include "urcu/tls-compat.h"
44
45 #include "urcu-die.h"
46 #include "urcu-wait.h"
47
48 /* Do not #define _LGPL_SOURCE to ensure we can emit the wrapper symbols */
49 #undef _LGPL_SOURCE
50 #include "urcu-qsbr.h"
51 #define _LGPL_SOURCE
52
53 void __attribute__((destructor)) rcu_exit(void);
54
55 static pthread_mutex_t rcu_gp_lock = PTHREAD_MUTEX_INITIALIZER;
56 struct rcu_gp rcu_gp = { .ctr = RCU_GP_ONLINE };
57
58 /*
59 * Active attempts to check for reader Q.S. before calling futex().
60 */
61 #define RCU_QS_ACTIVE_ATTEMPTS 100
62
63 /*
64 * Written to only by each individual reader. Read by both the reader and the
65 * writers.
66 */
67 DEFINE_URCU_TLS(struct rcu_reader, rcu_reader);
68
69 static CDS_LIST_HEAD(registry);
70
71 /*
72 * Queue keeping threads awaiting to wait for a grace period. Contains
73 * struct gp_waiters_thread objects.
74 */
75 static DEFINE_URCU_WAIT_QUEUE(gp_waiters);
76
77 static void mutex_lock(pthread_mutex_t *mutex)
78 {
79 int ret;
80
81 #ifndef DISTRUST_SIGNALS_EXTREME
82 ret = pthread_mutex_lock(mutex);
83 if (ret)
84 urcu_die(ret);
85 #else /* #ifndef DISTRUST_SIGNALS_EXTREME */
86 while ((ret = pthread_mutex_trylock(mutex)) != 0) {
87 if (ret != EBUSY && ret != EINTR)
88 urcu_die(ret);
89 poll(NULL,0,10);
90 }
91 #endif /* #else #ifndef DISTRUST_SIGNALS_EXTREME */
92 }
93
94 static void mutex_unlock(pthread_mutex_t *mutex)
95 {
96 int ret;
97
98 ret = pthread_mutex_unlock(mutex);
99 if (ret)
100 urcu_die(ret);
101 }
102
103 /*
104 * synchronize_rcu() waiting. Single thread.
105 */
106 static void wait_gp(void)
107 {
108 /* Read reader_gp before read futex */
109 cmm_smp_rmb();
110 if (uatomic_read(&rcu_gp.futex) == -1)
111 futex_noasync(&rcu_gp.futex, FUTEX_WAIT, -1,
112 NULL, NULL, 0);
113 }
114
115 static void wait_for_readers(struct cds_list_head *input_readers,
116 struct cds_list_head *cur_snap_readers,
117 struct cds_list_head *qsreaders)
118 {
119 unsigned int wait_loops = 0;
120 struct rcu_reader *index, *tmp;
121
122 /*
123 * Wait for each thread URCU_TLS(rcu_reader).ctr to either
124 * indicate quiescence (offline), or for them to observe the
125 * current rcu_gp.ctr value.
126 */
127 for (;;) {
128 if (wait_loops < RCU_QS_ACTIVE_ATTEMPTS)
129 wait_loops++;
130 if (wait_loops >= RCU_QS_ACTIVE_ATTEMPTS) {
131 uatomic_set(&rcu_gp.futex, -1);
132 /*
133 * Write futex before write waiting (the other side
134 * reads them in the opposite order).
135 */
136 cmm_smp_wmb();
137 cds_list_for_each_entry(index, input_readers, node) {
138 _CMM_STORE_SHARED(index->waiting, 1);
139 }
140 /* Write futex before read reader_gp */
141 cmm_smp_mb();
142 }
143 cds_list_for_each_entry_safe(index, tmp, input_readers, node) {
144 switch (rcu_reader_state(&index->ctr)) {
145 case RCU_READER_ACTIVE_CURRENT:
146 if (cur_snap_readers) {
147 cds_list_move(&index->node,
148 cur_snap_readers);
149 break;
150 }
151 /* Fall-through */
152 case RCU_READER_INACTIVE:
153 cds_list_move(&index->node, qsreaders);
154 break;
155 case RCU_READER_ACTIVE_OLD:
156 /*
157 * Old snapshot. Leaving node in
158 * input_readers will make us busy-loop
159 * until the snapshot becomes current or
160 * the reader becomes inactive.
161 */
162 break;
163 }
164 }
165
166 if (cds_list_empty(input_readers)) {
167 if (wait_loops >= RCU_QS_ACTIVE_ATTEMPTS) {
168 /* Read reader_gp before write futex */
169 cmm_smp_mb();
170 uatomic_set(&rcu_gp.futex, 0);
171 }
172 break;
173 } else {
174 if (wait_loops >= RCU_QS_ACTIVE_ATTEMPTS) {
175 wait_gp();
176 } else {
177 #ifndef HAS_INCOHERENT_CACHES
178 caa_cpu_relax();
179 #else /* #ifndef HAS_INCOHERENT_CACHES */
180 cmm_smp_mb();
181 #endif /* #else #ifndef HAS_INCOHERENT_CACHES */
182 }
183 }
184 }
185 }
186
187 /*
188 * Using a two-subphases algorithm for architectures with smaller than 64-bit
189 * long-size to ensure we do not encounter an overflow bug.
190 */
191
192 #if (CAA_BITS_PER_LONG < 64)
193 void synchronize_rcu(void)
194 {
195 CDS_LIST_HEAD(cur_snap_readers);
196 CDS_LIST_HEAD(qsreaders);
197 unsigned long was_online;
198 DEFINE_URCU_WAIT_NODE(wait, URCU_WAIT_WAITING);
199 struct urcu_waiters waiters;
200
201 was_online = rcu_read_ongoing();
202
203 /* All threads should read qparity before accessing data structure
204 * where new ptr points to. In the "then" case, rcu_thread_offline
205 * includes a memory barrier.
206 *
207 * Mark the writer thread offline to make sure we don't wait for
208 * our own quiescent state. This allows using synchronize_rcu()
209 * in threads registered as readers.
210 */
211 if (was_online)
212 rcu_thread_offline();
213 else
214 cmm_smp_mb();
215
216 /*
217 * Add ourself to gp_waiters queue of threads awaiting to wait
218 * for a grace period. Proceed to perform the grace period only
219 * if we are the first thread added into the queue.
220 */
221 if (urcu_wait_add(&gp_waiters, &wait) != 0) {
222 /* Not first in queue: will be awakened by another thread. */
223 urcu_adaptative_busy_wait(&wait);
224 goto gp_end;
225 }
226 /* We won't need to wake ourself up */
227 urcu_wait_set_state(&wait, URCU_WAIT_RUNNING);
228
229 mutex_lock(&rcu_gp_lock);
230
231 /*
232 * Move all waiters into our local queue.
233 */
234 urcu_move_waiters(&waiters, &gp_waiters);
235
236 if (cds_list_empty(&registry))
237 goto out;
238
239 /*
240 * Wait for readers to observe original parity or be quiescent.
241 */
242 wait_for_readers(&registry, &cur_snap_readers, &qsreaders);
243
244 /*
245 * Must finish waiting for quiescent state for original parity
246 * before committing next rcu_gp.ctr update to memory. Failure
247 * to do so could result in the writer waiting forever while new
248 * readers are always accessing data (no progress). Enforce
249 * compiler-order of load URCU_TLS(rcu_reader).ctr before store
250 * to rcu_gp.ctr.
251 */
252 cmm_barrier();
253
254 /*
255 * Adding a cmm_smp_mb() which is _not_ formally required, but makes the
256 * model easier to understand. It does not have a big performance impact
257 * anyway, given this is the write-side.
258 */
259 cmm_smp_mb();
260
261 /* Switch parity: 0 -> 1, 1 -> 0 */
262 CMM_STORE_SHARED(rcu_gp.ctr, rcu_gp.ctr ^ RCU_GP_CTR);
263
264 /*
265 * Must commit rcu_gp.ctr update to memory before waiting for
266 * quiescent state. Failure to do so could result in the writer
267 * waiting forever while new readers are always accessing data
268 * (no progress). Enforce compiler-order of store to rcu_gp.ctr
269 * before load URCU_TLS(rcu_reader).ctr.
270 */
271 cmm_barrier();
272
273 /*
274 * Adding a cmm_smp_mb() which is _not_ formally required, but makes the
275 * model easier to understand. It does not have a big performance impact
276 * anyway, given this is the write-side.
277 */
278 cmm_smp_mb();
279
280 /*
281 * Wait for readers to observe new parity or be quiescent.
282 */
283 wait_for_readers(&cur_snap_readers, NULL, &qsreaders);
284
285 /*
286 * Put quiescent reader list back into registry.
287 */
288 cds_list_splice(&qsreaders, &registry);
289 out:
290 mutex_unlock(&rcu_gp_lock);
291 urcu_wake_all_waiters(&waiters);
292 gp_end:
293 /*
294 * Finish waiting for reader threads before letting the old ptr being
295 * freed.
296 */
297 if (was_online)
298 rcu_thread_online();
299 else
300 cmm_smp_mb();
301 }
302 #else /* !(CAA_BITS_PER_LONG < 64) */
303 void synchronize_rcu(void)
304 {
305 CDS_LIST_HEAD(qsreaders);
306 unsigned long was_online;
307 DEFINE_URCU_WAIT_NODE(wait, URCU_WAIT_WAITING);
308 struct urcu_waiters waiters;
309
310 was_online = rcu_read_ongoing();
311
312 /*
313 * Mark the writer thread offline to make sure we don't wait for
314 * our own quiescent state. This allows using synchronize_rcu()
315 * in threads registered as readers.
316 */
317 if (was_online)
318 rcu_thread_offline();
319 else
320 cmm_smp_mb();
321
322 /*
323 * Add ourself to gp_waiters queue of threads awaiting to wait
324 * for a grace period. Proceed to perform the grace period only
325 * if we are the first thread added into the queue.
326 */
327 if (urcu_wait_add(&gp_waiters, &wait) != 0) {
328 /* Not first in queue: will be awakened by another thread. */
329 urcu_adaptative_busy_wait(&wait);
330 goto gp_end;
331 }
332 /* We won't need to wake ourself up */
333 urcu_wait_set_state(&wait, URCU_WAIT_RUNNING);
334
335 mutex_lock(&rcu_gp_lock);
336
337 /*
338 * Move all waiters into our local queue.
339 */
340 urcu_move_waiters(&waiters, &gp_waiters);
341
342 if (cds_list_empty(&registry))
343 goto out;
344
345 /* Increment current G.P. */
346 CMM_STORE_SHARED(rcu_gp.ctr, rcu_gp.ctr + RCU_GP_CTR);
347
348 /*
349 * Must commit rcu_gp.ctr update to memory before waiting for
350 * quiescent state. Failure to do so could result in the writer
351 * waiting forever while new readers are always accessing data
352 * (no progress). Enforce compiler-order of store to rcu_gp.ctr
353 * before load URCU_TLS(rcu_reader).ctr.
354 */
355 cmm_barrier();
356
357 /*
358 * Adding a cmm_smp_mb() which is _not_ formally required, but makes the
359 * model easier to understand. It does not have a big performance impact
360 * anyway, given this is the write-side.
361 */
362 cmm_smp_mb();
363
364 /*
365 * Wait for readers to observe new count of be quiescent.
366 */
367 wait_for_readers(&registry, NULL, &qsreaders);
368
369 /*
370 * Put quiescent reader list back into registry.
371 */
372 cds_list_splice(&qsreaders, &registry);
373 out:
374 mutex_unlock(&rcu_gp_lock);
375 urcu_wake_all_waiters(&waiters);
376 gp_end:
377 if (was_online)
378 rcu_thread_online();
379 else
380 cmm_smp_mb();
381 }
382 #endif /* !(CAA_BITS_PER_LONG < 64) */
383
384 /*
385 * library wrappers to be used by non-LGPL compatible source code.
386 */
387
388 void rcu_read_lock(void)
389 {
390 _rcu_read_lock();
391 }
392
393 void rcu_read_unlock(void)
394 {
395 _rcu_read_unlock();
396 }
397
398 int rcu_read_ongoing(void)
399 {
400 return _rcu_read_ongoing();
401 }
402
403 void rcu_quiescent_state(void)
404 {
405 _rcu_quiescent_state();
406 }
407
408 void rcu_thread_offline(void)
409 {
410 _rcu_thread_offline();
411 }
412
413 void rcu_thread_online(void)
414 {
415 _rcu_thread_online();
416 }
417
418 void rcu_register_thread(void)
419 {
420 URCU_TLS(rcu_reader).tid = pthread_self();
421 assert(URCU_TLS(rcu_reader).ctr == 0);
422
423 mutex_lock(&rcu_gp_lock);
424 cds_list_add(&URCU_TLS(rcu_reader).node, &registry);
425 mutex_unlock(&rcu_gp_lock);
426 _rcu_thread_online();
427 }
428
429 void rcu_unregister_thread(void)
430 {
431 /*
432 * We have to make the thread offline otherwise we end up dealocking
433 * with a waiting writer.
434 */
435 _rcu_thread_offline();
436 mutex_lock(&rcu_gp_lock);
437 cds_list_del(&URCU_TLS(rcu_reader).node);
438 mutex_unlock(&rcu_gp_lock);
439 }
440
441 void rcu_exit(void)
442 {
443 /*
444 * Assertion disabled because call_rcu threads are now rcu
445 * readers, and left running at exit.
446 * assert(cds_list_empty(&registry));
447 */
448 }
449
450 DEFINE_RCU_FLAVOR(rcu_flavor);
451
452 #include "urcu-call-rcu-impl.h"
453 #include "urcu-defer-impl.h"
This page took 0.05822 seconds and 4 git commands to generate.