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