X-Git-Url: http://git.liburcu.org/?p=urcu.git;a=blobdiff_plain;f=urcu.c;h=0e42207e2bcd59d82127dc308def83b85c424798;hp=23a985b9f9f4cdaf992c227322720e38bc139d85;hb=40e140c9f237bbe8c50d4a5355561139f01afb70;hpb=e462817ecdce42b00f0cb0825e2425b08943105b diff --git a/urcu.c b/urcu.c index 23a985b..0e42207 100644 --- a/urcu.c +++ b/urcu.c @@ -19,17 +19,21 @@ pthread_mutex_t urcu_mutex = PTHREAD_MUTEX_INITIALIZER; -/* Global grace period counter */ -int urcu_gp_ctr; +/* + * 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. + */ +long urcu_gp_ctr = RCU_GP_COUNT; -int __thread urcu_active_readers; +long __thread urcu_active_readers; /* Thread IDs of registered readers */ #define INIT_NUM_THREADS 4 struct reader_data { pthread_t tid; - int *urcu_active_readers; + long *urcu_active_readers; }; #ifdef DEBUG_YIELD @@ -39,7 +43,9 @@ unsigned int __thread rand_yield; static struct reader_data *reader_data; static int num_readers, alloc_readers; +#ifndef DEBUG_FULL_MB static int sig_done; +#endif void internal_urcu_lock(void) { @@ -70,34 +76,55 @@ static void switch_next_urcu_qparity(void) 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; + smp_mb(); /* 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 (sig_done < 1) + smp_rmb(); /* ensure we re-read sig-done */ + smp_mb(); /* read sig_done before ending the barrier */ +} + static void force_mb_all_threads(void) { struct reader_data *index; /* - * Ask for each threads to execute a mb() so we can consider the + * Ask for each threads to execute a smp_mb() so we can consider the * compiler barriers around rcu read lock as real memory barriers. */ if (!reader_data) return; - debug_yield_write(); sig_done = 0; - debug_yield_write(); - mb(); /* write sig_done before sending the signals */ - debug_yield_write(); - for (index = reader_data; index < reader_data + num_readers; index++) { + smp_mb(); /* write sig_done before sending the signals */ + for (index = reader_data; index < reader_data + num_readers; index++) pthread_kill(index->tid, SIGURCU); - debug_yield_write(); - } /* * Wait for sighandler (and thus mb()) to execute on every thread. * BUSY-LOOP. */ while (sig_done < num_readers) - barrier(); - debug_yield_write(); - mb(); /* read sig_done before ending the barrier */ - debug_yield_write(); + smp_rmb(); /* ensure we re-read sig-done */ + smp_mb(); /* read sig_done before ending the barrier */ } +#endif void wait_for_quiescent_state(void) { @@ -105,86 +132,78 @@ 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; + } + } } - /* - * Locally : read *index->urcu_active_readers before freeing old - * pointer. - * Remote (reader threads) : Order urcu_qparity update and other - * thread's quiescent state counter read. - */ - force_mb_all_threads(); } -static void switch_qparity(void) +void synchronize_rcu(void) { - /* All threads should read qparity before accessing data structure. */ - /* Write ptr before changing the qparity */ + /* All threads should read qparity before accessing data structure + * where new ptr points to. */ + /* Write new ptr before changing the qparity */ force_mb_all_threads(); - debug_yield_write(); - switch_next_urcu_qparity(); - debug_yield_write(); + + internal_urcu_lock(); + + switch_next_urcu_qparity(); /* 0 -> 1 */ + + /* + * Must commit qparity update to memory before waiting for parity + * 0 quiescent state. Failure to do so could result in the writer + * waiting forever while new readers are always accessing data (no + * progress). + */ + smp_mb(); /* * Wait for previous parity to be empty of readers. */ - wait_for_quiescent_state(); -} + wait_for_quiescent_state(); /* Wait readers in parity 0 */ -void synchronize_rcu(void) -{ - debug_yield_write(); - internal_urcu_lock(); - debug_yield_write(); - switch_qparity(); - debug_yield_write(); - switch_qparity(); - debug_yield_write(); - internal_urcu_lock(); - debug_yield_write(); -} + /* + * Must finish waiting for quiescent state for parity 0 before + * 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). + */ + smp_mb(); -/* - * Return old pointer, OK to free, no more reference exist. - * Called under rcu_write_lock. - */ -void *urcu_publish_content(void **ptr, void *new) -{ - void *oldptr; + switch_next_urcu_qparity(); /* 1 -> 0 */ - debug_yield_write(); - internal_urcu_lock(); - debug_yield_write(); /* - * We can publish the new pointer before we change the current qparity. - * Readers seeing the new pointer while being in the previous qparity - * window will make us wait until the end of the quiescent state before - * we release the unrelated memory area. However, given we hold the - * urcu_mutex, we are making sure that no further garbage collection can - * occur until we release the mutex, therefore we guarantee that this - * given reader will have completed its execution using the new pointer - * when the next quiescent state window will be over. + * Must commit qparity update to memory before waiting for parity + * 1 quiescent state. Failure to do so could result in the writer + * waiting forever while new readers are always accessing data (no + * progress). */ - oldptr = *ptr; - debug_yield_write(); - rcu_assign_pointer(*ptr, new); - - debug_yield_write(); - switch_qparity(); - debug_yield_write(); - switch_qparity(); - debug_yield_write(); + smp_mb(); + + /* + * Wait for previous parity to be empty of readers. + */ + wait_for_quiescent_state(); /* Wait readers in parity 1 */ + internal_urcu_unlock(); - debug_yield_write(); - return oldptr; + /* 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. */ + force_mb_all_threads(); } void urcu_add_reader(pthread_t id) @@ -249,9 +268,15 @@ void urcu_unregister_thread(void) internal_urcu_unlock(); } +#ifndef DEBUG_FULL_MB void sigurcu_handler(int signo, siginfo_t *siginfo, void *context) { - mb(); + /* + * 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); } @@ -281,3 +306,4 @@ void __attribute__((destructor)) urcu_exit(void) assert(act.sa_sigaction == sigurcu_handler); free(reader_data); } +#endif