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