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