src/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 _LGPL_SOURCE
  27 #include <stdio.h>
  28 #include <pthread.h>
  29 #include <signal.h>
  30 #include <assert.h>
  31 #include <stdlib.h>
  32 #include <stdint.h>
  33 #include <string.h>
  34 #include <errno.h>
  35 #include <poll.h>
  36
  37 #include <urcu/wfcqueue.h>
  38 #include <urcu/map/urcu-qsbr.h>
  39 #define BUILD_QSBR_LIB
  40 #include <urcu/static/urcu-qsbr.h>
  41 #include <urcu/pointer.h>
  42 #include <urcu/tls-compat.h>
  43
  44 #include "urcu-die.h"
  45 #include "urcu-wait.h"
  46 #include "urcu-utils.h"
  47
  48 #define URCU_API_MAP
  49 /* Do not #define _LGPL_SOURCE to ensure we can emit the wrapper symbols */
  50 #undef _LGPL_SOURCE
  51 #include <urcu/urcu-qsbr.h>
  52 #define _LGPL_SOURCE
  53
  54 void __attribute__((destructor)) urcu_qsbr_exit(void);
  55
  56 /*
  57  * rcu_gp_lock ensures mutual exclusion between threads calling
  58  * synchronize_rcu().
  59  */
  60 static pthread_mutex_t rcu_gp_lock = PTHREAD_MUTEX_INITIALIZER;
  61 /*
  62  * rcu_registry_lock ensures mutual exclusion between threads
  63  * registering and unregistering themselves to/from the registry, and
  64  * with threads reading that registry from synchronize_rcu(). However,
  65  * this lock is not held all the way through the completion of awaiting
  66  * for the grace period. It is sporadically released between iterations
  67  * on the registry.
  68  * rcu_registry_lock may nest inside rcu_gp_lock.
  69  */
  70 static pthread_mutex_t rcu_registry_lock = PTHREAD_MUTEX_INITIALIZER;
  71 struct urcu_gp urcu_qsbr_gp = { .ctr = URCU_QSBR_GP_ONLINE };
  72 URCU_ATTR_ALIAS("urcu_qsbr_gp") extern struct urcu_gp rcu_gp_qsbr;
  73
  74 /*
  75  * Active attempts to check for reader Q.S. before calling futex().
  76  */
  77 #define RCU_QS_ACTIVE_ATTEMPTS 100
  78
  79 /*
  80  * Written to only by each individual reader. Read by both the reader and the
  81  * writers.
  82  */
  83 DEFINE_URCU_TLS(struct urcu_qsbr_reader, urcu_qsbr_reader);
  84 DEFINE_URCU_TLS_ALIAS(struct urcu_qsbr_reader, urcu_qsbr_reader, rcu_reader_qsbr);
  85
  86 static CDS_LIST_HEAD(registry);
  87
  88 /*
  89  * Queue keeping threads awaiting to wait for a grace period. Contains
  90  * struct gp_waiters_thread objects.
  91  */
  92 static DEFINE_URCU_WAIT_QUEUE(gp_waiters);
  93
  94 static void mutex_lock(pthread_mutex_t *mutex)
  95 {
  96         int ret;
  97
  98 #ifndef DISTRUST_SIGNALS_EXTREME
  99         ret = pthread_mutex_lock(mutex);
 100         if (ret)
 101                 urcu_die(ret);
 102 #else /* #ifndef DISTRUST_SIGNALS_EXTREME */
 103         while ((ret = pthread_mutex_trylock(mutex)) != 0) {
 104                 if (ret != EBUSY && ret != EINTR)
 105                         urcu_die(ret);
 106                 poll(NULL,0,10);
 107         }
 108 #endif /* #else #ifndef DISTRUST_SIGNALS_EXTREME */
 109 }
 110
 111 static void mutex_unlock(pthread_mutex_t *mutex)
 112 {
 113         int ret;
 114
 115         ret = pthread_mutex_unlock(mutex);
 116         if (ret)
 117                 urcu_die(ret);
 118 }
 119
 120 /*
 121  * synchronize_rcu() waiting. Single thread.
 122  */
 123 static void wait_gp(void)
 124 {
 125         /* Read reader_gp before read futex */
 126         cmm_smp_rmb();
 127         if (uatomic_read(&urcu_qsbr_gp.futex) != -1)
 128                 return;
 129         while (futex_noasync(&urcu_qsbr_gp.futex, FUTEX_WAIT, -1,
 130                         NULL, NULL, 0)) {
 131                 switch (errno) {
 132                 case EWOULDBLOCK:
 133                         /* Value already changed. */
 134                         return;
 135                 case EINTR:
 136                         /* Retry if interrupted by signal. */
 137                         break;  /* Get out of switch. */
 138                 default:
 139                         /* Unexpected error. */
 140                         urcu_die(errno);
 141                 }
 142         }
 143 }
 144
 145 /*
 146  * Always called with rcu_registry lock held. Releases this lock between
 147  * iterations and grabs it again. Holds the lock when it returns.
 148  */
 149 static void wait_for_readers(struct cds_list_head *input_readers,
 150                         struct cds_list_head *cur_snap_readers,
 151                         struct cds_list_head *qsreaders)
 152 {
 153         unsigned int wait_loops = 0;
 154         struct urcu_qsbr_reader *index, *tmp;
 155
 156         /*
 157          * Wait for each thread URCU_TLS(urcu_qsbr_reader).ctr to either
 158          * indicate quiescence (offline), or for them to observe the
 159          * current urcu_qsbr_gp.ctr value.
 160          */
 161         for (;;) {
 162                 if (wait_loops < RCU_QS_ACTIVE_ATTEMPTS)
 163                         wait_loops++;
 164                 if (wait_loops >= RCU_QS_ACTIVE_ATTEMPTS) {
 165                         uatomic_set(&urcu_qsbr_gp.futex, -1);
 166                         /*
 167                          * Write futex before write waiting (the other side
 168                          * reads them in the opposite order).
 169                          */
 170                         cmm_smp_wmb();
 171                         cds_list_for_each_entry(index, input_readers, node) {
 172                                 _CMM_STORE_SHARED(index->waiting, 1);
 173                         }
 174                         /* Write futex before read reader_gp */
 175                         cmm_smp_mb();
 176                 }
 177                 cds_list_for_each_entry_safe(index, tmp, input_readers, node) {
 178                         switch (urcu_qsbr_reader_state(&index->ctr)) {
 179                         case URCU_READER_ACTIVE_CURRENT:
 180                                 if (cur_snap_readers) {
 181                                         cds_list_move(&index->node,
 182                                                 cur_snap_readers);
 183                                         break;
 184                                 }
 185                                 /* Fall-through */
 186                         case URCU_READER_INACTIVE:
 187                                 cds_list_move(&index->node, qsreaders);
 188                                 break;
 189                         case URCU_READER_ACTIVE_OLD:
 190                                 /*
 191                                  * Old snapshot. Leaving node in
 192                                  * input_readers will make us busy-loop
 193                                  * until the snapshot becomes current or
 194                                  * the reader becomes inactive.
 195                                  */
 196                                 break;
 197                         }
 198                 }
 199
 200                 if (cds_list_empty(input_readers)) {
 201                         if (wait_loops >= RCU_QS_ACTIVE_ATTEMPTS) {
 202                                 /* Read reader_gp before write futex */
 203                                 cmm_smp_mb();
 204                                 uatomic_set(&urcu_qsbr_gp.futex, 0);
 205                         }
 206                         break;
 207                 } else {
 208                         /* Temporarily unlock the registry lock. */
 209                         mutex_unlock(&rcu_registry_lock);
 210                         if (wait_loops >= RCU_QS_ACTIVE_ATTEMPTS) {
 211                                 wait_gp();
 212                         } else {
 213 #ifndef HAS_INCOHERENT_CACHES
 214                                 caa_cpu_relax();
 215 #else /* #ifndef HAS_INCOHERENT_CACHES */
 216                                 cmm_smp_mb();
 217 #endif /* #else #ifndef HAS_INCOHERENT_CACHES */
 218                         }
 219                         /* Re-lock the registry lock before the next loop. */
 220                         mutex_lock(&rcu_registry_lock);
 221                 }
 222         }
 223 }
 224
 225 /*
 226  * Using a two-subphases algorithm for architectures with smaller than 64-bit
 227  * long-size to ensure we do not encounter an overflow bug.
 228  */
 229
 230 #if (CAA_BITS_PER_LONG < 64)
 231 void urcu_qsbr_synchronize_rcu(void)
 232 {
 233         CDS_LIST_HEAD(cur_snap_readers);
 234         CDS_LIST_HEAD(qsreaders);
 235         unsigned long was_online;
 236         DEFINE_URCU_WAIT_NODE(wait, URCU_WAIT_WAITING);
 237         struct urcu_waiters waiters;
 238
 239         was_online = urcu_qsbr_read_ongoing();
 240
 241         /* All threads should read qparity before accessing data structure
 242          * where new ptr points to.  In the "then" case, rcu_thread_offline
 243          * includes a memory barrier.
 244          *
 245          * Mark the writer thread offline to make sure we don't wait for
 246          * our own quiescent state. This allows using synchronize_rcu()
 247          * in threads registered as readers.
 248          */
 249         if (was_online)
 250                 urcu_qsbr_thread_offline();
 251         else
 252                 cmm_smp_mb();
 253
 254         /*
 255          * Add ourself to gp_waiters queue of threads awaiting to wait
 256          * for a grace period. Proceed to perform the grace period only
 257          * if we are the first thread added into the queue.
 258          */
 259         if (urcu_wait_add(&gp_waiters, &wait) != 0) {
 260                 /* Not first in queue: will be awakened by another thread. */
 261                 urcu_adaptative_busy_wait(&wait);
 262                 goto gp_end;
 263         }
 264         /* We won't need to wake ourself up */
 265         urcu_wait_set_state(&wait, URCU_WAIT_RUNNING);
 266
 267         mutex_lock(&rcu_gp_lock);
 268
 269         /*
 270          * Move all waiters into our local queue.
 271          */
 272         urcu_move_waiters(&waiters, &gp_waiters);
 273
 274         mutex_lock(&rcu_registry_lock);
 275
 276         if (cds_list_empty(&registry))
 277                 goto out;
 278
 279         /*
 280          * Wait for readers to observe original parity or be quiescent.
 281          * wait_for_readers() can release and grab again rcu_registry_lock
 282          * interally.
 283          */
 284         wait_for_readers(&registry, &cur_snap_readers, &qsreaders);
 285
 286         /*
 287          * Must finish waiting for quiescent state for original parity
 288          * before committing next urcu_qsbr_gp.ctr update to memory. Failure
 289          * to do so could result in the writer waiting forever while new
 290          * readers are always accessing data (no progress).  Enforce
 291          * compiler-order of load URCU_TLS(urcu_qsbr_reader).ctr before store
 292          * to urcu_qsbr_gp.ctr.
 293          */
 294         cmm_barrier();
 295
 296         /*
 297          * Adding a cmm_smp_mb() which is _not_ formally required, but makes the
 298          * model easier to understand. It does not have a big performance impact
 299          * anyway, given this is the write-side.
 300          */
 301         cmm_smp_mb();
 302
 303         /* Switch parity: 0 -> 1, 1 -> 0 */
 304         CMM_STORE_SHARED(urcu_qsbr_gp.ctr, urcu_qsbr_gp.ctr ^ URCU_QSBR_GP_CTR);
 305
 306         /*
 307          * Must commit urcu_qsbr_gp.ctr update to memory before waiting for
 308          * quiescent state. Failure to do so could result in the writer
 309          * waiting forever while new readers are always accessing data
 310          * (no progress). Enforce compiler-order of store to urcu_qsbr_gp.ctr
 311          * before load URCU_TLS(urcu_qsbr_reader).ctr.
 312          */
 313         cmm_barrier();
 314
 315         /*
 316          * Adding a cmm_smp_mb() which is _not_ formally required, but makes the
 317          * model easier to understand. It does not have a big performance impact
 318          * anyway, given this is the write-side.
 319          */
 320         cmm_smp_mb();
 321
 322         /*
 323          * Wait for readers to observe new parity or be quiescent.
 324          * wait_for_readers() can release and grab again rcu_registry_lock
 325          * interally.
 326          */
 327         wait_for_readers(&cur_snap_readers, NULL, &qsreaders);
 328
 329         /*
 330          * Put quiescent reader list back into registry.
 331          */
 332         cds_list_splice(&qsreaders, &registry);
 333 out:
 334         mutex_unlock(&rcu_registry_lock);
 335         mutex_unlock(&rcu_gp_lock);
 336         urcu_wake_all_waiters(&waiters);
 337 gp_end:
 338         /*
 339          * Finish waiting for reader threads before letting the old ptr being
 340          * freed.
 341          */
 342         if (was_online)
 343                 urcu_qsbr_thread_online();
 344         else
 345                 cmm_smp_mb();
 346 }
 347 #else /* !(CAA_BITS_PER_LONG < 64) */
 348 void urcu_qsbr_synchronize_rcu(void)
 349 {
 350         CDS_LIST_HEAD(qsreaders);
 351         unsigned long was_online;
 352         DEFINE_URCU_WAIT_NODE(wait, URCU_WAIT_WAITING);
 353         struct urcu_waiters waiters;
 354
 355         was_online = urcu_qsbr_read_ongoing();
 356
 357         /*
 358          * Mark the writer thread offline to make sure we don't wait for
 359          * our own quiescent state. This allows using synchronize_rcu()
 360          * in threads registered as readers.
 361          */
 362         if (was_online)
 363                 urcu_qsbr_thread_offline();
 364         else
 365                 cmm_smp_mb();
 366
 367         /*
 368          * Add ourself to gp_waiters queue of threads awaiting to wait
 369          * for a grace period. Proceed to perform the grace period only
 370          * if we are the first thread added into the queue.
 371          */
 372         if (urcu_wait_add(&gp_waiters, &wait) != 0) {
 373                 /* Not first in queue: will be awakened by another thread. */
 374                 urcu_adaptative_busy_wait(&wait);
 375                 goto gp_end;
 376         }
 377         /* We won't need to wake ourself up */
 378         urcu_wait_set_state(&wait, URCU_WAIT_RUNNING);
 379
 380         mutex_lock(&rcu_gp_lock);
 381
 382         /*
 383          * Move all waiters into our local queue.
 384          */
 385         urcu_move_waiters(&waiters, &gp_waiters);
 386
 387         mutex_lock(&rcu_registry_lock);
 388
 389         if (cds_list_empty(&registry))
 390                 goto out;
 391
 392         /* Increment current G.P. */
 393         CMM_STORE_SHARED(urcu_qsbr_gp.ctr, urcu_qsbr_gp.ctr + URCU_QSBR_GP_CTR);
 394
 395         /*
 396          * Must commit urcu_qsbr_gp.ctr update to memory before waiting for
 397          * quiescent state. Failure to do so could result in the writer
 398          * waiting forever while new readers are always accessing data
 399          * (no progress). Enforce compiler-order of store to urcu_qsbr_gp.ctr
 400          * before load URCU_TLS(urcu_qsbr_reader).ctr.
 401          */
 402         cmm_barrier();
 403
 404         /*
 405          * Adding a cmm_smp_mb() which is _not_ formally required, but makes the
 406          * model easier to understand. It does not have a big performance impact
 407          * anyway, given this is the write-side.
 408          */
 409         cmm_smp_mb();
 410
 411         /*
 412          * Wait for readers to observe new count of be quiescent.
 413          * wait_for_readers() can release and grab again rcu_registry_lock
 414          * interally.
 415          */
 416         wait_for_readers(&registry, NULL, &qsreaders);
 417
 418         /*
 419          * Put quiescent reader list back into registry.
 420          */
 421         cds_list_splice(&qsreaders, &registry);
 422 out:
 423         mutex_unlock(&rcu_registry_lock);
 424         mutex_unlock(&rcu_gp_lock);
 425         urcu_wake_all_waiters(&waiters);
 426 gp_end:
 427         if (was_online)
 428                 urcu_qsbr_thread_online();
 429         else
 430                 cmm_smp_mb();
 431 }
 432 #endif  /* !(CAA_BITS_PER_LONG < 64) */
 433 URCU_ATTR_ALIAS("urcu_qsbr_synchronize_rcu")
 434 void synchronize_rcu_qsbr();
 435
 436 /*
 437  * library wrappers to be used by non-LGPL compatible source code.
 438  */
 439
 440 void urcu_qsbr_read_lock(void)
 441 {
 442         _urcu_qsbr_read_lock();
 443 }
 444 URCU_ATTR_ALIAS("urcu_qsbr_read_lock") void rcu_read_lock_qsbr();
 445
 446 void urcu_qsbr_read_unlock(void)
 447 {
 448         _urcu_qsbr_read_unlock();
 449 }
 450 URCU_ATTR_ALIAS("urcu_qsbr_read_unlock") void rcu_read_unlock_qsbr();
 451
 452 int urcu_qsbr_read_ongoing(void)
 453 {
 454         return _urcu_qsbr_read_ongoing();
 455 }
 456 URCU_ATTR_ALIAS("urcu_qsbr_read_ongoing")
 457 void rcu_read_ongoing_qsbr();
 458
 459 void urcu_qsbr_quiescent_state(void)
 460 {
 461         _urcu_qsbr_quiescent_state();
 462 }
 463 URCU_ATTR_ALIAS("urcu_qsbr_quiescent_state")
 464 void rcu_quiescent_state_qsbr();
 465
 466 void urcu_qsbr_thread_offline(void)
 467 {
 468         _urcu_qsbr_thread_offline();
 469 }
 470 URCU_ATTR_ALIAS("urcu_qsbr_thread_offline")
 471 void rcu_thread_offline_qsbr();
 472
 473 void urcu_qsbr_thread_online(void)
 474 {
 475         _urcu_qsbr_thread_online();
 476 }
 477 URCU_ATTR_ALIAS("urcu_qsbr_thread_online")
 478 void rcu_thread_online_qsbr();
 479
 480 void urcu_qsbr_register_thread(void)
 481 {
 482         URCU_TLS(urcu_qsbr_reader).tid = pthread_self();
 483         assert(URCU_TLS(urcu_qsbr_reader).ctr == 0);
 484
 485         mutex_lock(&rcu_registry_lock);
 486         assert(!URCU_TLS(urcu_qsbr_reader).registered);
 487         URCU_TLS(urcu_qsbr_reader).registered = 1;
 488         cds_list_add(&URCU_TLS(urcu_qsbr_reader).node, &registry);
 489         mutex_unlock(&rcu_registry_lock);
 490         _urcu_qsbr_thread_online();
 491 }
 492 URCU_ATTR_ALIAS("urcu_qsbr_register_thread")
 493 void rcu_register_thread_qsbr();
 494
 495 void urcu_qsbr_unregister_thread(void)
 496 {
 497         /*
 498          * We have to make the thread offline otherwise we end up dealocking
 499          * with a waiting writer.
 500          */
 501         _urcu_qsbr_thread_offline();
 502         assert(URCU_TLS(urcu_qsbr_reader).registered);
 503         URCU_TLS(urcu_qsbr_reader).registered = 0;
 504         mutex_lock(&rcu_registry_lock);
 505         cds_list_del(&URCU_TLS(urcu_qsbr_reader).node);
 506         mutex_unlock(&rcu_registry_lock);
 507 }
 508 URCU_ATTR_ALIAS("urcu_qsbr_unregister_thread")
 509 void rcu_unregister_thread_qsbr();
 510
 511 void urcu_qsbr_exit(void)
 512 {
 513         /*
 514          * Assertion disabled because call_rcu threads are now rcu
 515          * readers, and left running at exit.
 516          * assert(cds_list_empty(&registry));
 517          */
 518 }
 519 URCU_ATTR_ALIAS("urcu_qsbr_exit") void rcu_exit_qsbr();
 520
 521 DEFINE_RCU_FLAVOR(rcu_flavor);
 522 DEFINE_RCU_FLAVOR_ALIAS(rcu_flavor, alias_rcu_flavor);
 523
 524 #include "urcu-call-rcu-impl.h"
 525 #include "urcu-defer-impl.h"