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 static void switch_qparity(void)
 117 {
 118         int prev_parity;
 119
 120         /* All threads should read qparity before accessing data structure. */
 121         /* Write ptr before changing the qparity */
 122         force_mb_all_threads();
 123         prev_parity = switch_next_urcu_qparity();
 124
 125         /*
 126          * Wait for previous parity to be empty of readers.
 127          */
 128         wait_for_quiescent_state(prev_parity);
 129 }
 130
 131 void synchronize_rcu(void)
 132 {
 133         rcu_write_lock();
 134         switch_qparity();
 135         switch_qparity();
 136         rcu_write_unlock();
 137 }
 138
 139 /*
 140  * Return old pointer, OK to free, no more reference exist.
 141  * Called under rcu_write_lock.
 142  */
 143 void *urcu_publish_content(void **ptr, void *new)
 144 {
 145         void *oldptr;
 146
 147         /*
 148          * We can publish the new pointer before we change the current qparity.
 149          * Readers seeing the new pointer while being in the previous qparity
 150          * window will make us wait until the end of the quiescent state before
 151          * we release the unrelated memory area. However, given we hold the
 152          * urcu_mutex, we are making sure that no further garbage collection can
 153          * occur until we release the mutex, therefore we guarantee that this
 154          * given reader will have completed its execution using the new pointer
 155          * when the next quiescent state window will be over.
 156          */
 157         oldptr = *ptr;
 158         *ptr = new;
 159
 160         switch_qparity();
 161         switch_qparity();
 162
 163         return oldptr;
 164 }
 165
 166 void urcu_add_reader(pthread_t id)
 167 {
 168         struct reader_data *oldarray;
 169
 170         if (!reader_data) {
 171                 alloc_readers = INIT_NUM_THREADS;
 172                 num_readers = 0;
 173                 reader_data =
 174                         malloc(sizeof(struct reader_data) * alloc_readers);
 175         }
 176         if (alloc_readers < num_readers + 1) {
 177                 oldarray = reader_data;
 178                 reader_data = malloc(sizeof(struct reader_data)
 179                                 * (alloc_readers << 1));
 180                 memcpy(reader_data, oldarray,
 181                         sizeof(struct reader_data) * alloc_readers);
 182                 alloc_readers <<= 1;
 183                 free(oldarray);
 184         }
 185         reader_data[num_readers].tid = id;
 186         /* reference to the TLS of _this_ reader thread. */
 187         reader_data[num_readers].urcu_active_readers = urcu_active_readers;
 188         num_readers++;
 189 }
 190
 191 /*
 192  * Never shrink (implementation limitation).
 193  * This is O(nb threads). Eventually use a hash table.
 194  */
 195 void urcu_remove_reader(pthread_t id)
 196 {
 197         struct reader_data *index;
 198
 199         assert(reader_data != NULL);
 200         for (index = reader_data; index < reader_data + num_readers; index++) {
 201                 if (pthread_equal(index->tid, id)) {
 202                         memcpy(index, &reader_data[num_readers - 1],
 203                                 sizeof(struct reader_data));
 204                         reader_data[num_readers - 1].tid = 0;
 205                         reader_data[num_readers - 1].urcu_active_readers = NULL;
 206                         num_readers--;
 207                         return;
 208                 }
 209         }
 210         /* Hrm not found, forgot to register ? */
 211         assert(0);
 212 }
 213
 214 void urcu_register_thread(void)
 215 {
 216         rcu_write_lock();
 217         urcu_add_reader(pthread_self());
 218         rcu_write_unlock();
 219 }
 220
 221 void urcu_unregister_thread(void)
 222 {
 223         rcu_write_lock();
 224         urcu_remove_reader(pthread_self());
 225         rcu_write_unlock();
 226 }
 227
 228 void sigurcu_handler(int signo, siginfo_t *siginfo, void *context)
 229 {
 230         mb();
 231         atomic_inc(&sig_done);
 232 }
 233
 234 void __attribute__((constructor)) urcu_init(void)
 235 {
 236         struct sigaction act;
 237         int ret;
 238
 239         act.sa_sigaction = sigurcu_handler;
 240         ret = sigaction(SIGURCU, &act, NULL);
 241         if (ret) {
 242                 perror("Error in sigaction");
 243                 exit(-1);
 244         }
 245 }
 246
 247 void __attribute__((destructor)) urcu_exit(void)
 248 {
 249         struct sigaction act;
 250         int ret;
 251
 252         ret = sigaction(SIGURCU, NULL, &act);
 253         if (ret) {
 254                 perror("Error in sigaction");
 255                 exit(-1);
 256         }
 257         assert(act.sa_sigaction == sigurcu_handler);
 258         free(reader_data);
 259 }