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