urcu.c

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