This analysis is based on Linux kernel 4.19.195.
Software Implementation Principles
The RCU implementation in software revolves around idnetifying the beginning of a Grace Period (GP), detecting Quiescent States (QS), determining when a grace period ends, handling post-grace-period cleanup, and initiating the next grace period—a large state machine, in essence. The Linux kernel primarily relies on whether the scheduler can run during a tick to determine whether a CPU has exited its critical section (i.e., whether it has passed through a quiescent state). Whether all CPUs have experienced a quiescent state is then used to determine if a grace period has elapsed. Once a grace period ends, memory release operations and initialization of the next grace period can proceed.
Hardware Implementation Prerequisites
The kernel's RCU implementation has hardware prerequisites. As noted in literature: RCU-based updaters typically take advantage of the fact that writes to single aligned pointers are atomic on modern CPUs, allowing atomic insertion, removal, and replacement of data in a linked structure without disrupting readers.
The reason for this prerequisite becomes clear when examining the kernel's RCU-enabled doubly-linked list implementation. If the writer side operations are not atomic, a lock-free reader could potentially observe data in a transitional state between old and new versions, causing correctness issues. Modern 64-bit CPUs can guarantee atomicity for aligned variables smaller than 8 bytes, so in principle, aligned variables under 8 bytes can be protected by RCU. However, practically speaking, non-pointer variables are rarely (if ever) protected by RCU. This is because: first, non-pointer variables are too small to hold much information; second, data validity typically requires a reference count to indicate whether the data is valid or stale. Many RCU-protected data structures include a reference count for this purpose. Without a reference count, it's impossible to know whether data is valid or outdated. When accessing data, reference counts must be incremented to prevent memory from being freed. Since a reference count is typically 4 bytes or larger, protecting pointers through RCU to access data makes perfect sense.
This hardware prerequisite also applies to code that displays data—reading per-CPU variables from another CPU for output purposes. Reading another CPU's per-CPU variable is unreasonable in normal contexts, but for read-only display code where precision isn't critical, the worst case is reading stale data, which doesn't cause serious consequences.
RCU Implementation Analysis
This section uses tree RCU—specifically rcu_sched—as an example to explain the Linux kernel's RCU implementation.
Core Data Structures
The rcu_state structure describes the global state of RCU, with one instance defined for each RCU flavor. The rcu_node structure describes RCU state for a group of processors and serves as organizational nodes in Tree RCU. The rcu_data structure describes RCU state for each processor, with one instance per CPU. Each CPU has its own perspective on the grace period it is experiencing—not necessarily globally synchronized—because some CPUs may have their ticks stopped, be in long IDLE periods, or go offline/online, creating differences from other CPUs. Additionally, some CPUs may have longer grace periods, causing their grace period numbers to lag behind others.
The rcu_segcblist structure manages the RCU callback链表, divided into four segments with specific meanings defined in code comments.
struct rcu_state {
/* Global RCU state management */
};
struct rcu_node {
/* Per-node RCU state for tree organization */
};
struct rcu_data {
/* Per-CPU RCU state tracking */
};
/*
* Segment indices for rcu_segcblist structure.
*
* Callbacks are organized as follows:
* [head, *tails[RCU_DONE_TAIL]): Ready to invoke, grace period elapsed
* [*tails[RCU_DONE_TAIL], *tails[RCU_WAIT_TAIL]): Waiting for current GP
* [*tails[RCU_WAIT_TAIL], *tails[RCU_NEXT_READY_TAIL]): Arrived before next GP
* [*tails[RCU_NEXT_READY_TAIL], *tails[RCU_NEXT_TAIL]): May have arrived after next GP
*/
#define RCU_DONE_TAIL 0
#define RCU_WAIT_TAIL 1
#define RCU_NEXT_READY_TAIL 2
#define RCU_NEXT_TAIL 3
#define RCU_CBLIST_NSEGS 4
struct rcu_segcblist {
struct rcu_head *head;
struct rcu_head **tails[RCU_CBLIST_NSEGS];
unsigned long gp_seq[RCU_CBLIST_NSEGS];
long len;
long len_lazy;
};
Background Thread
Each RCU flavor has a background thread responsible for starting and stopping grace periods. This grace period thread executes the rcu_gp_kthread() function, which performs three main operations: waiting for and creating a new grace period, waiting for forced quiescent states with timeout handling, and grace period cleanup.
The first part waits for the RCU_GP_FLAG_INIT flag to be set in rsp->gp_flags, which indicates a new grace period should begin. This flag is set in rcu_gp_cleanup() and rcu_start_this_gp().
The second part waits for all CPUs to pass through the grace period. If all CPUs pass before timeout, processing proceeds to the third phase. Otherwise, when the timeout expires, CPUs that haven't passed through are forced to generate a quiescent state.
The third part handles cleanup after the grace period ends.
static int __noreturn rcu_gp_kthread(void *arg)
{
bool initial_fqs_pass;
int grace_flags;
unsigned long tick_counter;
int force_result;
struct rcu_state *rsp = arg;
struct rcu_node *root_node = rcu_get_root(rsp);
rcu_bind_gp_kthread();
for (;;) {
/* Phase 1: Grace period initialization */
for (;;) {
trace_rcu_grace_period(rsp->name,
READ_ONCE(rsp->gp_seq),
TPS("reqwait"));
rsp->gp_state = RCU_GP_WAIT_GPS;
swait_event_idle_exclusive(rsp->gp_wq,
READ_ONCE(rsp->gp_flags) &
RCU_GP_FLAG_INIT);
rsp->gp_state = RCU_GP_DONE_GPS;
if (rcu_gp_init(rsp))
break;
cond_resched_tasks_rcu_qs();
WRITE_ONCE(rsp->gp_activity, jiffies);
WARN_ON(signal_pending(current));
trace_rcu_grace_period(rsp->name,
READ_ONCE(rsp->gp_seq),
TPS("reqwaitsig"));
}
/* Phase 2: Quiescent state forcing */
initial_fqs_pass = true;
tick_counter = jiffies_till_first_fqs;
force_result = 0;
for (;;) {
if (!force_result) {
rsp->jiffies_force_qs = jiffies + tick_counter;
WRITE_ONCE(rsp->jiffies_kick_kthreads,
jiffies + 3 * tick_counter);
}
trace_rcu_grace_period(rsp->name,
READ_ONCE(rsp->gp_seq),
TPS("fqswait"));
rsp->gp_state = RCU_GP_WAIT_FQS;
force_result = swait_event_idle_timeout_exclusive(
rsp->gp_wq,
rcu_gp_fqs_check_wake(rsp, &grace_flags),
tick_counter);
rsp->gp_state = RCU_GP_DOING_FQS;
if (!READ_ONCE(root_node->qsmask) &&
!rcu_preempt_blocked_readers_cgp(root_node))
break;
if (ULONG_CMP_GE(jiffies, rsp->jiffies_force_qs) ||
(grace_flags & RCU_GP_FLAG_FQS)) {
trace_rcu_grace_period(rsp->name,
READ_ONCE(rsp->gp_seq),
TPS("fqsstart"));
rcu_gp_fqs(rsp, initial_fqs_pass);
initial_fqs_pass = false;
trace_rcu_grace_period(rsp->name,
READ_ONCE(rsp->gp_seq),
TPS("fqsend"));
cond_resched_tasks_rcu_qs();
WRITE_ONCE(rsp->gp_activity, jiffies);
force_result = 0;
tick_counter = jiffies_till_next_fqs;
} else {
cond_resched_tasks_rcu_qs();
WRITE_ONCE(rsp->gp_activity, jiffies);
WARN_ON(signal_pending(current));
trace_rcu_grace_period(rsp->name,
READ_ONCE(rsp->gp_seq),
TPS("fqswaitsig"));
force_result = 1;
tick_counter = jiffies;
if (time_after(jiffies, rsp->jiffies_force_qs))
tick_counter = 1;
else
tick_counter = rsp->jiffies_force_qs - jiffies;
}
}
/* Phase 3: Grace period cleanup */
rsp->gp_state = RCU_GP_CLEANUP;
rcu_gp_cleanup(rsp);
rsp->gp_state = RCU_GP_CLEANED;
}
}
The second part first sleeps until rcu_gp_fqs_check_wake() returns true. This function returns true in two cases: either quiescent state forcing is required, or the current grace period has completed. If no forced quiescent state is needed, the grace period can end. Otherwise, forced state generation proceeds in a loop using jiffies_till_first_fqse as the interval, calling rcu_gp_fqs() to force processors that haven't passed through to experience a quiescent state.
For the first pass of forced quiescent state processing in the current grace period, rcu_gp_fqs() calls force_qs_rnp(rsp, dyntick_save_progress_counter). For subsequent passes, it calls force_qs_rnp(rsp, rcu_implicit_dynticks_qs).
static void rcu_gp_fqs(struct rcu_state *rsp, bool first_pass)
{
struct rcu_node *root = rcu_get_root(rsp);
WRITE_ONCE(rsp->gp_activity, jiffies);
rsp->n_force_qs++;
if (first_pass) {
force_qs_rnp(rsp, dyntick_save_progress_counter);
} else {
force_qs_rnp(rsp, rcu_implicit_dynticks_qs);
}
if (READ_ONCE(rsp->gp_flags) & RCU_GP_FLAG_FQS) {
raw_spin_lock_irq_rcu_node(root);
WRITE_ONCE(rsp->gp_flags,
READ_ONCE(rsp->gp_flags) & ~RCU_GP_FLAG_FQS);
raw_spin_unlock_irq_rcu_node(root);
}
}
static void force_qs_rnp(struct rcu_state *rsp,
int (*quiescent_func)(struct rcu_data *))
{
int cpu;
unsigned long flags;
unsigned long cpu_mask;
struct rcu_node *node;
rcu_for_each_leaf_node(rsp, node) {
cond_resched_tasks_rcu_qs();
cpu_mask = 0;
raw_spin_lock_irqsave_rcu_node(node, flags);
if (node->qsmask == 0) {
if (rcu_state_p == &rcu_sched_state ||
rsp != rcu_state_p ||
rcu_preempt_blocked_readers_cgp(node)) {
rcu_initiate_boost(node, flags);
continue;
}
raw_spin_unlock_irqrestore_rcu_node(node, flags);
continue;
}
for_each_leaf_node_possible_cpu(node, cpu) {
unsigned long bit = leaf_node_cpu_bit(node, cpu);
if ((node->qsmask & bit) != 0) {
if (quiescent_func(per_cpu_ptr(rsp->rda, cpu)))
cpu_mask |= bit;
}
}
if (cpu_mask != 0) {
rcu_report_qs_rnp(cpu_mask, rsp, node, node->gp_seq, flags);
} else {
raw_spin_unlock_irqrestore_rcu_node(node, flags);
}
}
}
static int rcu_implicit_dynticks_qs(struct rcu_data *rdp)
{
unsigned long sched_qs_threshold;
bool *urgent_flag;
bool *heavy_qs_flag;
struct rcu_node *node = rdp->mynode;
if (rcu_dynticks_in_eqs_since(rdp->dynticks, rdp->dynticks_snap)) {
trace_rcu_fqs(rdp->rsp->name, rdp->gp_seq, rdp->cpu, TPS("dti"));
rdp->dynticks_fqs++;
rcu_gpnum_ovf(node, rdp);
return 1;
}
sched_qs_threshold = jiffies_till_sched_qs;
urgent_flag = per_cpu_ptr(&rcu_dynticks.rcu_urgent_qs, rdp->cpu);
if (time_after(jiffies, rdp->rsp->gp_start + sched_qs_threshold) &&
READ_ONCE(rdp->rcu_qs_ctr_snap) !=
per_cpu(rcu_dynticks.rcu_qs_ctr, rdp->cpu) &&
rcu_seq_current(&rdp->gp_seq) == node->gp_seq && !rdp->gpwrap) {
trace_rcu_fqs(rdp->rsp->name, rdp->gp_seq, rdp->cpu, TPS("rqc"));
rcu_gpnum_ovf(node, rdp);
return 1;
} else if (time_after(jiffies, rdp->rsp->gp_start + sched_qs_threshold)) {
smp_store_release(urgent_flag, true);
}
if (!(rdp->grpmask & rcu_rnp_online_cpus(node)) &&
time_after(jiffies, rdp->rsp->gp_start + HZ)) {
bool is_online;
struct rcu_node *tmp_node;
WARN_ON(1);
pr_info("%s: grp: %d-%d level: %d ->gp_seq %ld ->completedqs %ld\n",
__func__, node->grplo, node->grphi, node->level,
(long)node->gp_seq, (long)node->completedqs);
for (tmp_node = node; tmp_node; tmp_node = tmp_node->parent)
pr_info("%s: %d:%d ->qsmask %#lx ->qsmaskinit %#lx ->qsmaskinitnext %#lx\n",
__func__, tmp_node->grplo, tmp_node->grphi,
tmp_node->qsmask, tmp_node->qsmaskinit,
tmp_node->qsmaskinitnext);
is_online = !!(rdp->grpmask & rcu_rnp_online_cpus(node));
pr_info("%s %d: %c online: %ld(%d) offline: %ld(%d)\n",
__func__, rdp->cpu, ".o"[is_online],
(long)rdp->rcu_onl_gp_seq, rdp->rcu_onl_gp_flags,
(long)rdp->rcu_ofl_gp_seq, rdp->rcu_ofl_gp_flags);
return 1;
}
heavy_qs_flag = &per_cpu(rcu_dynticks.rcu_need_heavy_qs, rdp->cpu);
if (!READ_ONCE(*heavy_qs_flag) &&
(time_after(jiffies, rdp->rsp->gp_start + sched_qs_threshold) ||
time_after(jiffies, rdp->rsp->jiffies_resched))) {
WRITE_ONCE(*heavy_qs_flag, true);
smp_store_release(urgent_flag, true);
rdp->rsp->jiffies_resched += sched_qs_threshold;
}
if (jiffies - rdp->rsp->gp_start > rcu_jiffies_till_stall_check() / 2) {
resched_cpu(rdp->cpu);
if (IS_ENABLED(CONFIG_IRQ_WORK) &&
!rdp->rcu_iw_pending && rdp->rcu_iw_gp_seq != node->gp_seq &&
(node->ffmask & rdp->grpmask)) {
init_irq_work(&rdp->rcu_iw, rcu_iw_handler);
rdp->rcu_iw_pending = true;
rdp->rcu_iw_gp_seq = node->gp_seq;
irq_work_queue_on(&rdp->rcu_iw, rdp->cpu);
}
}
return 0;
}
The force_qs_rnp function iterates through each node. If all CPUs represented by a node have passed through the grace period, it skips that node. For leaf nodes that haven't passed through, rcu_implicit_dynticks_qs() is called to determine whether the relevant CPU has experienced a quiescent state since the grace period began. If not, resched_cpu() is called to request the corresponding CPU to reschedule.
The third part sets relevant state after a grace period ends and calls rcu_gp_cleanup() to complete processing.
static void rcu_gp_cleanup(struct rcu_state *rsp)
{
unsigned long gp_time_elapsed;
bool need_new_gp = false;
unsigned long new_gp_sequence;
struct rcu_data *rdp;
struct rcu_node *root = rcu_get_root(rsp);
struct swait_queue_head *wait_queue;
WRITE_ONCE(rsp->gp_activity, jiffies);
raw_spin_lock_irq_rcu_node(root);
gp_time_elapsed = jiffies - rsp->gp_start;
if (gp_time_elapsed > rsp->gp_max)
rsp->gp_max = gp_time_elapsed;
raw_spin_unlock_irq_rcu_node(root);
new_gp_sequence = rsp->gp_seq;
rcu_seq_end(&new_gp_sequence);
rcu_for_each_node_breadth_first(rsp, root) {
raw_spin_lock_irq_rcu_node(root);
if (WARN_ON_ONCE(rcu_preempt_blocked_readers_cgp(root)))
dump_blkd_tasks(rsp, root, 10);
WARN_ON_ONCE(root->qsmask);
WRITE_ONCE(root->gp_seq, new_gp_sequence);
rdp = this_cpu_ptr(rsp->rda);
if (root == rdp->mynode)
need_new_gp = __note_gp_changes(rsp, root, rdp) || need_new_gp;
need_new_gp = rcu_future_gp_cleanup(rsp, root) || need_new_gp;
wait_queue = rcu_nocb_gp_get(root);
raw_spin_unlock_irq_rcu_node(root);
rcu_nocb_gp_cleanup(wait_queue);
cond_resched_tasks_rcu_qs();
WRITE_ONCE(rsp->gp_activity, jiffies);
rcu_gp_slow(rsp, gp_cleanup_delay);
}
root = rcu_get_root(rsp);
raw_spin_lock_irq_rcu_node(root);
rcu_seq_end(&rsp->gp_seq);
trace_rcu_grace_period(rsp->name, rsp->gp_seq, TPS("end"));
rsp->gp_state = RCU_GP_IDLE;
rdp = this_cpu_ptr(rsp->rda);
if (!need_new_gp && ULONG_CMP_LT(root->gp_seq, root->gp_seq_needed)) {
trace_rcu_this_gp(root, rdp, root->gp_seq_needed,
TPS("CleanupMore"));
need_new_gp = true;
}
if (!rcu_accelerate_cbs(rsp, root, rdp) && need_new_gp) {
WRITE_ONCE(rsp->gp_flags, RCU_GP_FLAG_INIT);
rsp->gp_req_activity = jiffies;
trace_rcu_grace_period(rsp->name, READ_ONCE(rsp->gp_seq),
TPS("newreq"));
} else {
WRITE_ONCE(rsp->gp_flags, rsp->gp_flags & RCU_GP_FLAG_INIT);
}
raw_spin_unlock_irq_rcu_node(root);
}
static bool rcu_accelerate_cbs(struct rcu_state *rsp, struct rcu_node *node,
struct rcu_data *rdp)
{
unsigned long required_gp_seq;
bool should_wake = false;
raw_lockdep_assert_held_rcu_node(node);
if (!rcu_segcblist_pend_cbs(&rdp->cblist))
return false;
required_gp_seq = rcu_seq_snap(&rsp->gp_seq);
if (rcu_segcblist_accelerate(&rdp->cblist, required_gp_seq))
should_wake = rcu_start_this_gp(node, rdp, required_gp_seq);
if (rcu_segcblist_restempty(&rdp->cblist, RCU_WAIT_TAIL))
trace_rcu_grace_period(rsp->name, rdp->gp_seq, TPS("AccWaitCB"));
else
trace_rcu_grace_period(rsp->name, rdp->gp_seq, TPS("AccReadyCB"));
return should_wake;
}
The grace period termination sequence proceeds as follows:
- Starting from the root node, traverse the RCU tree level by level, updating each node's completed grace period number to the current grace period number.
- Assuming the grace period thread runs on processor n, process grace period completion for that processor's
rcu_datainstance—moving callbacks registered up to the current grace period into theRCU_DONE_TAIL链表 and updating the current grace period number.
The rcu_accelerate_cbs() function handles callback acceleration, moving callbacks from the last segment (RCU_next_tail) to earlier segments. This function manages callback movement within rcu_segcblist and determines whether a new grace period needs to be started.
Tick Handler Implementation
Each CPU has its own tick interrupt, with RCU handling in the tick primarily dealing with per-CPU work. Global work is handled by the RCU background thread.
In the tick handler, rcu_check_callbacks() is called to perform RCU-related checks.
void rcu_check_callbacks(int user)
{
trace_rcu_utilization(TPS("Start scheduler-tick"));
increment_cpu_stall_ticks();
if (user || rcu_is_cpu_rrupt_from_idle()) {
rcu_sched_qs();
rcu_bh_qs();
rcu_note_voluntary_context_switch(current);
} else if (!in_softirq()) {
rcu_bh_qs();
}
rcu_preempt_check_callbacks();
if (smp_load_acquire(this_cpu_ptr(&rcu_dynticks.rcu_urgent_qs))) {
if (!rcu_is_cpu_rrupt_from_idle() && !user) {
set_tsk_need_resched(current);
set_preempt_need_resched();
}
__this_cpu_write(rcu_dynticks.rcu_urgent_qs, false);
}
if (rcu_pending())
invoke_rcu_core();
trace_rcu_utilization(TPS("End scheduler-tick"));
}
The logic is straightforward: if the tick triggers in user mode or idle state, the CPU is considered to have passed through a quiescent state, and rcu_sched_qs() is called to record it. This only records the state; reporting the CPU's quiescent state is completed in the softirq context.
rcu_init()->open_softirq(RCU_SOFTIRQ, rcu_process_callbacks);
rcu_process_callbacks()
->__rcu_process_callbacks()
->rcu_check_quiescent_state()
->rcu_report_qs_rdp()
->rcu_report_qs_rnp()
->rcu_report_qs_rsp()
The rcu_report_qs_rnp() function reports quiescent states upward from leaf nodes to the tree root.
The rcu_check_callbacks() function concludes by using rcu_pending() to check if the local processor has related work to process. If so, it invokes invoke_rcu_core() to trigger the RCU_SOFTIRQ softirq for processing, rather than handling it in the tick context.
static int rcu_pending(void)
{
struct rcu_state *rsp;
for_each_rcu_flavor(rsp)
if (__rcu_pending(rsp, this_cpu_ptr(rsp->rda)))
return 1;
return 0;
}
static int __rcu_pending(struct rcu_state *rsp, struct rcu_data *rdp)
{
struct rcu_node *node = rdp->mynode;
check_cpu_stall(rsp, rdp);
if (rcu_nohz_full_cpu(rsp))
return 0;
if (rdp->core_needs_qs && !rdp->cpu_no_qs.b.norm)
return 1;
if (rcu_segcblist_ready_cbs(&rdp->cblist))
return 1;
if (!rcu_gp_in_progress(rsp) &&
rcu_segcblist_is_enabled(&rdp->cblist) &&
!rcu_segcblist_restempty(&rdp->cblist, RCU_NEXT_READY_TAIL))
return 1;
if (rcu_seq_current(&node->gp_seq) != rdp->gp_seq ||
unlikely(READ_ONCE(rdp->gpwrap)))
return 1;
if (rcu_nocb_need_deferred_wakeup(rdp))
return 1;
return 0;
}
The rdp->cpu_no_qs.b.norm variable is set to false after passing through a quiescent state in rcu_sched_qs(). When this判断 succeeds, it returns 1, indicating quiescent state reporting is needed. If there are callback functions to invoke, it also returns 1 indicating work needs to be done. If the callback链表 indicates a new grace period is needed, it also returns 1.
Idle State Implementation
When a CPU enters the idle state, it ultimately reaches the rcu_idle_enter() function to handle RCU-related matters.
void rcu_idle_enter(void)
{
lockdep_assert_irqs_disabled();
rcu_eqs_enter(false);
}
The rcu_idle_enter() function simply calls rcu_eqs_enter(), which marks the CPU as having entered an extended quiescent state. In this state, the CPU is continuously considered to have passed through a quiescent state, meaning it can effectively be excluded from the global determination of whether all CPUs have experienced a quiescent state. However, the kernel still performs a simple check—when a CPU enters an extended quiescent state, other CPUs assist in reporting its quiescent state, allowing the CPU in extended quiescent state to remain in idle or guest mode without interruption.
static void rcu_eqs_enter(bool user)
{
struct rcu_state *rsp;
struct rcu_data *rdp;
struct rcu_dynticks *tdp;
tdp = this_cpu_ptr(&rcu_dynticks);
WRITE_ONCE(tdp->dynticks_nmi_nesting, 0);
WARN_ON_ONCE(IS_ENABLED(CONFIG_RCU_EQS_DEBUG) &&
tdp->dynticks_nesting == 0);
if (tdp->dynticks_nesting != 1) {
tdp->dynticks_nesting--;
return;
}
lockdep_assert_irqs_disabled();
trace_rcu_dyntick(TPS("Start"), tdp->dynticks_nesting, 0, tdp->dynticks);
WARN_ON_ONCE(IS_ENABLED(CONFIG_RCU_EQS_DEBUG) &&
!user && !is_idle_task(current));
for_each_rcu_flavor(rsp) {
rdp = this_cpu_ptr(rsp->rda);
do_nocb_deferred_wakeup(rdp);
}
rcu_prepare_for_idle();
WRITE_ONCE(tdp->dynticks_nesting, 0);
rcu_dynticks_eqs_enter();
rcu_dynticks_task_enter();
}
The key function is rcu_dynticks_eqs_enter(), which adds 0x10 to the CPU's rcu_dynticks variable.
static void rcu_dynticks_eqs_enter(void)
{
struct rcu_dynticks *tdp = this_cpu_ptr(&rcu_dynticks);
int seq;
seq = atomic_add_return(RCU_DYNTICK_CTRL_CTR, &tdp->dynticks);
WARN_ON_ONCE(IS_ENABLED(CONFIG_RCU_EQS_DEBUG) &&
(seq & RCU_DYNTICK_CTRL_CTR));
WARN_ON_ONCE(IS_ENABLED(CONFIG_RCU_EQS_DEBUG) &&
(seq & RCU_DYNTICK_CTRL_MASK));
}
Where is the rcu_dynticks value used? The answer lies in the grace period thread.
rcu_gp_kthread() phase two
->rcu_gp_fqs()
->dyntick_save_progress_counter()
->force_qs_rnp()
Reader and Writer Implementation
The reader and writer implementations center on a few core functions. Using RCU-enabled doubly-linked list operations as an example:
#define list_next_rcu(list) (*((struct list_head __rcu **)(&(list)->next)))
static inline void __list_add_rcu(struct list_head *new_entry,
struct list_head *prev,
struct list_head *next)
{
if (!__list_add_valid(new_entry, prev, next))
return;
new_entry->next = next;
new_entry->prev = prev;
rcu_assign_pointer(list_next_rcu(prev), new_entry);
next->prev = new_entry;
}
static inline void list_add_rcu(struct list_head *new_entry,
struct list_head *head)
{
__list_add_rcu(new_entry, head, head->next);
}
#define list_entry_rcu(ptr, type, member) \
container_of(READ_ONCE(ptr), type, member)
#define list_for_each_entry_rcu(pos, head, member) \
for (pos = list_entry_rcu((head)->next, typeof(*pos), member); \
&pos->member != (head); \
pos = list_entry_rcu(pos->member.next, typeof(*pos), member))
Examining the __list_add_rcu function implementation reveals it differs from the non-RCU version. The __list_add_rcu function first initializes the new node, then uses rcu_assign_pointer to update the previous node's next pointer, followed by updating the next node's prev pointer. The rcu_assign_pointer/rcu_dereference macros ensure proper ordering. The reason for this approach is to guarantee that when readers observe the new node (whether in forward or backward traversal), both its prev and next pointers are correct, ensuring reader correctness.
After the writer completes updates, it must call either synchronize_rcu or call_rcu to reclaim the old data. The implementations of these two functions are similar.
void synchronize_rcu(void)
{
RCU_LOCKDEP_WARN(lock_is_held(&rcu_bh_lock_map) ||
lock_is_held(&rcu_lock_map) ||
lock_is_held(&rcu_sched_lock_map),
"Illegal synchronize_rcu() in RCU read-side critical section");
if (rcu_scheduler_active == RCU_SCHEDULER_INACTIVE)
return;
if (rcu_gp_is_expedited())
synchronize_rcu_expedited();
else
wait_rcu_gp(call_rcu);
}
EXPORT_SYMBOL_GPL(synchronize_rcu);
Under normal circumstances, the else branch is taken.
void __wait_rcu_gp(bool checktiny, int count, call_rcu_func_t *func_array,
struct rcu_synchronize *sync_array)
{
int i;
int j;
for (i = 0; i < count; i++) {
if (checktiny &&
(func_array[i] == call_rcu ||
func_array[i] == call_rcu_bh)) {
might_sleep();
continue;
}
init_rcu_head_on_stack(&sync_array[i].head);
init_completion(&sync_array[i].completion);
for (j = 0; j < i; j++)
if (func_array[j] == func_array[i])
break;
if (j == i)
(func_array[i])(&sync_array[i].head, wakeme_after_rcu);
}
for (i = 0; i < count; i++) {
if (checktiny &&
(func_array[i] == call_rcu ||
func_array[i] == call_rcu_bh))
continue;
for (j = 0; j < i; j++)
if (func_array[j] == func_array[i])
break;
if (j == i)
wait_for_completion(&sync_array[i].completion);
destroy_rcu_head_on_stack(&sync_array[i].head);
}
}
EXPORT_SYMBOL_GPL(__wait_rcu_gp);
void wakeme_after_rcu(struct rcu_head *head)
{
struct rcu_synchronize *rs;
rs = container_of(head, struct rcu_synchronize, head);
complete(&rs->completion);
}
EXPORT_SYMBOL_GPL(wakeme_after_rcu);
As the code shows, synchronize_rcu is implemented using call_rcu, with the callback function being wakeme_after_rcu. After the current grace period passes, wakeme_after_rcu wakes up the process that called synchronize_rcu(), allowing it to safely release related resources.
How is call_rcu implemented?
void call_rcu(struct rcu_head *head, rcu_callback_t func)
{
__call_rcu(head, func, rcu_state_p, -1, 0);
}
EXPORT_SYMBOL_GPL(call_rcu);
static void __call_rcu(struct rcu_head *head, rcu_callback_t func,
struct rcu_state *rsp, int cpu, bool lazy)
{
unsigned long flags;
struct rcu_data *rdp;
WARN_ON_ONCE((unsigned long)head & (sizeof(void *) - 1));
if (debug_rcu_head_queue(head)) {
WARN_ONCE(1, "__call_rcu(): Double-freed CB %p->%pF()!!!\n",
head, head->func);
WRITE_ONCE(head->func, rcu_leak_callback);
return;
}
head->func = func;
head->next = NULL;
local_irq_save(flags);
rdp = this_cpu_ptr(rsp->rda);
if (unlikely(!rcu_segcblist_is_enabled(&rdp->cblist)) || cpu != -1) {
int offline_status;
if (cpu != -1)
rdp = per_cpu_ptr(rsp->rda, cpu);
if (likely(rdp->mynode)) {
offline_status = !__call_rcu_nocb(rdp, head, lazy, flags);
WARN_ON_ONCE(offline_status);
local_irq_restore(flags);
return;
}
BUG_ON(cpu != -1);
WARN_ON_ONCE(!rcu_is_watching());
if (rcu_segcblist_empty(&rdp->cblist))
rcu_segcblist_init(&rdp->cblist);
}
rcu_segcblist_enqueue(&rdp->cblist, head, lazy);
if (!lazy)
rcu_idle_count_callbacks_posted();
if (__is_kfree_rcu_offset((unsigned long)func))
trace_rcu_kfree_callback(rsp->name, head, (unsigned long)func,
rcu_segcblist_n_lazy_cbs(&rdp->cblist),
rcu_segcblist_n_cbs(&rdp->cblist));
else
trace_rcu_callback(rsp->name, head,
rcu_segcblist_n_lazy_cbs(&rdp->cblist),
rcu_segcblist_n_cbs(&rdp->cblist));
__call_rcu_core(rsp, rdp, head, flags);
local_irq_restore(flags);
}
static void __call_rcu_core(struct rcu_state *rsp, struct rcu_data *rdp,
struct rcu_head *head, unsigned long flags)
{
if (!rcu_is_watching())
invoke_rcu_core();
if (irqs_disabled_flags(flags) || cpu_is_offline(smp_processor_id()))
return;
if (unlikely(rcu_segcblist_n_cbs(&rdp->cblist) >
rdp->qlen_last_fqs_check + qhimark)) {
note_gp_changes(rsp, rdp);
if (!rcu_gp_in_progress(rsp)) {
rcu_accelerate_cbs_unlocked(rsp, rdp->mynode, rdp);
} else {
rdp->blimit = LONG_MAX;
if (rsp->n_force_qs == rdp->n_force_qs_snap &&
rcu_segcblist_first_pend_cb(&rdp->cblist) != head)
force_quiescent_state(rsp);
rdp->n_force_qs_snap = rsp->n_force_qs;
rdp->qlen_last_fqs_check = rcu_segcblist_n_cbs(&rdp->cblist);
}
}
}
The call_rcu function has two primary functions. First, it registers a callback function, with callbacks maintained in the struct rcu_segcblist cblist field of the rcu_data structure. Second, it determines whether a new grace period needs to be started (by notifying the grace period thread). As analyzed above, these callback functions are invoked in the softirq context.