From 52c75091b18f3cc398b765570a94aaa870afcd51 Mon Sep 17 00:00:00 2001 From: Mathieu Desnoyers Date: Sun, 11 Nov 2012 22:33:34 -0500 Subject: [PATCH] urcu-bp: move quiescent threads to separate list Accelerate 2-phase grace period by not having to iterate twice on threads not within RCU read-side critical section. CC: Paul E. McKenney CC: Lai Jiangshan CC: Alan Stern Signed-off-by: Mathieu Desnoyers --- urcu-bp.c | 46 ++++++++++++++++++++++++++++++++----------- urcu/static/urcu-bp.h | 29 +++++++++++++++++---------- 2 files changed, 54 insertions(+), 21 deletions(-) diff --git a/urcu-bp.c b/urcu-bp.c index 4b3cf01..f99c0e5 100644 --- a/urcu-bp.c +++ b/urcu-bp.c @@ -115,7 +115,7 @@ DEFINE_URCU_TLS(unsigned int, rcu_rand_yield); * Also has a RCU_GP_COUNT of 1, to accelerate the reader fast path. * Written to only by writer with mutex taken. Read by both writer and readers. */ -long rcu_gp_ctr = RCU_GP_COUNT; +unsigned long rcu_gp_ctr = RCU_GP_COUNT; /* * Pointer to registry elements. Written to only by each individual reader. Read @@ -164,9 +164,10 @@ static void mutex_unlock(pthread_mutex_t *mutex) urcu_die(ret); } -static void wait_for_readers(void) +static void wait_for_readers(struct cds_list_head *input_readers, + struct cds_list_head *cur_snap_readers, + struct cds_list_head *qsreaders) { - CDS_LIST_HEAD(qsreaders); int wait_loops = 0; struct rcu_reader *index, *tmp; @@ -177,12 +178,30 @@ static void wait_for_readers(void) */ for (;;) { wait_loops++; - cds_list_for_each_entry_safe(index, tmp, ®istry, node) { - if (!rcu_old_gp_ongoing(&index->ctr)) - cds_list_move(&index->node, &qsreaders); + cds_list_for_each_entry_safe(index, tmp, input_readers, node) { + switch (rcu_reader_state(&index->ctr)) { + case RCU_READER_ACTIVE_CURRENT: + if (cur_snap_readers) { + cds_list_move(&index->node, + cur_snap_readers); + break; + } + /* Fall-through */ + case RCU_READER_INACTIVE: + cds_list_move(&index->node, qsreaders); + break; + case RCU_READER_ACTIVE_OLD: + /* + * Old snapshot. Leaving node in + * input_readers will make us busy-loop + * until the snapshot becomes current or + * the reader becomes inactive. + */ + break; + } } - if (cds_list_empty(®istry)) { + if (cds_list_empty(input_readers)) { break; } else { if (wait_loops == RCU_QS_ACTIVE_ATTEMPTS) @@ -191,12 +210,12 @@ static void wait_for_readers(void) caa_cpu_relax(); } } - /* put back the reader list in the registry */ - cds_list_splice(&qsreaders, ®istry); } void synchronize_rcu(void) { + CDS_LIST_HEAD(cur_snap_readers); + CDS_LIST_HEAD(qsreaders); sigset_t newmask, oldmask; int ret; @@ -221,7 +240,7 @@ void synchronize_rcu(void) /* * Wait for readers to observe original parity or be quiescent. */ - wait_for_readers(); + wait_for_readers(®istry, &cur_snap_readers, &qsreaders); /* * Adding a cmm_smp_mb() which is _not_ formally required, but makes the @@ -250,7 +269,12 @@ void synchronize_rcu(void) /* * Wait for readers to observe new parity or be quiescent. */ - wait_for_readers(); + wait_for_readers(&cur_snap_readers, NULL, &qsreaders); + + /* + * Put quiescent reader list back into registry. + */ + cds_list_splice(&qsreaders, ®istry); /* * Finish waiting for reader threads before letting the old ptr being diff --git a/urcu/static/urcu-bp.h b/urcu/static/urcu-bp.h index c52a688..c7f5326 100644 --- a/urcu/static/urcu-bp.h +++ b/urcu/static/urcu-bp.h @@ -58,6 +58,12 @@ extern "C" { #define rcu_assert(args...) #endif +enum rcu_state { + RCU_READER_ACTIVE_CURRENT, + RCU_READER_ACTIVE_OLD, + RCU_READER_INACTIVE, +}; + #ifdef DEBUG_YIELD #include #include @@ -129,11 +135,11 @@ extern void rcu_bp_register(void); * Using a int rather than a char to eliminate false register dependencies * causing stalls on some architectures. */ -extern long rcu_gp_ctr; +extern unsigned long rcu_gp_ctr; struct rcu_reader { /* Data used by both reader and synchronize_rcu() */ - long ctr; + unsigned long ctr; /* Data used for registry */ struct cds_list_head node __attribute__((aligned(CAA_CACHE_LINE_SIZE))); pthread_t tid; @@ -147,19 +153,22 @@ struct rcu_reader { */ extern DECLARE_URCU_TLS(struct rcu_reader *, rcu_reader); -static inline int rcu_old_gp_ongoing(long *value) +static inline enum rcu_state rcu_reader_state(unsigned long *ctr) { - long v; + unsigned long v; - if (value == NULL) - return 0; + if (ctr == NULL) + return RCU_READER_INACTIVE; /* * Make sure both tests below are done on the same version of *value * to insure consistency. */ - v = CMM_LOAD_SHARED(*value); - return (v & RCU_GP_CTR_NEST_MASK) && - ((v ^ rcu_gp_ctr) & RCU_GP_CTR_PHASE); + v = CMM_LOAD_SHARED(*ctr); + if (!(v & RCU_GP_CTR_NEST_MASK)) + return RCU_READER_INACTIVE; + if (!((v ^ rcu_gp_ctr) & RCU_GP_CTR_PHASE)) + return RCU_READER_ACTIVE_CURRENT; + return RCU_READER_ACTIVE_OLD; } /* @@ -190,7 +199,7 @@ static inline void _rcu_read_lock_update(unsigned long tmp) */ static inline void _rcu_read_lock(void) { - long tmp; + unsigned long tmp; if (caa_unlikely(!URCU_TLS(rcu_reader))) rcu_bp_register(); /* If not yet registered. */ -- 2.34.1