urcu.c

   1 /*
   2  * urcu.c
   3  *
   4  * Userspace RCU library
   5  *
   6  * Copyright February 2009 - Mathieu Desnoyers <mathieu.desnoyers@polymtl.ca>
   7  *
   8  * Distributed under GPLv2
   9  */
  10
  11 #include <stdio.h>
  12 #include <pthread.h>
  13 #include <signal.h>
  14 #include <assert.h>
  15 #include <stdlib.h>
  16 #include <string.h>
  17
  18 #include "urcu.h"
  19
  20 pthread_mutex_t urcu_mutex = PTHREAD_MUTEX_INITIALIZER;
  21
  22 /*
  23  * Global grace period counter.
  24  * Contains the current RCU_GP_CTR_BIT.
  25  * Also has a RCU_GP_CTR_BIT of 1, to accelerate the reader fast path.
  26  */
  27 long urcu_gp_ctr = RCU_GP_COUNT;
  28
  29 long __thread urcu_active_readers;
  30
  31 /* Thread IDs of registered readers */
  32 #define INIT_NUM_THREADS 4
  33
  34 struct reader_data {
  35         pthread_t tid;
  36         long *urcu_active_readers;
  37 };
  38
  39 #ifdef DEBUG_YIELD
  40 unsigned int yield_active;
  41 unsigned int __thread rand_yield;
  42 #endif
  43
  44 static struct reader_data *reader_data;
  45 static int num_readers, alloc_readers;
  46 #ifndef DEBUG_FULL_MB
  47 static int sig_done;
  48 #endif
  49
  50 void internal_urcu_lock(void)
  51 {
  52         int ret;
  53         ret = pthread_mutex_lock(&urcu_mutex);
  54         if (ret) {
  55                 perror("Error in pthread mutex lock");
  56                 exit(-1);
  57         }
  58 }
  59
  60 void internal_urcu_unlock(void)
  61 {
  62         int ret;
  63
  64         ret = pthread_mutex_unlock(&urcu_mutex);
  65         if (ret) {
  66                 perror("Error in pthread mutex unlock");
  67                 exit(-1);
  68         }
  69 }
  70
  71 /*
  72  * called with urcu_mutex held.
  73  */
  74 static void switch_next_urcu_qparity(void)
  75 {
  76         urcu_gp_ctr ^= RCU_GP_CTR_BIT;
  77 }
  78
  79 #ifdef DEBUG_FULL_MB
  80 static void force_mb_single_thread(pthread_t tid)
  81 {
  82         smp_mb();
  83 }
  84
  85 static void force_mb_all_threads(void)
  86 {
  87         smp_mb();
  88 }
  89 #else
  90
  91 static void force_mb_single_thread(pthread_t tid)
  92 {
  93         assert(reader_data);
  94         sig_done = 0;
  95         smp_mb();       /* write sig_done before sending the signals */
  96         pthread_kill(tid, SIGURCU);
  97         /*
  98          * Wait for sighandler (and thus mb()) to execute on every thread.
  99          * BUSY-LOOP.
 100          */
 101         while (sig_done < 1)
 102                 smp_rmb();      /* ensure we re-read sig-done */
 103         smp_mb();       /* read sig_done before ending the barrier */
 104 }
 105
 106 static void force_mb_all_threads(void)
 107 {
 108         struct reader_data *index;
 109         /*
 110          * Ask for each threads to execute a smp_mb() so we can consider the
 111          * compiler barriers around rcu read lock as real memory barriers.
 112          */
 113         if (!reader_data)
 114                 return;
 115         sig_done = 0;
 116         smp_mb();       /* write sig_done before sending the signals */
 117         for (index = reader_data; index < reader_data + num_readers; index++)
 118                 pthread_kill(index->tid, SIGURCU);
 119         /*
 120          * Wait for sighandler (and thus mb()) to execute on every thread.
 121          * BUSY-LOOP.
 122          */
 123         while (sig_done < num_readers)
 124                 smp_rmb();      /* ensure we re-read sig-done */
 125         smp_mb();       /* read sig_done before ending the barrier */
 126 }
 127 #endif
 128
 129 void wait_for_quiescent_state(void)
 130 {
 131         struct reader_data *index;
 132
 133         if (!reader_data)
 134                 return;
 135         /*
 136          * Wait for each thread urcu_active_readers count to become 0.
 137          */
 138         for (index = reader_data; index < reader_data + num_readers; index++) {
 139                 int wait_loops = 0;
 140                 /*
 141                  * BUSY-LOOP. Force the reader thread to commit its
 142                  * urcu_active_readers update to memory if we wait for too long.
 143                  */
 144                 while (rcu_old_gp_ongoing(index->urcu_active_readers)) {
 145                         if (wait_loops++ == KICK_READER_LOOPS) {
 146                                 force_mb_single_thread(index->tid);
 147                                 wait_loops = 0;
 148                         }
 149                 }
 150         }
 151 }
 152
 153 void synchronize_rcu(void)
 154 {
 155         internal_urcu_lock();
 156
 157         /* All threads should read qparity before accessing data structure
 158          * where new ptr points to. Must be done within internal_urcu_lock
 159          * because it iterates on reader threads.*/
 160         /* Write new ptr before changing the qparity */
 161         force_mb_all_threads();
 162
 163         switch_next_urcu_qparity();     /* 0 -> 1 */
 164
 165         /*
 166          * Must commit qparity update to memory before waiting for parity
 167          * 0 quiescent state. Failure to do so could result in the writer
 168          * waiting forever while new readers are always accessing data (no
 169          * progress).
 170          */
 171         smp_mb();
 172
 173         /*
 174          * Wait for previous parity to be empty of readers.
 175          */
 176         wait_for_quiescent_state();     /* Wait readers in parity 0 */
 177
 178         /*
 179          * Must finish waiting for quiescent state for parity 0 before
 180          * committing qparity update to memory. Failure to do so could result in
 181          * the writer waiting forever while new readers are always accessing
 182          * data (no progress).
 183          */
 184         smp_mb();
 185
 186         switch_next_urcu_qparity();     /* 1 -> 0 */
 187
 188         /*
 189          * Must commit qparity update to memory before waiting for parity
 190          * 1 quiescent state. Failure to do so could result in the writer
 191          * waiting forever while new readers are always accessing data (no
 192          * progress).
 193          */
 194         smp_mb();
 195
 196         /*
 197          * Wait for previous parity to be empty of readers.
 198          */
 199         wait_for_quiescent_state();     /* Wait readers in parity 1 */
 200
 201         /* Finish waiting for reader threads before letting the old ptr being
 202          * freed. Must be done within internal_urcu_lock because it iterates on
 203          * reader threads. */
 204         force_mb_all_threads();
 205
 206         internal_urcu_unlock();
 207 }
 208
 209 void urcu_add_reader(pthread_t id)
 210 {
 211         struct reader_data *oldarray;
 212
 213         if (!reader_data) {
 214                 alloc_readers = INIT_NUM_THREADS;
 215                 num_readers = 0;
 216                 reader_data =
 217                         malloc(sizeof(struct reader_data) * alloc_readers);
 218         }
 219         if (alloc_readers < num_readers + 1) {
 220                 oldarray = reader_data;
 221                 reader_data = malloc(sizeof(struct reader_data)
 222                                 * (alloc_readers << 1));
 223                 memcpy(reader_data, oldarray,
 224                         sizeof(struct reader_data) * alloc_readers);
 225                 alloc_readers <<= 1;
 226                 free(oldarray);
 227         }
 228         reader_data[num_readers].tid = id;
 229         /* reference to the TLS of _this_ reader thread. */
 230         reader_data[num_readers].urcu_active_readers = &urcu_active_readers;
 231         num_readers++;
 232 }
 233
 234 /*
 235  * Never shrink (implementation limitation).
 236  * This is O(nb threads). Eventually use a hash table.
 237  */
 238 void urcu_remove_reader(pthread_t id)
 239 {
 240         struct reader_data *index;
 241
 242         assert(reader_data != NULL);
 243         for (index = reader_data; index < reader_data + num_readers; index++) {
 244                 if (pthread_equal(index->tid, id)) {
 245                         memcpy(index, &reader_data[num_readers - 1],
 246                                 sizeof(struct reader_data));
 247                         reader_data[num_readers - 1].tid = 0;
 248                         reader_data[num_readers - 1].urcu_active_readers = NULL;
 249                         num_readers--;
 250                         return;
 251                 }
 252         }
 253         /* Hrm not found, forgot to register ? */
 254         assert(0);
 255 }
 256
 257 void urcu_register_thread(void)
 258 {
 259         internal_urcu_lock();
 260         urcu_add_reader(pthread_self());
 261         internal_urcu_unlock();
 262 }
 263
 264 void urcu_unregister_thread(void)
 265 {
 266         internal_urcu_lock();
 267         urcu_remove_reader(pthread_self());
 268         internal_urcu_unlock();
 269 }
 270
 271 #ifndef DEBUG_FULL_MB
 272 void sigurcu_handler(int signo, siginfo_t *siginfo, void *context)
 273 {
 274         /*
 275          * Executing this smp_mb() is the only purpose of this signal handler.
 276          * It punctually promotes barrier() into smp_mb() on every thread it is
 277          * executed on.
 278          */
 279         smp_mb();
 280         atomic_inc(&sig_done);
 281 }
 282
 283 void __attribute__((constructor)) urcu_init(void)
 284 {
 285         struct sigaction act;
 286         int ret;
 287
 288         act.sa_sigaction = sigurcu_handler;
 289         ret = sigaction(SIGURCU, &act, NULL);
 290         if (ret) {
 291                 perror("Error in sigaction");
 292                 exit(-1);
 293         }
 294 }
 295
 296 void __attribute__((destructor)) urcu_exit(void)
 297 {
 298         struct sigaction act;
 299         int ret;
 300
 301         ret = sigaction(SIGURCU, NULL, &act);
 302         if (ret) {
 303                 perror("Error in sigaction");
 304                 exit(-1);
 305         }
 306         assert(act.sa_sigaction == sigurcu_handler);
 307         free(reader_data);
 308 }
 309 #endif