Introduce kvfree_rcu(1 or 2 arguments)

From: "Joel Fernandes (Google)" <[email protected]>

Simple clean up of comments in kfree_rcu() code to keep it consistent with majority of commenting styles.

Reviewed-by Uladzislau Rezki <[email protected]>

Signed-off-by Joel Fernandes (Google) <[email protected]>

Signed-off-by Uladzislau Rezki (Sony) <[email protected]>

---
 kernel/rcu/tree.c | 16 ++++++++--------
 1 file changed, 8 insertions(+), 8 deletions(-)

kernel/rcu/tree.c +8 -8

index cd61649e1b00..1487af8e11e8 100644
@@ -3043,15 +3043,15 @@ static inline bool queue_kfree_rcu_work(struct kfree_rcu_cpu *krcp)
 static inline void kfree_rcu_drain_unlock(struct kfree_rcu_cpu *krcp,
 					  unsigned long flags)
 {
-	// Attempt to start a new batch.
+	/* Attempt to start a new batch. */
 	krcp->monitor_todo = false;
 	if (queue_kfree_rcu_work(krcp)) {
-		// Success! Our job is done here.
+		/* Success! Our job is done here. */
 		raw_spin_unlock_irqrestore(&krcp->lock, flags);
 		return;
 	}
 
-	// Previous RCU batch still in progress, try again later.
+	/* Previous RCU batch still in progress, try again later. */
 	krcp->monitor_todo = true;
 	schedule_delayed_work(&krcp->monitor_work, KFREE_DRAIN_JIFFIES);
 	raw_spin_unlock_irqrestore(&krcp->lock, flags);

@@ -3151,14 +3151,14 @@ void kfree_call_rcu(struct rcu_head *head, rcu_callback_t func)
 	unsigned long flags;
 	struct kfree_rcu_cpu *krcp;
 
-	local_irq_save(flags);	// For safely calling this_cpu_ptr().
+	local_irq_save(flags);	/* For safely calling this_cpu_ptr(). */
 	krcp = this_cpu_ptr(&krc);
 	if (krcp->initialized)
 		raw_spin_lock(&krcp->lock);
 
-	// Queue the object but don't yet schedule the batch.
+	/* Queue the object but don't yet schedule the batch. */
 	if (debug_rcu_head_queue(head)) {
-		// Probable double kfree_rcu(), just leak.
+		/* Probable double kfree_rcu(), just leak. */
 		WARN_ONCE(1, "%s(): Double-freed call. rcu_head %p\n",
 			  __func__, head);
 		goto unlock_return;

@@ -3176,7 +3176,7 @@ void kfree_call_rcu(struct rcu_head *head, rcu_callback_t func)
 
 	WRITE_ONCE(krcp->count, krcp->count + 1);
 
-	// Set timer to drain after KFREE_DRAIN_JIFFIES.
+	/* Set timer to drain after KFREE_DRAIN_JIFFIES. */
 	if (rcu_scheduler_active == RCU_SCHEDULER_RUNNING &&
 	    !krcp->monitor_todo) {
 		krcp->monitor_todo = true;

@@ -3722,7 +3722,7 @@ int rcutree_offline_cpu(unsigned int cpu)
 
 	rcutree_affinity_setting(cpu, cpu);
 
-	// nohz_full CPUs need the tick for stop-machine to work quickly
+	/* nohz_full CPUs need the tick for stop-machine to work quickly */
 	tick_dep_set(TICK_DEP_BIT_RCU);
 	return 0;
 }

Signature · 1 line

2.20.1

Quoted 2 lines · Uladzislau Rezki (Sony)

Simple clean up of comments in kfree_rcu() code to keep it consistent with majority of commenting styles.

on /* */ style?

I am (slowly) moving RCU to "//" for those reasons. ;-)

Quoted 3 lines · Paul E. McKenney

Quoted 2 lines · Uladzislau Rezki (Sony)

Simple clean up of comments in kfree_rcu() code to keep it consistent with majority of commenting styles.

[]

Quoted 3 lines

on /* */ style?

I am (slowly) moving RCU to "//" for those reasons. ;-)

I hope c99 comment styles are more commonly used soon too. checkpatch doesn't care.

Perhaps a change to coding-style.rst

---
 Documentation/process/coding-style.rst | 5 +++++
 1 file changed, 5 insertions(+)

Documentation/process/coding-style.rst +0 -0

index acb2f1b..fee647 100644
@@ -565,6 +565,11 @@ comments is a little different.
 	 * but there is no initial almost-blank line.

Quoted 25 lines · Joe Perches

Quoted 3 lines · Paul E. McKenney

Quoted 2 lines · Uladzislau Rezki (Sony)

Simple clean up of comments in kfree_rcu() code to keep it consistent with majority of commenting styles.

[]

Quoted 3 lines

on /* */ style?

I am (slowly) moving RCU to "//" for those reasons. ;-)

I hope c99 comment styles are more commonly used soon too. checkpatch doesn't care.

Perhaps a change to coding-style.rst

---
 Documentation/process/coding-style.rst | 5 +++++
 1 file changed, 5 insertions(+)

Documentation/process/coding-style.rst +0 -0

index acb2f1b..fee647 100644
@@ -565,6 +565,11 @@ comments is a little different.
 	 * but there is no initial almost-blank line.
 	 */
 

Documentation/process/coding-style.rst comments is a little different. · line 575

Quoted 8 lines

Documentation/process/coding-style.rst +5 -0

+.. code-block:: c
+
+	// Single line and inline comments may also use the c99 // style
+	// Block comments as well
+
 It's also important to comment data, whether they are basic types or derived
 types.  To this end, use just one data declaration per line (no commas for
 multiple data declarations).  This leaves you room for a small comment on each

Yeah that's fine with me. This patch just tries to keep it consistent. I am Ok with either style.

Quoted 1 line · Joel Fernandes

On Fri, May 01, 2020 at 01:52:46PM -0700, Joe Perches wrote:

Quoted 3 lines

Documentation/process/coding-style.rst

Quoted 3 lines

Perhaps a change to coding-style.rst
---

Documentation/process/coding-style.rst +0 -0

Quoted 4 lines

Quoted 4 lines

Documentation/process/coding-style.rst +0 -0

@@ -565,6 +565,11 @@ comments is a little different.
 	 * but there is no initial almost-blank line.
 	 */
 

Quoted 11 lines

Documentation/process/coding-style.rst comments is a little different. · line 575

Quoted 8 lines

Documentation/process/coding-style.rst +5 -0

+.. code-block:: c
+
+	// Single line and inline comments may also use the c99 // style
+	// Block comments as well
+
 It's also important to comment data, whether they are basic types or derived
 types.  To this end, use just one data declaration per line (no commas for
 multiple data declarations).  This leaves you room for a small comment on each

Yeah that's fine with me. This patch just tries to keep it consistent. I am Ok with either style.

My approach has been gradual change. Big-bang changes of this sort cause quite a bit of trouble. So I use "//" in new code and (sometimes) convert nearby ones when making a change.

Quoted 26 lines · Joel Fernandes

Quoted 25 lines · Joe Perches

Quoted 3 lines · Paul E. McKenney

Quoted 2 lines · Uladzislau Rezki (Sony)

Simple clean up of comments in kfree_rcu() code to keep it consistent with majority of commenting styles.

[]

Quoted 3 lines

on /* */ style?

I am (slowly) moving RCU to "//" for those reasons. ;-)

I hope c99 comment styles are more commonly used soon too. checkpatch doesn't care.

Perhaps a change to coding-style.rst

---
 Documentation/process/coding-style.rst | 5 +++++
 1 file changed, 5 insertions(+)

Documentation/process/coding-style.rst +0 -0

index acb2f1b..fee647 100644
@@ -565,6 +565,11 @@ comments is a little different.
 	 * but there is no initial almost-blank line.
 	 */
 

Quoted 11 lines

Documentation/process/coding-style.rst comments is a little different. · line 575

Quoted 8 lines

Documentation/process/coding-style.rst +5 -0

+.. code-block:: c
+
+	// Single line and inline comments may also use the c99 // style
+	// Block comments as well
+
 It's also important to comment data, whether they are basic types or derived
 types.  To this end, use just one data declaration per line (no commas for
 multiple data declarations).  This leaves you room for a small comment on each

Yeah that's fine with me. This patch just tries to keep it consistent. I am Ok with either style.

My approach has been gradual change. Big-bang changes of this sort cause quite a bit of trouble. So I use "//" in new code and (sometimes) convert nearby ones when making a change.

Quoted 8 lines · Uladzislau Rezki (Sony)

From: "Joel Fernandes (Google)" <[email protected]>

Simple clean up of comments in kfree_rcu() code to keep it consistent with majority of commenting styles.

Reviewed-by Uladzislau Rezki <[email protected]>

Signed-off-by Joel Fernandes (Google) <[email protected]>

Signed-off-by Uladzislau Rezki (Sony) <[email protected]>

Hmmm...

Exactly why is three additional characters per line preferable? Or in the case of block comments, either one or two additional lines, depending on /* */ style?

I am (slowly) moving RCU to "//" for those reasons. ;-)

Thanx, Paul

Quoted 66 lines

---
 kernel/rcu/tree.c | 16 ++++++++--------
 1 file changed, 8 insertions(+), 8 deletions(-)

kernel/rcu/tree.c +8 -8

index cd61649e1b00..1487af8e11e8 100644
@@ -3043,15 +3043,15 @@ static inline bool queue_kfree_rcu_work(struct kfree_rcu_cpu *krcp)
 static inline void kfree_rcu_drain_unlock(struct kfree_rcu_cpu *krcp,
 					  unsigned long flags)
 {
-	// Attempt to start a new batch.
+	/* Attempt to start a new batch. */
 	krcp->monitor_todo = false;
 	if (queue_kfree_rcu_work(krcp)) {
-		// Success! Our job is done here.
+		/* Success! Our job is done here. */
 		raw_spin_unlock_irqrestore(&krcp->lock, flags);
 		return;
 	}
 
-	// Previous RCU batch still in progress, try again later.
+	/* Previous RCU batch still in progress, try again later. */
 	krcp->monitor_todo = true;
 	schedule_delayed_work(&krcp->monitor_work, KFREE_DRAIN_JIFFIES);
 	raw_spin_unlock_irqrestore(&krcp->lock, flags);

@@ -3151,14 +3151,14 @@ void kfree_call_rcu(struct rcu_head *head, rcu_callback_t func)
 	unsigned long flags;
 	struct kfree_rcu_cpu *krcp;
 
-	local_irq_save(flags);	// For safely calling this_cpu_ptr().
+	local_irq_save(flags);	/* For safely calling this_cpu_ptr(). */
 	krcp = this_cpu_ptr(&krc);
 	if (krcp->initialized)
 		raw_spin_lock(&krcp->lock);
 
-	// Queue the object but don't yet schedule the batch.
+	/* Queue the object but don't yet schedule the batch. */
 	if (debug_rcu_head_queue(head)) {
-		// Probable double kfree_rcu(), just leak.
+		/* Probable double kfree_rcu(), just leak. */
 		WARN_ONCE(1, "%s(): Double-freed call. rcu_head %p\n",
 			  __func__, head);
 		goto unlock_return;

@@ -3176,7 +3176,7 @@ void kfree_call_rcu(struct rcu_head *head, rcu_callback_t func)
 
 	WRITE_ONCE(krcp->count, krcp->count + 1);
 
-	// Set timer to drain after KFREE_DRAIN_JIFFIES.
+	/* Set timer to drain after KFREE_DRAIN_JIFFIES. */
 	if (rcu_scheduler_active == RCU_SCHEDULER_RUNNING &&
 	    !krcp->monitor_todo) {
 		krcp->monitor_todo = true;

@@ -3722,7 +3722,7 @@ int rcutree_offline_cpu(unsigned int cpu)
 
 	rcutree_affinity_setting(cpu, cpu);
 
-	// nohz_full CPUs need the tick for stop-machine to work quickly
+	/* nohz_full CPUs need the tick for stop-machine to work quickly */
 	tick_dep_set(TICK_DEP_BIT_RCU);
 	return 0;
 }

Signature · 1 line

2.20.1

Quoted 10 lines · Paul E. McKenney

Quoted 8 lines · Uladzislau Rezki (Sony)

From: "Joel Fernandes (Google)" <[email protected]>

Simple clean up of comments in kfree_rcu() code to keep it consistent with majority of commenting styles.

Reviewed-by Uladzislau Rezki <[email protected]>

Signed-off-by Joel Fernandes (Google) <[email protected]>

Signed-off-by Uladzislau Rezki (Sony) <[email protected]>

Quoted 5 lines

Hmmm...

Exactly why is three additional characters per line preferable? Or in the case of block comments, either one or two additional lines, depending on /* */ style?

I prefer to keep the code consistent and then bulk convert it later. Its a bit ugly to read when its mixed up with "//" and "/* */" right now. We can convert it to // all at once later but until then it'll be good to keep it consistent in this file IMO. When I checked the kfree_rcu() code, it had more "/* */" than not, so this small change is less churn for now.

thanks,

- Joel

Quoted 71 lines

I am (slowly) moving RCU to "//" for those reasons. ;-)

Thanx, Paul

Quoted 66 lines

---
 kernel/rcu/tree.c | 16 ++++++++--------
 1 file changed, 8 insertions(+), 8 deletions(-)

kernel/rcu/tree.c +8 -8

index cd61649e1b00..1487af8e11e8 100644
@@ -3043,15 +3043,15 @@ static inline bool queue_kfree_rcu_work(struct kfree_rcu_cpu *krcp)
 static inline void kfree_rcu_drain_unlock(struct kfree_rcu_cpu *krcp,
 					  unsigned long flags)
 {
-	// Attempt to start a new batch.
+	/* Attempt to start a new batch. */
 	krcp->monitor_todo = false;
 	if (queue_kfree_rcu_work(krcp)) {
-		// Success! Our job is done here.
+		/* Success! Our job is done here. */
 		raw_spin_unlock_irqrestore(&krcp->lock, flags);
 		return;
 	}
 
-	// Previous RCU batch still in progress, try again later.
+	/* Previous RCU batch still in progress, try again later. */
 	krcp->monitor_todo = true;
 	schedule_delayed_work(&krcp->monitor_work, KFREE_DRAIN_JIFFIES);
 	raw_spin_unlock_irqrestore(&krcp->lock, flags);

@@ -3151,14 +3151,14 @@ void kfree_call_rcu(struct rcu_head *head, rcu_callback_t func)
 	unsigned long flags;
 	struct kfree_rcu_cpu *krcp;
 
-	local_irq_save(flags);	// For safely calling this_cpu_ptr().
+	local_irq_save(flags);	/* For safely calling this_cpu_ptr(). */
 	krcp = this_cpu_ptr(&krc);
 	if (krcp->initialized)
 		raw_spin_lock(&krcp->lock);
 
-	// Queue the object but don't yet schedule the batch.
+	/* Queue the object but don't yet schedule the batch. */
 	if (debug_rcu_head_queue(head)) {
-		// Probable double kfree_rcu(), just leak.
+		/* Probable double kfree_rcu(), just leak. */
 		WARN_ONCE(1, "%s(): Double-freed call. rcu_head %p\n",
 			  __func__, head);
 		goto unlock_return;

@@ -3176,7 +3176,7 @@ void kfree_call_rcu(struct rcu_head *head, rcu_callback_t func)
 
 	WRITE_ONCE(krcp->count, krcp->count + 1);
 
-	// Set timer to drain after KFREE_DRAIN_JIFFIES.
+	/* Set timer to drain after KFREE_DRAIN_JIFFIES. */
 	if (rcu_scheduler_active == RCU_SCHEDULER_RUNNING &&
 	    !krcp->monitor_todo) {
 		krcp->monitor_todo = true;

@@ -3722,7 +3722,7 @@ int rcutree_offline_cpu(unsigned int cpu)
 
 	rcutree_affinity_setting(cpu, cpu);
 
-	// nohz_full CPUs need the tick for stop-machine to work quickly
+	/* nohz_full CPUs need the tick for stop-machine to work quickly */
 	tick_dep_set(TICK_DEP_BIT_RCU);
 	return 0;
 }

Signature · 1 line

2.20.1

Quoted 17 lines · Joel Fernandes

Quoted 15 lines · Paul E. McKenney

Quoted 8 lines · Uladzislau Rezki (Sony)

From: "Joel Fernandes (Google)" <[email protected]>

Simple clean up of comments in kfree_rcu() code to keep it consistent with majority of commenting styles.

Reviewed-by Uladzislau Rezki <[email protected]>

Signed-off-by Joel Fernandes (Google) <[email protected]>

Signed-off-by Uladzislau Rezki (Sony) <[email protected]>

Hmmm...

Exactly why is three additional characters per line preferable? Or in the case of block comments, either one or two additional lines, depending on /* */ style?

Quoted 5 lines

I prefer to keep the code consistent and then bulk convert it later. Its a bit ugly to read when its mixed up with "//" and "/* */" right now. We can convert it to // all at once later but until then it'll be good to keep it consistent in this file IMO. When I checked the kfree_rcu() code, it had more "/* */" than not, so this small change is less churn for now.

Please just drop this patch along with the other "//"-to-"/* */" regressions.

If you want to convert more comments to "//" within the confines of the kfree_rcu() code, I am probably OK with that. But again, a big-bang change of this sort often causes problems due to lots of potential rebase/merge conflicts.

Thanx, Paul

Quoted 75 lines

thanks,

- Joel

Quoted 71 lines

I am (slowly) moving RCU to "//" for those reasons. ;-)

Thanx, Paul

Quoted 66 lines

---
 kernel/rcu/tree.c | 16 ++++++++--------
 1 file changed, 8 insertions(+), 8 deletions(-)

kernel/rcu/tree.c +8 -8

index cd61649e1b00..1487af8e11e8 100644
@@ -3043,15 +3043,15 @@ static inline bool queue_kfree_rcu_work(struct kfree_rcu_cpu *krcp)
 static inline void kfree_rcu_drain_unlock(struct kfree_rcu_cpu *krcp,
 					  unsigned long flags)
 {
-	// Attempt to start a new batch.
+	/* Attempt to start a new batch. */
 	krcp->monitor_todo = false;
 	if (queue_kfree_rcu_work(krcp)) {
-		// Success! Our job is done here.
+		/* Success! Our job is done here. */
 		raw_spin_unlock_irqrestore(&krcp->lock, flags);
 		return;
 	}
 
-	// Previous RCU batch still in progress, try again later.
+	/* Previous RCU batch still in progress, try again later. */
 	krcp->monitor_todo = true;
 	schedule_delayed_work(&krcp->monitor_work, KFREE_DRAIN_JIFFIES);
 	raw_spin_unlock_irqrestore(&krcp->lock, flags);

@@ -3151,14 +3151,14 @@ void kfree_call_rcu(struct rcu_head *head, rcu_callback_t func)
 	unsigned long flags;
 	struct kfree_rcu_cpu *krcp;
 
-	local_irq_save(flags);	// For safely calling this_cpu_ptr().
+	local_irq_save(flags);	/* For safely calling this_cpu_ptr(). */
 	krcp = this_cpu_ptr(&krc);
 	if (krcp->initialized)
 		raw_spin_lock(&krcp->lock);
 
-	// Queue the object but don't yet schedule the batch.
+	/* Queue the object but don't yet schedule the batch. */
 	if (debug_rcu_head_queue(head)) {
-		// Probable double kfree_rcu(), just leak.
+		/* Probable double kfree_rcu(), just leak. */
 		WARN_ONCE(1, "%s(): Double-freed call. rcu_head %p\n",
 			  __func__, head);
 		goto unlock_return;

@@ -3176,7 +3176,7 @@ void kfree_call_rcu(struct rcu_head *head, rcu_callback_t func)
 
 	WRITE_ONCE(krcp->count, krcp->count + 1);
 
-	// Set timer to drain after KFREE_DRAIN_JIFFIES.
+	/* Set timer to drain after KFREE_DRAIN_JIFFIES. */
 	if (rcu_scheduler_active == RCU_SCHEDULER_RUNNING &&
 	    !krcp->monitor_todo) {
 		krcp->monitor_todo = true;

@@ -3722,7 +3722,7 @@ int rcutree_offline_cpu(unsigned int cpu)
 
 	rcutree_affinity_setting(cpu, cpu);
 
-	// nohz_full CPUs need the tick for stop-machine to work quickly
+	/* nohz_full CPUs need the tick for stop-machine to work quickly */
 	tick_dep_set(TICK_DEP_BIT_RCU);
 	return 0;
 }

Signature · 1 line

2.20.1

From: Sebastian Andrzej Siewior <[email protected]>

The per-CPU variable is initialized at runtime in kfree_rcu_batch_init(). This function is invoked before 'rcu_scheduler_active' is set to 'RCU_SCHEDULER_RUNNING'. After the initialisation, '->initialized' is to true.

The raw_spin_lock is only acquired if '->initialized' is set to true. The worqueue item is only used if 'rcu_scheduler_active' set to RCU_SCHEDULER_RUNNING which happens after initialisation.

Use a static initializer for krc.lock and remove the runtime initialisation of the lock. Since the lock can now be always acquired, remove the '->initialized' check.

Signed-off-by Sebastian Andrzej Siewior <[email protected]>

Signed-off-by Uladzislau Rezki (Sony) <[email protected]>

---
 kernel/rcu/tree.c | 15 +++++++--------
 1 file changed, 7 insertions(+), 8 deletions(-)

kernel/rcu/tree.c +4 -2

index bc6c2bc8fa32..89e9ca3f4e3e 100644
@@ -2892,7 +2892,7 @@ struct kfree_rcu_cpu_work {
  * @lock: Synchronize access to this structure
  * @monitor_work: Promote @head to @head_free after KFREE_DRAIN_JIFFIES
  * @monitor_todo: Tracks whether a @monitor_work delayed work is pending
- * @initialized: The @lock and @rcu_work fields have been initialized
+ * @initialized: The @rcu_work fields have been initialized
  *
  * This is a per-CPU structure.  The reason that it is not included in
  * the rcu_data structure is to permit this code to be extracted from

@@ -2912,7 +2912,9 @@ struct kfree_rcu_cpu {
 	int count;
 };
 
-static DEFINE_PER_CPU(struct kfree_rcu_cpu, krc);
+static DEFINE_PER_CPU(struct kfree_rcu_cpu, krc) = {
+	.lock = __RAW_SPIN_LOCK_UNLOCKED(krc.lock),
+};
 
 static __always_inline void
 debug_rcu_bhead_unqueue(struct kfree_rcu_bulk_data *bhead)

kernel/rcu/tree.c +2 -5

 	krcp = this_cpu_ptr(&krc);
-	if (likely(krcp->initialized))
-		raw_spin_lock(&krcp->lock);
+	raw_spin_lock(&krcp->lock);
 
 	return krcp;
 }

@@ -2941,8 +2942,7 @@ krc_this_cpu_lock(unsigned long *flags)
 static inline void
 krc_this_cpu_unlock(struct kfree_rcu_cpu *krcp, unsigned long flags)
 {
-	if (likely(krcp->initialized))
-		raw_spin_unlock(&krcp->lock);
+	raw_spin_unlock(&krcp->lock);
 	local_irq_restore(flags);
 }
 

@@ -4168,7 +4168,6 @@ static void __init kfree_rcu_batch_init(void)
 	for_each_possible_cpu(cpu) {
 		struct kfree_rcu_cpu *krcp = per_cpu_ptr(&krc, cpu);
 
-		raw_spin_lock_init(&krcp->lock);
 		for (i = 0; i < KFREE_N_BATCHES; i++) {
 			INIT_RCU_WORK(&krcp->krw_arr[i].rcu_work, kfree_rcu_work);
 			krcp->krw_arr[i].krcp = krcp;

Signature · 1 line

2.20.1

kernel/rcu/tree.c krc_this_cpu_lock(unsigned long *flags) · line 2935

Quoted 3 lines

kernel/rcu/tree.c +1 -1

-	local_irq_save(*flags);	// For safely calling this_cpu_ptr().
+	local_irq_save(*flags);	/* For safely calling this_cpu_ptr(). */

And here as well. ;-)

Cache some extra objects per-CPU. During reclaim process some pages are cached instead of releasing by linking them into the list. Such approach provides O(1) access time to the cache.

That reduces number of requests to the page allocator, also that makes it more helpful if a low memory condition occurs.

A parameter reflecting the minimum allowed pages to be cached per one CPU is propagated via sysfs, it is read only, the name is "rcu_min_cached_objs".

Signed-off-by Uladzislau Rezki (Sony) <[email protected]>

---
 kernel/rcu/tree.c | 64 ++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 60 insertions(+), 4 deletions(-)

kernel/rcu/tree.c +0 -0

index 89e9ca3f4e3e..d8975819b1c9 100644

kernel/rcu/tree.c +40 -2

+ * It reflects the minimum allowed number of objects which
+ * can be cached per-CPU. Object size is equal to one page.
+ */
+int rcu_min_cached_objs = 2;
+module_param(rcu_min_cached_objs, int, 0444);
+
 /* Retrieve RCU kthreads priority for rcutorture */
 int rcu_get_gp_kthreads_prio(void)
 {

@@ -2887,7 +2895,6 @@ struct kfree_rcu_cpu_work {
  * struct kfree_rcu_cpu - batch up kfree_rcu() requests for RCU grace period
  * @head: List of kfree_rcu() objects not yet waiting for a grace period
  * @bhead: Bulk-List of kfree_rcu() objects not yet waiting for a grace period
- * @bcached: Keeps at most one object for later reuse when build chain blocks
  * @krw_arr: Array of batches of kfree_rcu() objects waiting for a grace period
  * @lock: Synchronize access to this structure
  * @monitor_work: Promote @head to @head_free after KFREE_DRAIN_JIFFIES

@@ -2902,7 +2909,6 @@ struct kfree_rcu_cpu_work {
 struct kfree_rcu_cpu {
 	struct rcu_head *head;
 	struct kfree_rcu_bulk_data *bhead;
-	struct kfree_rcu_bulk_data *bcached;
 	struct kfree_rcu_cpu_work krw_arr[KFREE_N_BATCHES];
 	raw_spinlock_t lock;
 	struct delayed_work monitor_work;

@@ -2910,6 +2916,15 @@ struct kfree_rcu_cpu {
 	bool initialized;
 	// Number of objects for which GP not started
 	int count;
+
+	/*
+	 * Number of cached objects which are queued into
+	 * the lock-less list. This cache is used by the
+	 * kvfree_call_rcu() function and as of now its
+	 * size is static.
+	 */
+	struct llist_head bkvcache;
+	int nr_bkv_objs;
 };
 
 static DEFINE_PER_CPU(struct kfree_rcu_cpu, krc) = {

@@ -2946,6 +2961,31 @@ krc_this_cpu_unlock(struct kfree_rcu_cpu *krcp, unsigned long flags)
 	local_irq_restore(flags);
 }
 
+static inline struct kfree_rcu_bulk_data *
+get_cached_bnode(struct kfree_rcu_cpu *krcp)
+{
+	if (!krcp->nr_bkv_objs)
+		return NULL;
+
+	krcp->nr_bkv_objs--;
+	return (struct kfree_rcu_bulk_data *)
+		llist_del_first(&krcp->bkvcache);
+}
+
+static inline bool
+put_cached_bnode(struct kfree_rcu_cpu *krcp,
+	struct kfree_rcu_bulk_data *bnode)
+{
+	/* Check the limit. */
+	if (krcp->nr_bkv_objs >= rcu_min_cached_objs)
+		return false;
+
+	llist_add((struct llist_node *) bnode, &krcp->bkvcache);
+	krcp->nr_bkv_objs++;
+	return true;
+
+}
+
 /*
  * This function is invoked in workqueue context after a grace period.
  * It frees all the objects queued on ->bhead_free or ->head_free.

kernel/rcu/tree.c +17 -1

+		if (put_cached_bnode(krcp, bhead))
+			bhead = NULL;
+		krc_this_cpu_unlock(krcp, flags);
+
+		if (bhead)
 			free_page((unsigned long) bhead);
 
 		cond_resched_tasks_rcu_qs();

@@ -3114,7 +3159,7 @@ kfree_call_rcu_add_ptr_to_bulk(struct kfree_rcu_cpu *krcp,
 	/* Check if a new block is required. */
 	if (!krcp->bhead ||
 			krcp->bhead->nr_records == KFREE_BULK_MAX_ENTR) {
-		bnode = xchg(&krcp->bcached, NULL);
+		bnode = get_cached_bnode(krcp);
 		if (!bnode) {
 			WARN_ON_ONCE(sizeof(struct kfree_rcu_bulk_data) > PAGE_SIZE);
 

@@ -4167,12 +4212,23 @@ static void __init kfree_rcu_batch_init(void)
 
 	for_each_possible_cpu(cpu) {
 		struct kfree_rcu_cpu *krcp = per_cpu_ptr(&krc, cpu);
+		struct kfree_rcu_bulk_data *bnode;
 
 		for (i = 0; i < KFREE_N_BATCHES; i++) {
 			INIT_RCU_WORK(&krcp->krw_arr[i].rcu_work, kfree_rcu_work);
 			krcp->krw_arr[i].krcp = krcp;
 		}
 
+		for (i = 0; i < rcu_min_cached_objs; i++) {
+			bnode = (struct kfree_rcu_bulk_data *)
+				__get_free_page(GFP_NOWAIT | __GFP_NOWARN);
+
+			if (bnode)
+				put_cached_bnode(krcp, bnode);
+			else
+				pr_err("Failed to preallocate for %d CPU!\n", cpu);
+		}
+
 		INIT_DELAYED_WORK(&krcp->monitor_work, kfree_rcu_monitor);
 		krcp->initialized = true;
 	}

Signature · 1 line

2.20.1

Quoted 25 lines · Uladzislau Rezki (Sony)

Cache some extra objects per-CPU. During reclaim process some pages are cached instead of releasing by linking them into the list. Such approach provides O(1) access time to the cache.

That reduces number of requests to the page allocator, also that makes it more helpful if a low memory condition occurs.

A parameter reflecting the minimum allowed pages to be cached per one CPU is propagated via sysfs, it is read only, the name is "rcu_min_cached_objs".

Signed-off-by Uladzislau Rezki (Sony) <[email protected]>

---
 kernel/rcu/tree.c | 64 ++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 60 insertions(+), 4 deletions(-)

kernel/rcu/tree.c +0 -0

index 89e9ca3f4e3e..d8975819b1c9 100644
@@ -178,6 +178,14 @@ module_param(gp_init_delay, int, 0444);
 static int gp_cleanup_delay;
 module_param(gp_cleanup_delay, int, 0444);
 

kernel/rcu/tree.c module_param(gp_init_delay, int, 0444); · line 182

Quoted 2 lines

kernel/rcu/tree.c +2 -0

+/*
+ * This rcu parameter is read-only, but can be write also.

You mean that although the parameter is read-only, you see no reason why it could not be converted to writeable?

If it was writeable, and a given CPU had the maximum numbr of cached objects, the rcu_min_cached_objs value was decreased, but that CPU never saw another kfree_rcu(), would the number of cached objects change?

(Just curious, not asking for a change in functionality.)

Quoted 77 lines

kernel/rcu/tree.c +40 -2

+ * It reflects the minimum allowed number of objects which
+ * can be cached per-CPU. Object size is equal to one page.
+ */
+int rcu_min_cached_objs = 2;
+module_param(rcu_min_cached_objs, int, 0444);
+
 /* Retrieve RCU kthreads priority for rcutorture */
 int rcu_get_gp_kthreads_prio(void)
 {

@@ -2887,7 +2895,6 @@ struct kfree_rcu_cpu_work {
  * struct kfree_rcu_cpu - batch up kfree_rcu() requests for RCU grace period
  * @head: List of kfree_rcu() objects not yet waiting for a grace period
  * @bhead: Bulk-List of kfree_rcu() objects not yet waiting for a grace period
- * @bcached: Keeps at most one object for later reuse when build chain blocks
  * @krw_arr: Array of batches of kfree_rcu() objects waiting for a grace period
  * @lock: Synchronize access to this structure
  * @monitor_work: Promote @head to @head_free after KFREE_DRAIN_JIFFIES

@@ -2902,7 +2909,6 @@ struct kfree_rcu_cpu_work {
 struct kfree_rcu_cpu {
 	struct rcu_head *head;
 	struct kfree_rcu_bulk_data *bhead;
-	struct kfree_rcu_bulk_data *bcached;
 	struct kfree_rcu_cpu_work krw_arr[KFREE_N_BATCHES];
 	raw_spinlock_t lock;
 	struct delayed_work monitor_work;

@@ -2910,6 +2916,15 @@ struct kfree_rcu_cpu {
 	bool initialized;
 	// Number of objects for which GP not started
 	int count;
+
+	/*
+	 * Number of cached objects which are queued into
+	 * the lock-less list. This cache is used by the
+	 * kvfree_call_rcu() function and as of now its
+	 * size is static.
+	 */
+	struct llist_head bkvcache;
+	int nr_bkv_objs;
 };
 
 static DEFINE_PER_CPU(struct kfree_rcu_cpu, krc) = {

@@ -2946,6 +2961,31 @@ krc_this_cpu_unlock(struct kfree_rcu_cpu *krcp, unsigned long flags)
 	local_irq_restore(flags);
 }
 
+static inline struct kfree_rcu_bulk_data *
+get_cached_bnode(struct kfree_rcu_cpu *krcp)
+{
+	if (!krcp->nr_bkv_objs)
+		return NULL;
+
+	krcp->nr_bkv_objs--;
+	return (struct kfree_rcu_bulk_data *)
+		llist_del_first(&krcp->bkvcache);
+}
+
+static inline bool
+put_cached_bnode(struct kfree_rcu_cpu *krcp,
+	struct kfree_rcu_bulk_data *bnode)
+{
+	/* Check the limit. */
+	if (krcp->nr_bkv_objs >= rcu_min_cached_objs)
+		return false;
+
+	llist_add((struct llist_node *) bnode, &krcp->bkvcache);
+	krcp->nr_bkv_objs++;
+	return true;
+
+}
+
 /*
  * This function is invoked in workqueue context after a grace period.
  * It frees all the objects queued on ->bhead_free or ->head_free.

@@ -2981,7 +3021,12 @@ static void kfree_rcu_work(struct work_struct *work)
 		kfree_bulk(bhead->nr_records, bhead->records);
 		rcu_lock_release(&rcu_callback_map);
 

kernel/rcu/tree.c static void kfree_rcu_work(struct work_struct *work) · line 3024

Quoted 2 lines

kernel/rcu/tree.c +1 -1

-		if (cmpxchg(&krcp->bcached, NULL, bhead))
+		krcp = krc_this_cpu_lock(&flags);

Presumably the list can also be accessed without holding this lock, because otherwise we shouldn't need llist...

Quoted 26 lines · Paul E. McKenney

Quoted 25 lines · Uladzislau Rezki (Sony)

Cache some extra objects per-CPU. During reclaim process some pages are cached instead of releasing by linking them into the list. Such approach provides O(1) access time to the cache.

That reduces number of requests to the page allocator, also that makes it more helpful if a low memory condition occurs.

A parameter reflecting the minimum allowed pages to be cached per one CPU is propagated via sysfs, it is read only, the name is "rcu_min_cached_objs".

Signed-off-by Uladzislau Rezki (Sony) <[email protected]>

---
 kernel/rcu/tree.c | 64 ++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 60 insertions(+), 4 deletions(-)

kernel/rcu/tree.c +0 -0

index 89e9ca3f4e3e..d8975819b1c9 100644
@@ -178,6 +178,14 @@ module_param(gp_init_delay, int, 0444);
 static int gp_cleanup_delay;
 module_param(gp_cleanup_delay, int, 0444);
 

Quoted 6 lines

kernel/rcu/tree.c module_param(gp_init_delay, int, 0444); · line 182

Quoted 2 lines

kernel/rcu/tree.c +2 -0

+/*
+ * This rcu parameter is read-only, but can be write also.

You mean that although the parameter is read-only, you see no reason why it could not be converted to writeable?

I added just a note. If it is writable, then we can change the size of the per-CPU cache dynamically, i.e. "echo 5 > /sys/.../rcu_min_cached_objs" would cache 5 pages. But i do not have a strong opinion if it should be writable.

Quoted 4 lines

If it was writeable, and a given CPU had the maximum numbr of cached objects, the rcu_min_cached_objs value was decreased, but that CPU never saw another kfree_rcu(), would the number of cached objects change?

No. It works the way: unqueue the page from cache in the kfree_rcu(), whereas "rcu work" will put it back if number of objects < rcu_min_cached_objs, if >= will free the page.

Quoted 79 lines

(Just curious, not asking for a change in functionality.)

Quoted 77 lines

kernel/rcu/tree.c +40 -2

+ * It reflects the minimum allowed number of objects which
+ * can be cached per-CPU. Object size is equal to one page.
+ */
+int rcu_min_cached_objs = 2;
+module_param(rcu_min_cached_objs, int, 0444);
+
 /* Retrieve RCU kthreads priority for rcutorture */
 int rcu_get_gp_kthreads_prio(void)
 {

@@ -2887,7 +2895,6 @@ struct kfree_rcu_cpu_work {
  * struct kfree_rcu_cpu - batch up kfree_rcu() requests for RCU grace period
  * @head: List of kfree_rcu() objects not yet waiting for a grace period
  * @bhead: Bulk-List of kfree_rcu() objects not yet waiting for a grace period
- * @bcached: Keeps at most one object for later reuse when build chain blocks
  * @krw_arr: Array of batches of kfree_rcu() objects waiting for a grace period
  * @lock: Synchronize access to this structure
  * @monitor_work: Promote @head to @head_free after KFREE_DRAIN_JIFFIES

@@ -2902,7 +2909,6 @@ struct kfree_rcu_cpu_work {
 struct kfree_rcu_cpu {
 	struct rcu_head *head;
 	struct kfree_rcu_bulk_data *bhead;
-	struct kfree_rcu_bulk_data *bcached;
 	struct kfree_rcu_cpu_work krw_arr[KFREE_N_BATCHES];
 	raw_spinlock_t lock;
 	struct delayed_work monitor_work;

@@ -2910,6 +2916,15 @@ struct kfree_rcu_cpu {
 	bool initialized;
 	// Number of objects for which GP not started
 	int count;
+
+	/*
+	 * Number of cached objects which are queued into
+	 * the lock-less list. This cache is used by the
+	 * kvfree_call_rcu() function and as of now its
+	 * size is static.
+	 */
+	struct llist_head bkvcache;
+	int nr_bkv_objs;
 };
 
 static DEFINE_PER_CPU(struct kfree_rcu_cpu, krc) = {

@@ -2946,6 +2961,31 @@ krc_this_cpu_unlock(struct kfree_rcu_cpu *krcp, unsigned long flags)
 	local_irq_restore(flags);
 }
 
+static inline struct kfree_rcu_bulk_data *
+get_cached_bnode(struct kfree_rcu_cpu *krcp)
+{
+	if (!krcp->nr_bkv_objs)
+		return NULL;
+
+	krcp->nr_bkv_objs--;
+	return (struct kfree_rcu_bulk_data *)
+		llist_del_first(&krcp->bkvcache);
+}
+
+static inline bool
+put_cached_bnode(struct kfree_rcu_cpu *krcp,
+	struct kfree_rcu_bulk_data *bnode)
+{
+	/* Check the limit. */
+	if (krcp->nr_bkv_objs >= rcu_min_cached_objs)
+		return false;
+
+	llist_add((struct llist_node *) bnode, &krcp->bkvcache);
+	krcp->nr_bkv_objs++;
+	return true;
+
+}
+
 /*
  * This function is invoked in workqueue context after a grace period.
  * It frees all the objects queued on ->bhead_free or ->head_free.

@@ -2981,7 +3021,12 @@ static void kfree_rcu_work(struct work_struct *work)
 		kfree_bulk(bhead->nr_records, bhead->records);
 		rcu_lock_release(&rcu_callback_map);
 

Quoted 6 lines

kernel/rcu/tree.c static void kfree_rcu_work(struct work_struct *work) · line 3024

Quoted 2 lines

kernel/rcu/tree.c +1 -1

-		if (cmpxchg(&krcp->bcached, NULL, bhead))
+		krcp = krc_this_cpu_lock(&flags);

Presumably the list can also be accessed without holding this lock, because otherwise we shouldn't need llist...

Hm... We increase the number of elements in cache, therefore it is not lockless. From the other hand i used llist_head to maintain the cache because it is single linked list, we do not need "*prev" link. Also we do not need to init the list.

But i can change it to list_head. Please let me know if i need :)

Quoted 42 lines · Uladzislau Rezki

Quoted 26 lines · Paul E. McKenney

Quoted 25 lines · Uladzislau Rezki (Sony)

Cache some extra objects per-CPU. During reclaim process some pages are cached instead of releasing by linking them into the list. Such approach provides O(1) access time to the cache.

That reduces number of requests to the page allocator, also that makes it more helpful if a low memory condition occurs.

A parameter reflecting the minimum allowed pages to be cached per one CPU is propagated via sysfs, it is read only, the name is "rcu_min_cached_objs".

Signed-off-by Uladzislau Rezki (Sony) <[email protected]>

---
 kernel/rcu/tree.c | 64 ++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 60 insertions(+), 4 deletions(-)

kernel/rcu/tree.c +0 -0

index 89e9ca3f4e3e..d8975819b1c9 100644
@@ -178,6 +178,14 @@ module_param(gp_init_delay, int, 0444);
 static int gp_cleanup_delay;
 module_param(gp_cleanup_delay, int, 0444);
 

Quoted 6 lines

kernel/rcu/tree.c module_param(gp_init_delay, int, 0444); · line 182

Quoted 2 lines

kernel/rcu/tree.c +2 -0

+/*
+ * This rcu parameter is read-only, but can be write also.

You mean that although the parameter is read-only, you see no reason why it could not be converted to writeable?

I added just a note. If it is writable, then we can change the size of the per-CPU cache dynamically, i.e. "echo 5 > /sys/.../rcu_min_cached_objs" would cache 5 pages. But i do not have a strong opinion if it should be writable.

Quoted 4 lines

If it was writeable, and a given CPU had the maximum numbr of cached objects, the rcu_min_cached_objs value was decreased, but that CPU never saw another kfree_rcu(), would the number of cached objects change?

Quoted 3 lines

No. It works the way: unqueue the page from cache in the kfree_rcu(), whereas "rcu work" will put it back if number of objects < rcu_min_cached_objs, if >= will free the page.

Just to make sure I understand... If someone writes a smaller number to the sysfs variable, the per-CPU caches will be decreased at that point, immediately during that sysfs write? Or are you saying something else?

Quoted 9 lines

Quoted 3 lines

Presumably the list can also be accessed without holding this lock, because otherwise we shouldn't need llist...

Hm... We increase the number of elements in cache, therefore it is not lockless. From the other hand i used llist_head to maintain the cache because it is single linked list, we do not need "*prev" link. Also we do not need to init the list.

But i can change it to list_head. Please let me know if i need :)

Hmmm... Maybe it is time for a non-atomic singly linked list? In the RCU callback processing, the operations were open-coded, but they have been pushed into include/linux/rcu_segcblist.h and kernel/rcu/rcu_segcblist.*.

Maybe some non-atomic/protected/whatever macros in the llist.h file? Or maybe just open-code the singly linked list? (Probably not the best choice, though.) Add comments stating that the atomic properties of the llist functions aren't neded? Something else?

The comments would be a good start. Just to take pity on people seeing the potential for concurrency and wondering how the concurrent accesses actually happen. ;-)

Quoted 47 lines · Paul E. McKenney

Quoted 45 lines · Uladzislau Rezki

Quoted 26 lines · Paul E. McKenney

Quoted 25 lines · Uladzislau Rezki (Sony)

Cache some extra objects per-CPU. During reclaim process some pages are cached instead of releasing by linking them into the list. Such approach provides O(1) access time to the cache.

That reduces number of requests to the page allocator, also that makes it more helpful if a low memory condition occurs.

A parameter reflecting the minimum allowed pages to be cached per one CPU is propagated via sysfs, it is read only, the name is "rcu_min_cached_objs".

Signed-off-by Uladzislau Rezki (Sony) <[email protected]>

---
 kernel/rcu/tree.c | 64 ++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 60 insertions(+), 4 deletions(-)

kernel/rcu/tree.c +0 -0

index 89e9ca3f4e3e..d8975819b1c9 100644
@@ -178,6 +178,14 @@ module_param(gp_init_delay, int, 0444);
 static int gp_cleanup_delay;
 module_param(gp_cleanup_delay, int, 0444);
 

Quoted 6 lines

kernel/rcu/tree.c module_param(gp_init_delay, int, 0444); · line 182

Quoted 2 lines

kernel/rcu/tree.c +2 -0

+/*
+ * This rcu parameter is read-only, but can be write also.

You mean that although the parameter is read-only, you see no reason why it could not be converted to writeable?

I added just a note. If it is writable, then we can change the size of the per-CPU cache dynamically, i.e. "echo 5 > /sys/.../rcu_min_cached_objs" would cache 5 pages. But i do not have a strong opinion if it should be writable.

Quoted 4 lines

If it was writeable, and a given CPU had the maximum numbr of cached objects, the rcu_min_cached_objs value was decreased, but that CPU never saw another kfree_rcu(), would the number of cached objects change?

No. It works the way: unqueue the page from cache in the kfree_rcu(), whereas "rcu work" will put it back if number of objects < rcu_min_cached_objs, if >= will free the page.

Quoted 4 lines

Just to make sure I understand... If someone writes a smaller number to the sysfs variable, the per-CPU caches will be decreased at that point, immediately during that sysfs write? Or are you saying something else?

This patch defines it as read-only. It defines the minimum threshold that controls number of elements in the per-CPU cache. If we decide to make it write also, then we will have full of freedom how to define its behavior, i.e. it is not defined because it is read only.

Quoted 14 lines

Quoted 9 lines

Quoted 3 lines

Presumably the list can also be accessed without holding this lock, because otherwise we shouldn't need llist...

Hm... We increase the number of elements in cache, therefore it is not lockless. From the other hand i used llist_head to maintain the cache because it is single linked list, we do not need "*prev" link. Also we do not need to init the list.

But i can change it to list_head. Please let me know if i need :)

Hmmm... Maybe it is time for a non-atomic singly linked list? In the RCU callback processing, the operations were open-coded, but they have been pushed into include/linux/rcu_segcblist.h and kernel/rcu/rcu_segcblist.*.

Quoted 5 lines

Maybe some non-atomic/protected/whatever macros in the llist.h file? Or maybe just open-code the singly linked list? (Probably not the best choice, though.) Add comments stating that the atomic properties of the llist functions aren't neded? Something else?

In order to keep it simple i can replace llist_head by the list_head?

Quoted 4 lines

The comments would be a good start. Just to take pity on people seeing the potential for concurrency and wondering how the concurrent accesses actually happen. ;-)

Sounds like you are kidding me :)

On Mon, May 04, 2020 at 08:24:37AM -0700, Paul E. McKenney wrote:

Quoted 14 lines

Quoted 9 lines

Quoted 3 lines

Presumably the list can also be accessed without holding this lock, because otherwise we shouldn't need llist...

Hm... We increase the number of elements in cache, therefore it is not lockless. From the other hand i used llist_head to maintain the cache because it is single linked list, we do not need "*prev" link. Also we do not need to init the list.

But i can change it to list_head. Please let me know if i need :)

Hmmm... Maybe it is time for a non-atomic singly linked list? In the RCU callback processing, the operations were open-coded, but they have been pushed into include/linux/rcu_segcblist.h and kernel/rcu/rcu_segcblist.*.

Quoted 5 lines

Maybe some non-atomic/protected/whatever macros in the llist.h file? Or maybe just open-code the singly linked list? (Probably not the best choice, though.) Add comments stating that the atomic properties of the llist functions aren't neded? Something else?

In order to keep it simple i can replace llist_head by the list_head?

Quoted 5 lines

The comments would be a good start. Just to take pity on people seeing the potential for concurrency and wondering how the concurrent accesses actually happen. ;-)

Sounds like you are kidding me :)

--
Vlad Rezki

Quoted 1 line · Uladzislau Rezki

On Mon, May 04, 2020 at 08:24:37AM -0700, Paul E. McKenney wrote:

[..]

Quoted 14 lines

Quoted 14 lines

Quoted 9 lines

Quoted 3 lines

Presumably the list can also be accessed without holding this lock, because otherwise we shouldn't need llist...

Hm... We increase the number of elements in cache, therefore it is not lockless. From the other hand i used llist_head to maintain the cache because it is single linked list, we do not need "*prev" link. Also we do not need to init the list.

But i can change it to list_head. Please let me know if i need :)

Hmmm... Maybe it is time for a non-atomic singly linked list? In the RCU callback processing, the operations were open-coded, but they have been pushed into include/linux/rcu_segcblist.h and kernel/rcu/rcu_segcblist.*.

Quoted 6 lines

Quoted 5 lines

Maybe some non-atomic/protected/whatever macros in the llist.h file? Or maybe just open-code the singly linked list? (Probably not the best choice, though.) Add comments stating that the atomic properties of the llist functions aren't neded? Something else?

In order to keep it simple i can replace llist_head by the list_head?

Just to clarify for me, what is the disadvantage of using llist here?

Since we don't care about traversing backwards, isn't it better to use llist for this usecase?

I think Vlad is using locking as we're also tracking the size of the llist to know when to free pages. This tracking could suffer from the lost-update problem without any locking, 2 lockless llist_add happened simulatenously.

Also if list_head is used, it will take more space and still use locking.

Thoughts?

thanks,

- Joel

Quoted 9 lines

Quoted 5 lines

The comments would be a good start. Just to take pity on people seeing the potential for concurrency and wondering how the concurrent accesses actually happen. ;-)

Sounds like you are kidding me :)

--
Vlad Rezki

Quoted 22 lines · Joel Fernandes

Quoted 1 line · Uladzislau Rezki

On Mon, May 04, 2020 at 08:24:37AM -0700, Paul E. McKenney wrote:

[..]

Quoted 19 lines

Quoted 19 lines

Quoted 9 lines

Quoted 3 lines

Presumably the list can also be accessed without holding this lock, because otherwise we shouldn't need llist...

Hm... We increase the number of elements in cache, therefore it is not lockless. From the other hand i used llist_head to maintain the cache because it is single linked list, we do not need "*prev" link. Also we do not need to init the list.

But i can change it to list_head. Please let me know if i need :)

Hmmm... Maybe it is time for a non-atomic singly linked list? In the RCU callback processing, the operations were open-coded, but they have been pushed into include/linux/rcu_segcblist.h and kernel/rcu/rcu_segcblist.*.

Maybe some non-atomic/protected/whatever macros in the llist.h file? Or maybe just open-code the singly linked list? (Probably not the best choice, though.) Add comments stating that the atomic properties of the llist functions aren't neded? Something else?

Quoted 3 lines

Quoted 1 line

In order to keep it simple i can replace llist_head by the list_head?

Just to clarify for me, what is the disadvantage of using llist here?

Are there some llist APIs that are not set up for concurrency? I am not seeing any.

The overhead isn't that much of a concern, given that these are not on the hotpath, but people reading the code and seeing the cmpxchg operations might be forgiven for believing that there is some concurrency involved somewhere.

Or am I confused and there are now single-threaded add/delete operations for llist?

Quoted 8 lines

Since we don't care about traversing backwards, isn't it better to use llist for this usecase?

I think Vlad is using locking as we're also tracking the size of the llist to know when to free pages. This tracking could suffer from the lost-update problem without any locking, 2 lockless llist_add happened simulatenously.

Also if list_head is used, it will take more space and still use locking.

Indeed, it would be best to use a non-concurrent singly linked list.

Quoted 7 lines

Quoted 7 lines

Since we don't care about traversing backwards, isn't it better to use llist for this usecase?

I think Vlad is using locking as we're also tracking the size of the llist to know when to free pages. This tracking could suffer from the lost-update problem without any locking, 2 lockless llist_add happened simulatenously.

Quoted 3 lines

Quoted 1 line

Also if list_head is used, it will take more space and still use locking.

Indeed, it would be best to use a non-concurrent singly linked list.

Ok cool :-)

Is it safe to say something like the following is ruled out? ;-) :-D #define kfree_rcu_list_add llist_add

Quoted 1 line

Quoted 1 line

Quoted 1 line

Since we don't care about traversing backwards, isn't it better

to use llist

Quoted 3 lines

Quoted 3 lines

Quoted 3 lines

for this usecase?

I think Vlad is using locking as we're also tracking the size of

the llist to

Quoted 1 line

Quoted 1 line

Quoted 1 line

know when to free pages. This tracking could suffer from the

lost-update

Quoted 1 line

Quoted 1 line

Quoted 1 line

problem without any locking, 2 lockless llist_add happened

simulatenously.

Quoted 2 lines

Quoted 2 lines

Quoted 2 lines

Also if list_head is used, it will take more space and still use

locking.

Quoted 2 lines

Quoted 2 lines

Indeed, it would be best to use a non-concurrent singly linked

list.

Quoted 3 lines

Ok cool :-)

Quoted 3 lines

Is it safe to say something like the following is ruled out? ;-) :-D #define kfree_rcu_list_add llist_add

In that case i think it is better just to add a comment about using llist_head. To state that it used as a singular list to save space and the access is synchronized by the lock :)

IMHO.

--
Vlad Rezki

Quoted 13 lines

Quoted 10 lines

Quoted 8 lines

Since we don't care about traversing backwards, isn't it better to use llist for this usecase?

I think Vlad is using locking as we're also tracking the size of the llist to know when to free pages. This tracking could suffer from the lost-update problem without any locking, 2 lockless llist_add happened simulatenously.

Also if list_head is used, it will take more space and still use locking.

Indeed, it would be best to use a non-concurrent singly linked list.

Ok cool :-)

Quoted 3 lines

Is it safe to say something like the following is ruled out? ;-) :-D #define kfree_rcu_list_add llist_add

In that case i think it is better just to add a comment about using llist_head. To state that it used as a singular list to save space and the access is synchronized by the lock :)

IMHO.

Quoted 16 lines · Uladzislau Rezki

Quoted 16 lines

Quoted 10 lines

Quoted 8 lines

Since we don't care about traversing backwards, isn't it better to use llist for this usecase?

I think Vlad is using locking as we're also tracking the size of the llist to know when to free pages. This tracking could suffer from the lost-update problem without any locking, 2 lockless llist_add happened simulatenously.

Also if list_head is used, it will take more space and still use locking.

Indeed, it would be best to use a non-concurrent singly linked list.

Ok cool :-)

Is it safe to say something like the following is ruled out? ;-) :-D #define kfree_rcu_list_add llist_add

Quoted 5 lines

In that case i think it is better just to add a comment about using llist_head. To state that it used as a singular list to save space and the access is synchronized by the lock :)

IMHO.

But adding such a comment would be fine as well.

Document the rcutree.rcu_min_cached_objs sysfs kernel parameter.

Signed-off-by Uladzislau Rezki (Sony) <[email protected]>

---
 Documentation/admin-guide/kernel-parameters.txt | 8 ++++++++
 1 file changed, 8 insertions(+)

Documentation/admin-guide/kernel-parameters.txt +8 -0

index 828ff975fbc6..b2b7022374af 100644
@@ -3977,6 +3977,14 @@
 			latencies, which will choose a value aligned
 			with the appropriate hardware boundaries.
 
+	rcutree.rcu_min_cached_objs= [KNL]
+			Minimum number of objects which are cached and
+			maintained per one CPU. Object size is equal
+			to PAGE_SIZE. The cache allows to reduce the
+			pressure to page allocator, also it makes the
+			whole algorithm to behave better in low memory
+			condition.
+
 	rcutree.jiffies_till_first_fqs= [KNL]
 			Set delay from grace-period initialization to
 			first attempt to force quiescent states.

Signature · 1 line

2.20.1

Quoted 3 lines · Uladzislau Rezki (Sony)

Document the rcutree.rcu_min_cached_objs sysfs kernel parameter.

Signed-off-by Uladzislau Rezki (Sony) <[email protected]>

Could you please combine this wtih the patch that created this sysfs parameter?

To do so we use an array of common kvfree_rcu_bulk_data structure. It consists of two elements, index number 0 corresponds to SLAB ptrs., whereas vmalloc pointers can be accessed by using index number 1.

The reason of not mixing pointers is to have an easy way to to distinguish them.

It is also the preparation patch for head-less objects support. When an object is head-less we can not queue it into any list, instead a pointer is placed directly into an array.

Signed-off-by Uladzislau Rezki (Sony) <[email protected]>

Signed-off-by Joel Fernandes (Google) <[email protected]>

Reviewed-by Joel Fernandes (Google) <[email protected]>

Co-developed-by Joel Fernandes (Google) <[email protected]>

---
 kernel/rcu/tree.c | 172 +++++++++++++++++++++++++++++-----------------
 1 file changed, 109 insertions(+), 63 deletions(-)

kernel/rcu/tree.c +28 -19

index d8975819b1c9..7983926af95b 100644
@@ -57,6 +57,7 @@
 #include <linux/slab.h>
 #include <linux/sched/isolation.h>
 #include <linux/sched/clock.h>
+#include <linux/mm.h>
 #include "../time/tick-internal.h"
 
 #include "tree.h"

@@ -2857,44 +2858,44 @@ EXPORT_SYMBOL_GPL(call_rcu);
 #define KFREE_N_BATCHES 2
 
 /**
- * struct kfree_rcu_bulk_data - single block to store kfree_rcu() pointers
+ * struct kvfree_rcu_bulk_data - single block to store kvfree_rcu() pointers
  * @nr_records: Number of active pointers in the array
- * @records: Array of the kfree_rcu() pointers
  * @next: Next bulk object in the block chain
+ * @records: Array of the kvfree_rcu() pointers
  */
-struct kfree_rcu_bulk_data {
+struct kvfree_rcu_bulk_data {
 	unsigned long nr_records;
-	struct kfree_rcu_bulk_data *next;
+	struct kvfree_rcu_bulk_data *next;
 	void *records[];
 };
 
 /*
  * This macro defines how many entries the "records" array
  * will contain. It is based on the fact that the size of
- * kfree_rcu_bulk_data structure becomes exactly one page.
+ * kvfree_rcu_bulk_data structure becomes exactly one page.
  */
-#define KFREE_BULK_MAX_ENTR \
-	((PAGE_SIZE - sizeof(struct kfree_rcu_bulk_data)) / sizeof(void *))
+#define KVFREE_BULK_MAX_ENTR \
+	((PAGE_SIZE - sizeof(struct kvfree_rcu_bulk_data)) / sizeof(void *))
 
 /**
  * struct kfree_rcu_cpu_work - single batch of kfree_rcu() requests
  * @rcu_work: Let queue_rcu_work() invoke workqueue handler after grace period
  * @head_free: List of kfree_rcu() objects waiting for a grace period
- * @bhead_free: Bulk-List of kfree_rcu() objects waiting for a grace period
+ * @bkvhead_free: Bulk-List of kvfree_rcu() objects waiting for a grace period
  * @krcp: Pointer to @kfree_rcu_cpu structure
  */
 
 struct kfree_rcu_cpu_work {
 	struct rcu_work rcu_work;
 	struct rcu_head *head_free;
-	struct kfree_rcu_bulk_data *bhead_free;
+	struct kvfree_rcu_bulk_data *bkvhead_free[2];
 	struct kfree_rcu_cpu *krcp;
 };
 
 /**
  * struct kfree_rcu_cpu - batch up kfree_rcu() requests for RCU grace period
  * @head: List of kfree_rcu() objects not yet waiting for a grace period
- * @bhead: Bulk-List of kfree_rcu() objects not yet waiting for a grace period
+ * @bkvhead: Bulk-List of kvfree_rcu() objects not yet waiting for a grace period
  * @krw_arr: Array of batches of kfree_rcu() objects waiting for a grace period
  * @lock: Synchronize access to this structure
  * @monitor_work: Promote @head to @head_free after KFREE_DRAIN_JIFFIES

@@ -2908,7 +2909,7 @@ struct kfree_rcu_cpu_work {
  */
 struct kfree_rcu_cpu {
 	struct rcu_head *head;
-	struct kfree_rcu_bulk_data *bhead;
+	struct kvfree_rcu_bulk_data *bkvhead[2];
 	struct kfree_rcu_cpu_work krw_arr[KFREE_N_BATCHES];
 	raw_spinlock_t lock;
 	struct delayed_work monitor_work;

@@ -2932,7 +2933,7 @@ static DEFINE_PER_CPU(struct kfree_rcu_cpu, krc) = {
 };
 
 static __always_inline void
-debug_rcu_bhead_unqueue(struct kfree_rcu_bulk_data *bhead)
+debug_rcu_bhead_unqueue(struct kvfree_rcu_bulk_data *bhead)
 {
 #ifdef CONFIG_DEBUG_OBJECTS_RCU_HEAD
 	int i;

@@ -2961,20 +2962,20 @@ krc_this_cpu_unlock(struct kfree_rcu_cpu *krcp, unsigned long flags)
 	local_irq_restore(flags);
 }
 
-static inline struct kfree_rcu_bulk_data *
+static inline struct kvfree_rcu_bulk_data *
 get_cached_bnode(struct kfree_rcu_cpu *krcp)
 {
 	if (!krcp->nr_bkv_objs)
 		return NULL;
 
 	krcp->nr_bkv_objs--;
-	return (struct kfree_rcu_bulk_data *)
+	return (struct kvfree_rcu_bulk_data *)
 		llist_del_first(&krcp->bkvcache);
 }
 
 static inline bool
 put_cached_bnode(struct kfree_rcu_cpu *krcp,
-	struct kfree_rcu_bulk_data *bnode)
+	struct kvfree_rcu_bulk_data *bnode)
 {
 	/* Check the limit. */
 	if (krcp->nr_bkv_objs >= rcu_min_cached_objs)

@@ -2993,41 +2994,73 @@ put_cached_bnode(struct kfree_rcu_cpu *krcp,
 static void kfree_rcu_work(struct work_struct *work)
 {
 	unsigned long flags;
+	struct kvfree_rcu_bulk_data *bkhead, *bvhead, *bnext;
 	struct rcu_head *head, *next;
-	struct kfree_rcu_bulk_data *bhead, *bnext;
 	struct kfree_rcu_cpu *krcp;
 	struct kfree_rcu_cpu_work *krwp;
+	int i;
 
 	krwp = container_of(to_rcu_work(work),
 			    struct kfree_rcu_cpu_work, rcu_work);
 	krcp = krwp->krcp;
+
 	raw_spin_lock_irqsave(&krcp->lock, flags);
+	/* Channel 1. */
+	bkhead = krwp->bkvhead_free[0];
+	krwp->bkvhead_free[0] = NULL;
+
+	/* Channel 2. */
+	bvhead = krwp->bkvhead_free[1];
+	krwp->bkvhead_free[1] = NULL;
+
+	/* Channel 3. */
 	head = krwp->head_free;
 	krwp->head_free = NULL;
-	bhead = krwp->bhead_free;
-	krwp->bhead_free = NULL;
 	raw_spin_unlock_irqrestore(&krcp->lock, flags);
 
-	/* "bhead" is now private, so traverse locklessly. */

kernel/rcu/tree.c +13 -1

 		trace_rcu_invoke_kfree_bulk_callback(rcu_state.name,
-			bhead->nr_records, bhead->records);
+			bkhead->nr_records, bkhead->records);
+
+		kfree_bulk(bkhead->nr_records, bkhead->records);
+		rcu_lock_release(&rcu_callback_map);
+
+		krcp = krc_this_cpu_lock(&flags);
+		if (put_cached_bnode(krcp, bkhead))
+			bkhead = NULL;
+		krc_this_cpu_unlock(krcp, flags);
+
+		if (bkhead)
+			free_page((unsigned long) bkhead);
+

kernel/rcu/tree.c +19 -11

+		for (i = 0; i < bvhead->nr_records; i++) {
+			trace_rcu_invoke_kfree_callback(rcu_state.name,
+				(struct rcu_head *) bvhead->records[i], 0);
+			vfree(bvhead->records[i]);
+		}
 		rcu_lock_release(&rcu_callback_map);
 
 		krcp = krc_this_cpu_lock(&flags);
-		if (put_cached_bnode(krcp, bhead))
-			bhead = NULL;
+		if (put_cached_bnode(krcp, bvhead))
+			bvhead = NULL;
 		krc_this_cpu_unlock(krcp, flags);
 
-		if (bhead)
-			free_page((unsigned long) bhead);
+		if (bvhead)
+			free_page((unsigned long) bvhead);
 
 		cond_resched_tasks_rcu_qs();
 	}

@@ -3047,7 +3080,7 @@ static void kfree_rcu_work(struct work_struct *work)
 		trace_rcu_invoke_kfree_callback(rcu_state.name, head, offset);
 
 		if (!WARN_ON_ONCE(!__is_kfree_rcu_offset(offset)))
-			kfree(ptr);
+			kvfree(ptr);
 
 		rcu_lock_release(&rcu_callback_map);
 		cond_resched_tasks_rcu_qs();

@@ -3072,21 +3105,34 @@ static inline bool queue_kfree_rcu_work(struct kfree_rcu_cpu *krcp)
 		krwp = &(krcp->krw_arr[i]);
 
 		/*
-		 * Try to detach bhead or head and attach it over any
+		 * Try to detach bkvhead or head and attach it over any
 		 * available corresponding free channel. It can be that
 		 * a previous RCU batch is in progress, it means that
 		 * immediately to queue another one is not possible so
 		 * return false to tell caller to retry.
 		 */
-		if ((krcp->bhead && !krwp->bhead_free) ||
+		if ((krcp->bkvhead[0] && !krwp->bkvhead_free[0]) ||
+			(krcp->bkvhead[1] && !krwp->bkvhead_free[1]) ||
 				(krcp->head && !krwp->head_free)) {
-			/* Channel 1. */
-			if (!krwp->bhead_free) {
-				krwp->bhead_free = krcp->bhead;
-				krcp->bhead = NULL;
+			/*
+			 * Channel 1 corresponds to SLAB ptrs.
+			 */
+			if (!krwp->bkvhead_free[0]) {
+				krwp->bkvhead_free[0] = krcp->bkvhead[0];
+				krcp->bkvhead[0] = NULL;

kernel/rcu/tree.c +29 -26

+			/*
+			 * Channel 3 corresponds to emergency path.
+			 */
 			if (!krwp->head_free) {
 				krwp->head_free = krcp->head;
 				krcp->head = NULL;

@@ -3095,16 +3141,17 @@ static inline bool queue_kfree_rcu_work(struct kfree_rcu_cpu *krcp)
 			WRITE_ONCE(krcp->count, 0);
 
 			/*
-			 * One work is per one batch, so there are two "free channels",
-			 * "bhead_free" and "head_free" the batch can handle. It can be
-			 * that the work is in the pending state when two channels have
-			 * been detached following each other, one by one.
+			 * One work is per one batch, so there are three
+			 * "free channels", the batch can handle. It can
+			 * be that the work is in the pending state when
+			 * channels have been detached following by each
+			 * other.
 			 */
 			queue_rcu_work(system_wq, &krwp->rcu_work);
 		}
 
 		/* Repeat if any "free" corresponding channel is still busy. */
-		if (krcp->bhead || krcp->head)
+		if (krcp->bkvhead[0] || krcp->bkvhead[1] || krcp->head)
 			repeat = true;
 	}
 

@@ -3146,23 +3193,22 @@ static void kfree_rcu_monitor(struct work_struct *work)
 }
 
 static inline bool
-kfree_call_rcu_add_ptr_to_bulk(struct kfree_rcu_cpu *krcp,
-	struct rcu_head *head, rcu_callback_t func)
+kvfree_call_rcu_add_ptr_to_bulk(struct kfree_rcu_cpu *krcp, void *ptr)
 {
-	struct kfree_rcu_bulk_data *bnode;
+	struct kvfree_rcu_bulk_data *bnode;
+	int idx;
 
 	if (unlikely(!krcp->initialized))
 		return false;
 
 	lockdep_assert_held(&krcp->lock);
+	idx = !!is_vmalloc_addr(ptr);
 
 	/* Check if a new block is required. */
-	if (!krcp->bhead ||
-			krcp->bhead->nr_records == KFREE_BULK_MAX_ENTR) {
+	if (!krcp->bkvhead[idx] ||
+			krcp->bkvhead[idx]->nr_records == KVFREE_BULK_MAX_ENTR) {
 		bnode = get_cached_bnode(krcp);
 		if (!bnode) {
-			WARN_ON_ONCE(sizeof(struct kfree_rcu_bulk_data) > PAGE_SIZE);
-
 			/*
 			 * To keep this path working on raw non-preemptible
 			 * sections, prevent the optional entry into the

@@ -3175,7 +3221,7 @@ kfree_call_rcu_add_ptr_to_bulk(struct kfree_rcu_cpu *krcp,
 			if (IS_ENABLED(CONFIG_PREEMPT_RT))
 				return false;
 
-			bnode = (struct kfree_rcu_bulk_data *)
+			bnode = (struct kvfree_rcu_bulk_data *)
 				__get_free_page(GFP_NOWAIT | __GFP_NOWARN);
 		}
 

@@ -3185,30 +3231,30 @@ kfree_call_rcu_add_ptr_to_bulk(struct kfree_rcu_cpu *krcp,
 
 		/* Initialize the new block. */
 		bnode->nr_records = 0;
-		bnode->next = krcp->bhead;
+		bnode->next = krcp->bkvhead[idx];
 
 		/* Attach it to the head. */
-		krcp->bhead = bnode;
+		krcp->bkvhead[idx] = bnode;
 	}
 
 	/* Finally insert. */
-	krcp->bhead->records[krcp->bhead->nr_records++] =
-		(void *) head - (unsigned long) func;
+	krcp->bkvhead[idx]->records
+		[krcp->bkvhead[idx]->nr_records++] = ptr;
 
 	return true;
 }
 
 /*
- * Queue a request for lazy invocation of kfree_bulk()/kfree() after a grace
- * period. Please note there are two paths are maintained, one is the main one
- * that uses kfree_bulk() interface and second one is emergency one, that is
- * used only when the main path can not be maintained temporary, due to memory
- * pressure.
+ * Queue a request for lazy invocation of appropriate free routine after a
+ * grace period. Please note there are three paths are maintained, two are the
+ * main ones that use array of pointers interface and third one is emergency
+ * one, that is used only when the main path can not be maintained temporary,
+ * due to memory pressure.
  *
  * Each kfree_call_rcu() request is added to a batch. The batch will be drained
  * every KFREE_DRAIN_JIFFIES number of jiffies. All the objects in the batch will
  * be free'd in workqueue context. This allows us to: batch requests together to
- * reduce the number of grace periods during heavy kfree_rcu() load.
+ * reduce the number of grace periods during heavy kfree_rcu()/kvfree_rcu() load.
  */
 void kfree_call_rcu(struct rcu_head *head, rcu_callback_t func)
 {

@@ -3231,7 +3277,7 @@ void kfree_call_rcu(struct rcu_head *head, rcu_callback_t func)
 	 * Under high memory pressure GFP_NOWAIT can fail,
 	 * in that case the emergency path is maintained.
 	 */
-	if (unlikely(!kfree_call_rcu_add_ptr_to_bulk(krcp, head, func))) {
+	if (unlikely(!kvfree_call_rcu_add_ptr_to_bulk(krcp, ptr))) {
 		head->func = func;
 		head->next = krcp->head;
 		krcp->head = head;

@@ -4212,7 +4258,7 @@ static void __init kfree_rcu_batch_init(void)
 
 	for_each_possible_cpu(cpu) {
 		struct kfree_rcu_cpu *krcp = per_cpu_ptr(&krc, cpu);
-		struct kfree_rcu_bulk_data *bnode;
+		struct kvfree_rcu_bulk_data *bnode;
 
 		for (i = 0; i < KFREE_N_BATCHES; i++) {
 			INIT_RCU_WORK(&krcp->krw_arr[i].rcu_work, kfree_rcu_work);

@@ -4220,7 +4266,7 @@ static void __init kfree_rcu_batch_init(void)
 		}
 
 		for (i = 0; i < rcu_min_cached_objs; i++) {
-			bnode = (struct kfree_rcu_bulk_data *)
+			bnode = (struct kvfree_rcu_bulk_data *)
 				__get_free_page(GFP_NOWAIT | __GFP_NOWARN);
 
 			if (bnode)

Signature · 1 line

2.20.1

Quoted 41 lines

kernel/rcu/tree.c +16 -8

@@ -2993,41 +2994,73 @@ put_cached_bnode(struct kfree_rcu_cpu *krcp,
 static void kfree_rcu_work(struct work_struct *work)
 {
 	unsigned long flags;
+	struct kvfree_rcu_bulk_data *bkhead, *bvhead, *bnext;
 	struct rcu_head *head, *next;
-	struct kfree_rcu_bulk_data *bhead, *bnext;
 	struct kfree_rcu_cpu *krcp;
 	struct kfree_rcu_cpu_work *krwp;
+	int i;
 
 	krwp = container_of(to_rcu_work(work),
 			    struct kfree_rcu_cpu_work, rcu_work);
 	krcp = krwp->krcp;
+
 	raw_spin_lock_irqsave(&krcp->lock, flags);
+	/* Channel 1. */
+	bkhead = krwp->bkvhead_free[0];
+	krwp->bkvhead_free[0] = NULL;
+
+	/* Channel 2. */
+	bvhead = krwp->bkvhead_free[1];
+	krwp->bkvhead_free[1] = NULL;
+
+	/* Channel 3. */
 	head = krwp->head_free;
 	krwp->head_free = NULL;
-	bhead = krwp->bhead_free;
-	krwp->bhead_free = NULL;
 	raw_spin_unlock_irqrestore(&krcp->lock, flags);
 
-	/* "bhead" is now private, so traverse locklessly. */
-	for (; bhead; bhead = bnext) {
-		bnext = bhead->next;
-
-		debug_rcu_bhead_unqueue(bhead);
+	/* kmalloc()/kfree() channel. */
+	for (; bkhead; bkhead = bnext) {
+		bnext = bkhead->next;
+		debug_rcu_bhead_unqueue(bkhead);
 

kernel/rcu/tree.c put_cached_bnode(struct kfree_rcu_cpu *krcp, · line 3026

Quoted 1 line

kernel/rcu/tree.c +0 -0

 		rcu_lock_acquire(&rcu_callback_map);

Given that rcu_lock_acquire() only affects lockdep, I have to ask exactly what concurrency design you are using here...

Quoted 15 lines

kernel/rcu/tree.c +13 -1

 		trace_rcu_invoke_kfree_bulk_callback(rcu_state.name,
-			bhead->nr_records, bhead->records);
+			bkhead->nr_records, bkhead->records);
+
+		kfree_bulk(bkhead->nr_records, bkhead->records);
+		rcu_lock_release(&rcu_callback_map);
+
+		krcp = krc_this_cpu_lock(&flags);
+		if (put_cached_bnode(krcp, bkhead))
+			bkhead = NULL;
+		krc_this_cpu_unlock(krcp, flags);
+
+		if (bkhead)
+			free_page((unsigned long) bkhead);
+

kernel/rcu/tree.c put_cached_bnode(struct kfree_rcu_cpu *krcp, · line 3049

Quoted 10 lines

kernel/rcu/tree.c +8 -1

+		cond_resched_tasks_rcu_qs();
+	}
+
+	/* vmalloc()/vfree() channel. */
+	for (; bvhead; bvhead = bnext) {
+		bnext = bvhead->next;
+		debug_rcu_bhead_unqueue(bvhead);
 
-		kfree_bulk(bhead->nr_records, bhead->records);
+		rcu_lock_acquire(&rcu_callback_map);

And the same here.

Quoted 55 lines

kernel/rcu/tree.c +19 -11

+		for (i = 0; i < bvhead->nr_records; i++) {
+			trace_rcu_invoke_kfree_callback(rcu_state.name,
+				(struct rcu_head *) bvhead->records[i], 0);
+			vfree(bvhead->records[i]);
+		}
 		rcu_lock_release(&rcu_callback_map);
 
 		krcp = krc_this_cpu_lock(&flags);
-		if (put_cached_bnode(krcp, bhead))
-			bhead = NULL;
+		if (put_cached_bnode(krcp, bvhead))
+			bvhead = NULL;
 		krc_this_cpu_unlock(krcp, flags);
 
-		if (bhead)
-			free_page((unsigned long) bhead);
+		if (bvhead)
+			free_page((unsigned long) bvhead);
 
 		cond_resched_tasks_rcu_qs();
 	}

@@ -3047,7 +3080,7 @@ static void kfree_rcu_work(struct work_struct *work)
 		trace_rcu_invoke_kfree_callback(rcu_state.name, head, offset);
 
 		if (!WARN_ON_ONCE(!__is_kfree_rcu_offset(offset)))
-			kfree(ptr);
+			kvfree(ptr);
 
 		rcu_lock_release(&rcu_callback_map);
 		cond_resched_tasks_rcu_qs();

@@ -3072,21 +3105,34 @@ static inline bool queue_kfree_rcu_work(struct kfree_rcu_cpu *krcp)
 		krwp = &(krcp->krw_arr[i]);
 
 		/*
-		 * Try to detach bhead or head and attach it over any
+		 * Try to detach bkvhead or head and attach it over any
 		 * available corresponding free channel. It can be that
 		 * a previous RCU batch is in progress, it means that
 		 * immediately to queue another one is not possible so
 		 * return false to tell caller to retry.
 		 */
-		if ((krcp->bhead && !krwp->bhead_free) ||
+		if ((krcp->bkvhead[0] && !krwp->bkvhead_free[0]) ||
+			(krcp->bkvhead[1] && !krwp->bkvhead_free[1]) ||
 				(krcp->head && !krwp->head_free)) {
-			/* Channel 1. */
-			if (!krwp->bhead_free) {
-				krwp->bhead_free = krcp->bhead;
-				krcp->bhead = NULL;
+			/*
+			 * Channel 1 corresponds to SLAB ptrs.
+			 */
+			if (!krwp->bkvhead_free[0]) {
+				krwp->bkvhead_free[0] = krcp->bkvhead[0];
+				krcp->bkvhead[0] = NULL;

kernel/rcu/tree.c static inline bool queue_kfree_rcu_work(struct kfree_rcu_cpu *krcp) · line 3131

Quoted 10 lines

kernel/rcu/tree.c +7 -1

 			}
 
-			/* Channel 2. */
+			/*
+			 * Channel 2 corresponds to vmalloc ptrs.
+			 */
+			if (!krwp->bkvhead_free[1]) {
+				krwp->bkvhead_free[1] = krcp->bkvhead[1];
+				krcp->bkvhead[1] = NULL;
+			}

Why not a "for" loop here? Duplicate code is most certainly not what we want, as it can cause all sorts of trouble down the road.

Thanx, Paul

Quoted 139 lines

kernel/rcu/tree.c +29 -26

+			/*
+			 * Channel 3 corresponds to emergency path.
+			 */
 			if (!krwp->head_free) {
 				krwp->head_free = krcp->head;
 				krcp->head = NULL;

@@ -3095,16 +3141,17 @@ static inline bool queue_kfree_rcu_work(struct kfree_rcu_cpu *krcp)
 			WRITE_ONCE(krcp->count, 0);
 
 			/*
-			 * One work is per one batch, so there are two "free channels",
-			 * "bhead_free" and "head_free" the batch can handle. It can be
-			 * that the work is in the pending state when two channels have
-			 * been detached following each other, one by one.
+			 * One work is per one batch, so there are three
+			 * "free channels", the batch can handle. It can
+			 * be that the work is in the pending state when
+			 * channels have been detached following by each
+			 * other.
 			 */
 			queue_rcu_work(system_wq, &krwp->rcu_work);
 		}
 
 		/* Repeat if any "free" corresponding channel is still busy. */
-		if (krcp->bhead || krcp->head)
+		if (krcp->bkvhead[0] || krcp->bkvhead[1] || krcp->head)
 			repeat = true;
 	}
 

@@ -3146,23 +3193,22 @@ static void kfree_rcu_monitor(struct work_struct *work)
 }
 
 static inline bool
-kfree_call_rcu_add_ptr_to_bulk(struct kfree_rcu_cpu *krcp,
-	struct rcu_head *head, rcu_callback_t func)
+kvfree_call_rcu_add_ptr_to_bulk(struct kfree_rcu_cpu *krcp, void *ptr)
 {
-	struct kfree_rcu_bulk_data *bnode;
+	struct kvfree_rcu_bulk_data *bnode;
+	int idx;
 
 	if (unlikely(!krcp->initialized))
 		return false;
 
 	lockdep_assert_held(&krcp->lock);
+	idx = !!is_vmalloc_addr(ptr);
 
 	/* Check if a new block is required. */
-	if (!krcp->bhead ||
-			krcp->bhead->nr_records == KFREE_BULK_MAX_ENTR) {
+	if (!krcp->bkvhead[idx] ||
+			krcp->bkvhead[idx]->nr_records == KVFREE_BULK_MAX_ENTR) {
 		bnode = get_cached_bnode(krcp);
 		if (!bnode) {
-			WARN_ON_ONCE(sizeof(struct kfree_rcu_bulk_data) > PAGE_SIZE);
-
 			/*
 			 * To keep this path working on raw non-preemptible
 			 * sections, prevent the optional entry into the

@@ -3175,7 +3221,7 @@ kfree_call_rcu_add_ptr_to_bulk(struct kfree_rcu_cpu *krcp,
 			if (IS_ENABLED(CONFIG_PREEMPT_RT))
 				return false;
 
-			bnode = (struct kfree_rcu_bulk_data *)
+			bnode = (struct kvfree_rcu_bulk_data *)
 				__get_free_page(GFP_NOWAIT | __GFP_NOWARN);
 		}
 

@@ -3185,30 +3231,30 @@ kfree_call_rcu_add_ptr_to_bulk(struct kfree_rcu_cpu *krcp,
 
 		/* Initialize the new block. */
 		bnode->nr_records = 0;
-		bnode->next = krcp->bhead;
+		bnode->next = krcp->bkvhead[idx];
 
 		/* Attach it to the head. */
-		krcp->bhead = bnode;
+		krcp->bkvhead[idx] = bnode;
 	}
 
 	/* Finally insert. */
-	krcp->bhead->records[krcp->bhead->nr_records++] =
-		(void *) head - (unsigned long) func;
+	krcp->bkvhead[idx]->records
+		[krcp->bkvhead[idx]->nr_records++] = ptr;
 
 	return true;
 }
 
 /*
- * Queue a request for lazy invocation of kfree_bulk()/kfree() after a grace
- * period. Please note there are two paths are maintained, one is the main one
- * that uses kfree_bulk() interface and second one is emergency one, that is
- * used only when the main path can not be maintained temporary, due to memory
- * pressure.
+ * Queue a request for lazy invocation of appropriate free routine after a
+ * grace period. Please note there are three paths are maintained, two are the
+ * main ones that use array of pointers interface and third one is emergency
+ * one, that is used only when the main path can not be maintained temporary,
+ * due to memory pressure.
  *
  * Each kfree_call_rcu() request is added to a batch. The batch will be drained
  * every KFREE_DRAIN_JIFFIES number of jiffies. All the objects in the batch will
  * be free'd in workqueue context. This allows us to: batch requests together to
- * reduce the number of grace periods during heavy kfree_rcu() load.
+ * reduce the number of grace periods during heavy kfree_rcu()/kvfree_rcu() load.
  */
 void kfree_call_rcu(struct rcu_head *head, rcu_callback_t func)
 {

@@ -3231,7 +3277,7 @@ void kfree_call_rcu(struct rcu_head *head, rcu_callback_t func)
 	 * Under high memory pressure GFP_NOWAIT can fail,
 	 * in that case the emergency path is maintained.
 	 */
-	if (unlikely(!kfree_call_rcu_add_ptr_to_bulk(krcp, head, func))) {
+	if (unlikely(!kvfree_call_rcu_add_ptr_to_bulk(krcp, ptr))) {
 		head->func = func;
 		head->next = krcp->head;
 		krcp->head = head;

@@ -4212,7 +4258,7 @@ static void __init kfree_rcu_batch_init(void)
 
 	for_each_possible_cpu(cpu) {
 		struct kfree_rcu_cpu *krcp = per_cpu_ptr(&krc, cpu);
-		struct kfree_rcu_bulk_data *bnode;
+		struct kvfree_rcu_bulk_data *bnode;
 
 		for (i = 0; i < KFREE_N_BATCHES; i++) {
 			INIT_RCU_WORK(&krcp->krw_arr[i].rcu_work, kfree_rcu_work);

@@ -4220,7 +4266,7 @@ static void __init kfree_rcu_batch_init(void)
 		}
 
 		for (i = 0; i < rcu_min_cached_objs; i++) {
-			bnode = (struct kfree_rcu_bulk_data *)
+			bnode = (struct kvfree_rcu_bulk_data *)
 				__get_free_page(GFP_NOWAIT | __GFP_NOWARN);
 
 			if (bnode)

Signature · 1 line

2.20.1

On Fri, May 01, 2020 at 02:37:53PM -0700, Paul E. McKenney wrote:
[...]

Quoted 41 lines

Quoted 41 lines

kernel/rcu/tree.c +16 -8

@@ -2993,41 +2994,73 @@ put_cached_bnode(struct kfree_rcu_cpu *krcp,
 static void kfree_rcu_work(struct work_struct *work)
 {
 	unsigned long flags;
+	struct kvfree_rcu_bulk_data *bkhead, *bvhead, *bnext;
 	struct rcu_head *head, *next;
-	struct kfree_rcu_bulk_data *bhead, *bnext;
 	struct kfree_rcu_cpu *krcp;
 	struct kfree_rcu_cpu_work *krwp;
+	int i;
 
 	krwp = container_of(to_rcu_work(work),
 			    struct kfree_rcu_cpu_work, rcu_work);
 	krcp = krwp->krcp;
+
 	raw_spin_lock_irqsave(&krcp->lock, flags);
+	/* Channel 1. */
+	bkhead = krwp->bkvhead_free[0];
+	krwp->bkvhead_free[0] = NULL;
+
+	/* Channel 2. */
+	bvhead = krwp->bkvhead_free[1];
+	krwp->bkvhead_free[1] = NULL;
+
+	/* Channel 3. */
 	head = krwp->head_free;
 	krwp->head_free = NULL;
-	bhead = krwp->bhead_free;
-	krwp->bhead_free = NULL;
 	raw_spin_unlock_irqrestore(&krcp->lock, flags);
 
-	/* "bhead" is now private, so traverse locklessly. */
-	for (; bhead; bhead = bnext) {
-		bnext = bhead->next;
-
-		debug_rcu_bhead_unqueue(bhead);
+	/* kmalloc()/kfree() channel. */
+	for (; bkhead; bkhead = bnext) {
+		bnext = bkhead->next;
+		debug_rcu_bhead_unqueue(bkhead);
 

Quoted 4 lines

kernel/rcu/tree.c put_cached_bnode(struct kfree_rcu_cpu *krcp, · line 3026

Quoted 1 line

kernel/rcu/tree.c +0 -0

 		rcu_lock_acquire(&rcu_callback_map);

Given that rcu_lock_acquire() only affects lockdep, I have to ask exactly what concurrency design you are using here...

I believe the rcu_callback_map usage above follows a similar pattern from old code where the rcu_callback_map is acquired before doing the kfree.

static inline bool __rcu_reclaim(const char *rn, struct rcu_head *head)

{
        rcu_callback_t f;
        unsigned long offset = (unsigned long)head->func;

        rcu_lock_acquire(&rcu_callback_map);
        if (__is_kfree_rcu_offset(offset)) {
                trace_rcu_invoke_kfree_callback(rn, head, offset);
                kfree((void *)head - offset);
                rcu_lock_release(&rcu_callback_map);

So when kfree_rcu() was rewritten, the rcu_lock_acquire() of rcu_callback_map got carried.

I believe it is for detecting recursion where we possibly try to free RCU-held memory while already freeing memory. Or was there anoher purpose of the rcu_callback_map?

thanks,

- Joel

Quoted 238 lines

Quoted 15 lines

kernel/rcu/tree.c +13 -1

 		trace_rcu_invoke_kfree_bulk_callback(rcu_state.name,
-			bhead->nr_records, bhead->records);
+			bkhead->nr_records, bkhead->records);
+
+		kfree_bulk(bkhead->nr_records, bkhead->records);
+		rcu_lock_release(&rcu_callback_map);
+
+		krcp = krc_this_cpu_lock(&flags);
+		if (put_cached_bnode(krcp, bkhead))
+			bkhead = NULL;
+		krc_this_cpu_unlock(krcp, flags);
+
+		if (bkhead)
+			free_page((unsigned long) bkhead);
+

kernel/rcu/tree.c put_cached_bnode(struct kfree_rcu_cpu *krcp, · line 3049

Quoted 10 lines

kernel/rcu/tree.c +8 -1

+		cond_resched_tasks_rcu_qs();
+	}
+
+	/* vmalloc()/vfree() channel. */
+	for (; bvhead; bvhead = bnext) {
+		bnext = bvhead->next;
+		debug_rcu_bhead_unqueue(bvhead);
 
-		kfree_bulk(bhead->nr_records, bhead->records);
+		rcu_lock_acquire(&rcu_callback_map);

And the same here.

Quoted 55 lines

kernel/rcu/tree.c +19 -11

+		for (i = 0; i < bvhead->nr_records; i++) {
+			trace_rcu_invoke_kfree_callback(rcu_state.name,
+				(struct rcu_head *) bvhead->records[i], 0);
+			vfree(bvhead->records[i]);
+		}
 		rcu_lock_release(&rcu_callback_map);
 
 		krcp = krc_this_cpu_lock(&flags);
-		if (put_cached_bnode(krcp, bhead))
-			bhead = NULL;
+		if (put_cached_bnode(krcp, bvhead))
+			bvhead = NULL;
 		krc_this_cpu_unlock(krcp, flags);
 
-		if (bhead)
-			free_page((unsigned long) bhead);
+		if (bvhead)
+			free_page((unsigned long) bvhead);
 
 		cond_resched_tasks_rcu_qs();
 	}

@@ -3047,7 +3080,7 @@ static void kfree_rcu_work(struct work_struct *work)
 		trace_rcu_invoke_kfree_callback(rcu_state.name, head, offset);
 
 		if (!WARN_ON_ONCE(!__is_kfree_rcu_offset(offset)))
-			kfree(ptr);
+			kvfree(ptr);
 
 		rcu_lock_release(&rcu_callback_map);
 		cond_resched_tasks_rcu_qs();

@@ -3072,21 +3105,34 @@ static inline bool queue_kfree_rcu_work(struct kfree_rcu_cpu *krcp)
 		krwp = &(krcp->krw_arr[i]);
 
 		/*
-		 * Try to detach bhead or head and attach it over any
+		 * Try to detach bkvhead or head and attach it over any
 		 * available corresponding free channel. It can be that
 		 * a previous RCU batch is in progress, it means that
 		 * immediately to queue another one is not possible so
 		 * return false to tell caller to retry.
 		 */
-		if ((krcp->bhead && !krwp->bhead_free) ||
+		if ((krcp->bkvhead[0] && !krwp->bkvhead_free[0]) ||
+			(krcp->bkvhead[1] && !krwp->bkvhead_free[1]) ||
 				(krcp->head && !krwp->head_free)) {
-			/* Channel 1. */
-			if (!krwp->bhead_free) {
-				krwp->bhead_free = krcp->bhead;
-				krcp->bhead = NULL;
+			/*
+			 * Channel 1 corresponds to SLAB ptrs.
+			 */
+			if (!krwp->bkvhead_free[0]) {
+				krwp->bkvhead_free[0] = krcp->bkvhead[0];
+				krcp->bkvhead[0] = NULL;

kernel/rcu/tree.c static inline bool queue_kfree_rcu_work(struct kfree_rcu_cpu *krcp) · line 3131

Quoted 10 lines

kernel/rcu/tree.c +7 -1

 			}
 
-			/* Channel 2. */
+			/*
+			 * Channel 2 corresponds to vmalloc ptrs.
+			 */
+			if (!krwp->bkvhead_free[1]) {
+				krwp->bkvhead_free[1] = krcp->bkvhead[1];
+				krcp->bkvhead[1] = NULL;
+			}

Why not a "for" loop here? Duplicate code is most certainly not what we want, as it can cause all sorts of trouble down the road.

Thanx, Paul

Quoted 139 lines

kernel/rcu/tree.c +29 -26

+			/*
+			 * Channel 3 corresponds to emergency path.
+			 */
 			if (!krwp->head_free) {
 				krwp->head_free = krcp->head;
 				krcp->head = NULL;

@@ -3095,16 +3141,17 @@ static inline bool queue_kfree_rcu_work(struct kfree_rcu_cpu *krcp)
 			WRITE_ONCE(krcp->count, 0);
 
 			/*
-			 * One work is per one batch, so there are two "free channels",
-			 * "bhead_free" and "head_free" the batch can handle. It can be
-			 * that the work is in the pending state when two channels have
-			 * been detached following each other, one by one.
+			 * One work is per one batch, so there are three
+			 * "free channels", the batch can handle. It can
+			 * be that the work is in the pending state when
+			 * channels have been detached following by each
+			 * other.
 			 */
 			queue_rcu_work(system_wq, &krwp->rcu_work);
 		}
 
 		/* Repeat if any "free" corresponding channel is still busy. */
-		if (krcp->bhead || krcp->head)
+		if (krcp->bkvhead[0] || krcp->bkvhead[1] || krcp->head)
 			repeat = true;
 	}
 

@@ -3146,23 +3193,22 @@ static void kfree_rcu_monitor(struct work_struct *work)
 }
 
 static inline bool
-kfree_call_rcu_add_ptr_to_bulk(struct kfree_rcu_cpu *krcp,
-	struct rcu_head *head, rcu_callback_t func)
+kvfree_call_rcu_add_ptr_to_bulk(struct kfree_rcu_cpu *krcp, void *ptr)
 {
-	struct kfree_rcu_bulk_data *bnode;
+	struct kvfree_rcu_bulk_data *bnode;
+	int idx;
 
 	if (unlikely(!krcp->initialized))
 		return false;
 
 	lockdep_assert_held(&krcp->lock);
+	idx = !!is_vmalloc_addr(ptr);
 
 	/* Check if a new block is required. */
-	if (!krcp->bhead ||
-			krcp->bhead->nr_records == KFREE_BULK_MAX_ENTR) {
+	if (!krcp->bkvhead[idx] ||
+			krcp->bkvhead[idx]->nr_records == KVFREE_BULK_MAX_ENTR) {
 		bnode = get_cached_bnode(krcp);
 		if (!bnode) {
-			WARN_ON_ONCE(sizeof(struct kfree_rcu_bulk_data) > PAGE_SIZE);
-
 			/*
 			 * To keep this path working on raw non-preemptible
 			 * sections, prevent the optional entry into the

@@ -3175,7 +3221,7 @@ kfree_call_rcu_add_ptr_to_bulk(struct kfree_rcu_cpu *krcp,
 			if (IS_ENABLED(CONFIG_PREEMPT_RT))
 				return false;
 
-			bnode = (struct kfree_rcu_bulk_data *)
+			bnode = (struct kvfree_rcu_bulk_data *)
 				__get_free_page(GFP_NOWAIT | __GFP_NOWARN);
 		}
 

@@ -3185,30 +3231,30 @@ kfree_call_rcu_add_ptr_to_bulk(struct kfree_rcu_cpu *krcp,
 
 		/* Initialize the new block. */
 		bnode->nr_records = 0;
-		bnode->next = krcp->bhead;
+		bnode->next = krcp->bkvhead[idx];
 
 		/* Attach it to the head. */
-		krcp->bhead = bnode;
+		krcp->bkvhead[idx] = bnode;
 	}
 
 	/* Finally insert. */
-	krcp->bhead->records[krcp->bhead->nr_records++] =
-		(void *) head - (unsigned long) func;
+	krcp->bkvhead[idx]->records
+		[krcp->bkvhead[idx]->nr_records++] = ptr;
 
 	return true;
 }
 
 /*
- * Queue a request for lazy invocation of kfree_bulk()/kfree() after a grace
- * period. Please note there are two paths are maintained, one is the main one
- * that uses kfree_bulk() interface and second one is emergency one, that is
- * used only when the main path can not be maintained temporary, due to memory
- * pressure.
+ * Queue a request for lazy invocation of appropriate free routine after a
+ * grace period. Please note there are three paths are maintained, two are the
+ * main ones that use array of pointers interface and third one is emergency
+ * one, that is used only when the main path can not be maintained temporary,
+ * due to memory pressure.
  *
  * Each kfree_call_rcu() request is added to a batch. The batch will be drained
  * every KFREE_DRAIN_JIFFIES number of jiffies. All the objects in the batch will
  * be free'd in workqueue context. This allows us to: batch requests together to
- * reduce the number of grace periods during heavy kfree_rcu() load.
+ * reduce the number of grace periods during heavy kfree_rcu()/kvfree_rcu() load.
  */
 void kfree_call_rcu(struct rcu_head *head, rcu_callback_t func)
 {

@@ -3231,7 +3277,7 @@ void kfree_call_rcu(struct rcu_head *head, rcu_callback_t func)
 	 * Under high memory pressure GFP_NOWAIT can fail,
 	 * in that case the emergency path is maintained.
 	 */
-	if (unlikely(!kfree_call_rcu_add_ptr_to_bulk(krcp, head, func))) {
+	if (unlikely(!kvfree_call_rcu_add_ptr_to_bulk(krcp, ptr))) {
 		head->func = func;
 		head->next = krcp->head;
 		krcp->head = head;

@@ -4212,7 +4258,7 @@ static void __init kfree_rcu_batch_init(void)
 
 	for_each_possible_cpu(cpu) {
 		struct kfree_rcu_cpu *krcp = per_cpu_ptr(&krc, cpu);
-		struct kfree_rcu_bulk_data *bnode;
+		struct kvfree_rcu_bulk_data *bnode;
 
 		for (i = 0; i < KFREE_N_BATCHES; i++) {
 			INIT_RCU_WORK(&krcp->krw_arr[i].rcu_work, kfree_rcu_work);

@@ -4220,7 +4266,7 @@ static void __init kfree_rcu_batch_init(void)
 		}
 
 		for (i = 0; i < rcu_min_cached_objs; i++) {
-			bnode = (struct kfree_rcu_bulk_data *)
+			bnode = (struct kvfree_rcu_bulk_data *)
 				__get_free_page(GFP_NOWAIT | __GFP_NOWARN);
 
 			if (bnode)

Signature · 1 line

2.20.1