Changeset 82719589 in mainline

Timestamp:

2012-11-19T21:14:26Z (11 years ago)

Author:

Adam Hraska <adam.hraska+hos@…>

Branches:

lfn, master, serial, ticket/834-toolchain-update, topic/msim-upgrade, topic/simplify-dev-export

Children:

0adfc9d

Parents:

6831475

Message:

rcu: Made both A-RCU and Podzimek-Preempt-RCU exception safe.

Location:

kernel/generic

Files:

• include/synch/rcu_types.h (modified) (1 diff)
• src/synch/rcu.c (modified) (25 diffs)

kernel/generic/include/synch/rcu_types.h

@@ -153 +153 @@
         /** 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(). 
          * 

kernel/generic/src/synch/rcu.c

@@ -67 +67 @@
  * 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 +291 @@
 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 +511 @@
         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 +521 @@
          * 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 +530 @@
          * 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 +538 @@
         /* If there was something to signal to the detector we have done so. */
         CPU->rcu.signal_unlock = false;
-        
-        interrupts_restore(ipl);
 }
 
@@ -671 +663 @@
         
         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 +677 @@
         }
         
+        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 +843 @@
         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 +1060 @@
          * 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 +1076 @@
         /* 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 +1110 @@
         } 
         
-        upd_max_cbs_in_slice();
+        upd_max_cbs_in_slice(CPU->rcu.arriving_cbs_cnt);
 }
 
@@ -1101 +1116 @@
 void rcu_before_thread_runs(void)
 {
-        ASSERT(PREEMPTION_DISABLED || interrupts_disabled());
         ASSERT(!rcu_read_locked());
         
@@ -1115 +1129 @@
 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 +1140 @@
                 /* 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 +1157 @@
 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 +1431 @@
                 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 +1493 @@
 {
         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 +1521 @@
         _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 +1553 @@
         } 
         
-        upd_max_cbs_in_slice();
+        upd_max_cbs_in_slice(CPU->rcu.arriving_cbs_cnt);
 }
 
@@ -1542 +1564 @@
         /* 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 +1577 @@
          * 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 +1762 @@
 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 +1782 @@
         }
 
-        irq_spinlock_unlock(&rcu.preempt_lock, false);
+        irq_spinlock_unlock(&rcu.preempt_lock, true);
 }
 
@@ -1775 +1805 @@
 }
 
-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;
 }