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