X-Git-Url: http://git.liburcu.org/?p=urcu.git;a=blobdiff_plain;f=urcu.c;h=475bebf526dfd1b9cc8cf78b87d8804ded7a1b6f;hp=018e09b8601b041621fc6f72afdaeb57839e895c;hb=2d6debff95ad695255d2ea9d590d1e418590b238;hpb=ae878d0dcc291dba7c5722eed13294195a64409e diff --git a/urcu.c b/urcu.c index 018e09b..475bebf 100644 --- a/urcu.c +++ b/urcu.c @@ -23,9 +23,14 @@ pthread_mutex_t urcu_mutex = PTHREAD_MUTEX_INITIALIZER; * Global grace period counter. * Contains the current RCU_GP_CTR_BIT. * Also has a RCU_GP_CTR_BIT of 1, to accelerate the reader fast path. + * Written to only by writer with mutex taken. Read by both writer and readers. */ long urcu_gp_ctr = RCU_GP_COUNT; +/* + * Written to only by each individual reader. Read by both the reader and the + * writers. + */ long __thread urcu_active_readers; /* Thread IDs of registered readers */ @@ -49,12 +54,17 @@ static int sig_done; void internal_urcu_lock(void) { +#if 0 int ret; + /* Mutex sleeping does not play well with busy-waiting loop. */ ret = pthread_mutex_lock(&urcu_mutex); if (ret) { perror("Error in pthread mutex lock"); exit(-1); } +#endif + while (pthread_mutex_trylock(&urcu_mutex) != 0) + cpu_relax(); } void internal_urcu_unlock(void) @@ -73,15 +83,43 @@ void internal_urcu_unlock(void) */ static void switch_next_urcu_qparity(void) { - urcu_gp_ctr ^= RCU_GP_CTR_BIT; + STORE_SHARED(urcu_gp_ctr, urcu_gp_ctr ^ RCU_GP_CTR_BIT); } #ifdef DEBUG_FULL_MB +static void force_mb_single_thread(pthread_t tid) +{ + smp_mb(); +} + static void force_mb_all_threads(void) { smp_mb(); } #else + +static void force_mb_single_thread(pthread_t tid) +{ + assert(reader_data); + sig_done = 0; + /* + * pthread_kill has a smp_mb(). But beware, we assume it performs + * a cache flush on architectures with non-coherent cache. Let's play + * safe and don't assume anything : we use smp_mc() to make sure the + * cache flush is enforced. + * smp_mb(); write sig_done before sending the signals + */ + smp_mc(); /* write sig_done before sending the signals */ + pthread_kill(tid, SIGURCU); + /* + * Wait for sighandler (and thus mb()) to execute on every thread. + * BUSY-LOOP. + */ + while (LOAD_SHARED(sig_done) < 1) + cpu_relax(); + smp_mb(); /* read sig_done before ending the barrier */ +} + static void force_mb_all_threads(void) { struct reader_data *index; @@ -92,15 +130,22 @@ static void force_mb_all_threads(void) if (!reader_data) return; sig_done = 0; - smp_mb(); /* write sig_done before sending the signals */ + /* + * pthread_kill has a smp_mb(). But beware, we assume it performs + * a cache flush on architectures with non-coherent cache. Let's play + * safe and don't assume anything : we use smp_mc() to make sure the + * cache flush is enforced. + * smp_mb(); write sig_done before sending the signals + */ + smp_mc(); /* write sig_done before sending the signals */ for (index = reader_data; index < reader_data + num_readers; index++) pthread_kill(index->tid, SIGURCU); /* * Wait for sighandler (and thus mb()) to execute on every thread. * BUSY-LOOP. */ - while (sig_done < num_readers) - barrier(); + while (LOAD_SHARED(sig_done) < num_readers) + cpu_relax(); smp_mb(); /* read sig_done before ending the barrier */ } #endif @@ -111,26 +156,36 @@ void wait_for_quiescent_state(void) if (!reader_data) return; - /* Wait for each thread urcu_active_readers count to become 0. + /* + * Wait for each thread urcu_active_readers count to become 0. */ for (index = reader_data; index < reader_data + num_readers; index++) { + int wait_loops = 0; /* - * BUSY-LOOP. + * BUSY-LOOP. Force the reader thread to commit its + * urcu_active_readers update to memory if we wait for too long. */ - while (rcu_old_gp_ongoing(index->urcu_active_readers)) - barrier(); + while (rcu_old_gp_ongoing(index->urcu_active_readers)) { + if (wait_loops++ == KICK_READER_LOOPS) { + force_mb_single_thread(index->tid); + wait_loops = 0; + } else { + cpu_relax(); + } + } } } void synchronize_rcu(void) { + internal_urcu_lock(); + /* All threads should read qparity before accessing data structure - * where new ptr points to. */ + * where new ptr points to. Must be done within internal_urcu_lock + * because it iterates on reader threads.*/ /* Write new ptr before changing the qparity */ force_mb_all_threads(); - internal_urcu_lock(); - switch_next_urcu_qparity(); /* 0 -> 1 */ /* @@ -138,8 +193,8 @@ void synchronize_rcu(void) * 0 quiescent state. Failure to do so could result in the writer * waiting forever while new readers are always accessing data (no * progress). + * Ensured by STORE_SHARED and LOAD_SHARED. */ - smp_mb(); /* * Wait for previous parity to be empty of readers. @@ -151,8 +206,8 @@ void synchronize_rcu(void) * committing qparity update to memory. Failure to do so could result in * the writer waiting forever while new readers are always accessing * data (no progress). + * Ensured by STORE_SHARED and LOAD_SHARED. */ - smp_mb(); switch_next_urcu_qparity(); /* 1 -> 0 */ @@ -161,21 +216,20 @@ void synchronize_rcu(void) * 1 quiescent state. Failure to do so could result in the writer * waiting forever while new readers are always accessing data (no * progress). + * Ensured by STORE_SHARED and LOAD_SHARED. */ - smp_mb(); /* * Wait for previous parity to be empty of readers. */ wait_for_quiescent_state(); /* Wait readers in parity 1 */ - internal_urcu_unlock(); - - /* All threads should finish using the data referred to by old ptr - * before decrementing their urcu_active_readers count */ /* Finish waiting for reader threads before letting the old ptr being - * freed. */ + * freed. Must be done within internal_urcu_lock because it iterates on + * reader threads. */ force_mb_all_threads(); + + internal_urcu_unlock(); } void urcu_add_reader(pthread_t id) @@ -243,6 +297,11 @@ void urcu_unregister_thread(void) #ifndef DEBUG_FULL_MB void sigurcu_handler(int signo, siginfo_t *siginfo, void *context) { + /* + * Executing this smp_mb() is the only purpose of this signal handler. + * It punctually promotes barrier() into smp_mb() on every thread it is + * executed on. + */ smp_mb(); atomic_inc(&sig_done); }