rcu: Make call_rcu() safe to call from any context

+/*
+ * Defer a call_rcu()/call_srcu() callback rather than enqueue it now?  Defer
+ * whenever interrupts are disabled, since a callback-list operation may be in
+ * flight on this CPU.  Not while the scheduler is down, though: irq_work is
+ * unusable before init_IRQ(), yet rcu_init() already calls call_rcu().
+ */
+static inline bool should_rcu_defer(void)
+{
+	if (!IS_ENABLED(CONFIG_RCU_DEFER))
+		return false;
+
+	return irqs_disabled() && rcu_scheduler_active != RCU_SCHEDULER_INACTIVE;
+}

+/* Stage @head for this CPU's irq_work when call_rcu() cannot enqueue now. */
+static void call_rcu_defer(struct rcu_head *head, rcu_callback_t func)
+{
+	struct rcu_data *rdp = this_cpu_ptr(&rcu_data);
+
+	/*
+	 * Instrumentation on the enqueue path can re-enter here from inside
+	 * rcu_defer_drain().  Re-queuing would livelock the drain, so drop the
+	 * callback; an NMI is one-shot and cannot loop, so let it through.
+	 */
+	if (this_cpu_read(rcu_defer_draining) && !in_nmi()) {
+		WARN_ONCE(1, "call_rcu() re-entered during callback drain; leaking callback\n");
+		return;
+	}
+	head->func = func;
+	if (llist_add((struct llist_node *)head, &rdp->defer_head))
+		irq_work_queue(&rdp->defer_work);
+}

+static void rcu_defer_drain(struct irq_work *iw)
+{
+	struct rcu_data *rdp = container_of(iw, struct rcu_data, defer_work);
+	struct llist_node *node, *next;
+	unsigned long flags;
+
+	raw_spin_lock_irqsave(&rdp->defer_lock, flags);
+	this_cpu_write(rcu_defer_draining, true);
+	llist_for_each_safe(node, next, llist_del_all(&rdp->defer_head)) {
+		struct rcu_head *head = (struct rcu_head *)node;
+
+		rcu_do_enqueue(head, head->func, false);
+	}
+	this_cpu_write(rcu_defer_draining, false);
+	raw_spin_unlock_irqrestore(&rdp->defer_lock, flags);
+}

+	if (should_rcu_defer()) {
+		call_rcu_defer(head, func);
+		return;
+	}
+
+	/* An NMI reaching here entered with irqs enabled, so the enqueue can race. */
+	WARN_ON_ONCE(IS_ENABLED(CONFIG_PROVE_RCU) && in_nmi());
+
+	rcu_do_enqueue(head, func, lazy_in);
+}

@@ void rcu_barrier(void)
 	unsigned long s;

+	/* Register any deferred callbacks before snapshotting the sequence. */
+	rcu_defer_flush();
+
 	s = rcu_seq_snap(&rcu_state.barrier_sequence);

@@ void rcutree_migrate_callbacks(int cpu)
+	/*
+	 * Callbacks the outgoing CPU deferred late in the offline path (past the
+	 * point its irq_work can run) sit on ->defer_head, which the ->cblist
+	 * migration below does not cover.  Drain them here, before the early
+	 * returns; the re-issue lands on this CPU.
+	 */
+	rcu_defer_drain(&rdp->defer_work);
+
 	if (rcu_rdp_is_offloaded(rdp))
 		return;