Fix: Add failover for platforms without nproc
[urcu.git] / urcu-bp.c
1 /*
2 * urcu-bp.c
3 *
4 * Userspace RCU library, "bulletproof" version.
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 <string.h>
34 #include <errno.h>
35 #include <poll.h>
36 #include <unistd.h>
37 #include <sys/mman.h>
38
39 #include "urcu/arch.h"
40 #include "urcu/wfcqueue.h"
41 #include "urcu/map/urcu-bp.h"
42 #include "urcu/static/urcu-bp.h"
43 #include "urcu-pointer.h"
44 #include "urcu/tls-compat.h"
45
46 #include "urcu-die.h"
47
48 /* Do not #define _LGPL_SOURCE to ensure we can emit the wrapper symbols */
49 #undef _LGPL_SOURCE
50 #include "urcu-bp.h"
51 #define _LGPL_SOURCE
52
53 #ifndef MAP_ANONYMOUS
54 #define MAP_ANONYMOUS MAP_ANON
55 #endif
56
57 #ifdef __linux__
58 static
59 void *mremap_wrapper(void *old_address, size_t old_size,
60 size_t new_size, int flags)
61 {
62 return mremap(old_address, old_size, new_size, flags);
63 }
64 #else
65
66 #define MREMAP_MAYMOVE 1
67 #define MREMAP_FIXED 2
68
69 /*
70 * mremap wrapper for non-Linux systems not allowing MAYMOVE.
71 * This is not generic.
72 */
73 static
74 void *mremap_wrapper(void *old_address, size_t old_size,
75 size_t new_size, int flags)
76 {
77 assert(!(flags & MREMAP_MAYMOVE));
78
79 return MAP_FAILED;
80 }
81 #endif
82
83 /* Sleep delay in ms */
84 #define RCU_SLEEP_DELAY_MS 10
85 #define INIT_NR_THREADS 8
86 #define ARENA_INIT_ALLOC \
87 sizeof(struct registry_chunk) \
88 + INIT_NR_THREADS * sizeof(struct rcu_reader)
89
90 /*
91 * Active attempts to check for reader Q.S. before calling sleep().
92 */
93 #define RCU_QS_ACTIVE_ATTEMPTS 100
94
95 static
96 int rcu_bp_refcount;
97
98 /* If the headers do not support membarrier system call, fall back smp_mb. */
99 #ifdef __NR_membarrier
100 # define membarrier(...) syscall(__NR_membarrier, __VA_ARGS__)
101 #else
102 # define membarrier(...) -ENOSYS
103 #endif
104
105 enum membarrier_cmd {
106 MEMBARRIER_CMD_QUERY = 0,
107 MEMBARRIER_CMD_SHARED = (1 << 0),
108 };
109
110 static
111 void __attribute__((constructor)) rcu_bp_init(void);
112 static
113 void __attribute__((destructor)) rcu_bp_exit(void);
114
115 int urcu_bp_has_sys_membarrier;
116
117 /*
118 * rcu_gp_lock ensures mutual exclusion between threads calling
119 * synchronize_rcu().
120 */
121 static pthread_mutex_t rcu_gp_lock = PTHREAD_MUTEX_INITIALIZER;
122 /*
123 * rcu_registry_lock ensures mutual exclusion between threads
124 * registering and unregistering themselves to/from the registry, and
125 * with threads reading that registry from synchronize_rcu(). However,
126 * this lock is not held all the way through the completion of awaiting
127 * for the grace period. It is sporadically released between iterations
128 * on the registry.
129 * rcu_registry_lock may nest inside rcu_gp_lock.
130 */
131 static pthread_mutex_t rcu_registry_lock = PTHREAD_MUTEX_INITIALIZER;
132
133 static pthread_mutex_t init_lock = PTHREAD_MUTEX_INITIALIZER;
134 static int initialized;
135
136 static pthread_key_t urcu_bp_key;
137
138 struct rcu_gp rcu_gp = { .ctr = RCU_GP_COUNT };
139
140 /*
141 * Pointer to registry elements. Written to only by each individual reader. Read
142 * by both the reader and the writers.
143 */
144 DEFINE_URCU_TLS(struct rcu_reader *, rcu_reader);
145
146 static CDS_LIST_HEAD(registry);
147
148 struct registry_chunk {
149 size_t data_len; /* data length */
150 size_t used; /* amount of data used */
151 struct cds_list_head node; /* chunk_list node */
152 char data[];
153 };
154
155 struct registry_arena {
156 struct cds_list_head chunk_list;
157 };
158
159 static struct registry_arena registry_arena = {
160 .chunk_list = CDS_LIST_HEAD_INIT(registry_arena.chunk_list),
161 };
162
163 /* Saved fork signal mask, protected by rcu_gp_lock */
164 static sigset_t saved_fork_signal_mask;
165
166 static void mutex_lock(pthread_mutex_t *mutex)
167 {
168 int ret;
169
170 #ifndef DISTRUST_SIGNALS_EXTREME
171 ret = pthread_mutex_lock(mutex);
172 if (ret)
173 urcu_die(ret);
174 #else /* #ifndef DISTRUST_SIGNALS_EXTREME */
175 while ((ret = pthread_mutex_trylock(mutex)) != 0) {
176 if (ret != EBUSY && ret != EINTR)
177 urcu_die(ret);
178 poll(NULL,0,10);
179 }
180 #endif /* #else #ifndef DISTRUST_SIGNALS_EXTREME */
181 }
182
183 static void mutex_unlock(pthread_mutex_t *mutex)
184 {
185 int ret;
186
187 ret = pthread_mutex_unlock(mutex);
188 if (ret)
189 urcu_die(ret);
190 }
191
192 static void smp_mb_master(void)
193 {
194 if (caa_likely(urcu_bp_has_sys_membarrier))
195 (void) membarrier(MEMBARRIER_CMD_SHARED, 0);
196 else
197 cmm_smp_mb();
198 }
199
200 /*
201 * Always called with rcu_registry lock held. Releases this lock between
202 * iterations and grabs it again. Holds the lock when it returns.
203 */
204 static void wait_for_readers(struct cds_list_head *input_readers,
205 struct cds_list_head *cur_snap_readers,
206 struct cds_list_head *qsreaders)
207 {
208 unsigned int wait_loops = 0;
209 struct rcu_reader *index, *tmp;
210
211 /*
212 * Wait for each thread URCU_TLS(rcu_reader).ctr to either
213 * indicate quiescence (not nested), or observe the current
214 * rcu_gp.ctr value.
215 */
216 for (;;) {
217 if (wait_loops < RCU_QS_ACTIVE_ATTEMPTS)
218 wait_loops++;
219
220 cds_list_for_each_entry_safe(index, tmp, input_readers, node) {
221 switch (rcu_reader_state(&index->ctr)) {
222 case RCU_READER_ACTIVE_CURRENT:
223 if (cur_snap_readers) {
224 cds_list_move(&index->node,
225 cur_snap_readers);
226 break;
227 }
228 /* Fall-through */
229 case RCU_READER_INACTIVE:
230 cds_list_move(&index->node, qsreaders);
231 break;
232 case RCU_READER_ACTIVE_OLD:
233 /*
234 * Old snapshot. Leaving node in
235 * input_readers will make us busy-loop
236 * until the snapshot becomes current or
237 * the reader becomes inactive.
238 */
239 break;
240 }
241 }
242
243 if (cds_list_empty(input_readers)) {
244 break;
245 } else {
246 /* Temporarily unlock the registry lock. */
247 mutex_unlock(&rcu_registry_lock);
248 if (wait_loops >= RCU_QS_ACTIVE_ATTEMPTS)
249 (void) poll(NULL, 0, RCU_SLEEP_DELAY_MS);
250 else
251 caa_cpu_relax();
252 /* Re-lock the registry lock before the next loop. */
253 mutex_lock(&rcu_registry_lock);
254 }
255 }
256 }
257
258 void synchronize_rcu(void)
259 {
260 CDS_LIST_HEAD(cur_snap_readers);
261 CDS_LIST_HEAD(qsreaders);
262 sigset_t newmask, oldmask;
263 int ret;
264
265 ret = sigfillset(&newmask);
266 assert(!ret);
267 ret = pthread_sigmask(SIG_BLOCK, &newmask, &oldmask);
268 assert(!ret);
269
270 mutex_lock(&rcu_gp_lock);
271
272 mutex_lock(&rcu_registry_lock);
273
274 if (cds_list_empty(&registry))
275 goto out;
276
277 /* All threads should read qparity before accessing data structure
278 * where new ptr points to. */
279 /* Write new ptr before changing the qparity */
280 smp_mb_master();
281
282 /*
283 * Wait for readers to observe original parity or be quiescent.
284 * wait_for_readers() can release and grab again rcu_registry_lock
285 * interally.
286 */
287 wait_for_readers(&registry, &cur_snap_readers, &qsreaders);
288
289 /*
290 * Adding a cmm_smp_mb() which is _not_ formally required, but makes the
291 * model easier to understand. It does not have a big performance impact
292 * anyway, given this is the write-side.
293 */
294 cmm_smp_mb();
295
296 /* Switch parity: 0 -> 1, 1 -> 0 */
297 CMM_STORE_SHARED(rcu_gp.ctr, rcu_gp.ctr ^ RCU_GP_CTR_PHASE);
298
299 /*
300 * Must commit qparity update to memory before waiting for other parity
301 * quiescent state. Failure to do so could result in the writer waiting
302 * forever while new readers are always accessing data (no progress).
303 * Ensured by CMM_STORE_SHARED and CMM_LOAD_SHARED.
304 */
305
306 /*
307 * Adding a cmm_smp_mb() which is _not_ formally required, but makes the
308 * model easier to understand. It does not have a big performance impact
309 * anyway, given this is the write-side.
310 */
311 cmm_smp_mb();
312
313 /*
314 * Wait for readers to observe new parity or be quiescent.
315 * wait_for_readers() can release and grab again rcu_registry_lock
316 * interally.
317 */
318 wait_for_readers(&cur_snap_readers, NULL, &qsreaders);
319
320 /*
321 * Put quiescent reader list back into registry.
322 */
323 cds_list_splice(&qsreaders, &registry);
324
325 /*
326 * Finish waiting for reader threads before letting the old ptr being
327 * freed.
328 */
329 smp_mb_master();
330 out:
331 mutex_unlock(&rcu_registry_lock);
332 mutex_unlock(&rcu_gp_lock);
333 ret = pthread_sigmask(SIG_SETMASK, &oldmask, NULL);
334 assert(!ret);
335 }
336
337 /*
338 * library wrappers to be used by non-LGPL compatible source code.
339 */
340
341 void rcu_read_lock(void)
342 {
343 _rcu_read_lock();
344 }
345
346 void rcu_read_unlock(void)
347 {
348 _rcu_read_unlock();
349 }
350
351 int rcu_read_ongoing(void)
352 {
353 return _rcu_read_ongoing();
354 }
355
356 /*
357 * Only grow for now. If empty, allocate a ARENA_INIT_ALLOC sized chunk.
358 * Else, try expanding the last chunk. If this fails, allocate a new
359 * chunk twice as big as the last chunk.
360 * Memory used by chunks _never_ moves. A chunk could theoretically be
361 * freed when all "used" slots are released, but we don't do it at this
362 * point.
363 */
364 static
365 void expand_arena(struct registry_arena *arena)
366 {
367 struct registry_chunk *new_chunk, *last_chunk;
368 size_t old_chunk_len, new_chunk_len;
369
370 /* No chunk. */
371 if (cds_list_empty(&arena->chunk_list)) {
372 assert(ARENA_INIT_ALLOC >=
373 sizeof(struct registry_chunk)
374 + sizeof(struct rcu_reader));
375 new_chunk_len = ARENA_INIT_ALLOC;
376 new_chunk = mmap(NULL, new_chunk_len,
377 PROT_READ | PROT_WRITE,
378 MAP_ANONYMOUS | MAP_PRIVATE,
379 -1, 0);
380 if (new_chunk == MAP_FAILED)
381 abort();
382 memset(new_chunk, 0, new_chunk_len);
383 new_chunk->data_len =
384 new_chunk_len - sizeof(struct registry_chunk);
385 cds_list_add_tail(&new_chunk->node, &arena->chunk_list);
386 return; /* We're done. */
387 }
388
389 /* Try expanding last chunk. */
390 last_chunk = cds_list_entry(arena->chunk_list.prev,
391 struct registry_chunk, node);
392 old_chunk_len =
393 last_chunk->data_len + sizeof(struct registry_chunk);
394 new_chunk_len = old_chunk_len << 1;
395
396 /* Don't allow memory mapping to move, just expand. */
397 new_chunk = mremap_wrapper(last_chunk, old_chunk_len,
398 new_chunk_len, 0);
399 if (new_chunk != MAP_FAILED) {
400 /* Should not have moved. */
401 assert(new_chunk == last_chunk);
402 memset((char *) last_chunk + old_chunk_len, 0,
403 new_chunk_len - old_chunk_len);
404 last_chunk->data_len =
405 new_chunk_len - sizeof(struct registry_chunk);
406 return; /* We're done. */
407 }
408
409 /* Remap did not succeed, we need to add a new chunk. */
410 new_chunk = mmap(NULL, new_chunk_len,
411 PROT_READ | PROT_WRITE,
412 MAP_ANONYMOUS | MAP_PRIVATE,
413 -1, 0);
414 if (new_chunk == MAP_FAILED)
415 abort();
416 memset(new_chunk, 0, new_chunk_len);
417 new_chunk->data_len =
418 new_chunk_len - sizeof(struct registry_chunk);
419 cds_list_add_tail(&new_chunk->node, &arena->chunk_list);
420 }
421
422 static
423 struct rcu_reader *arena_alloc(struct registry_arena *arena)
424 {
425 struct registry_chunk *chunk;
426 struct rcu_reader *rcu_reader_reg;
427 int expand_done = 0; /* Only allow to expand once per alloc */
428 size_t len = sizeof(struct rcu_reader);
429
430 retry:
431 cds_list_for_each_entry(chunk, &arena->chunk_list, node) {
432 if (chunk->data_len - chunk->used < len)
433 continue;
434 /* Find spot */
435 for (rcu_reader_reg = (struct rcu_reader *) &chunk->data[0];
436 rcu_reader_reg < (struct rcu_reader *) &chunk->data[chunk->data_len];
437 rcu_reader_reg++) {
438 if (!rcu_reader_reg->alloc) {
439 rcu_reader_reg->alloc = 1;
440 chunk->used += len;
441 return rcu_reader_reg;
442 }
443 }
444 }
445
446 if (!expand_done) {
447 expand_arena(arena);
448 expand_done = 1;
449 goto retry;
450 }
451
452 return NULL;
453 }
454
455 /* Called with signals off and mutex locked */
456 static
457 void add_thread(void)
458 {
459 struct rcu_reader *rcu_reader_reg;
460 int ret;
461
462 rcu_reader_reg = arena_alloc(&registry_arena);
463 if (!rcu_reader_reg)
464 abort();
465 ret = pthread_setspecific(urcu_bp_key, rcu_reader_reg);
466 if (ret)
467 abort();
468
469 /* Add to registry */
470 rcu_reader_reg->tid = pthread_self();
471 assert(rcu_reader_reg->ctr == 0);
472 cds_list_add(&rcu_reader_reg->node, &registry);
473 /*
474 * Reader threads are pointing to the reader registry. This is
475 * why its memory should never be relocated.
476 */
477 URCU_TLS(rcu_reader) = rcu_reader_reg;
478 }
479
480 /* Called with mutex locked */
481 static
482 void cleanup_thread(struct registry_chunk *chunk,
483 struct rcu_reader *rcu_reader_reg)
484 {
485 rcu_reader_reg->ctr = 0;
486 cds_list_del(&rcu_reader_reg->node);
487 rcu_reader_reg->tid = 0;
488 rcu_reader_reg->alloc = 0;
489 chunk->used -= sizeof(struct rcu_reader);
490 }
491
492 static
493 struct registry_chunk *find_chunk(struct rcu_reader *rcu_reader_reg)
494 {
495 struct registry_chunk *chunk;
496
497 cds_list_for_each_entry(chunk, &registry_arena.chunk_list, node) {
498 if (rcu_reader_reg < (struct rcu_reader *) &chunk->data[0])
499 continue;
500 if (rcu_reader_reg >= (struct rcu_reader *) &chunk->data[chunk->data_len])
501 continue;
502 return chunk;
503 }
504 return NULL;
505 }
506
507 /* Called with signals off and mutex locked */
508 static
509 void remove_thread(struct rcu_reader *rcu_reader_reg)
510 {
511 cleanup_thread(find_chunk(rcu_reader_reg), rcu_reader_reg);
512 URCU_TLS(rcu_reader) = NULL;
513 }
514
515 /* Disable signals, take mutex, add to registry */
516 void rcu_bp_register(void)
517 {
518 sigset_t newmask, oldmask;
519 int ret;
520
521 ret = sigfillset(&newmask);
522 if (ret)
523 abort();
524 ret = pthread_sigmask(SIG_BLOCK, &newmask, &oldmask);
525 if (ret)
526 abort();
527
528 /*
529 * Check if a signal concurrently registered our thread since
530 * the check in rcu_read_lock().
531 */
532 if (URCU_TLS(rcu_reader))
533 goto end;
534
535 /*
536 * Take care of early registration before urcu_bp constructor.
537 */
538 rcu_bp_init();
539
540 mutex_lock(&rcu_registry_lock);
541 add_thread();
542 mutex_unlock(&rcu_registry_lock);
543 end:
544 ret = pthread_sigmask(SIG_SETMASK, &oldmask, NULL);
545 if (ret)
546 abort();
547 }
548
549 /* Disable signals, take mutex, remove from registry */
550 static
551 void rcu_bp_unregister(struct rcu_reader *rcu_reader_reg)
552 {
553 sigset_t newmask, oldmask;
554 int ret;
555
556 ret = sigfillset(&newmask);
557 if (ret)
558 abort();
559 ret = pthread_sigmask(SIG_BLOCK, &newmask, &oldmask);
560 if (ret)
561 abort();
562
563 mutex_lock(&rcu_registry_lock);
564 remove_thread(rcu_reader_reg);
565 mutex_unlock(&rcu_registry_lock);
566 ret = pthread_sigmask(SIG_SETMASK, &oldmask, NULL);
567 if (ret)
568 abort();
569 rcu_bp_exit();
570 }
571
572 /*
573 * Remove thread from the registry when it exits, and flag it as
574 * destroyed so garbage collection can take care of it.
575 */
576 static
577 void urcu_bp_thread_exit_notifier(void *rcu_key)
578 {
579 rcu_bp_unregister(rcu_key);
580 }
581
582 static
583 void rcu_bp_init(void)
584 {
585 mutex_lock(&init_lock);
586 if (!rcu_bp_refcount++) {
587 int ret;
588
589 ret = pthread_key_create(&urcu_bp_key,
590 urcu_bp_thread_exit_notifier);
591 if (ret)
592 abort();
593 ret = membarrier(MEMBARRIER_CMD_QUERY, 0);
594 if (ret >= 0 && (ret & MEMBARRIER_CMD_SHARED)) {
595 urcu_bp_has_sys_membarrier = 1;
596 }
597 initialized = 1;
598 }
599 mutex_unlock(&init_lock);
600 }
601
602 static
603 void rcu_bp_exit(void)
604 {
605 mutex_lock(&init_lock);
606 if (!--rcu_bp_refcount) {
607 struct registry_chunk *chunk, *tmp;
608 int ret;
609
610 cds_list_for_each_entry_safe(chunk, tmp,
611 &registry_arena.chunk_list, node) {
612 munmap(chunk, chunk->data_len
613 + sizeof(struct registry_chunk));
614 }
615 CDS_INIT_LIST_HEAD(&registry_arena.chunk_list);
616 ret = pthread_key_delete(urcu_bp_key);
617 if (ret)
618 abort();
619 }
620 mutex_unlock(&init_lock);
621 }
622
623 /*
624 * Holding the rcu_gp_lock and rcu_registry_lock across fork will make
625 * sure we fork() don't race with a concurrent thread executing with
626 * any of those locks held. This ensures that the registry and data
627 * protected by rcu_gp_lock are in a coherent state in the child.
628 */
629 void rcu_bp_before_fork(void)
630 {
631 sigset_t newmask, oldmask;
632 int ret;
633
634 ret = sigfillset(&newmask);
635 assert(!ret);
636 ret = pthread_sigmask(SIG_BLOCK, &newmask, &oldmask);
637 assert(!ret);
638 mutex_lock(&rcu_gp_lock);
639 mutex_lock(&rcu_registry_lock);
640 saved_fork_signal_mask = oldmask;
641 }
642
643 void rcu_bp_after_fork_parent(void)
644 {
645 sigset_t oldmask;
646 int ret;
647
648 oldmask = saved_fork_signal_mask;
649 mutex_unlock(&rcu_registry_lock);
650 mutex_unlock(&rcu_gp_lock);
651 ret = pthread_sigmask(SIG_SETMASK, &oldmask, NULL);
652 assert(!ret);
653 }
654
655 /*
656 * Prune all entries from registry except our own thread. Fits the Linux
657 * fork behavior. Called with rcu_gp_lock and rcu_registry_lock held.
658 */
659 static
660 void urcu_bp_prune_registry(void)
661 {
662 struct registry_chunk *chunk;
663 struct rcu_reader *rcu_reader_reg;
664
665 cds_list_for_each_entry(chunk, &registry_arena.chunk_list, node) {
666 for (rcu_reader_reg = (struct rcu_reader *) &chunk->data[0];
667 rcu_reader_reg < (struct rcu_reader *) &chunk->data[chunk->data_len];
668 rcu_reader_reg++) {
669 if (!rcu_reader_reg->alloc)
670 continue;
671 if (rcu_reader_reg->tid == pthread_self())
672 continue;
673 cleanup_thread(chunk, rcu_reader_reg);
674 }
675 }
676 }
677
678 void rcu_bp_after_fork_child(void)
679 {
680 sigset_t oldmask;
681 int ret;
682
683 urcu_bp_prune_registry();
684 oldmask = saved_fork_signal_mask;
685 mutex_unlock(&rcu_registry_lock);
686 mutex_unlock(&rcu_gp_lock);
687 ret = pthread_sigmask(SIG_SETMASK, &oldmask, NULL);
688 assert(!ret);
689 }
690
691 void *rcu_dereference_sym_bp(void *p)
692 {
693 return _rcu_dereference(p);
694 }
695
696 void *rcu_set_pointer_sym_bp(void **p, void *v)
697 {
698 cmm_wmb();
699 uatomic_set(p, v);
700 return v;
701 }
702
703 void *rcu_xchg_pointer_sym_bp(void **p, void *v)
704 {
705 cmm_wmb();
706 return uatomic_xchg(p, v);
707 }
708
709 void *rcu_cmpxchg_pointer_sym_bp(void **p, void *old, void *_new)
710 {
711 cmm_wmb();
712 return uatomic_cmpxchg(p, old, _new);
713 }
714
715 DEFINE_RCU_FLAVOR(rcu_flavor);
716
717 #include "urcu-call-rcu-impl.h"
718 #include "urcu-defer-impl.h"
This page took 0.043404 seconds and 4 git commands to generate.