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