Index: kernel/generic/include/synch/rcu_types.h =================================================================== --- kernel/generic/include/synch/rcu_types.h (revision 68314757ea344753b0b4bc5489f6b6951ee87911) +++ kernel/generic/include/synch/rcu_types.h (revision 827195897bcb3224e1e2c0e46a9c8649c05bd714) @@ -153,5 +153,5 @@ /** True if the thread was preempted in a reader section. * - * The thread is place into rcu.cur_preempted or rcu.next_preempted + * The thread is placed into rcu.cur_preempted or rcu.next_preempted * and must remove itself in rcu_read_unlock(). * Index: kernel/generic/src/synch/rcu.c =================================================================== --- kernel/generic/src/synch/rcu.c (revision 68314757ea344753b0b4bc5489f6b6951ee87911) +++ kernel/generic/src/synch/rcu.c (revision 827195897bcb3224e1e2c0e46a9c8649c05bd714) @@ -67,5 +67,6 @@ * the detector; or the cpu is still in a CS) the cpu is interrupted * via an IPI. If the IPI handler finds the cpu still in a CS, it instructs - * the cpu to notify the detector that it had exited the CS via a semaphore. + * the cpu to notify the detector that it had exited the CS via a semaphore + * (CPU->rcu.is_delaying_gp). * The detector then waits on the semaphore for any cpus to exit their * CSs. Lastly, it waits for the last reader preempted in a CS to @@ -290,5 +291,5 @@ static void note_preempted_reader(void); static void rm_preempted_reader(void); -static void upd_max_cbs_in_slice(void); +static void upd_max_cbs_in_slice(size_t arriving_cbs_cnt); @@ -510,14 +511,9 @@ ASSERT(PREEMPTION_DISABLED || interrupts_disabled()); - /* todo: make NMI safe with cpu-local atomic ops. */ - /* - * We have to disable interrupts in order to make checking - * and resetting was_preempted and is_delaying_gp atomic - * with respect to local interrupt handlers. Otherwise - * an interrupt could beat us to calling semaphore_up() - * before we reset the appropriate flag. - */ - ipl_t ipl = interrupts_disable(); + * If an interrupt occurs here (even a NMI) it may beat us to + * resetting .is_delaying_gp or .was_preempted and up the semaphore + * for us. + */ /* @@ -525,6 +521,5 @@ * notify it that the reader did so. */ - if (CPU->rcu.is_delaying_gp) { - CPU->rcu.is_delaying_gp = false; + if (local_atomic_exchange(&CPU->rcu.is_delaying_gp, false)) { semaphore_up(&rcu.remaining_readers); } @@ -535,7 +530,6 @@ * detector if so. */ - if (THREAD && THREAD->rcu.was_preempted) { + if (THREAD && local_atomic_exchange(&THREAD->rcu.was_preempted, false)) { ASSERT(link_used(&THREAD->rcu.preempt_link)); - THREAD->rcu.was_preempted = false; rm_preempted_reader(); @@ -544,6 +538,4 @@ /* If there was something to signal to the detector we have done so. */ CPU->rcu.signal_unlock = false; - - interrupts_restore(ipl); } @@ -671,13 +663,13 @@ preemption_disable(); - - ipl_t ipl = interrupts_disable(); rcu_cpu_data_t *r = &CPU->rcu; - *r->parriving_cbs_tail = rcu_item; - r->parriving_cbs_tail = &rcu_item->next; - - size_t cnt = ++r->arriving_cbs_cnt; - interrupts_restore(ipl); + + rcu_item_t **prev_tail + = local_atomic_exchange(&r->parriving_cbs_tail, &rcu_item->next); + *prev_tail = rcu_item; + + /* Approximate the number of callbacks present. */ + ++r->arriving_cbs_cnt; if (expedite) { @@ -685,6 +677,8 @@ } + bool first_cb = (prev_tail == &CPU->rcu.arriving_cbs); + /* Added first callback - notify the reclaimer. */ - if (cnt == 1 && !semaphore_count_get(&r->arrived_flag)) { + if (first_cb && !semaphore_count_get(&r->arrived_flag)) { semaphore_up(&r->arrived_flag); } @@ -849,24 +843,40 @@ CPU->rcu.cur_cbs_gp = CPU->rcu.next_cbs_gp; - /* Move arriving_cbs to next_cbs. Empties arriving_cbs. */ - ipl_t ipl = interrupts_disable(); - + /* Move arriving_cbs to next_cbs. */ + + CPU->rcu.next_cbs_cnt = CPU->rcu.arriving_cbs_cnt; + CPU->rcu.arriving_cbs_cnt = 0; + /* * Too many callbacks queued. Better speed up the detection * or risk exhausting all system memory. */ - bool expedite = (EXPEDITE_THRESHOLD < CPU->rcu.arriving_cbs_cnt) + bool expedite = (EXPEDITE_THRESHOLD < CPU->rcu.next_cbs_cnt) || CPU->rcu.expedite_arriving; - CPU->rcu.expedite_arriving = false; - + + /* Start moving the arriving_cbs list to next_cbs. */ CPU->rcu.next_cbs = CPU->rcu.arriving_cbs; - CPU->rcu.next_cbs_cnt = CPU->rcu.arriving_cbs_cnt; - - CPU->rcu.arriving_cbs = NULL; - CPU->rcu.parriving_cbs_tail = &CPU->rcu.arriving_cbs; - CPU->rcu.arriving_cbs_cnt = 0; - - interrupts_restore(ipl); + + /* + * At least one callback arrived. The tail therefore does not point + * to the head of arriving_cbs and we can safely reset it to NULL. + */ + if (CPU->rcu.next_cbs) { + ASSERT(CPU->rcu.parriving_cbs_tail != &CPU->rcu.arriving_cbs); + + CPU->rcu.arriving_cbs = NULL; + /* Reset arriving_cbs before updating the tail pointer. */ + compiler_barrier(); + /* Updating the tail pointer completes the move of arriving_cbs. */ + ACCESS_ONCE(CPU->rcu.parriving_cbs_tail) = &CPU->rcu.arriving_cbs; + } else { + /* + * arriving_cbs was null and parriving_cbs_tail pointed to it + * so leave it that way. Note that interrupt handlers may have + * added a callback in the meantime so it is not safe to reset + * arriving_cbs or parriving_cbs. + */ + } /* Update statistics of arrived callbacks. */ @@ -1050,5 +1060,11 @@ * with a local copy. */ - size_t nesting_cnt = ACCESS_ONCE(THE->rcu_nesting); + size_t nesting_cnt = local_atomic_exchange(&THE->rcu_nesting, 0); + + /* + * Ensures NMIs see .rcu_nesting without the WAS_PREEMPTED mark and + * do not accidentally call rm_preempted_reader() from unlock(). + */ + compiler_barrier(); /* Preempted a reader critical section for the first time. */ @@ -1060,5 +1076,4 @@ /* Save the thread's nesting count when it is not running. */ THREAD->rcu.nesting_cnt = nesting_cnt; - ACCESS_ONCE(THE->rcu_nesting) = 0; if (CPU->rcu.last_seen_gp != _rcu_cur_gp) { @@ -1095,5 +1110,5 @@ } - upd_max_cbs_in_slice(); + upd_max_cbs_in_slice(CPU->rcu.arriving_cbs_cnt); } @@ -1101,5 +1116,4 @@ void rcu_before_thread_runs(void) { - ASSERT(PREEMPTION_DISABLED || interrupts_disabled()); ASSERT(!rcu_read_locked()); @@ -1115,5 +1129,6 @@ void rcu_thread_exiting(void) { - ASSERT(PREEMPTION_DISABLED || interrupts_disabled()); + ASSERT(THE->rcu_nesting == 0); + /* * The thread forgot to exit its reader critical section. @@ -1125,7 +1140,5 @@ /* Emulate _rcu_preempted_unlock() with the proper nesting count. */ if (THREAD->rcu.nesting_cnt & RCU_WAS_PREEMPTED) { - ipl_t ipl = interrupts_disable(); rm_preempted_reader(); - interrupts_restore(ipl); } @@ -1144,19 +1157,22 @@ void _rcu_preempted_unlock(void) { - ipl_t ipl = interrupts_disable(); - - /* todo: Replace with cpu-local atomics to be NMI-safe */ - if (THE->rcu_nesting == RCU_WAS_PREEMPTED) { - THE->rcu_nesting = 0; + ASSERT(0 == THE->rcu_nesting || RCU_WAS_PREEMPTED == THE->rcu_nesting); + + size_t prev = local_atomic_exchange(&THE->rcu_nesting, 0); + if (prev == RCU_WAS_PREEMPTED) { /* * NMI handlers are never preempted but may call rm_preempted_reader() * if a NMI occurred in _rcu_preempted_unlock() of a preempted thread. + * The only other rcu code that may have been interrupted by the NMI + * in _rcu_preempted_unlock() is: an IPI/sample_local_cpu() and + * the initial part of rcu_after_thread_ran(). + * * rm_preempted_reader() will not deadlock because none of the locks - * it uses are locked in this case. + * it uses are locked in this case. Neither _rcu_preempted_unlock() + * nor sample_local_cpu() nor the initial part of rcu_after_thread_ran() + * acquire any locks. */ rm_preempted_reader(); } - - interrupts_restore(ipl); } @@ -1415,12 +1431,11 @@ if (0 < CPU->rcu.nesting_cnt) { ASSERT(!CPU->idle); - /* Note to notify the detector from rcu_read_unlock(). */ - CPU->rcu.is_delaying_gp = true; /* - * Set signal_unlock only after setting is_delaying_gp so - * that NMI handlers do not accidentally clear it in unlock() - * before seeing and acting upon is_delaying_gp. + * Note to notify the detector from rcu_read_unlock(). + * + * ACCESS_ONCE ensures the compiler writes to is_delaying_gp + * only after it determines that we are in a reader CS. */ - compiler_barrier(); + ACCESS_ONCE(CPU->rcu.is_delaying_gp) = true; CPU->rcu.signal_unlock = true; @@ -1478,19 +1493,19 @@ { ASSERT(interrupts_disabled()); - /* todo: make is_delaying_gp and was_preempted NMI safe via local atomics.*/ /* * Prevent NMI handlers from interfering. The detector will be notified - * here if CPU->rcu.is_delaying_gp and the current thread is no longer - * running so there is nothing to signal to the detector. + * in this function if CPU->rcu.is_delaying_gp. The current thread is + * no longer running so there is nothing else to signal to the detector. */ CPU->rcu.signal_unlock = false; - /* Separates clearing of .signal_unlock from CPU->rcu.nesting_cnt = 0. */ + /* + * Separates clearing of .signal_unlock from accesses to + * THREAD->rcu.was_preempted and CPU->rcu.nesting_cnt. + */ compiler_barrier(); /* Save the thread's nesting count when it is not running. */ THREAD->rcu.nesting_cnt = CPU->rcu.nesting_cnt; - /* Interrupt handlers might use RCU while idle in scheduler(). */ - CPU->rcu.nesting_cnt = 0; /* Preempted a reader critical section for the first time. */ @@ -1506,4 +1521,11 @@ _rcu_record_qs(); + /* + * Interrupt handlers might use RCU while idle in scheduler(). + * The preempted reader has been noted globally, so the handlers + * may now start announcing quiescent states. + */ + CPU->rcu.nesting_cnt = 0; + /* * This cpu is holding up the current GP. Let the detector know @@ -1531,5 +1553,5 @@ } - upd_max_cbs_in_slice(); + upd_max_cbs_in_slice(CPU->rcu.arriving_cbs_cnt); } @@ -1542,4 +1564,12 @@ /* Load the thread's saved nesting count from before it was preempted. */ CPU->rcu.nesting_cnt = THREAD->rcu.nesting_cnt; + + /* + * Ensures NMI see the proper nesting count before .signal_unlock. + * Otherwise the NMI may incorrectly signal that a preempted reader + * exited its reader section. + */ + compiler_barrier(); + /* * In the unlikely event that a NMI occurs between the loading of the @@ -1547,5 +1577,5 @@ * rcu_read_unlock() and clear signal_unlock. In that case we will * incorrectly overwrite signal_unlock from false to true. This event - * situation benign and the next rcu_read_unlock() will at worst + * is benign and the next rcu_read_unlock() will at worst * needlessly invoke _rcu_signal_unlock(). */ @@ -1732,5 +1762,5 @@ static void rm_preempted_reader(void) { - irq_spinlock_lock(&rcu.preempt_lock, false); + irq_spinlock_lock(&rcu.preempt_lock, true); ASSERT(link_used(&THREAD->rcu.preempt_link)); @@ -1752,5 +1782,5 @@ } - irq_spinlock_unlock(&rcu.preempt_lock, false); + irq_spinlock_unlock(&rcu.preempt_lock, true); } @@ -1775,14 +1805,14 @@ } -static void upd_max_cbs_in_slice(void) +static void upd_max_cbs_in_slice(size_t arriving_cbs_cnt) { rcu_cpu_data_t *cr = &CPU->rcu; - if (cr->arriving_cbs_cnt > cr->last_arriving_cnt) { - size_t arrived_cnt = cr->arriving_cbs_cnt - cr->last_arriving_cnt; + if (arriving_cbs_cnt > cr->last_arriving_cnt) { + size_t arrived_cnt = arriving_cbs_cnt - cr->last_arriving_cnt; cr->stat_max_slice_cbs = max(arrived_cnt, cr->stat_max_slice_cbs); } - cr->last_arriving_cnt = cr->arriving_cbs_cnt; + cr->last_arriving_cnt = arriving_cbs_cnt; }