Skip to content
Permalink
Browse files
squash! srcu: Make Tree SRCU able to operate without snp_node array 
When there is no snp_node array, all SRCU callbacks use CPU 0's callback
queue.  This is optimal in the common case of low update-side load.

Signed-off-by: Paul E. McKenney <paulmck@kernel.org>
  • Loading branch information
@paulmckrcu
paulmckrcu committed Jan 24, 2022
1 parent 44b29ab commit 2c773bcaf4dd5031d0b2c980cc623ecfbcc6ab62
Show file tree
Hide file tree
Showing 2 changed files with 89 additions and 54 deletions.
13 include/linux/srcutree.h
View file
@@ -63,8 +63,9 @@ struct srcu_struct {
struct srcu_node *node; /* Combining tree. */
struct srcu_node *level[RCU_NUM_LVLS + 1];
/* First node at each level. */
int srcu_size_state; /* Small-to-big transition state. */
struct mutex srcu_cb_mutex; /* Serialize CB preparation. */
spinlock_t __private lock; /* Protect counters */
spinlock_t __private lock; /* Protect counters and size state. */
struct mutex srcu_gp_mutex; /* Serialize GP work. */
unsigned int srcu_idx; /* Current rdr array element. */
unsigned long srcu_gp_seq; /* Grace-period seq #. */
@@ -83,6 +84,16 @@ struct srcu_struct {
struct lockdep_map dep_map;
};

/* Values for size state variable (->srcu_size_state). */
#define SRCU_SIZE_SMALL 0
#define SRCU_SIZE_ALLOC 1
#define SRCU_SIZE_WAIT_GP 2
#define SRCU_SIZE_WAIT_CBS1 3
#define SRCU_SIZE_WAIT_CBS2 4
#define SRCU_SIZE_WAIT_CBS3 5
#define SRCU_SIZE_WAIT_CBS4 6
#define SRCU_SIZE_BIG 7

/* Values for state variable (bottom bits of ->srcu_gp_seq). */
#define SRCU_STATE_IDLE 0
#define SRCU_STATE_SCAN1 1
130 kernel/rcu/srcutree.c
View file
@@ -177,14 +177,14 @@ static bool init_srcu_struct_nodes(struct srcu_struct *ssp)
}
sdp->grpmask = 1 << (cpu - sdp->mynode->grplo);
}
smp_store_release(&ssp->srcu_size_state, SRCU_SIZE_BIG);
return true;
}

/*
* Initialize non-compile-time initialized fields, including the
* associated srcu_node and srcu_data structures. The is_static
* parameter is passed through to init_srcu_struct_nodes(), and
* also tells us that ->sda has already been wired up to srcu_data.
* associated srcu_node and srcu_data structures. The is_static parameter
* tells us that ->sda has already been wired up to srcu_data.
*/
static int init_srcu_struct_fields(struct srcu_struct *ssp, bool is_static)
{
@@ -467,6 +467,10 @@ static void srcu_gp_start(struct srcu_struct *ssp)
struct srcu_data *sdp = this_cpu_ptr(ssp->sda);
int state;

if (smp_load_acquire(&ssp->srcu_size_state))
sdp = this_cpu_ptr(ssp->sda);
else
sdp = per_cpu_ptr(ssp->sda, 0);
lockdep_assert_held(&ACCESS_PRIVATE(ssp, lock));
WARN_ON_ONCE(ULONG_CMP_GE(ssp->srcu_gp_seq, ssp->srcu_gp_seq_needed));
spin_lock_rcu_node(sdp); /* Interrupts already disabled. */
@@ -567,38 +571,40 @@ static void srcu_gp_end(struct srcu_struct *ssp)
/* A new grace period can start at this point. But only one. */

/* Initiate callback invocation as needed. */
idx = rcu_seq_ctr(gpseq) % ARRAY_SIZE(snp->srcu_have_cbs);
srcu_for_each_node_breadth_first(ssp, snp) {
spin_lock_irq_rcu_node(snp);
cbs = false;
last_lvl = snp >= ssp->level[rcu_num_lvls - 1];
if (last_lvl)
cbs = snp->srcu_have_cbs[idx] == gpseq;
snp->srcu_have_cbs[idx] = gpseq;
rcu_seq_set_state(&snp->srcu_have_cbs[idx], 1);
if (ULONG_CMP_LT(snp->srcu_gp_seq_needed_exp, gpseq))
WRITE_ONCE(snp->srcu_gp_seq_needed_exp, gpseq);
mask = snp->srcu_data_have_cbs[idx];
snp->srcu_data_have_cbs[idx] = 0;
spin_unlock_irq_rcu_node(snp);
if (cbs)
srcu_schedule_cbs_snp(ssp, snp, mask, cbdelay);

/* Occasionally prevent srcu_data counter wrap. */
if (!(gpseq & counter_wrap_check) && last_lvl)
for (cpu = snp->grplo; cpu <= snp->grphi; cpu++) {
sdp = per_cpu_ptr(ssp->sda, cpu);
spin_lock_irqsave_rcu_node(sdp, flags);
if (ULONG_CMP_GE(gpseq,
sdp->srcu_gp_seq_needed + 100))
sdp->srcu_gp_seq_needed = gpseq;
if (ULONG_CMP_GE(gpseq,
sdp->srcu_gp_seq_needed_exp + 100))
sdp->srcu_gp_seq_needed_exp = gpseq;
spin_unlock_irqrestore_rcu_node(sdp, flags);
}
if (smp_load_acquire(&ssp->srcu_size_state) == SRCU_SIZE_SMALL) {
srcu_schedule_cbs_sdp(per_cpu_ptr(ssp->sda, 0), cbdelay);
} else {
idx = rcu_seq_ctr(gpseq) % ARRAY_SIZE(snp->srcu_have_cbs);
srcu_for_each_node_breadth_first(ssp, snp) {
spin_lock_irq_rcu_node(snp);
cbs = false;
last_lvl = snp >= ssp->level[rcu_num_lvls - 1];
if (last_lvl)
cbs = snp->srcu_have_cbs[idx] == gpseq;
snp->srcu_have_cbs[idx] = gpseq;
rcu_seq_set_state(&snp->srcu_have_cbs[idx], 1);
if (ULONG_CMP_LT(snp->srcu_gp_seq_needed_exp, gpseq))
WRITE_ONCE(snp->srcu_gp_seq_needed_exp, gpseq);
mask = snp->srcu_data_have_cbs[idx];
snp->srcu_data_have_cbs[idx] = 0;
spin_unlock_irq_rcu_node(snp);
if (cbs)
srcu_schedule_cbs_snp(ssp, snp, mask, cbdelay);
}
}

/* Occasionally prevent srcu_data counter wrap. */
if (!(gpseq & counter_wrap_check))
for_each_possible_cpu(cpu) {
sdp = per_cpu_ptr(ssp->sda, cpu);
spin_lock_irqsave_rcu_node(sdp, flags);
if (ULONG_CMP_GE(gpseq, sdp->srcu_gp_seq_needed + 100))
sdp->srcu_gp_seq_needed = gpseq;
if (ULONG_CMP_GE(gpseq, sdp->srcu_gp_seq_needed_exp + 100))
sdp->srcu_gp_seq_needed_exp = gpseq;
spin_unlock_irqrestore_rcu_node(sdp, flags);
}

/* Callback initiation done, allow grace periods after next. */
mutex_unlock(&ssp->srcu_cb_mutex);

@@ -850,7 +856,11 @@ static unsigned long srcu_gp_start_if_needed(struct srcu_struct *ssp,

check_init_srcu_struct(ssp);
idx = srcu_read_lock(ssp);
sdp = raw_cpu_ptr(ssp->sda);
if (!smp_load_acquire(&ssp->srcu_size_state)) {
sdp = per_cpu_ptr(ssp->sda, 0);
} else {
sdp = raw_cpu_ptr(ssp->sda);
}
spin_lock_irqsave_rcu_node(sdp, flags);
if (rhp)
rcu_segcblist_enqueue(&sdp->srcu_cblist, rhp);
@@ -1130,6 +1140,24 @@ static void srcu_barrier_cb(struct rcu_head *rhp)
complete(&ssp->srcu_barrier_completion);
}

/*
* If needed, enqueue an srcu_barrier() callback on the specified
* srcu_data structure's ->cblist.
*/
static void srcu_barrier_one_cpu(struct srcu_struct *ssp, struct srcu_data *sdp)
{
spin_lock_irq_rcu_node(sdp);
atomic_inc(&ssp->srcu_barrier_cpu_cnt);
sdp->srcu_barrier_head.func = srcu_barrier_cb;
debug_rcu_head_queue(&sdp->srcu_barrier_head);
if (!rcu_segcblist_entrain(&sdp->srcu_cblist,
&sdp->srcu_barrier_head)) {
debug_rcu_head_unqueue(&sdp->srcu_barrier_head);
atomic_dec(&ssp->srcu_barrier_cpu_cnt);
}
spin_unlock_irq_rcu_node(sdp);
}

/**
* srcu_barrier - Wait until all in-flight call_srcu() callbacks complete.
* @ssp: srcu_struct on which to wait for in-flight callbacks.
@@ -1153,26 +1181,22 @@ void srcu_barrier(struct srcu_struct *ssp)
/* Initial count prevents reaching zero until all CBs are posted. */
atomic_set(&ssp->srcu_barrier_cpu_cnt, 1);

/*
* Each pass through this loop enqueues a callback, but only
* on CPUs already having callbacks enqueued. Note that if
* a CPU already has callbacks enqueue, it must have already
* registered the need for a future grace period, so all we
* need do is enqueue a callback that will use the same
* grace period as the last callback already in the queue.
*/
for_each_possible_cpu(cpu) {
sdp = per_cpu_ptr(ssp->sda, cpu);
spin_lock_irq_rcu_node(sdp);
atomic_inc(&ssp->srcu_barrier_cpu_cnt);
sdp->srcu_barrier_head.func = srcu_barrier_cb;
debug_rcu_head_queue(&sdp->srcu_barrier_head);
if (!rcu_segcblist_entrain(&sdp->srcu_cblist,
&sdp->srcu_barrier_head)) {
debug_rcu_head_unqueue(&sdp->srcu_barrier_head);
atomic_dec(&ssp->srcu_barrier_cpu_cnt);
if (!smp_load_acquire(&ssp->srcu_size_state)) {
srcu_barrier_one_cpu(ssp, per_cpu_ptr(ssp->sda, 0));
} else {

/*
* Each pass through this loop enqueues a callback, but only
* on CPUs already having callbacks enqueued. Note that if
* a CPU already has callbacks enqueue, it must have already
* registered the need for a future grace period, so all we
* need do is enqueue a callback that will use the same
* grace period as the last callback already in the queue.
*/
for_each_possible_cpu(cpu) {
sdp = per_cpu_ptr(ssp->sda, cpu);
srcu_barrier_one_cpu(ssp, per_cpu_ptr(ssp->sda, cpu));
}
spin_unlock_irq_rcu_node(sdp);
}

/* Remove the initial count, at which point reaching zero can happen. */

0 comments on commit 2c773bc

Please sign in to comment.