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_all_threads(void)
  81 {
  82         smp_mb();
  83 }
  84 #else
  85 static void force_mb_all_threads(void)
  86 {
  87         struct reader_data *index;
  88         /*
  89          * Ask for each threads to execute a smp_mb() so we can consider the
  90          * compiler barriers around rcu read lock as real memory barriers.
  91          */
  92         if (!reader_data)
  93                 return;
  94         sig_done = 0;
  95         smp_mb();       /* write sig_done before sending the signals */
  96         for (index = reader_data; index < reader_data + num_readers; index++)
  97                 pthread_kill(index->tid, SIGURCU);
  98         /*
  99          * Wait for sighandler (and thus mb()) to execute on every thread.
 100          * BUSY-LOOP.
 101          */
 102         while (sig_done < num_readers)
 103                 barrier();
 104         smp_mb();       /* read sig_done before ending the barrier */
 105 }
 106 #endif
 107
 108 void wait_for_quiescent_state(void)
 109 {
 110         struct reader_data *index;
 111
 112         if (!reader_data)
 113                 return;
 114         /* Wait for each thread urcu_active_readers count to become 0.
 115          */
 116         for (index = reader_data; index < reader_data + num_readers; index++) {
 117                 /*
 118                  * BUSY-LOOP.
 119                  */
 120                 while (rcu_old_gp_ongoing(index->urcu_active_readers))
 121                         barrier();
 122         }
 123 }
 124
 125 void synchronize_rcu(void)
 126 {
 127         /* All threads should read qparity before accessing data structure
 128          * where new ptr points to. */
 129         /* Write new ptr before changing the qparity */
 130         force_mb_all_threads();
 131
 132         internal_urcu_lock();
 133
 134         switch_next_urcu_qparity();     /* 0 -> 1 */
 135
 136         /*
 137          * Must commit qparity update to memory before waiting for parity
 138          * 0 quiescent state. Failure to do so could result in the writer
 139          * waiting forever while new readers are always accessing data (no
 140          * progress).
 141          */
 142         smp_mb();
 143
 144         /*
 145          * Wait for previous parity to be empty of readers.
 146          */
 147         wait_for_quiescent_state();     /* Wait readers in parity 0 */
 148
 149         /*
 150          * Must finish waiting for quiescent state for parity 0 before
 151          * committing qparity update to memory. Failure to do so could result in
 152          * the writer waiting forever while new readers are always accessing
 153          * data (no progress).
 154          */
 155         smp_mb();
 156
 157         switch_next_urcu_qparity();     /* 1 -> 0 */
 158
 159         /*
 160          * Must commit qparity update to memory before waiting for parity
 161          * 1 quiescent state. Failure to do so could result in the writer
 162          * waiting forever while new readers are always accessing data (no
 163          * progress).
 164          */
 165         smp_mb();
 166
 167         /*
 168          * Wait for previous parity to be empty of readers.
 169          */
 170         wait_for_quiescent_state();     /* Wait readers in parity 1 */
 171
 172         internal_urcu_unlock();
 173
 174         /* All threads should finish using the data referred to by old ptr
 175          * before decrementing their urcu_active_readers count */
 176         /* Finish waiting for reader threads before letting the old ptr being
 177          * freed. */
 178         force_mb_all_threads();
 179 }
 180
 181 void urcu_add_reader(pthread_t id)
 182 {
 183         struct reader_data *oldarray;
 184
 185         if (!reader_data) {
 186                 alloc_readers = INIT_NUM_THREADS;
 187                 num_readers = 0;
 188                 reader_data =
 189                         malloc(sizeof(struct reader_data) * alloc_readers);
 190         }
 191         if (alloc_readers < num_readers + 1) {
 192                 oldarray = reader_data;
 193                 reader_data = malloc(sizeof(struct reader_data)
 194                                 * (alloc_readers << 1));
 195                 memcpy(reader_data, oldarray,
 196                         sizeof(struct reader_data) * alloc_readers);
 197                 alloc_readers <<= 1;
 198                 free(oldarray);
 199         }
 200         reader_data[num_readers].tid = id;
 201         /* reference to the TLS of _this_ reader thread. */
 202         reader_data[num_readers].urcu_active_readers = &urcu_active_readers;
 203         num_readers++;
 204 }
 205
 206 /*
 207  * Never shrink (implementation limitation).
 208  * This is O(nb threads). Eventually use a hash table.
 209  */
 210 void urcu_remove_reader(pthread_t id)
 211 {
 212         struct reader_data *index;
 213
 214         assert(reader_data != NULL);
 215         for (index = reader_data; index < reader_data + num_readers; index++) {
 216                 if (pthread_equal(index->tid, id)) {
 217                         memcpy(index, &reader_data[num_readers - 1],
 218                                 sizeof(struct reader_data));
 219                         reader_data[num_readers - 1].tid = 0;
 220                         reader_data[num_readers - 1].urcu_active_readers = NULL;
 221                         num_readers--;
 222                         return;
 223                 }
 224         }
 225         /* Hrm not found, forgot to register ? */
 226         assert(0);
 227 }
 228
 229 void urcu_register_thread(void)
 230 {
 231         internal_urcu_lock();
 232         urcu_add_reader(pthread_self());
 233         internal_urcu_unlock();
 234 }
 235
 236 void urcu_unregister_thread(void)
 237 {
 238         internal_urcu_lock();
 239         urcu_remove_reader(pthread_self());
 240         internal_urcu_unlock();
 241 }
 242
 243 #ifndef DEBUG_FULL_MB
 244 void sigurcu_handler(int signo, siginfo_t *siginfo, void *context)
 245 {
 246         smp_mb();
 247         atomic_inc(&sig_done);
 248 }
 249
 250 void __attribute__((constructor)) urcu_init(void)
 251 {
 252         struct sigaction act;
 253         int ret;
 254
 255         act.sa_sigaction = sigurcu_handler;
 256         ret = sigaction(SIGURCU, &act, NULL);
 257         if (ret) {
 258                 perror("Error in sigaction");
 259                 exit(-1);
 260         }
 261 }
 262
 263 void __attribute__((destructor)) urcu_exit(void)
 264 {
 265         struct sigaction act;
 266         int ret;
 267
 268         ret = sigaction(SIGURCU, NULL, &act);
 269         if (ret) {
 270                 perror("Error in sigaction");
 271                 exit(-1);
 272         }
 273         assert(act.sa_sigaction == sigurcu_handler);
 274         free(reader_data);
 275 }
 276 #endif