urcu.c

   1 /*
   2  * urcu.c
   3  *
   4  * Userspace RCU library
   5  *
   6  * Copyright (c) 2009 Mathieu Desnoyers <mathieu.desnoyers@polymtl.ca>
   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 _BSD_SOURCE
  27 #include <stdio.h>
  28 #include <pthread.h>
  29 #include <signal.h>
  30 #include <assert.h>
  31 #include <stdlib.h>
  32 #include <string.h>
  33 #include <errno.h>
  34 #include <poll.h>
  35
  36 #include "urcu-static.h"
  37 /* Do not #define _LGPL_SOURCE to ensure we can emit the wrapper symbols */
  38 #include "urcu.h"
  39
  40 #ifdef RCU_MEMBARRIER
  41 static int init_done;
  42 int has_sys_membarrier;
  43
  44 void __attribute__((constructor)) rcu_init(void);
  45 #endif
  46
  47 #ifdef RCU_MB
  48 void rcu_init(void)
  49 {
  50 }
  51 #endif
  52
  53 #ifdef RCU_SIGNAL
  54 static int init_done;
  55
  56 void __attribute__((constructor)) rcu_init(void);
  57 void __attribute__((destructor)) rcu_exit(void);
  58 #endif
  59
  60 static pthread_mutex_t rcu_gp_lock = PTHREAD_MUTEX_INITIALIZER;
  61
  62 int gp_futex;
  63
  64 /*
  65  * Global grace period counter.
  66  * Contains the current RCU_GP_CTR_PHASE.
  67  * Also has a RCU_GP_COUNT of 1, to accelerate the reader fast path.
  68  * Written to only by writer with mutex taken. Read by both writer and readers.
  69  */
  70 long rcu_gp_ctr = RCU_GP_COUNT;
  71
  72 /*
  73  * Written to only by each individual reader. Read by both the reader and the
  74  * writers.
  75  */
  76 struct rcu_reader __thread rcu_reader;
  77
  78 #ifdef DEBUG_YIELD
  79 unsigned int yield_active;
  80 unsigned int __thread rand_yield;
  81 #endif
  82
  83 static LIST_HEAD(registry);
  84
  85 static void mutex_lock(pthread_mutex_t *mutex)
  86 {
  87         int ret;
  88
  89 #ifndef DISTRUST_SIGNALS_EXTREME
  90         ret = pthread_mutex_lock(mutex);
  91         if (ret) {
  92                 perror("Error in pthread mutex lock");
  93                 exit(-1);
  94         }
  95 #else /* #ifndef DISTRUST_SIGNALS_EXTREME */
  96         while ((ret = pthread_mutex_trylock(mutex)) != 0) {
  97                 if (ret != EBUSY && ret != EINTR) {
  98                         printf("ret = %d, errno = %d\n", ret, errno);
  99                         perror("Error in pthread mutex lock");
 100                         exit(-1);
 101                 }
 102                 if (rcu_reader.need_mb) {
 103                         smp_mb();
 104                         rcu_reader.need_mb = 0;
 105                         smp_mb();
 106                 }
 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                 perror("Error in pthread mutex unlock");
 119                 exit(-1);
 120         }
 121 }
 122
 123 /*
 124  * called with rcu_gp_lock held.
 125  */
 126 static void switch_next_rcu_qparity(void)
 127 {
 128         STORE_SHARED(rcu_gp_ctr, rcu_gp_ctr ^ RCU_GP_CTR_PHASE);
 129 }
 130
 131 #ifdef RCU_MEMBARRIER
 132 static void smp_mb_heavy(void)
 133 {
 134         if (likely(has_sys_membarrier))
 135                 membarrier(MEMBARRIER_EXPEDITED);
 136         else
 137                 smp_mb();
 138 }
 139 #endif
 140
 141 #ifdef RCU_MB
 142 static void smp_mb_heavy(void)
 143 {
 144         smp_mb();
 145 }
 146 #endif
 147
 148 #ifdef RCU_SIGNAL
 149 static void force_mb_all_readers(void)
 150 {
 151         struct rcu_reader *index;
 152
 153         /*
 154          * Ask for each threads to execute a smp_mb() so we can consider the
 155          * compiler barriers around rcu read lock as real memory barriers.
 156          */
 157         if (list_empty(&registry))
 158                 return;
 159         /*
 160          * pthread_kill has a smp_mb(). But beware, we assume it performs
 161          * a cache flush on architectures with non-coherent cache. Let's play
 162          * safe and don't assume anything : we use smp_mc() to make sure the
 163          * cache flush is enforced.
 164          */
 165         list_for_each_entry(index, &registry, head) {
 166                 index->need_mb = 1;
 167                 smp_mc();       /* write need_mb before sending the signal */
 168                 pthread_kill(index->tid, SIGRCU);
 169         }
 170         /*
 171          * Wait for sighandler (and thus mb()) to execute on every thread.
 172          *
 173          * Note that the pthread_kill() will never be executed on systems
 174          * that correctly deliver signals in a timely manner.  However, it
 175          * is not uncommon for kernels to have bugs that can result in
 176          * lost or unduly delayed signals.
 177          *
 178          * If you are seeing the below pthread_kill() executing much at
 179          * all, we suggest testing the underlying kernel and filing the
 180          * relevant bug report.  For Linux kernels, we recommend getting
 181          * the Linux Test Project (LTP).
 182          */
 183         list_for_each_entry(index, &registry, head) {
 184                 while (index->need_mb) {
 185                         pthread_kill(index->tid, SIGRCU);
 186                         poll(NULL, 0, 1);
 187                 }
 188         }
 189         smp_mb();       /* read ->need_mb before ending the barrier */
 190 }
 191
 192 static void smp_mb_heavy(void)
 193 {
 194         force_mb_all_readers();
 195 }
 196 #endif /* #ifdef RCU_SIGNAL */
 197
 198 /*
 199  * synchronize_rcu() waiting. Single thread.
 200  */
 201 static void wait_gp(void)
 202 {
 203         /* Read reader_gp before read futex */
 204         smp_mb_heavy();
 205         if (uatomic_read(&gp_futex) == -1)
 206                 futex_async(&gp_futex, FUTEX_WAIT, -1,
 207                       NULL, NULL, 0);
 208 }
 209
 210 void wait_for_quiescent_state(void)
 211 {
 212         LIST_HEAD(qsreaders);
 213         int wait_loops = 0;
 214         struct rcu_reader *index, *tmp;
 215
 216         if (list_empty(&registry))
 217                 return;
 218         /*
 219          * Wait for each thread rcu_reader.ctr count to become 0.
 220          */
 221         for (;;) {
 222                 wait_loops++;
 223                 if (wait_loops == RCU_QS_ACTIVE_ATTEMPTS) {
 224                         uatomic_dec(&gp_futex);
 225                         /* Write futex before read reader_gp */
 226                         smp_mb_heavy();
 227                 }
 228
 229                 list_for_each_entry_safe(index, tmp, &registry, head) {
 230                         if (!rcu_old_gp_ongoing(&index->ctr))
 231                                 list_move(&index->head, &qsreaders);
 232                 }
 233
 234 #ifndef HAS_INCOHERENT_CACHES
 235                 if (list_empty(&registry)) {
 236                         if (wait_loops == RCU_QS_ACTIVE_ATTEMPTS) {
 237                                 /* Read reader_gp before write futex */
 238                                 smp_mb_heavy();
 239                                 uatomic_set(&gp_futex, 0);
 240                         }
 241                         break;
 242                 } else {
 243                         if (wait_loops == RCU_QS_ACTIVE_ATTEMPTS)
 244                                 wait_gp();
 245                         else
 246                                 cpu_relax();
 247                 }
 248 #else /* #ifndef HAS_INCOHERENT_CACHES */
 249                 /*
 250                  * BUSY-LOOP. Force the reader thread to commit its
 251                  * rcu_reader.ctr update to memory if we wait for too long.
 252                  */
 253                 if (list_empty(&registry)) {
 254                         if (wait_loops == RCU_QS_ACTIVE_ATTEMPTS) {
 255                                 /* Read reader_gp before write futex */
 256                                 smp_mb_heavy();
 257                                 uatomic_set(&gp_futex, 0);
 258                         }
 259                         break;
 260                 } else {
 261                         switch (wait_loops) {
 262                         case RCU_QS_ACTIVE_ATTEMPTS:
 263                                 wait_gp();
 264                                 break; /* only escape switch */
 265                         case KICK_READER_LOOPS:
 266                                 smp_mb_heavy();
 267                                 wait_loops = 0;
 268                                 break; /* only escape switch */
 269                         default:
 270                                 cpu_relax();
 271                         }
 272                 }
 273 #endif /* #else #ifndef HAS_INCOHERENT_CACHES */
 274         }
 275         /* put back the reader list in the registry */
 276         list_splice(&qsreaders, &registry);
 277 }
 278
 279 void synchronize_rcu(void)
 280 {
 281         mutex_lock(&rcu_gp_lock);
 282
 283         /* All threads should read qparity before accessing data structure
 284          * where new ptr points to. Must be done within rcu_gp_lock because it
 285          * iterates on reader threads.*/
 286         /* Write new ptr before changing the qparity */
 287         smp_mb_heavy();
 288
 289         switch_next_rcu_qparity();      /* 0 -> 1 */
 290
 291         /*
 292          * Must commit qparity update to memory before waiting for parity
 293          * 0 quiescent state. Failure to do so could result in the writer
 294          * waiting forever while new readers are always accessing data (no
 295          * progress).
 296          * Ensured by STORE_SHARED and LOAD_SHARED.
 297          */
 298
 299         /*
 300          * Adding a smp_mb() which is _not_ formally required, but makes the
 301          * model easier to understand. It does not have a big performance impact
 302          * anyway, given this is the write-side.
 303          */
 304         smp_mb();
 305
 306         /*
 307          * Wait for previous parity to be empty of readers.
 308          */
 309         wait_for_quiescent_state();     /* Wait readers in parity 0 */
 310
 311         /*
 312          * Must finish waiting for quiescent state for parity 0 before
 313          * committing qparity update to memory. Failure to do so could result in
 314          * the writer waiting forever while new readers are always accessing
 315          * data (no progress).
 316          * Ensured by STORE_SHARED and LOAD_SHARED.
 317          */
 318
 319         /*
 320          * Adding a smp_mb() which is _not_ formally required, but makes the
 321          * model easier to understand. It does not have a big performance impact
 322          * anyway, given this is the write-side.
 323          */
 324         smp_mb();
 325
 326         switch_next_rcu_qparity();      /* 1 -> 0 */
 327
 328         /*
 329          * Must commit qparity update to memory before waiting for parity
 330          * 1 quiescent state. Failure to do so could result in the writer
 331          * waiting forever while new readers are always accessing data (no
 332          * progress).
 333          * Ensured by STORE_SHARED and LOAD_SHARED.
 334          */
 335
 336         /*
 337          * Adding a smp_mb() which is _not_ formally required, but makes the
 338          * model easier to understand. It does not have a big performance impact
 339          * anyway, given this is the write-side.
 340          */
 341         smp_mb();
 342
 343         /*
 344          * Wait for previous parity to be empty of readers.
 345          */
 346         wait_for_quiescent_state();     /* Wait readers in parity 1 */
 347
 348         /* Finish waiting for reader threads before letting the old ptr being
 349          * freed. Must be done within rcu_gp_lock because it iterates on reader
 350          * threads. */
 351         smp_mb_heavy();
 352
 353         mutex_unlock(&rcu_gp_lock);
 354 }
 355
 356 /*
 357  * library wrappers to be used by non-LGPL compatible source code.
 358  */
 359
 360 void rcu_read_lock(void)
 361 {
 362         _rcu_read_lock();
 363 }
 364
 365 void rcu_read_unlock(void)
 366 {
 367         _rcu_read_unlock();
 368 }
 369
 370 void rcu_register_thread(void)
 371 {
 372         rcu_reader.tid = pthread_self();
 373         assert(rcu_reader.need_mb == 0);
 374         assert(rcu_reader.ctr == 0);
 375
 376         mutex_lock(&rcu_gp_lock);
 377         rcu_init();     /* In case gcc does not support constructor attribute */
 378         list_add(&rcu_reader.head, &registry);
 379         mutex_unlock(&rcu_gp_lock);
 380 }
 381
 382 void rcu_unregister_thread(void)
 383 {
 384         mutex_lock(&rcu_gp_lock);
 385         list_del(&rcu_reader.head);
 386         mutex_unlock(&rcu_gp_lock);
 387 }
 388
 389 #ifdef RCU_MEMBARRIER
 390 void rcu_init(void)
 391 {
 392         if (init_done)
 393                 return;
 394         init_done = 1;
 395         if (!membarrier(MEMBARRIER_EXPEDITED | MEMBARRIER_QUERY))
 396                 has_sys_membarrier = 1;
 397 }
 398 #endif
 399
 400 #ifdef RCU_SIGNAL
 401 static void sigrcu_handler(int signo, siginfo_t *siginfo, void *context)
 402 {
 403         /*
 404          * Executing this smp_mb() is the only purpose of this signal handler.
 405          * It punctually promotes barrier() into smp_mb() on every thread it is
 406          * executed on.
 407          */
 408         smp_mb();
 409         rcu_reader.need_mb = 0;
 410         smp_mb();
 411 }
 412
 413 /*
 414  * rcu_init constructor. Called when the library is linked, but also when
 415  * reader threads are calling rcu_register_thread().
 416  * Should only be called by a single thread at a given time. This is ensured by
 417  * holing the rcu_gp_lock from rcu_register_thread() or by running at library
 418  * load time, which should not be executed by multiple threads nor concurrently
 419  * with rcu_register_thread() anyway.
 420  */
 421 void rcu_init(void)
 422 {
 423         struct sigaction act;
 424         int ret;
 425
 426         if (init_done)
 427                 return;
 428         init_done = 1;
 429
 430         act.sa_sigaction = sigrcu_handler;
 431         act.sa_flags = SA_SIGINFO | SA_RESTART;
 432         sigemptyset(&act.sa_mask);
 433         ret = sigaction(SIGRCU, &act, NULL);
 434         if (ret) {
 435                 perror("Error in sigaction");
 436                 exit(-1);
 437         }
 438 }
 439
 440 void rcu_exit(void)
 441 {
 442         struct sigaction act;
 443         int ret;
 444
 445         ret = sigaction(SIGRCU, NULL, &act);
 446         if (ret) {
 447                 perror("Error in sigaction");
 448                 exit(-1);
 449         }
 450         assert(act.sa_sigaction == sigrcu_handler);
 451         assert(list_empty(&registry));
 452 }
 453 #endif /* #ifdef RCU_SIGNAL */