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 #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/map/urcu-qsbr.h"
  38
  39 #define BUILD_QSBR_LIB
  40 #include "urcu/static/urcu-qsbr.h"
  41 /* Do not #define _LGPL_SOURCE to ensure we can emit the wrapper symbols */
  42 #include "urcu-qsbr.h"
  43
  44 void __attribute__((destructor)) rcu_exit(void);
  45
  46 static pthread_mutex_t rcu_gp_lock = PTHREAD_MUTEX_INITIALIZER;
  47
  48 int32_t gp_futex;
  49
  50 /*
  51  * Global grace period counter.
  52  */
  53 unsigned long rcu_gp_ctr = RCU_GP_ONLINE;
  54
  55 /*
  56  * Written to only by each individual reader. Read by both the reader and the
  57  * writers.
  58  */
  59 struct rcu_reader __thread rcu_reader;
  60
  61 #ifdef DEBUG_YIELD
  62 unsigned int yield_active;
  63 unsigned int __thread rand_yield;
  64 #endif
  65
  66 static CDS_LIST_HEAD(registry);
  67
  68 static void mutex_lock(pthread_mutex_t *mutex)
  69 {
  70         int ret;
  71
  72 #ifndef DISTRUST_SIGNALS_EXTREME
  73         ret = pthread_mutex_lock(mutex);
  74         if (ret) {
  75                 perror("Error in pthread mutex lock");
  76                 exit(-1);
  77         }
  78 #else /* #ifndef DISTRUST_SIGNALS_EXTREME */
  79         while ((ret = pthread_mutex_trylock(mutex)) != 0) {
  80                 if (ret != EBUSY && ret != EINTR) {
  81                         printf("ret = %d, errno = %d\n", ret, errno);
  82                         perror("Error in pthread mutex lock");
  83                         exit(-1);
  84                 }
  85                 poll(NULL,0,10);
  86         }
  87 #endif /* #else #ifndef DISTRUST_SIGNALS_EXTREME */
  88 }
  89
  90 static void mutex_unlock(pthread_mutex_t *mutex)
  91 {
  92         int ret;
  93
  94         ret = pthread_mutex_unlock(mutex);
  95         if (ret) {
  96                 perror("Error in pthread mutex unlock");
  97                 exit(-1);
  98         }
  99 }
 100
 101 /*
 102  * synchronize_rcu() waiting. Single thread.
 103  */
 104 static void wait_gp(void)
 105 {
 106         /* Read reader_gp before read futex */
 107         cmm_smp_rmb();
 108         if (uatomic_read(&gp_futex) == -1)
 109                 futex_noasync(&gp_futex, FUTEX_WAIT, -1,
 110                       NULL, NULL, 0);
 111 }
 112
 113 static void update_counter_and_wait(void)
 114 {
 115         CDS_LIST_HEAD(qsreaders);
 116         int wait_loops = 0;
 117         struct rcu_reader *index, *tmp;
 118
 119 #if (CAA_BITS_PER_LONG < 64)
 120         /* Switch parity: 0 -> 1, 1 -> 0 */
 121         CMM_STORE_SHARED(rcu_gp_ctr, rcu_gp_ctr ^ RCU_GP_CTR);
 122 #else   /* !(CAA_BITS_PER_LONG < 64) */
 123         /* Increment current G.P. */
 124         CMM_STORE_SHARED(rcu_gp_ctr, rcu_gp_ctr + RCU_GP_CTR);
 125 #endif  /* !(CAA_BITS_PER_LONG < 64) */
 126
 127         /*
 128          * Must commit rcu_gp_ctr update to memory before waiting for
 129          * quiescent state. Failure to do so could result in the writer
 130          * waiting forever while new readers are always accessing data
 131          * (no progress). Enforce compiler-order of store to rcu_gp_ctr
 132          * before load rcu_reader ctr.
 133          */
 134         cmm_barrier();
 135
 136         /*
 137          * Adding a cmm_smp_mb() which is _not_ formally required, but makes the
 138          * model easier to understand. It does not have a big performance impact
 139          * anyway, given this is the write-side.
 140          */
 141         cmm_smp_mb();
 142
 143         /*
 144          * Wait for each thread rcu_reader_qs_gp count to become 0.
 145          */
 146         for (;;) {
 147                 wait_loops++;
 148                 if (wait_loops == RCU_QS_ACTIVE_ATTEMPTS) {
 149                         uatomic_dec(&gp_futex);
 150                         /* Write futex before read reader_gp */
 151                         cmm_smp_mb();
 152                 }
 153
 154                 cds_list_for_each_entry_safe(index, tmp, &registry, node) {
 155                         if (!rcu_gp_ongoing(&index->ctr))
 156                                 cds_list_move(&index->node, &qsreaders);
 157                 }
 158
 159                 if (cds_list_empty(&registry)) {
 160                         if (wait_loops == RCU_QS_ACTIVE_ATTEMPTS) {
 161                                 /* Read reader_gp before write futex */
 162                                 cmm_smp_mb();
 163                                 uatomic_set(&gp_futex, 0);
 164                         }
 165                         break;
 166                 } else {
 167                         if (wait_loops == RCU_QS_ACTIVE_ATTEMPTS) {
 168                                 wait_gp();
 169                         } else {
 170 #ifndef HAS_INCOHERENT_CACHES
 171                                 caa_cpu_relax();
 172 #else /* #ifndef HAS_INCOHERENT_CACHES */
 173                                 cmm_smp_mb();
 174 #endif /* #else #ifndef HAS_INCOHERENT_CACHES */
 175                         }
 176                 }
 177         }
 178         /* put back the reader list in the registry */
 179         cds_list_splice(&qsreaders, &registry);
 180 }
 181
 182 /*
 183  * Using a two-subphases algorithm for architectures with smaller than 64-bit
 184  * long-size to ensure we do not encounter an overflow bug.
 185  */
 186
 187 #if (CAA_BITS_PER_LONG < 64)
 188 void synchronize_rcu(void)
 189 {
 190         unsigned long was_online;
 191
 192         was_online = rcu_reader.ctr;
 193
 194         /* All threads should read qparity before accessing data structure
 195          * where new ptr points to.
 196          */
 197         /* Write new ptr before changing the qparity */
 198         cmm_smp_mb();
 199
 200         /*
 201          * Mark the writer thread offline to make sure we don't wait for
 202          * our own quiescent state. This allows using synchronize_rcu()
 203          * in threads registered as readers.
 204          */
 205         if (was_online)
 206                 CMM_STORE_SHARED(rcu_reader.ctr, 0);
 207
 208         mutex_lock(&rcu_gp_lock);
 209
 210         if (cds_list_empty(&registry))
 211                 goto out;
 212
 213         /*
 214          * Wait for previous parity to be empty of readers.
 215          */
 216         update_counter_and_wait();      /* 0 -> 1, wait readers in parity 0 */
 217
 218         /*
 219          * Must finish waiting for quiescent state for parity 0 before
 220          * committing next rcu_gp_ctr update to memory. Failure to
 221          * do so could result in the writer waiting forever while new
 222          * readers are always accessing data (no progress).  Enforce
 223          * compiler-order of load rcu_reader ctr before store to
 224          * rcu_gp_ctr.
 225          */
 226         cmm_barrier();
 227
 228         /*
 229          * Adding a cmm_smp_mb() which is _not_ formally required, but makes the
 230          * model easier to understand. It does not have a big performance impact
 231          * anyway, given this is the write-side.
 232          */
 233         cmm_smp_mb();
 234
 235         /*
 236          * Wait for previous parity to be empty of readers.
 237          */
 238         update_counter_and_wait();      /* 1 -> 0, wait readers in parity 1 */
 239 out:
 240         mutex_unlock(&rcu_gp_lock);
 241
 242         /*
 243          * Finish waiting for reader threads before letting the old ptr being
 244          * freed.
 245          */
 246         if (was_online)
 247                 _CMM_STORE_SHARED(rcu_reader.ctr,
 248                                   CMM_LOAD_SHARED(rcu_gp_ctr));
 249         cmm_smp_mb();
 250 }
 251 #else /* !(CAA_BITS_PER_LONG < 64) */
 252 void synchronize_rcu(void)
 253 {
 254         unsigned long was_online;
 255
 256         was_online = rcu_reader.ctr;
 257
 258         /*
 259          * Mark the writer thread offline to make sure we don't wait for
 260          * our own quiescent state. This allows using synchronize_rcu()
 261          * in threads registered as readers.
 262          */
 263         cmm_smp_mb();
 264         if (was_online)
 265                 CMM_STORE_SHARED(rcu_reader.ctr, 0);
 266
 267         mutex_lock(&rcu_gp_lock);
 268         if (cds_list_empty(&registry))
 269                 goto out;
 270         update_counter_and_wait();
 271 out:
 272         mutex_unlock(&rcu_gp_lock);
 273
 274         if (was_online)
 275                 _CMM_STORE_SHARED(rcu_reader.ctr,
 276                                   CMM_LOAD_SHARED(rcu_gp_ctr));
 277         cmm_smp_mb();
 278 }
 279 #endif  /* !(CAA_BITS_PER_LONG < 64) */
 280
 281 /*
 282  * library wrappers to be used by non-LGPL compatible source code.
 283  */
 284
 285 void rcu_read_lock(void)
 286 {
 287         _rcu_read_lock();
 288 }
 289
 290 void rcu_read_unlock(void)
 291 {
 292         _rcu_read_unlock();
 293 }
 294
 295 void rcu_quiescent_state(void)
 296 {
 297         _rcu_quiescent_state();
 298 }
 299
 300 void rcu_thread_offline(void)
 301 {
 302         _rcu_thread_offline();
 303 }
 304
 305 void rcu_thread_online(void)
 306 {
 307         _rcu_thread_online();
 308 }
 309
 310 void rcu_register_thread(void)
 311 {
 312         rcu_reader.tid = pthread_self();
 313         assert(rcu_reader.ctr == 0);
 314
 315         mutex_lock(&rcu_gp_lock);
 316         cds_list_add(&rcu_reader.node, &registry);
 317         mutex_unlock(&rcu_gp_lock);
 318         _rcu_thread_online();
 319 }
 320
 321 void rcu_unregister_thread(void)
 322 {
 323         /*
 324          * We have to make the thread offline otherwise we end up dealocking
 325          * with a waiting writer.
 326          */
 327         _rcu_thread_offline();
 328         mutex_lock(&rcu_gp_lock);
 329         cds_list_del(&rcu_reader.node);
 330         mutex_unlock(&rcu_gp_lock);
 331 }
 332
 333 void rcu_exit(void)
 334 {
 335         assert(cds_list_empty(&registry));
 336 }
 337
 338 #include "urcu-call-rcu-impl.h"
 339 #include "urcu-defer-impl.h"