audit: fix info leak in AUDIT_GET requests
[linux-3.10.git] / kernel / rcutree.c
index 812d04b..3538001 100644 (file)
 static struct lock_class_key rcu_node_class[RCU_NUM_LVLS];
 static struct lock_class_key rcu_fqs_class[RCU_NUM_LVLS];
 
-#define RCU_STATE_INITIALIZER(sname, cr) { \
+#define RCU_STATE_INITIALIZER(sname, sabbr, cr) { \
        .level = { &sname##_state.node[0] }, \
        .call = cr, \
        .fqs_state = RCU_GP_IDLE, \
-       .gpnum = -300, \
-       .completed = -300, \
-       .onofflock = __RAW_SPIN_LOCK_UNLOCKED(&sname##_state.onofflock), \
+       .gpnum = 0UL - 300UL, \
+       .completed = 0UL - 300UL, \
+       .orphan_lock = __RAW_SPIN_LOCK_UNLOCKED(&sname##_state.orphan_lock), \
        .orphan_nxttail = &sname##_state.orphan_nxtlist, \
        .orphan_donetail = &sname##_state.orphan_donelist, \
        .barrier_mutex = __MUTEX_INITIALIZER(sname##_state.barrier_mutex), \
+       .onoff_mutex = __MUTEX_INITIALIZER(sname##_state.onoff_mutex), \
        .name = #sname, \
+       .abbr = sabbr, \
 }
 
 struct rcu_state rcu_sched_state =
-       RCU_STATE_INITIALIZER(rcu_sched, call_rcu_sched);
+       RCU_STATE_INITIALIZER(rcu_sched, 's', call_rcu_sched);
 DEFINE_PER_CPU(struct rcu_data, rcu_sched_data);
 
-struct rcu_state rcu_bh_state = RCU_STATE_INITIALIZER(rcu_bh, call_rcu_bh);
+struct rcu_state rcu_bh_state = RCU_STATE_INITIALIZER(rcu_bh, 'b', call_rcu_bh);
 DEFINE_PER_CPU(struct rcu_data, rcu_bh_data);
 
 static struct rcu_state *rcu_state;
@@ -104,7 +106,7 @@ int rcu_num_nodes __read_mostly = NUM_RCU_NODES; /* Total # rcu_nodes in use. */
  * The rcu_scheduler_active variable transitions from zero to one just
  * before the first task is spawned.  So when this variable is zero, RCU
  * can assume that there is but one task, allowing RCU to (for example)
- * optimized synchronize_sched() to a simple barrier().  When this variable
+ * optimize synchronize_sched() to a simple barrier().  When this variable
  * is one, RCU must actually do all the hard work required to detect real
  * grace periods.  This variable is also used to suppress boot-time false
  * positives from lockdep-RCU error checking.
@@ -206,24 +208,15 @@ EXPORT_SYMBOL_GPL(rcu_note_context_switch);
 DEFINE_PER_CPU(struct rcu_dynticks, rcu_dynticks) = {
        .dynticks_nesting = DYNTICK_TASK_EXIT_IDLE,
        .dynticks = ATOMIC_INIT(1),
-#if defined(CONFIG_RCU_USER_QS) && !defined(CONFIG_RCU_USER_QS_FORCE)
-       .ignore_user_qs = true,
-#endif
 };
 
-static int blimit = 10;                /* Maximum callbacks per rcu_do_batch. */
-static int qhimark = 10000;    /* If this many pending, ignore blimit. */
-static int qlowmark = 100;     /* Once only this many pending, use blimit. */
+static long blimit = 10;       /* Maximum callbacks per rcu_do_batch. */
+static long qhimark = 10000;   /* If this many pending, ignore blimit. */
+static long qlowmark = 100;    /* Once only this many pending, use blimit. */
 
-module_param(blimit, int, 0444);
-module_param(qhimark, int, 0444);
-module_param(qlowmark, int, 0444);
-
-int rcu_cpu_stall_suppress __read_mostly; /* 1 = suppress stall warnings. */
-int rcu_cpu_stall_timeout __read_mostly = CONFIG_RCU_CPU_STALL_TIMEOUT;
-
-module_param(rcu_cpu_stall_suppress, int, 0644);
-module_param(rcu_cpu_stall_timeout, int, 0644);
+module_param(blimit, long, 0444);
+module_param(qhimark, long, 0444);
+module_param(qlowmark, long, 0444);
 
 static ulong jiffies_till_first_fqs = RCU_JIFFIES_TILL_FORCE_QS;
 static ulong jiffies_till_next_fqs = RCU_JIFFIES_TILL_FORCE_QS;
@@ -231,6 +224,8 @@ static ulong jiffies_till_next_fqs = RCU_JIFFIES_TILL_FORCE_QS;
 module_param(jiffies_till_first_fqs, ulong, 0644);
 module_param(jiffies_till_next_fqs, ulong, 0644);
 
+static void rcu_start_gp_advanced(struct rcu_state *rsp, struct rcu_node *rnp,
+                                 struct rcu_data *rdp);
 static void force_qs_rnp(struct rcu_state *rsp, int (*f)(struct rcu_data *));
 static void force_quiescent_state(struct rcu_state *rsp);
 static int rcu_pending(int cpu);
@@ -302,18 +297,34 @@ EXPORT_SYMBOL_GPL(rcu_sched_force_quiescent_state);
 static int
 cpu_has_callbacks_ready_to_invoke(struct rcu_data *rdp)
 {
-       return &rdp->nxtlist != rdp->nxttail[RCU_DONE_TAIL];
+       return &rdp->nxtlist != rdp->nxttail[RCU_DONE_TAIL] &&
+              rdp->nxttail[RCU_DONE_TAIL] != NULL;
 }
 
 /*
- * Does the current CPU require a yet-as-unscheduled grace period?
+ * Does the current CPU require a not-yet-started grace period?
+ * The caller must have disabled interrupts to prevent races with
+ * normal callback registry.
  */
 static int
 cpu_needs_another_gp(struct rcu_state *rsp, struct rcu_data *rdp)
 {
-       return *rdp->nxttail[RCU_DONE_TAIL +
-                            ACCESS_ONCE(rsp->completed) != rdp->completed] &&
-              !rcu_gp_in_progress(rsp);
+       int i;
+
+       if (rcu_gp_in_progress(rsp))
+               return 0;  /* No, a grace period is already in progress. */
+       if (rcu_nocb_needs_gp(rsp))
+               return 1;  /* Yes, a no-CBs CPU needs one. */
+       if (!rdp->nxttail[RCU_NEXT_TAIL])
+               return 0;  /* No, this is a no-CBs (or offline) CPU. */
+       if (*rdp->nxttail[RCU_NEXT_READY_TAIL])
+               return 1;  /* Yes, this CPU has newly registered callbacks. */
+       for (i = RCU_WAIT_TAIL; i < RCU_NEXT_TAIL; i++)
+               if (rdp->nxttail[i - 1] != rdp->nxttail[i] &&
+                   ULONG_CMP_LT(ACCESS_ONCE(rsp->completed),
+                                rdp->nxtcompleted[i]))
+                       return 1;  /* Yes, CBs for future grace period. */
+       return 0; /* No grace period needed. */
 }
 
 /*
@@ -334,8 +345,8 @@ static struct rcu_node *rcu_get_root(struct rcu_state *rsp)
 static void rcu_eqs_enter_common(struct rcu_dynticks *rdtp, long long oldval,
                                bool user)
 {
-       trace_rcu_dyntick("Start", oldval, 0);
-       if (!is_idle_task(current) && !user) {
+       trace_rcu_dyntick("Start", oldval, rdtp->dynticks_nesting);
+       if (!user && !is_idle_task(current)) {
                struct task_struct *idle = idle_task(smp_processor_id());
 
                trace_rcu_dyntick("Error on entry: not idle task", oldval, 0);
@@ -399,7 +410,7 @@ void rcu_idle_enter(void)
        unsigned long flags;
 
        local_irq_save(flags);
-       rcu_eqs_enter(0);
+       rcu_eqs_enter(false);
        local_irq_restore(flags);
 }
 EXPORT_SYMBOL_GPL(rcu_idle_enter);
@@ -415,29 +426,7 @@ EXPORT_SYMBOL_GPL(rcu_idle_enter);
  */
 void rcu_user_enter(void)
 {
-       unsigned long flags;
-       struct rcu_dynticks *rdtp;
-
-       /*
-        * Some contexts may involve an exception occuring in an irq,
-        * leading to that nesting:
-        * rcu_irq_enter() rcu_user_exit() rcu_user_exit() rcu_irq_exit()
-        * This would mess up the dyntick_nesting count though. And rcu_irq_*()
-        * helpers are enough to protect RCU uses inside the exception. So
-        * just return immediately if we detect we are in an IRQ.
-        */
-       if (in_interrupt())
-               return;
-
-       WARN_ON_ONCE(!current->mm);
-
-       local_irq_save(flags);
-       rdtp = &__get_cpu_var(rcu_dynticks);
-       if (!rdtp->ignore_user_qs && !rdtp->in_user) {
-               rdtp->in_user = true;
-               rcu_eqs_enter(1);
-       }
-       local_irq_restore(flags);
+       rcu_eqs_enter(1);
 }
 
 /**
@@ -492,7 +481,7 @@ void rcu_irq_exit(void)
        if (rdtp->dynticks_nesting)
                trace_rcu_dyntick("--=", oldval, rdtp->dynticks_nesting);
        else
-               rcu_eqs_enter_common(rdtp, oldval, 1);
+               rcu_eqs_enter_common(rdtp, oldval, true);
        local_irq_restore(flags);
 }
 
@@ -513,7 +502,7 @@ static void rcu_eqs_exit_common(struct rcu_dynticks *rdtp, long long oldval,
        WARN_ON_ONCE(!(atomic_read(&rdtp->dynticks) & 0x1));
        rcu_cleanup_after_idle(smp_processor_id());
        trace_rcu_dyntick("End", oldval, rdtp->dynticks_nesting);
-       if (!is_idle_task(current) && !user) {
+       if (!user && !is_idle_task(current)) {
                struct task_struct *idle = idle_task(smp_processor_id());
 
                trace_rcu_dyntick("Error on exit: not idle task",
@@ -560,7 +549,7 @@ void rcu_idle_exit(void)
        unsigned long flags;
 
        local_irq_save(flags);
-       rcu_eqs_exit(0);
+       rcu_eqs_exit(false);
        local_irq_restore(flags);
 }
 EXPORT_SYMBOL_GPL(rcu_idle_exit);
@@ -574,27 +563,7 @@ EXPORT_SYMBOL_GPL(rcu_idle_exit);
  */
 void rcu_user_exit(void)
 {
-       unsigned long flags;
-       struct rcu_dynticks *rdtp;
-
-       /*
-        * Some contexts may involve an exception occuring in an irq,
-        * leading to that nesting:
-        * rcu_irq_enter() rcu_user_exit() rcu_user_exit() rcu_irq_exit()
-        * This would mess up the dyntick_nesting count though. And rcu_irq_*()
-        * helpers are enough to protect RCU uses inside the exception. So
-        * just return immediately if we detect we are in an IRQ.
-        */
-       if (in_interrupt())
-               return;
-
-       local_irq_save(flags);
-       rdtp = &__get_cpu_var(rcu_dynticks);
-       if (rdtp->in_user) {
-               rdtp->in_user = false;
-               rcu_eqs_exit(1);
-       }
-       local_irq_restore(flags);
+       rcu_eqs_exit(1);
 }
 
 /**
@@ -653,7 +622,7 @@ void rcu_irq_enter(void)
        if (oldval)
                trace_rcu_dyntick("++=", oldval, rdtp->dynticks_nesting);
        else
-               rcu_eqs_exit_common(rdtp, oldval, 1);
+               rcu_eqs_exit_common(rdtp, oldval, true);
        local_irq_restore(flags);
 }
 
@@ -717,21 +686,6 @@ int rcu_is_cpu_idle(void)
 }
 EXPORT_SYMBOL(rcu_is_cpu_idle);
 
-#ifdef CONFIG_RCU_USER_QS
-void rcu_user_hooks_switch(struct task_struct *prev,
-                          struct task_struct *next)
-{
-       struct rcu_dynticks *rdtp;
-
-       /* Interrupts are disabled in context switch */
-       rdtp = &__get_cpu_var(rcu_dynticks);
-       if (!rdtp->ignore_user_qs) {
-               clear_tsk_thread_flag(prev, TIF_NOHZ);
-               set_tsk_thread_flag(next, TIF_NOHZ);
-       }
-}
-#endif /* #ifdef CONFIG_RCU_USER_QS */
-
 #if defined(CONFIG_PROVE_RCU) && defined(CONFIG_HOTPLUG_CPU)
 
 /*
@@ -782,7 +736,7 @@ EXPORT_SYMBOL_GPL(rcu_lockdep_current_cpu_online);
  * interrupt from idle, return true.  The caller must have at least
  * disabled preemption.
  */
-int rcu_is_cpu_rrupt_from_idle(void)
+static int rcu_is_cpu_rrupt_from_idle(void)
 {
        return __get_cpu_var(rcu_dynticks).dynticks_nesting <= 1;
 }
@@ -845,31 +799,46 @@ static int rcu_implicit_dynticks_qs(struct rcu_data *rdp)
                rdp->offline_fqs++;
                return 1;
        }
-       return 0;
-}
-
-static int jiffies_till_stall_check(void)
-{
-       int till_stall_check = ACCESS_ONCE(rcu_cpu_stall_timeout);
 
        /*
-        * Limit check must be consistent with the Kconfig limits
-        * for CONFIG_RCU_CPU_STALL_TIMEOUT.
+        * There is a possibility that a CPU in adaptive-ticks state
+        * might run in the kernel with the scheduling-clock tick disabled
+        * for an extended time period.  Invoke rcu_kick_nohz_cpu() to
+        * force the CPU to restart the scheduling-clock tick in this
+        * CPU is in this state.
         */
-       if (till_stall_check < 3) {
-               ACCESS_ONCE(rcu_cpu_stall_timeout) = 3;
-               till_stall_check = 3;
-       } else if (till_stall_check > 300) {
-               ACCESS_ONCE(rcu_cpu_stall_timeout) = 300;
-               till_stall_check = 300;
-       }
-       return till_stall_check * HZ + RCU_STALL_DELAY_DELTA;
+       rcu_kick_nohz_cpu(rdp->cpu);
+
+       return 0;
 }
 
 static void record_gp_stall_check_time(struct rcu_state *rsp)
 {
        rsp->gp_start = jiffies;
-       rsp->jiffies_stall = jiffies + jiffies_till_stall_check();
+       rsp->jiffies_stall = jiffies + rcu_jiffies_till_stall_check();
+}
+
+/*
+ * Dump stacks of all tasks running on stalled CPUs.  This is a fallback
+ * for architectures that do not implement trigger_all_cpu_backtrace().
+ * The NMI-triggered stack traces are more accurate because they are
+ * printed by the target CPU.
+ */
+static void rcu_dump_cpu_stacks(struct rcu_state *rsp)
+{
+       int cpu;
+       unsigned long flags;
+       struct rcu_node *rnp;
+
+       rcu_for_each_leaf_node(rsp, rnp) {
+               raw_spin_lock_irqsave(&rnp->lock, flags);
+               if (rnp->qsmask != 0) {
+                       for (cpu = 0; cpu <= rnp->grphi - rnp->grplo; cpu++)
+                               if (rnp->qsmask & (1UL << cpu))
+                                       dump_cpu_task(rnp->grplo + cpu);
+               }
+               raw_spin_unlock_irqrestore(&rnp->lock, flags);
+       }
 }
 
 static void print_other_cpu_stall(struct rcu_state *rsp)
@@ -879,6 +848,7 @@ static void print_other_cpu_stall(struct rcu_state *rsp)
        unsigned long flags;
        int ndetected = 0;
        struct rcu_node *rnp = rcu_get_root(rsp);
+       long totqlen = 0;
 
        /* Only let one CPU complain about others per time interval. */
 
@@ -888,7 +858,7 @@ static void print_other_cpu_stall(struct rcu_state *rsp)
                raw_spin_unlock_irqrestore(&rnp->lock, flags);
                return;
        }
-       rsp->jiffies_stall = jiffies + 3 * jiffies_till_stall_check() + 3;
+       rsp->jiffies_stall = jiffies + 3 * rcu_jiffies_till_stall_check() + 3;
        raw_spin_unlock_irqrestore(&rnp->lock, flags);
 
        /*
@@ -923,12 +893,15 @@ static void print_other_cpu_stall(struct rcu_state *rsp)
        raw_spin_unlock_irqrestore(&rnp->lock, flags);
 
        print_cpu_stall_info_end();
-       printk(KERN_CONT "(detected by %d, t=%ld jiffies)\n",
-              smp_processor_id(), (long)(jiffies - rsp->gp_start));
+       for_each_possible_cpu(cpu)
+               totqlen += per_cpu_ptr(rsp->rda, cpu)->qlen;
+       pr_cont("(detected by %d, t=%ld jiffies, g=%lu, c=%lu, q=%lu)\n",
+              smp_processor_id(), (long)(jiffies - rsp->gp_start),
+              rsp->gpnum, rsp->completed, totqlen);
        if (ndetected == 0)
                printk(KERN_ERR "INFO: Stall ended before state dump start\n");
        else if (!trigger_all_cpu_backtrace())
-               dump_stack();
+               rcu_dump_cpu_stacks(rsp);
 
        /* Complain about tasks blocking the grace period. */
 
@@ -939,8 +912,10 @@ static void print_other_cpu_stall(struct rcu_state *rsp)
 
 static void print_cpu_stall(struct rcu_state *rsp)
 {
+       int cpu;
        unsigned long flags;
        struct rcu_node *rnp = rcu_get_root(rsp);
+       long totqlen = 0;
 
        /*
         * OK, time to rat on ourselves...
@@ -951,14 +926,17 @@ static void print_cpu_stall(struct rcu_state *rsp)
        print_cpu_stall_info_begin();
        print_cpu_stall_info(rsp, smp_processor_id());
        print_cpu_stall_info_end();
-       printk(KERN_CONT " (t=%lu jiffies)\n", jiffies - rsp->gp_start);
+       for_each_possible_cpu(cpu)
+               totqlen += per_cpu_ptr(rsp->rda, cpu)->qlen;
+       pr_cont(" (t=%lu jiffies g=%lu c=%lu q=%lu)\n",
+               jiffies - rsp->gp_start, rsp->gpnum, rsp->completed, totqlen);
        if (!trigger_all_cpu_backtrace())
                dump_stack();
 
        raw_spin_lock_irqsave(&rnp->lock, flags);
        if (ULONG_CMP_GE(jiffies, rsp->jiffies_stall))
                rsp->jiffies_stall = jiffies +
-                                    3 * jiffies_till_stall_check() + 3;
+                                    3 * rcu_jiffies_till_stall_check() + 3;
        raw_spin_unlock_irqrestore(&rnp->lock, flags);
 
        set_need_resched();  /* kick ourselves to get things going. */
@@ -989,12 +967,6 @@ static void check_cpu_stall(struct rcu_state *rsp, struct rcu_data *rdp)
        }
 }
 
-static int rcu_panic(struct notifier_block *this, unsigned long ev, void *ptr)
-{
-       rcu_cpu_stall_suppress = 1;
-       return NOTIFY_DONE;
-}
-
 /**
  * rcu_cpu_stall_reset - prevent further stall warnings in current grace period
  *
@@ -1012,15 +984,6 @@ void rcu_cpu_stall_reset(void)
                rsp->jiffies_stall = jiffies + ULONG_MAX / 2;
 }
 
-static struct notifier_block rcu_panic_block = {
-       .notifier_call = rcu_panic,
-};
-
-static void __init check_cpu_stall_init(void)
-{
-       atomic_notifier_chain_register(&panic_notifier_list, &rcu_panic_block);
-}
-
 /*
  * Update CPU-local rcu_data state to record the newly noticed grace period.
  * This is used both when we started the grace period and when we notice
@@ -1087,12 +1050,269 @@ static void init_callback_list(struct rcu_data *rdp)
 {
        int i;
 
+       if (init_nocb_callback_list(rdp))
+               return;
        rdp->nxtlist = NULL;
        for (i = 0; i < RCU_NEXT_SIZE; i++)
                rdp->nxttail[i] = &rdp->nxtlist;
 }
 
 /*
+ * Determine the value that ->completed will have at the end of the
+ * next subsequent grace period.  This is used to tag callbacks so that
+ * a CPU can invoke callbacks in a timely fashion even if that CPU has
+ * been dyntick-idle for an extended period with callbacks under the
+ * influence of RCU_FAST_NO_HZ.
+ *
+ * The caller must hold rnp->lock with interrupts disabled.
+ */
+static unsigned long rcu_cbs_completed(struct rcu_state *rsp,
+                                      struct rcu_node *rnp)
+{
+       /*
+        * If RCU is idle, we just wait for the next grace period.
+        * But we can only be sure that RCU is idle if we are looking
+        * at the root rcu_node structure -- otherwise, a new grace
+        * period might have started, but just not yet gotten around
+        * to initializing the current non-root rcu_node structure.
+        */
+       if (rcu_get_root(rsp) == rnp && rnp->gpnum == rnp->completed)
+               return rnp->completed + 1;
+
+       /*
+        * Otherwise, wait for a possible partial grace period and
+        * then the subsequent full grace period.
+        */
+       return rnp->completed + 2;
+}
+
+/*
+ * Trace-event helper function for rcu_start_future_gp() and
+ * rcu_nocb_wait_gp().
+ */
+static void trace_rcu_future_gp(struct rcu_node *rnp, struct rcu_data *rdp,
+                               unsigned long c, char *s)
+{
+       trace_rcu_future_grace_period(rdp->rsp->name, rnp->gpnum,
+                                     rnp->completed, c, rnp->level,
+                                     rnp->grplo, rnp->grphi, s);
+}
+
+/*
+ * Start some future grace period, as needed to handle newly arrived
+ * callbacks.  The required future grace periods are recorded in each
+ * rcu_node structure's ->need_future_gp field.
+ *
+ * The caller must hold the specified rcu_node structure's ->lock.
+ */
+static unsigned long __maybe_unused
+rcu_start_future_gp(struct rcu_node *rnp, struct rcu_data *rdp)
+{
+       unsigned long c;
+       int i;
+       struct rcu_node *rnp_root = rcu_get_root(rdp->rsp);
+
+       /*
+        * Pick up grace-period number for new callbacks.  If this
+        * grace period is already marked as needed, return to the caller.
+        */
+       c = rcu_cbs_completed(rdp->rsp, rnp);
+       trace_rcu_future_gp(rnp, rdp, c, "Startleaf");
+       if (rnp->need_future_gp[c & 0x1]) {
+               trace_rcu_future_gp(rnp, rdp, c, "Prestartleaf");
+               return c;
+       }
+
+       /*
+        * If either this rcu_node structure or the root rcu_node structure
+        * believe that a grace period is in progress, then we must wait
+        * for the one following, which is in "c".  Because our request
+        * will be noticed at the end of the current grace period, we don't
+        * need to explicitly start one.
+        */
+       if (rnp->gpnum != rnp->completed ||
+           ACCESS_ONCE(rnp->gpnum) != ACCESS_ONCE(rnp->completed)) {
+               rnp->need_future_gp[c & 0x1]++;
+               trace_rcu_future_gp(rnp, rdp, c, "Startedleaf");
+               return c;
+       }
+
+       /*
+        * There might be no grace period in progress.  If we don't already
+        * hold it, acquire the root rcu_node structure's lock in order to
+        * start one (if needed).
+        */
+       if (rnp != rnp_root)
+               raw_spin_lock(&rnp_root->lock);
+
+       /*
+        * Get a new grace-period number.  If there really is no grace
+        * period in progress, it will be smaller than the one we obtained
+        * earlier.  Adjust callbacks as needed.  Note that even no-CBs
+        * CPUs have a ->nxtcompleted[] array, so no no-CBs checks needed.
+        */
+       c = rcu_cbs_completed(rdp->rsp, rnp_root);
+       for (i = RCU_DONE_TAIL; i < RCU_NEXT_TAIL; i++)
+               if (ULONG_CMP_LT(c, rdp->nxtcompleted[i]))
+                       rdp->nxtcompleted[i] = c;
+
+       /*
+        * If the needed for the required grace period is already
+        * recorded, trace and leave.
+        */
+       if (rnp_root->need_future_gp[c & 0x1]) {
+               trace_rcu_future_gp(rnp, rdp, c, "Prestartedroot");
+               goto unlock_out;
+       }
+
+       /* Record the need for the future grace period. */
+       rnp_root->need_future_gp[c & 0x1]++;
+
+       /* If a grace period is not already in progress, start one. */
+       if (rnp_root->gpnum != rnp_root->completed) {
+               trace_rcu_future_gp(rnp, rdp, c, "Startedleafroot");
+       } else {
+               trace_rcu_future_gp(rnp, rdp, c, "Startedroot");
+               rcu_start_gp_advanced(rdp->rsp, rnp_root, rdp);
+       }
+unlock_out:
+       if (rnp != rnp_root)
+               raw_spin_unlock(&rnp_root->lock);
+       return c;
+}
+
+/*
+ * Clean up any old requests for the just-ended grace period.  Also return
+ * whether any additional grace periods have been requested.  Also invoke
+ * rcu_nocb_gp_cleanup() in order to wake up any no-callbacks kthreads
+ * waiting for this grace period to complete.
+ */
+static int rcu_future_gp_cleanup(struct rcu_state *rsp, struct rcu_node *rnp)
+{
+       int c = rnp->completed;
+       int needmore;
+       struct rcu_data *rdp = this_cpu_ptr(rsp->rda);
+
+       rcu_nocb_gp_cleanup(rsp, rnp);
+       rnp->need_future_gp[c & 0x1] = 0;
+       needmore = rnp->need_future_gp[(c + 1) & 0x1];
+       trace_rcu_future_gp(rnp, rdp, c, needmore ? "CleanupMore" : "Cleanup");
+       return needmore;
+}
+
+/*
+ * If there is room, assign a ->completed number to any callbacks on
+ * this CPU that have not already been assigned.  Also accelerate any
+ * callbacks that were previously assigned a ->completed number that has
+ * since proven to be too conservative, which can happen if callbacks get
+ * assigned a ->completed number while RCU is idle, but with reference to
+ * a non-root rcu_node structure.  This function is idempotent, so it does
+ * not hurt to call it repeatedly.
+ *
+ * The caller must hold rnp->lock with interrupts disabled.
+ */
+static void rcu_accelerate_cbs(struct rcu_state *rsp, struct rcu_node *rnp,
+                              struct rcu_data *rdp)
+{
+       unsigned long c;
+       int i;
+
+       /* If the CPU has no callbacks, nothing to do. */
+       if (!rdp->nxttail[RCU_NEXT_TAIL] || !*rdp->nxttail[RCU_DONE_TAIL])
+               return;
+
+       /*
+        * Starting from the sublist containing the callbacks most
+        * recently assigned a ->completed number and working down, find the
+        * first sublist that is not assignable to an upcoming grace period.
+        * Such a sublist has something in it (first two tests) and has
+        * a ->completed number assigned that will complete sooner than
+        * the ->completed number for newly arrived callbacks (last test).
+        *
+        * The key point is that any later sublist can be assigned the
+        * same ->completed number as the newly arrived callbacks, which
+        * means that the callbacks in any of these later sublist can be
+        * grouped into a single sublist, whether or not they have already
+        * been assigned a ->completed number.
+        */
+       c = rcu_cbs_completed(rsp, rnp);
+       for (i = RCU_NEXT_TAIL - 1; i > RCU_DONE_TAIL; i--)
+               if (rdp->nxttail[i] != rdp->nxttail[i - 1] &&
+                   !ULONG_CMP_GE(rdp->nxtcompleted[i], c))
+                       break;
+
+       /*
+        * If there are no sublist for unassigned callbacks, leave.
+        * At the same time, advance "i" one sublist, so that "i" will
+        * index into the sublist where all the remaining callbacks should
+        * be grouped into.
+        */
+       if (++i >= RCU_NEXT_TAIL)
+               return;
+
+       /*
+        * Assign all subsequent callbacks' ->completed number to the next
+        * full grace period and group them all in the sublist initially
+        * indexed by "i".
+        */
+       for (; i <= RCU_NEXT_TAIL; i++) {
+               rdp->nxttail[i] = rdp->nxttail[RCU_NEXT_TAIL];
+               rdp->nxtcompleted[i] = c;
+       }
+       /* Record any needed additional grace periods. */
+       rcu_start_future_gp(rnp, rdp);
+
+       /* Trace depending on how much we were able to accelerate. */
+       if (!*rdp->nxttail[RCU_WAIT_TAIL])
+               trace_rcu_grace_period(rsp->name, rdp->gpnum, "AccWaitCB");
+       else
+               trace_rcu_grace_period(rsp->name, rdp->gpnum, "AccReadyCB");
+}
+
+/*
+ * Move any callbacks whose grace period has completed to the
+ * RCU_DONE_TAIL sublist, then compact the remaining sublists and
+ * assign ->completed numbers to any callbacks in the RCU_NEXT_TAIL
+ * sublist.  This function is idempotent, so it does not hurt to
+ * invoke it repeatedly.  As long as it is not invoked -too- often...
+ *
+ * The caller must hold rnp->lock with interrupts disabled.
+ */
+static void rcu_advance_cbs(struct rcu_state *rsp, struct rcu_node *rnp,
+                           struct rcu_data *rdp)
+{
+       int i, j;
+
+       /* If the CPU has no callbacks, nothing to do. */
+       if (!rdp->nxttail[RCU_NEXT_TAIL] || !*rdp->nxttail[RCU_DONE_TAIL])
+               return;
+
+       /*
+        * Find all callbacks whose ->completed numbers indicate that they
+        * are ready to invoke, and put them into the RCU_DONE_TAIL sublist.
+        */
+       for (i = RCU_WAIT_TAIL; i < RCU_NEXT_TAIL; i++) {
+               if (ULONG_CMP_LT(rnp->completed, rdp->nxtcompleted[i]))
+                       break;
+               rdp->nxttail[RCU_DONE_TAIL] = rdp->nxttail[i];
+       }
+       /* Clean up any sublist tail pointers that were misordered above. */
+       for (j = RCU_WAIT_TAIL; j < i; j++)
+               rdp->nxttail[j] = rdp->nxttail[RCU_DONE_TAIL];
+
+       /* Copy down callbacks to fill in empty sublists. */
+       for (j = RCU_WAIT_TAIL; i < RCU_NEXT_TAIL; i++, j++) {
+               if (rdp->nxttail[j] == rdp->nxttail[RCU_NEXT_TAIL])
+                       break;
+               rdp->nxttail[j] = rdp->nxttail[i];
+               rdp->nxtcompleted[j] = rdp->nxtcompleted[i];
+       }
+
+       /* Classify any remaining callbacks. */
+       rcu_accelerate_cbs(rsp, rnp, rdp);
+}
+
+/*
  * Advance this CPU's callbacks, but only if the current grace period
  * has ended.  This may be called only from the CPU to whom the rdp
  * belongs.  In addition, the corresponding leaf rcu_node structure's
@@ -1102,12 +1322,15 @@ static void
 __rcu_process_gp_end(struct rcu_state *rsp, struct rcu_node *rnp, struct rcu_data *rdp)
 {
        /* Did another grace period end? */
-       if (rdp->completed != rnp->completed) {
+       if (rdp->completed == rnp->completed) {
 
-               /* Advance callbacks.  No harm if list empty. */
-               rdp->nxttail[RCU_DONE_TAIL] = rdp->nxttail[RCU_WAIT_TAIL];
-               rdp->nxttail[RCU_WAIT_TAIL] = rdp->nxttail[RCU_NEXT_READY_TAIL];
-               rdp->nxttail[RCU_NEXT_READY_TAIL] = rdp->nxttail[RCU_NEXT_TAIL];
+               /* No, so just accelerate recent callbacks. */
+               rcu_accelerate_cbs(rsp, rnp, rdp);
+
+       } else {
+
+               /* Advance callbacks. */
+               rcu_advance_cbs(rsp, rnp, rdp);
 
                /* Remember that we saw this grace-period completion. */
                rdp->completed = rnp->completed;
@@ -1197,7 +1420,7 @@ static int rcu_gp_init(struct rcu_state *rsp)
        raw_spin_unlock_irq(&rnp->lock);
 
        /* Exclude any concurrent CPU-hotplug operations. */
-       get_online_cpus();
+       mutex_lock(&rsp->onoff_mutex);
 
        /*
         * Set the quiescent-state-needed bits in all the rcu_node
@@ -1217,9 +1440,9 @@ static int rcu_gp_init(struct rcu_state *rsp)
                rdp = this_cpu_ptr(rsp->rda);
                rcu_preempt_check_blocked_tasks(rnp);
                rnp->qsmask = rnp->qsmaskinit;
-               rnp->gpnum = rsp->gpnum;
+               ACCESS_ONCE(rnp->gpnum) = rsp->gpnum;
                WARN_ON_ONCE(rnp->completed != rsp->completed);
-               rnp->completed = rsp->completed;
+               ACCESS_ONCE(rnp->completed) = rsp->completed;
                if (rnp == rdp->mynode)
                        rcu_start_gp_per_cpu(rsp, rnp, rdp);
                rcu_preempt_boost_start_gp(rnp);
@@ -1228,13 +1451,14 @@ static int rcu_gp_init(struct rcu_state *rsp)
                                            rnp->grphi, rnp->qsmask);
                raw_spin_unlock_irq(&rnp->lock);
 #ifdef CONFIG_PROVE_RCU_DELAY
-               if ((random32() % (rcu_num_nodes * 8)) == 0)
-                       schedule_timeout_uninterruptible(2);
+               if ((prandom_u32() % (rcu_num_nodes + 1)) == 0 &&
+                   system_state == SYSTEM_RUNNING)
+                       udelay(200);
 #endif /* #ifdef CONFIG_PROVE_RCU_DELAY */
                cond_resched();
        }
 
-       put_online_cpus();
+       mutex_unlock(&rsp->onoff_mutex);
        return 1;
 }
 
@@ -1270,6 +1494,7 @@ int rcu_gp_fqs(struct rcu_state *rsp, int fqs_state_in)
 static void rcu_gp_cleanup(struct rcu_state *rsp)
 {
        unsigned long gp_duration;
+       int nocb = 0;
        struct rcu_data *rdp;
        struct rcu_node *rnp = rcu_get_root(rsp);
 
@@ -1299,17 +1524,23 @@ static void rcu_gp_cleanup(struct rcu_state *rsp)
         */
        rcu_for_each_node_breadth_first(rsp, rnp) {
                raw_spin_lock_irq(&rnp->lock);
-               rnp->completed = rsp->gpnum;
+               ACCESS_ONCE(rnp->completed) = rsp->gpnum;
+               rdp = this_cpu_ptr(rsp->rda);
+               if (rnp == rdp->mynode)
+                       __rcu_process_gp_end(rsp, rnp, rdp);
+               nocb += rcu_future_gp_cleanup(rsp, rnp);
                raw_spin_unlock_irq(&rnp->lock);
                cond_resched();
        }
        rnp = rcu_get_root(rsp);
        raw_spin_lock_irq(&rnp->lock);
+       rcu_nocb_gp_set(rnp, nocb);
 
        rsp->completed = rsp->gpnum; /* Declare grace period done. */
        trace_rcu_grace_period(rsp->name, rsp->completed, "end");
        rsp->fqs_state = RCU_GP_IDLE;
        rdp = this_cpu_ptr(rsp->rda);
+       rcu_advance_cbs(rsp, rnp, rdp);  /* Reduce false positives below. */
        if (cpu_needs_another_gp(rsp, rdp))
                rsp->gp_flags = 1;
        raw_spin_unlock_irq(&rnp->lock);
@@ -1382,45 +1613,77 @@ static int __noreturn rcu_gp_kthread(void *arg)
        }
 }
 
+static void rsp_wakeup(struct irq_work *work)
+{
+       struct rcu_state *rsp = container_of(work, struct rcu_state, wakeup_work);
+
+       /* Wake up rcu_gp_kthread() to start the grace period. */
+       wake_up(&rsp->gp_wq);
+}
+
 /*
  * Start a new RCU grace period if warranted, re-initializing the hierarchy
  * in preparation for detecting the next grace period.  The caller must hold
- * the root node's ->lock, which is released before return.  Hard irqs must
- * be disabled.
+ * the root node's ->lock and hard irqs must be disabled.
  *
  * Note that it is legal for a dying CPU (which is marked as offline) to
  * invoke this function.  This can happen when the dying CPU reports its
  * quiescent state.
  */
 static void
-rcu_start_gp(struct rcu_state *rsp, unsigned long flags)
-       __releases(rcu_get_root(rsp)->lock)
+rcu_start_gp_advanced(struct rcu_state *rsp, struct rcu_node *rnp,
+                     struct rcu_data *rdp)
 {
-       struct rcu_data *rdp = this_cpu_ptr(rsp->rda);
-       struct rcu_node *rnp = rcu_get_root(rsp);
-
-       if (!rsp->gp_kthread ||
-           !cpu_needs_another_gp(rsp, rdp)) {
+       if (!rsp->gp_kthread || !cpu_needs_another_gp(rsp, rdp)) {
                /*
                 * Either we have not yet spawned the grace-period
-                * task or this CPU does not need another grace period.
+                * task, this CPU does not need another grace period,
+                * or a grace period is already in progress.
                 * Either way, don't start a new grace period.
                 */
-               raw_spin_unlock_irqrestore(&rnp->lock, flags);
                return;
        }
-
        rsp->gp_flags = RCU_GP_FLAG_INIT;
-       raw_spin_unlock_irqrestore(&rnp->lock, flags);
-       wake_up(&rsp->gp_wq);
+
+       /*
+        * We can't do wakeups while holding the rnp->lock, as that
+        * could cause possible deadlocks with the rq->lock. Deter
+        * the wakeup to interrupt context.
+        */
+       irq_work_queue(&rsp->wakeup_work);
+}
+
+/*
+ * Similar to rcu_start_gp_advanced(), but also advance the calling CPU's
+ * callbacks.  Note that rcu_start_gp_advanced() cannot do this because it
+ * is invoked indirectly from rcu_advance_cbs(), which would result in
+ * endless recursion -- or would do so if it wasn't for the self-deadlock
+ * that is encountered beforehand.
+ */
+static void
+rcu_start_gp(struct rcu_state *rsp)
+{
+       struct rcu_data *rdp = this_cpu_ptr(rsp->rda);
+       struct rcu_node *rnp = rcu_get_root(rsp);
+
+       /*
+        * If there is no grace period in progress right now, any
+        * callbacks we have up to this point will be satisfied by the
+        * next grace period.  Also, advancing the callbacks reduces the
+        * probability of false positives from cpu_needs_another_gp()
+        * resulting in pointless grace periods.  So, advance callbacks
+        * then start the grace period!
+        */
+       rcu_advance_cbs(rsp, rnp, rdp);
+       rcu_start_gp_advanced(rsp, rnp, rdp);
 }
 
 /*
  * Report a full set of quiescent states to the specified rcu_state
  * data structure.  This involves cleaning up after the prior grace
  * period and letting rcu_start_gp() start up the next grace period
- * if one is needed.  Note that the caller must hold rnp->lock, as
- * required by rcu_start_gp(), which will release it.
+ * if one is needed.  Note that the caller must hold rnp->lock, which
+ * is released before return.
  */
 static void rcu_report_qs_rsp(struct rcu_state *rsp, unsigned long flags)
        __releases(rcu_get_root(rsp)->lock)
@@ -1527,7 +1790,7 @@ rcu_report_qs_rdp(int cpu, struct rcu_state *rsp, struct rcu_data *rdp)
                 * This GP can't end until cpu checks in, so all of our
                 * callbacks can be processed during the next GP.
                 */
-               rdp->nxttail[RCU_NEXT_READY_TAIL] = rdp->nxttail[RCU_NEXT_TAIL];
+               rcu_accelerate_cbs(rsp, rnp, rdp);
 
                rcu_report_qs_rnp(mask, rsp, rnp, flags); /* rlses rnp->lock */
        }
@@ -1572,16 +1835,20 @@ rcu_check_quiescent_state(struct rcu_state *rsp, struct rcu_data *rdp)
 /*
  * Send the specified CPU's RCU callbacks to the orphanage.  The
  * specified CPU must be offline, and the caller must hold the
- * ->onofflock.
+ * ->orphan_lock.
  */
 static void
 rcu_send_cbs_to_orphanage(int cpu, struct rcu_state *rsp,
                          struct rcu_node *rnp, struct rcu_data *rdp)
 {
+       /* No-CBs CPUs do not have orphanable callbacks. */
+       if (rcu_is_nocb_cpu(rdp->cpu))
+               return;
+
        /*
         * Orphan the callbacks.  First adjust the counts.  This is safe
-        * because ->onofflock excludes _rcu_barrier()'s adoption of
-        * the callbacks, thus no memory barrier is required.
+        * because _rcu_barrier() excludes CPU-hotplug operations, so it
+        * cannot be running now.  Thus no memory barrier is required.
         */
        if (rdp->nxtlist != NULL) {
                rsp->qlen_lazy += rdp->qlen_lazy;
@@ -1622,13 +1889,17 @@ rcu_send_cbs_to_orphanage(int cpu, struct rcu_state *rsp,
 
 /*
  * Adopt the RCU callbacks from the specified rcu_state structure's
- * orphanage.  The caller must hold the ->onofflock.
+ * orphanage.  The caller must hold the ->orphan_lock.
  */
 static void rcu_adopt_orphan_cbs(struct rcu_state *rsp)
 {
        int i;
        struct rcu_data *rdp = __this_cpu_ptr(rsp->rda);
 
+       /* No-CBs CPUs are handled specially. */
+       if (rcu_nocb_adopt_orphan_cbs(rsp, rdp))
+               return;
+
        /* Do the accounting first. */
        rdp->qlen_lazy += rsp->qlen_lazy;
        rdp->qlen += rsp->qlen;
@@ -1700,7 +1971,8 @@ static void rcu_cleanup_dead_cpu(int cpu, struct rcu_state *rsp)
        /* Remove the dead CPU from the bitmasks in the rcu_node hierarchy. */
 
        /* Exclude any attempts to start a new grace period. */
-       raw_spin_lock_irqsave(&rsp->onofflock, flags);
+       mutex_lock(&rsp->onoff_mutex);
+       raw_spin_lock_irqsave(&rsp->orphan_lock, flags);
 
        /* Orphan the dead CPU's callbacks, and adopt them if appropriate. */
        rcu_send_cbs_to_orphanage(cpu, rsp, rnp, rdp);
@@ -1727,10 +1999,10 @@ static void rcu_cleanup_dead_cpu(int cpu, struct rcu_state *rsp)
        /*
         * We still hold the leaf rcu_node structure lock here, and
         * irqs are still disabled.  The reason for this subterfuge is
-        * because invoking rcu_report_unblock_qs_rnp() with ->onofflock
+        * because invoking rcu_report_unblock_qs_rnp() with ->orphan_lock
         * held leads to deadlock.
         */
-       raw_spin_unlock(&rsp->onofflock); /* irqs remain disabled. */
+       raw_spin_unlock(&rsp->orphan_lock); /* irqs remain disabled. */
        rnp = rdp->mynode;
        if (need_report & RCU_OFL_TASKS_NORM_GP)
                rcu_report_unblock_qs_rnp(rnp, flags);
@@ -1744,6 +2016,7 @@ static void rcu_cleanup_dead_cpu(int cpu, struct rcu_state *rsp)
        init_callback_list(rdp);
        /* Disallow further callbacks on this CPU. */
        rdp->nxttail[RCU_NEXT_TAIL] = NULL;
+       mutex_unlock(&rsp->onoff_mutex);
 }
 
 #else /* #ifdef CONFIG_HOTPLUG_CPU */
@@ -1766,9 +2039,10 @@ static void rcu_do_batch(struct rcu_state *rsp, struct rcu_data *rdp)
 {
        unsigned long flags;
        struct rcu_head *next, *list, **tail;
-       int bl, count, count_lazy, i;
+       long bl, count, count_lazy;
+       int i;
 
-       /* If no callbacks are ready, just return.*/
+       /* If no callbacks are ready, just return. */
        if (!cpu_has_callbacks_ready_to_invoke(rdp)) {
                trace_rcu_batch_start(rsp->name, rdp->qlen_lazy, rdp->qlen, 0);
                trace_rcu_batch_end(rsp->name, 0, !!ACCESS_ONCE(rdp->nxtlist),
@@ -1997,19 +2271,20 @@ __rcu_process_callbacks(struct rcu_state *rsp)
 
        WARN_ON_ONCE(rdp->beenonline == 0);
 
-       /*
-        * Advance callbacks in response to end of earlier grace
-        * period that some other CPU ended.
-        */
+       /* Handle the end of a grace period that some other CPU ended.  */
        rcu_process_gp_end(rsp, rdp);
 
        /* Update RCU state based on any recent quiescent states. */
        rcu_check_quiescent_state(rsp, rdp);
 
        /* Does this CPU require a not-yet-started grace period? */
+       local_irq_save(flags);
        if (cpu_needs_another_gp(rsp, rdp)) {
-               raw_spin_lock_irqsave(&rcu_get_root(rsp)->lock, flags);
-               rcu_start_gp(rsp, flags);  /* releases above lock */
+               raw_spin_lock(&rcu_get_root(rsp)->lock); /* irqs disabled. */
+               rcu_start_gp(rsp);
+               raw_spin_unlock_irqrestore(&rcu_get_root(rsp)->lock, flags);
+       } else {
+               local_irq_restore(flags);
        }
 
        /* If there are callbacks ready, invoke them. */
@@ -2052,7 +2327,8 @@ static void invoke_rcu_callbacks(struct rcu_state *rsp, struct rcu_data *rdp)
 
 static void invoke_rcu_core(void)
 {
-       raise_softirq(RCU_SOFTIRQ);
+       if (cpu_online(smp_processor_id()))
+               raise_softirq(RCU_SOFTIRQ);
 }
 
 /*
@@ -2087,11 +2363,11 @@ static void __call_rcu_core(struct rcu_state *rsp, struct rcu_data *rdp,
 
                /* Start a new grace period if one not already started. */
                if (!rcu_gp_in_progress(rsp)) {
-                       unsigned long nestflag;
                        struct rcu_node *rnp_root = rcu_get_root(rsp);
 
-                       raw_spin_lock_irqsave(&rnp_root->lock, nestflag);
-                       rcu_start_gp(rsp, nestflag);  /* rlses rnp_root->lock */
+                       raw_spin_lock(&rnp_root->lock);
+                       rcu_start_gp(rsp);
+                       raw_spin_unlock(&rnp_root->lock);
                } else {
                        /* Give the grace period a kick. */
                        rdp->blimit = LONG_MAX;
@@ -2104,9 +2380,15 @@ static void __call_rcu_core(struct rcu_state *rsp, struct rcu_data *rdp,
        }
 }
 
+/*
+ * Helper function for call_rcu() and friends.  The cpu argument will
+ * normally be -1, indicating "currently running CPU".  It may specify
+ * a CPU only if that CPU is a no-CBs CPU.  Currently, only _rcu_barrier()
+ * is expected to specify a CPU.
+ */
 static void
 __call_rcu(struct rcu_head *head, void (*func)(struct rcu_head *rcu),
-          struct rcu_state *rsp, bool lazy)
+          struct rcu_state *rsp, int cpu, bool lazy)
 {
        unsigned long flags;
        struct rcu_data *rdp;
@@ -2126,9 +2408,14 @@ __call_rcu(struct rcu_head *head, void (*func)(struct rcu_head *rcu),
        rdp = this_cpu_ptr(rsp->rda);
 
        /* Add the callback to our list. */
-       if (unlikely(rdp->nxttail[RCU_NEXT_TAIL] == NULL)) {
+       if (unlikely(rdp->nxttail[RCU_NEXT_TAIL] == NULL) || cpu != -1) {
+               int offline;
+
+               if (cpu != -1)
+                       rdp = per_cpu_ptr(rsp->rda, cpu);
+               offline = !__call_rcu_nocb(rdp, head, lazy);
+               WARN_ON_ONCE(offline);
                /* _call_rcu() is illegal on offline CPU; leak the callback. */
-               WARN_ON_ONCE(1);
                local_irq_restore(flags);
                return;
        }
@@ -2157,7 +2444,7 @@ __call_rcu(struct rcu_head *head, void (*func)(struct rcu_head *rcu),
  */
 void call_rcu_sched(struct rcu_head *head, void (*func)(struct rcu_head *rcu))
 {
-       __call_rcu(head, func, &rcu_sched_state, 0);
+       __call_rcu(head, func, &rcu_sched_state, -1, 0);
 }
 EXPORT_SYMBOL_GPL(call_rcu_sched);
 
@@ -2166,7 +2453,7 @@ EXPORT_SYMBOL_GPL(call_rcu_sched);
  */
 void call_rcu_bh(struct rcu_head *head, void (*func)(struct rcu_head *rcu))
 {
-       __call_rcu(head, func, &rcu_bh_state, 0);
+       __call_rcu(head, func, &rcu_bh_state, -1, 0);
 }
 EXPORT_SYMBOL_GPL(call_rcu_bh);
 
@@ -2202,10 +2489,28 @@ static inline int rcu_blocking_is_gp(void)
  * rcu_read_lock_sched().
  *
  * This means that all preempt_disable code sequences, including NMI and
- * hardware-interrupt handlers, in progress on entry will have completed
- * before this primitive returns.  However, this does not guarantee that
- * softirq handlers will have completed, since in some kernels, these
- * handlers can run in process context, and can block.
+ * non-threaded hardware-interrupt handlers, in progress on entry will
+ * have completed before this primitive returns.  However, this does not
+ * guarantee that softirq handlers will have completed, since in some
+ * kernels, these handlers can run in process context, and can block.
+ *
+ * Note that this guarantee implies further memory-ordering guarantees.
+ * On systems with more than one CPU, when synchronize_sched() returns,
+ * each CPU is guaranteed to have executed a full memory barrier since the
+ * end of its last RCU-sched read-side critical section whose beginning
+ * preceded the call to synchronize_sched().  In addition, each CPU having
+ * an RCU read-side critical section that extends beyond the return from
+ * synchronize_sched() is guaranteed to have executed a full memory barrier
+ * after the beginning of synchronize_sched() and before the beginning of
+ * that RCU read-side critical section.  Note that these guarantees include
+ * CPUs that are offline, idle, or executing in user mode, as well as CPUs
+ * that are executing in the kernel.
+ *
+ * Furthermore, if CPU A invoked synchronize_sched(), which returned
+ * to its caller on CPU B, then both CPU A and CPU B are guaranteed
+ * to have executed a full memory barrier during the execution of
+ * synchronize_sched() -- even if CPU A and CPU B are the same CPU (but
+ * again only if the system has more than one CPU).
  *
  * This primitive provides the guarantees made by the (now removed)
  * synchronize_kernel() API.  In contrast, synchronize_rcu() only
@@ -2221,7 +2526,10 @@ void synchronize_sched(void)
                           "Illegal synchronize_sched() in RCU-sched read-side critical section");
        if (rcu_blocking_is_gp())
                return;
-       wait_rcu_gp(call_rcu_sched);
+       if (rcu_expedited)
+               synchronize_sched_expedited();
+       else
+               wait_rcu_gp(call_rcu_sched);
 }
 EXPORT_SYMBOL_GPL(synchronize_sched);
 
@@ -2233,6 +2541,9 @@ EXPORT_SYMBOL_GPL(synchronize_sched);
  * read-side critical sections have completed.  RCU read-side critical
  * sections are delimited by rcu_read_lock_bh() and rcu_read_unlock_bh(),
  * and may be nested.
+ *
+ * See the description of synchronize_sched() for more detailed information
+ * on memory ordering guarantees.
  */
 void synchronize_rcu_bh(void)
 {
@@ -2242,13 +2553,13 @@ void synchronize_rcu_bh(void)
                           "Illegal synchronize_rcu_bh() in RCU-bh read-side critical section");
        if (rcu_blocking_is_gp())
                return;
-       wait_rcu_gp(call_rcu_bh);
+       if (rcu_expedited)
+               synchronize_rcu_bh_expedited();
+       else
+               wait_rcu_gp(call_rcu_bh);
 }
 EXPORT_SYMBOL_GPL(synchronize_rcu_bh);
 
-static atomic_t sync_sched_expedited_started = ATOMIC_INIT(0);
-static atomic_t sync_sched_expedited_done = ATOMIC_INIT(0);
-
 static int synchronize_sched_expedited_cpu_stop(void *data)
 {
        /*
@@ -2305,10 +2616,32 @@ static int synchronize_sched_expedited_cpu_stop(void *data)
  */
 void synchronize_sched_expedited(void)
 {
-       int firstsnap, s, snap, trycount = 0;
+       long firstsnap, s, snap;
+       int trycount = 0;
+       struct rcu_state *rsp = &rcu_sched_state;
 
-       /* Note that atomic_inc_return() implies full memory barrier. */
-       firstsnap = snap = atomic_inc_return(&sync_sched_expedited_started);
+       /*
+        * If we are in danger of counter wrap, just do synchronize_sched().
+        * By allowing sync_sched_expedited_started to advance no more than
+        * ULONG_MAX/8 ahead of sync_sched_expedited_done, we are ensuring
+        * that more than 3.5 billion CPUs would be required to force a
+        * counter wrap on a 32-bit system.  Quite a few more CPUs would of
+        * course be required on a 64-bit system.
+        */
+       if (ULONG_CMP_GE((ulong)atomic_long_read(&rsp->expedited_start),
+                        (ulong)atomic_long_read(&rsp->expedited_done) +
+                        ULONG_MAX / 8)) {
+               synchronize_sched();
+               atomic_long_inc(&rsp->expedited_wrap);
+               return;
+       }
+
+       /*
+        * Take a ticket.  Note that atomic_inc_return() implies a
+        * full memory barrier.
+        */
+       snap = atomic_long_inc_return(&rsp->expedited_start);
+       firstsnap = snap;
        get_online_cpus();
        WARN_ON_ONCE(cpu_is_offline(raw_smp_processor_id()));
 
@@ -2320,48 +2653,65 @@ void synchronize_sched_expedited(void)
                             synchronize_sched_expedited_cpu_stop,
                             NULL) == -EAGAIN) {
                put_online_cpus();
+               atomic_long_inc(&rsp->expedited_tryfail);
+
+               /* Check to see if someone else did our work for us. */
+               s = atomic_long_read(&rsp->expedited_done);
+               if (ULONG_CMP_GE((ulong)s, (ulong)firstsnap)) {
+                       /* ensure test happens before caller kfree */
+                       smp_mb__before_atomic_inc(); /* ^^^ */
+                       atomic_long_inc(&rsp->expedited_workdone1);
+                       return;
+               }
 
                /* No joy, try again later.  Or just synchronize_sched(). */
                if (trycount++ < 10) {
                        udelay(trycount * num_online_cpus());
                } else {
-                       synchronize_sched();
+                       wait_rcu_gp(call_rcu_sched);
+                       atomic_long_inc(&rsp->expedited_normal);
                        return;
                }
 
-               /* Check to see if someone else did our work for us. */
-               s = atomic_read(&sync_sched_expedited_done);
-               if (UINT_CMP_GE((unsigned)s, (unsigned)firstsnap)) {
-                       smp_mb(); /* ensure test happens before caller kfree */
+               /* Recheck to see if someone else did our work for us. */
+               s = atomic_long_read(&rsp->expedited_done);
+               if (ULONG_CMP_GE((ulong)s, (ulong)firstsnap)) {
+                       /* ensure test happens before caller kfree */
+                       smp_mb__before_atomic_inc(); /* ^^^ */
+                       atomic_long_inc(&rsp->expedited_workdone2);
                        return;
                }
 
                /*
                 * Refetching sync_sched_expedited_started allows later
-                * callers to piggyback on our grace period.  We subtract
-                * 1 to get the same token that the last incrementer got.
-                * We retry after they started, so our grace period works
-                * for them, and they started after our first try, so their
-                * grace period works for us.
+                * callers to piggyback on our grace period.  We retry
+                * after they started, so our grace period works for them,
+                * and they started after our first try, so their grace
+                * period works for us.
                 */
                get_online_cpus();
-               snap = atomic_read(&sync_sched_expedited_started);
+               snap = atomic_long_read(&rsp->expedited_start);
                smp_mb(); /* ensure read is before try_stop_cpus(). */
        }
+       atomic_long_inc(&rsp->expedited_stoppedcpus);
 
        /*
         * Everyone up to our most recent fetch is covered by our grace
         * period.  Update the counter, but only if our work is still
         * relevant -- which it won't be if someone who started later
-        * than we did beat us to the punch.
+        * than we did already did their update.
         */
        do {
-               s = atomic_read(&sync_sched_expedited_done);
-               if (UINT_CMP_GE((unsigned)s, (unsigned)snap)) {
-                       smp_mb(); /* ensure test happens before caller kfree */
+               atomic_long_inc(&rsp->expedited_done_tries);
+               s = atomic_long_read(&rsp->expedited_done);
+               if (ULONG_CMP_GE((ulong)s, (ulong)snap)) {
+                       /* ensure test happens before caller kfree */
+                       smp_mb__before_atomic_inc(); /* ^^^ */
+                       atomic_long_inc(&rsp->expedited_done_lost);
                        break;
                }
-       } while (atomic_cmpxchg(&sync_sched_expedited_done, s, snap) != s);
+       } while (atomic_long_cmpxchg(&rsp->expedited_done, s, snap) != s);
+       atomic_long_inc(&rsp->expedited_done_exit);
 
        put_online_cpus();
 }
@@ -2437,19 +2787,27 @@ static int rcu_pending(int cpu)
 }
 
 /*
- * Check to see if any future RCU-related work will need to be done
- * by the current CPU, even if none need be done immediately, returning
- * 1 if so.
+ * Return true if the specified CPU has any callback.  If all_lazy is
+ * non-NULL, store an indication of whether all callbacks are lazy.
+ * (If there are no callbacks, all of them are deemed to be lazy.)
  */
-static int rcu_cpu_has_callbacks(int cpu)
+static int rcu_cpu_has_callbacks(int cpu, bool *all_lazy)
 {
+       bool al = true;
+       bool hc = false;
+       struct rcu_data *rdp;
        struct rcu_state *rsp;
 
-       /* RCU callbacks either ready or pending? */
-       for_each_rcu_flavor(rsp)
-               if (per_cpu_ptr(rsp->rda, cpu)->nxtlist)
-                       return 1;
-       return 0;
+       for_each_rcu_flavor(rsp) {
+               rdp = per_cpu_ptr(rsp->rda, cpu);
+               if (rdp->qlen != rdp->qlen_lazy)
+                       al = false;
+               if (rdp->nxtlist)
+                       hc = true;
+       }
+       if (all_lazy)
+               *all_lazy = al;
+       return hc;
 }
 
 /*
@@ -2555,9 +2913,17 @@ static void _rcu_barrier(struct rcu_state *rsp)
         * When that callback is invoked, we will know that all of the
         * corresponding CPU's preceding callbacks have been invoked.
         */
-       for_each_online_cpu(cpu) {
+       for_each_possible_cpu(cpu) {
+               if (!cpu_online(cpu) && !rcu_is_nocb_cpu(cpu))
+                       continue;
                rdp = per_cpu_ptr(rsp->rda, cpu);
-               if (ACCESS_ONCE(rdp->qlen)) {
+               if (rcu_is_nocb_cpu(cpu)) {
+                       _rcu_barrier_trace(rsp, "OnlineNoCB", cpu,
+                                          rsp->n_barrier_done);
+                       atomic_inc(&rsp->barrier_cpu_count);
+                       __call_rcu(&rdp->barrier_head, rcu_barrier_callback,
+                                  rsp, cpu, 0);
+               } else if (ACCESS_ONCE(rdp->qlen)) {
                        _rcu_barrier_trace(rsp, "OnlineQ", cpu,
                                           rsp->n_barrier_done);
                        smp_call_function_single(cpu, rcu_barrier_func, rsp, 1);
@@ -2626,11 +2992,9 @@ rcu_boot_init_percpu_data(int cpu, struct rcu_state *rsp)
        rdp->dynticks = &per_cpu(rcu_dynticks, cpu);
        WARN_ON_ONCE(rdp->dynticks->dynticks_nesting != DYNTICK_TASK_EXIT_IDLE);
        WARN_ON_ONCE(atomic_read(&rdp->dynticks->dynticks) != 1);
-#ifdef CONFIG_RCU_USER_QS
-       WARN_ON_ONCE(rdp->dynticks->in_user);
-#endif
        rdp->cpu = cpu;
        rdp->rsp = rsp;
+       rcu_boot_init_nocb_percpu_data(rdp);
        raw_spin_unlock_irqrestore(&rnp->lock, flags);
 }
 
@@ -2648,6 +3012,9 @@ rcu_init_percpu_data(int cpu, struct rcu_state *rsp, int preemptible)
        struct rcu_data *rdp = per_cpu_ptr(rsp->rda, cpu);
        struct rcu_node *rnp = rcu_get_root(rsp);
 
+       /* Exclude new grace periods. */
+       mutex_lock(&rsp->onoff_mutex);
+
        /* Set up local state, ensuring consistent view of global state. */
        raw_spin_lock_irqsave(&rnp->lock, flags);
        rdp->beenonline = 1;     /* We have now been online. */
@@ -2659,17 +3026,8 @@ rcu_init_percpu_data(int cpu, struct rcu_state *rsp, int preemptible)
        rdp->dynticks->dynticks_nesting = DYNTICK_TASK_EXIT_IDLE;
        atomic_set(&rdp->dynticks->dynticks,
                   (atomic_read(&rdp->dynticks->dynticks) & ~0x1) + 1);
-       rcu_prepare_for_idle_init(cpu);
        raw_spin_unlock(&rnp->lock);            /* irqs remain disabled. */
 
-       /*
-        * A new grace period might start here.  If so, we won't be part
-        * of it, but that is OK, as we are currently in a quiescent state.
-        */
-
-       /* Exclude any attempts to start a new GP on large systems. */
-       raw_spin_lock(&rsp->onofflock);         /* irqs already disabled. */
-
        /* Add CPU to rcu_node bitmasks. */
        rnp = rdp->mynode;
        mask = rdp->grpmask;
@@ -2693,8 +3051,9 @@ rcu_init_percpu_data(int cpu, struct rcu_state *rsp, int preemptible)
                raw_spin_unlock(&rnp->lock); /* irqs already disabled. */
                rnp = rnp->parent;
        } while (rnp != NULL && !(rnp->qsmaskinit & mask));
+       local_irq_restore(flags);
 
-       raw_spin_unlock_irqrestore(&rsp->onofflock, flags);
+       mutex_unlock(&rsp->onoff_mutex);
 }
 
 static void __cpuinit rcu_prepare_cpu(int cpu)
@@ -2733,14 +3092,8 @@ static int __cpuinit rcu_cpu_notify(struct notifier_block *self,
                break;
        case CPU_DYING:
        case CPU_DYING_FROZEN:
-               /*
-                * The whole machine is "stopped" except this CPU, so we can
-                * touch any data without introducing corruption. We send the
-                * dying CPU's callbacks to an arbitrarily chosen online CPU.
-                */
                for_each_rcu_flavor(rsp)
                        rcu_cleanup_dying_cpu(rsp);
-               rcu_cleanup_after_idle(cpu);
                break;
        case CPU_DEAD:
        case CPU_DEAD_FROZEN:
@@ -2773,6 +3126,7 @@ static int __init rcu_spawn_gp_kthread(void)
                raw_spin_lock_irqsave(&rnp->lock, flags);
                rsp->gp_kthread = t;
                raw_spin_unlock_irqrestore(&rnp->lock, flags);
+               rcu_spawn_nocb_kthreads(rsp);
        }
        return 0;
 }
@@ -2843,6 +3197,10 @@ static void __init rcu_init_one(struct rcu_state *rsp,
 
        BUILD_BUG_ON(MAX_RCU_LVLS > ARRAY_SIZE(buf));  /* Fix buf[] init! */
 
+       /* Silence gcc 4.8 warning about array index out of range. */
+       if (rcu_num_lvls > RCU_NUM_LVLS)
+               panic("rcu_init_one: rcu_num_lvls overflow");
+
        /* Initialize the level-tracking arrays. */
 
        for (i = 0; i < rcu_num_lvls; i++)
@@ -2883,11 +3241,13 @@ static void __init rcu_init_one(struct rcu_state *rsp,
                        }
                        rnp->level = i;
                        INIT_LIST_HEAD(&rnp->blkd_tasks);
+                       rcu_init_one_nocb(rnp);
                }
        }
 
        rsp->rda = rda;
        init_waitqueue_head(&rsp->gp_wq);
+       init_irq_work(&rsp->wakeup_work, rsp_wakeup);
        rnp = rsp->level[rcu_num_lvls - 1];
        for_each_possible_cpu(i) {
                while (i > rnp->grphi)
@@ -2968,7 +3328,7 @@ void __init rcu_init(void)
        rcu_init_one(&rcu_sched_state, &rcu_sched_data);
        rcu_init_one(&rcu_bh_state, &rcu_bh_data);
        __rcu_init_preempt();
-        open_softirq(RCU_SOFTIRQ, rcu_process_callbacks);
+       open_softirq(RCU_SOFTIRQ, rcu_process_callbacks);
 
        /*
         * We don't need protection against CPU-hotplug here because
@@ -2978,7 +3338,6 @@ void __init rcu_init(void)
        cpu_notifier(rcu_cpu_notify, 0);
        for_each_online_cpu(cpu)
                rcu_cpu_notify(NULL, CPU_UP_PREPARE, (void *)(long)cpu);
-       check_cpu_stall_init();
 }
 
 #include "rcutree_plugin.h"