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