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 /* Global quiescent period parity */
  23 int urcu_qparity;
  24
  25 int __thread urcu_active_readers[2];
  26
  27 /* Thread IDs of registered readers */
  28 #define INIT_NUM_THREADS 4
  29
  30 struct reader_data {
  31         pthread_t tid;
  32         int *urcu_active_readers;
  33 };
  34
  35 #ifdef DEBUG_YIELD
  36 int yield_active;
  37 #endif
  38
  39 static struct reader_data *reader_data;
  40 static int num_readers, alloc_readers;
  41 static int sig_done;
  42
  43 void internal_urcu_lock(void)
  44 {
  45         int ret;
  46         ret = pthread_mutex_lock(&urcu_mutex);
  47         if (ret) {
  48                 perror("Error in pthread mutex lock");
  49                 exit(-1);
  50         }
  51 }
  52
  53 void internal_urcu_unlock(void)
  54 {
  55         int ret;
  56
  57         ret = pthread_mutex_unlock(&urcu_mutex);
  58         if (ret) {
  59                 perror("Error in pthread mutex unlock");
  60                 exit(-1);
  61         }
  62 }
  63
  64 /*
  65  * called with urcu_mutex held.
  66  */
  67 static int switch_next_urcu_qparity(void)
  68 {
  69         int old_parity = urcu_qparity;
  70         urcu_qparity = 1 - old_parity;
  71         return old_parity;
  72 }
  73
  74 static void force_mb_all_threads(void)
  75 {
  76         struct reader_data *index;
  77         /*
  78          * Ask for each threads to execute a mb() so we can consider the
  79          * compiler barriers around rcu read lock as real memory barriers.
  80          */
  81         if (!reader_data)
  82                 return;
  83         debug_yield_write();
  84         sig_done = 0;
  85         debug_yield_write();
  86         mb();   /* write sig_done before sending the signals */
  87         debug_yield_write();
  88         for (index = reader_data; index < reader_data + num_readers; index++) {
  89                 pthread_kill(index->tid, SIGURCU);
  90                 debug_yield_write();
  91         }
  92         /*
  93          * Wait for sighandler (and thus mb()) to execute on every thread.
  94          * BUSY-LOOP.
  95          */
  96         while (sig_done < num_readers)
  97                 barrier();
  98         debug_yield_write();
  99         mb();   /* read sig_done before ending the barrier */
 100         debug_yield_write();
 101 }
 102
 103 void wait_for_quiescent_state(int parity)
 104 {
 105         struct reader_data *index;
 106
 107         if (!reader_data)
 108                 return;
 109         /* Wait for each thread urcu_active_readers count to become 0.
 110          */
 111         for (index = reader_data; index < reader_data + num_readers; index++) {
 112                 /*
 113                  * BUSY-LOOP.
 114                  */
 115                 while (index->urcu_active_readers[parity] != 0)
 116                         barrier();
 117         }
 118         /*
 119          * Locally : read *index->urcu_active_readers before freeing old
 120          * pointer.
 121          * Remote (reader threads) : Order urcu_qparity update and other
 122          * thread's quiescent state counter read.
 123          */
 124         force_mb_all_threads();
 125 }
 126
 127 static void switch_qparity(void)
 128 {
 129         int prev_parity;
 130
 131         /* All threads should read qparity before accessing data structure. */
 132         /* Write ptr before changing the qparity */
 133         force_mb_all_threads();
 134         debug_yield_write();
 135         prev_parity = switch_next_urcu_qparity();
 136         debug_yield_write();
 137
 138         /*
 139          * Wait for previous parity to be empty of readers.
 140          */
 141         wait_for_quiescent_state(prev_parity);
 142 }
 143
 144 void synchronize_rcu(void)
 145 {
 146         debug_yield_write();
 147         internal_urcu_lock();
 148         debug_yield_write();
 149         switch_qparity();
 150         debug_yield_write();
 151         switch_qparity();
 152         debug_yield_write();
 153         internal_urcu_lock();
 154         debug_yield_write();
 155 }
 156
 157 /*
 158  * Return old pointer, OK to free, no more reference exist.
 159  * Called under rcu_write_lock.
 160  */
 161 void *urcu_publish_content(void **ptr, void *new)
 162 {
 163         void *oldptr;
 164
 165         debug_yield_write();
 166         internal_urcu_lock();
 167         debug_yield_write();
 168         /*
 169          * We can publish the new pointer before we change the current qparity.
 170          * Readers seeing the new pointer while being in the previous qparity
 171          * window will make us wait until the end of the quiescent state before
 172          * we release the unrelated memory area. However, given we hold the
 173          * urcu_mutex, we are making sure that no further garbage collection can
 174          * occur until we release the mutex, therefore we guarantee that this
 175          * given reader will have completed its execution using the new pointer
 176          * when the next quiescent state window will be over.
 177          */
 178         oldptr = *ptr;
 179         debug_yield_write();
 180         *ptr = new;
 181
 182         debug_yield_write();
 183         switch_qparity();
 184         debug_yield_write();
 185         switch_qparity();
 186         debug_yield_write();
 187         internal_urcu_unlock();
 188         debug_yield_write();
 189
 190         return oldptr;
 191 }
 192
 193 void urcu_add_reader(pthread_t id)
 194 {
 195         struct reader_data *oldarray;
 196
 197         if (!reader_data) {
 198                 alloc_readers = INIT_NUM_THREADS;
 199                 num_readers = 0;
 200                 reader_data =
 201                         malloc(sizeof(struct reader_data) * alloc_readers);
 202         }
 203         if (alloc_readers < num_readers + 1) {
 204                 oldarray = reader_data;
 205                 reader_data = malloc(sizeof(struct reader_data)
 206                                 * (alloc_readers << 1));
 207                 memcpy(reader_data, oldarray,
 208                         sizeof(struct reader_data) * alloc_readers);
 209                 alloc_readers <<= 1;
 210                 free(oldarray);
 211         }
 212         reader_data[num_readers].tid = id;
 213         /* reference to the TLS of _this_ reader thread. */
 214         reader_data[num_readers].urcu_active_readers = urcu_active_readers;
 215         num_readers++;
 216 }
 217
 218 /*
 219  * Never shrink (implementation limitation).
 220  * This is O(nb threads). Eventually use a hash table.
 221  */
 222 void urcu_remove_reader(pthread_t id)
 223 {
 224         struct reader_data *index;
 225
 226         assert(reader_data != NULL);
 227         for (index = reader_data; index < reader_data + num_readers; index++) {
 228                 if (pthread_equal(index->tid, id)) {
 229                         memcpy(index, &reader_data[num_readers - 1],
 230                                 sizeof(struct reader_data));
 231                         reader_data[num_readers - 1].tid = 0;
 232                         reader_data[num_readers - 1].urcu_active_readers = NULL;
 233                         num_readers--;
 234                         return;
 235                 }
 236         }
 237         /* Hrm not found, forgot to register ? */
 238         assert(0);
 239 }
 240
 241 void urcu_register_thread(void)
 242 {
 243         internal_urcu_lock();
 244         urcu_add_reader(pthread_self());
 245         internal_urcu_unlock();
 246 }
 247
 248 void urcu_unregister_thread(void)
 249 {
 250         internal_urcu_lock();
 251         urcu_remove_reader(pthread_self());
 252         internal_urcu_unlock();
 253 }
 254
 255 void sigurcu_handler(int signo, siginfo_t *siginfo, void *context)
 256 {
 257         mb();
 258         atomic_inc(&sig_done);
 259 }
 260
 261 void __attribute__((constructor)) urcu_init(void)
 262 {
 263         struct sigaction act;
 264         int ret;
 265
 266         act.sa_sigaction = sigurcu_handler;
 267         ret = sigaction(SIGURCU, &act, NULL);
 268         if (ret) {
 269                 perror("Error in sigaction");
 270                 exit(-1);
 271         }
 272 }
 273
 274 void __attribute__((destructor)) urcu_exit(void)
 275 {
 276         struct sigaction act;
 277         int ret;
 278
 279         ret = sigaction(SIGURCU, NULL, &act);
 280         if (ret) {
 281                 perror("Error in sigaction");
 282                 exit(-1);
 283         }
 284         assert(act.sa_sigaction == sigurcu_handler);
 285         free(reader_data);
 286 }