[PATCH v2 0/3] xen/rcu: rework the RCU logic

[PATCH v2 0/3] xen/rcu: rework the RCU logic

[Roger Pau Monne]

Hello,

Following series aims to solve two problems we have observed with the
RCU subsystem, complete details on patch 2.

The series is basically a re-write (and IMO simplification) of the RCU
logic, patch 2 containing most of the newly introduced logic.

It's been (slightly) tested locally and on the safety CI.  Maybe I'm
being naive, but this looks much easier to reason about and maintain
than the current logic.

This "simplification" is only possible after the introduction of
rcu_read_{lock,unlock}() helpers that identify RCU critical sections.

Thanks, Roger.

Roger Pau Monne (3):
  xen/rcu: introduce the concept of RCU epoch
  xen/rcu: simplify RCU implementation
  xen/rcu: remove rcu_quiesce_{disable,enable}()

 xen/common/rcupdate.c      | 497 +++++++++++--------------------------
 xen/common/softirq.c       |   4 +-
 xen/include/xen/rcupdate.h |  45 ++--
 3 files changed, 168 insertions(+), 378 deletions(-)

-- 
2.55.0

[PATCH v2 1/3] xen/rcu: introduce the concept of RCU epoch

[Roger Pau Monne]

An RCU epoch signals the lifetime of RCU references.  Each CPU records the
epoch in force when it enters an RCU critical section.  When a callback is
queued the current epoch is stored in the callback and the global epoch is
then bumped, as a way to know when all CPUs have moved past a specific
epoch.  At that point there can be no remaining references to objects
fetched during the callback's epoch.

The compiler barrier is switched to a full memory barrier, as future uses
of rcu_lock_cnt must ensure the count is increased before taking a
reference to any RCU protected object.

Access rcu_lock_cnt, rcu_lock_epoch and rcu_epoch through ACCESS_ONCE()
to stop the compiler shattering the loads and stores.  The reordering
prevention aspect of ACCESS_ONCE() is not what we rely on here (the memory
barriers cover that); what matters is that these variables now have remote
consumers, so each access must be a single, non-torn memory operation.

Signed-off-by: Roger Pau Monné <roger@xenproject.org>
---
Can possibly be folded into the next patch, as it's lacking context on its
own to understand the need to introduce the logic.
---
Changes since v1:
 - Do the setting of head->added in the interrupt disabled section.
 - Unconditionally do a memory barrier when enter/exit RCU critical
   regions.  Attempting to do selectively is too complex.
---
 xen/common/rcupdate.c      |  6 ++++++
 xen/include/xen/rcupdate.h | 17 +++++++++++++----
 2 files changed, 19 insertions(+), 4 deletions(-)

diff --git a/xen/common/rcupdate.c b/xen/common/rcupdate.c
index c1b6b2ae768b..babaabbbeefe 100644
--- a/xen/common/rcupdate.c
+++ b/xen/common/rcupdate.c
@@ -49,6 +49,11 @@
 #include <asm/atomic.h>
 
 DEFINE_PER_CPU(unsigned int, rcu_lock_cnt);
+/* Store epoch when CPU entered the RCU critical section. */
+DEFINE_PER_CPU(unsigned int, rcu_lock_epoch);
+
+/* Current RCU epoch, bumped every time a new callback is queued. */
+unsigned int rcu_epoch;
 
 /* Global control variables for rcupdate callback mechanism. */
 static struct rcu_ctrlblk {
@@ -283,6 +288,7 @@ void call_rcu(struct rcu_head *head,
     head->func = func;
     head->next = NULL;
     local_irq_save(flags);
+    head->added = arch_fetch_and_add(&rcu_epoch, 1);
     rdp = &this_cpu(rcu_data);
     *rdp->nxttail = head;
     rdp->nxttail = &head->next;
diff --git a/xen/include/xen/rcupdate.h b/xen/include/xen/rcupdate.h
index c57f628107cf..5846e8c169d7 100644
--- a/xen/include/xen/rcupdate.h
+++ b/xen/include/xen/rcupdate.h
@@ -34,24 +34,32 @@
 #include <xen/compiler.h>
 #include <xen/spinlock.h>
 #include <xen/cpumask.h>
+#include <xen/lib.h>
 #include <xen/percpu.h>
 #include <xen/preempt.h>
 
 #define __rcu
 
 DECLARE_PER_CPU(unsigned int, rcu_lock_cnt);
+DECLARE_PER_CPU(unsigned int, rcu_lock_epoch);
+
+extern unsigned int rcu_epoch;
 
 static inline void rcu_quiesce_disable(void)
 {
+    unsigned int cpu = smp_processor_id();
+
     preempt_disable();
-    this_cpu(rcu_lock_cnt)++;
-    barrier();
+    if ( !ACCESS_ONCE(per_cpu(rcu_lock_cnt, cpu))++ )
+        ACCESS_ONCE(per_cpu(rcu_lock_epoch, cpu)) = ACCESS_ONCE(rcu_epoch);
+    smp_mb();
 }
 
 static inline void rcu_quiesce_enable(void)
 {
-    barrier();
-    this_cpu(rcu_lock_cnt)--;
+
+    smp_mb();
+    ACCESS_ONCE(this_cpu(rcu_lock_cnt))--;
     preempt_enable();
 }
 
@@ -68,6 +76,7 @@ static inline bool rcu_quiesce_allowed(void)
 struct rcu_head {
     struct rcu_head *next;
     void (*func)(struct rcu_head *head);
+    unsigned int added;
 };
 
 #define RCU_HEAD_INIT   { .next = NULL, .func = NULL }
-- 
2.55.0

Re: [PATCH v2 1/3] xen/rcu: introduce the concept of RCU epoch

[Alejandro Vallejo]

On Fri Sep 25, 2026 at 6:30 PM CEST, Roger Pau Monne wrote:
> An RCU epoch signals the lifetime of RCU references.  Each CPU records the
> epoch in force when it enters an RCU critical section.  When a callback is
> queued the current epoch is stored in the callback and the global epoch is
> then bumped, as a way to know when all CPUs have moved past a specific
> epoch.  At that point there can be no remaining references to objects
> fetched during the callback's epoch.
>
> The compiler barrier is switched to a full memory barrier, as future uses
> of rcu_lock_cnt must ensure the count is increased before taking a
> reference to any RCU protected object.
>
> Access rcu_lock_cnt, rcu_lock_epoch and rcu_epoch through ACCESS_ONCE()
> to stop the compiler shattering the loads and stores.  The reordering
> prevention aspect of ACCESS_ONCE() is not what we rely on here (the memory
> barriers cover that); what matters is that these variables now have remote
> consumers, so each access must be a single, non-torn memory operation.
>
> Signed-off-by: Roger Pau Monné <roger@xenproject.org>
> ---
> Can possibly be folded into the next patch, as it's lacking context on its
> own to understand the need to introduce the logic.
> ---
> Changes since v1:
>  - Do the setting of head->added in the interrupt disabled section.
>  - Unconditionally do a memory barrier when enter/exit RCU critical
>    regions.  Attempting to do selectively is too complex.
> ---
>  xen/common/rcupdate.c      |  6 ++++++
>  xen/include/xen/rcupdate.h | 17 +++++++++++++----
>  2 files changed, 19 insertions(+), 4 deletions(-)
>
> diff --git a/xen/common/rcupdate.c b/xen/common/rcupdate.c
> index c1b6b2ae768b..babaabbbeefe 100644
> --- a/xen/common/rcupdate.c
> +++ b/xen/common/rcupdate.c
> @@ -49,6 +49,11 @@
>  #include <asm/atomic.h>
>  
>  DEFINE_PER_CPU(unsigned int, rcu_lock_cnt);
> +/* Store epoch when CPU entered the RCU critical section. */
> +DEFINE_PER_CPU(unsigned int, rcu_lock_epoch);
> +
> +/* Current RCU epoch, bumped every time a new callback is queued. */
> +unsigned int rcu_epoch;

nit: I'd suggest calling this rcu_cur_epoch, and then renaming "added"
below to plain "epoch". "added" doesn't convey what the field is at all.

>  
>  /* Global control variables for rcupdate callback mechanism. */
>  static struct rcu_ctrlblk {
> @@ -283,6 +288,7 @@ void call_rcu(struct rcu_head *head,
>      head->func = func;
>      head->next = NULL;
>      local_irq_save(flags);
> +    head->added = arch_fetch_and_add(&rcu_epoch, 1);
>      rdp = &this_cpu(rcu_data);
>      *rdp->nxttail = head;
>      rdp->nxttail = &head->next;
> diff --git a/xen/include/xen/rcupdate.h b/xen/include/xen/rcupdate.h
> index c57f628107cf..5846e8c169d7 100644
> --- a/xen/include/xen/rcupdate.h
> +++ b/xen/include/xen/rcupdate.h
> @@ -34,24 +34,32 @@
>  #include <xen/compiler.h>
>  #include <xen/spinlock.h>
>  #include <xen/cpumask.h>
> +#include <xen/lib.h>
>  #include <xen/percpu.h>
>  #include <xen/preempt.h>
>  
>  #define __rcu
>  
>  DECLARE_PER_CPU(unsigned int, rcu_lock_cnt);
> +DECLARE_PER_CPU(unsigned int, rcu_lock_epoch);
> +
> +extern unsigned int rcu_epoch;
>  
>  static inline void rcu_quiesce_disable(void)
>  {
> +    unsigned int cpu = smp_processor_id();
> +
>      preempt_disable();
> -    this_cpu(rcu_lock_cnt)++;
> -    barrier();
> +    if ( !ACCESS_ONCE(per_cpu(rcu_lock_cnt, cpu))++ )
> +        ACCESS_ONCE(per_cpu(rcu_lock_epoch, cpu)) = ACCESS_ONCE(rcu_epoch);

I'm not terribly convinced about using ACCESS_ONCE() for atomic
accesses. In particular...

> +    smp_mb();
>  }
>  
>  static inline void rcu_quiesce_enable(void)
>  {
> -    barrier();
> -    this_cpu(rcu_lock_cnt)--;
> +
> +    smp_mb();
> +    ACCESS_ONCE(this_cpu(rcu_lock_cnt))--;

... this variable is to be read by remote CPUs and autodecrement seems
like the wrong tool for an atomic decrement.

>      preempt_enable();
>  }
>  
> @@ -68,6 +76,7 @@ static inline bool rcu_quiesce_allowed(void)
>  struct rcu_head {
>      struct rcu_head *next;
>      void (*func)(struct rcu_head *head);
> +    unsigned int added;
>  };
>  
>  #define RCU_HEAD_INIT   { .next = NULL, .func = NULL }

Cheers,
Alejandro

Re: [PATCH v2 1/3] xen/rcu: introduce the concept of RCU epoch

[Roger Pau Monné]

On Mon, Sep 28, 2026 at 10:24:04AM +0200, Alejandro Vallejo wrote:
> On Fri Sep 25, 2026 at 6:30 PM CEST, Roger Pau Monne wrote:
> > An RCU epoch signals the lifetime of RCU references.  Each CPU records the
> > epoch in force when it enters an RCU critical section.  When a callback is
> > queued the current epoch is stored in the callback and the global epoch is
> > then bumped, as a way to know when all CPUs have moved past a specific
> > epoch.  At that point there can be no remaining references to objects
> > fetched during the callback's epoch.
> >
> > The compiler barrier is switched to a full memory barrier, as future uses
> > of rcu_lock_cnt must ensure the count is increased before taking a
> > reference to any RCU protected object.
> >
> > Access rcu_lock_cnt, rcu_lock_epoch and rcu_epoch through ACCESS_ONCE()
> > to stop the compiler shattering the loads and stores.  The reordering
> > prevention aspect of ACCESS_ONCE() is not what we rely on here (the memory
> > barriers cover that); what matters is that these variables now have remote
> > consumers, so each access must be a single, non-torn memory operation.
> >
> > Signed-off-by: Roger Pau Monné <roger@xenproject.org>
> > ---
> > Can possibly be folded into the next patch, as it's lacking context on its
> > own to understand the need to introduce the logic.
> > ---
> > Changes since v1:
> >  - Do the setting of head->added in the interrupt disabled section.
> >  - Unconditionally do a memory barrier when enter/exit RCU critical
> >    regions.  Attempting to do selectively is too complex.
> > ---
> >  xen/common/rcupdate.c      |  6 ++++++
> >  xen/include/xen/rcupdate.h | 17 +++++++++++++----
> >  2 files changed, 19 insertions(+), 4 deletions(-)
> >
> > diff --git a/xen/common/rcupdate.c b/xen/common/rcupdate.c
> > index c1b6b2ae768b..babaabbbeefe 100644
> > --- a/xen/common/rcupdate.c
> > +++ b/xen/common/rcupdate.c
> > @@ -49,6 +49,11 @@
> >  #include <asm/atomic.h>
> >  
> >  DEFINE_PER_CPU(unsigned int, rcu_lock_cnt);
> > +/* Store epoch when CPU entered the RCU critical section. */
> > +DEFINE_PER_CPU(unsigned int, rcu_lock_epoch);
> > +
> > +/* Current RCU epoch, bumped every time a new callback is queued. */
> > +unsigned int rcu_epoch;
> 
> nit: I'd suggest calling this rcu_cur_epoch, and then renaming "added"
> below to plain "epoch". "added" doesn't convey what the field is at all.

added_epoch? (or epoch_added?)

I would be fine with epoch also.  I think it's obvious that the
global variable holds the current epoch without it needing the "cur"
addition.

> >  
> >  /* Global control variables for rcupdate callback mechanism. */
> >  static struct rcu_ctrlblk {
> > @@ -283,6 +288,7 @@ void call_rcu(struct rcu_head *head,
> >      head->func = func;
> >      head->next = NULL;
> >      local_irq_save(flags);
> > +    head->added = arch_fetch_and_add(&rcu_epoch, 1);
> >      rdp = &this_cpu(rcu_data);
> >      *rdp->nxttail = head;
> >      rdp->nxttail = &head->next;
> > diff --git a/xen/include/xen/rcupdate.h b/xen/include/xen/rcupdate.h
> > index c57f628107cf..5846e8c169d7 100644
> > --- a/xen/include/xen/rcupdate.h
> > +++ b/xen/include/xen/rcupdate.h
> > @@ -34,24 +34,32 @@
> >  #include <xen/compiler.h>
> >  #include <xen/spinlock.h>
> >  #include <xen/cpumask.h>
> > +#include <xen/lib.h>
> >  #include <xen/percpu.h>
> >  #include <xen/preempt.h>
> >  
> >  #define __rcu
> >  
> >  DECLARE_PER_CPU(unsigned int, rcu_lock_cnt);
> > +DECLARE_PER_CPU(unsigned int, rcu_lock_epoch);
> > +
> > +extern unsigned int rcu_epoch;
> >  
> >  static inline void rcu_quiesce_disable(void)
> >  {
> > +    unsigned int cpu = smp_processor_id();
> > +
> >      preempt_disable();
> > -    this_cpu(rcu_lock_cnt)++;
> > -    barrier();
> > +    if ( !ACCESS_ONCE(per_cpu(rcu_lock_cnt, cpu))++ )
> > +        ACCESS_ONCE(per_cpu(rcu_lock_epoch, cpu)) = ACCESS_ONCE(rcu_epoch);
> 
> I'm not terribly convinced about using ACCESS_ONCE() for atomic
> accesses. In particular...
> 
> > +    smp_mb();
> >  }
> >  
> >  static inline void rcu_quiesce_enable(void)
> >  {
> > -    barrier();
> > -    this_cpu(rcu_lock_cnt)--;
> > +
> > +    smp_mb();
> > +    ACCESS_ONCE(this_cpu(rcu_lock_cnt))--;
> 
> ... this variable is to be read by remote CPUs and autodecrement seems
> like the wrong tool for an atomic decrement.

I can use {read,write}_atomic() instead, but there's been a tendency
to phase out the usage of those functions.  On x86 this should be a
plain `addl` or `subl` instruction (the non-locked version of
atomic_{inc,dec}()).

Thanks, Roger.

Re: [PATCH v2 1/3] xen/rcu: introduce the concept of RCU epoch

[Jan Beulich]

On 29.09.2026 16:20, Roger Pau Monné wrote:
> On Mon, Sep 28, 2026 at 10:24:04AM +0200, Alejandro Vallejo wrote:
>> On Fri Sep 25, 2026 at 6:30 PM CEST, Roger Pau Monne wrote:
>>>  static inline void rcu_quiesce_disable(void)
>>>  {
>>> +    unsigned int cpu = smp_processor_id();
>>> +
>>>      preempt_disable();
>>> -    this_cpu(rcu_lock_cnt)++;
>>> -    barrier();
>>> +    if ( !ACCESS_ONCE(per_cpu(rcu_lock_cnt, cpu))++ )
>>> +        ACCESS_ONCE(per_cpu(rcu_lock_epoch, cpu)) = ACCESS_ONCE(rcu_epoch);
>>
>> I'm not terribly convinced about using ACCESS_ONCE() for atomic
>> accesses. In particular...
>>
>>> +    smp_mb();
>>>  }
>>>  
>>>  static inline void rcu_quiesce_enable(void)
>>>  {
>>> -    barrier();
>>> -    this_cpu(rcu_lock_cnt)--;
>>> +
>>> +    smp_mb();
>>> +    ACCESS_ONCE(this_cpu(rcu_lock_cnt))--;
>>
>> ... this variable is to be read by remote CPUs and autodecrement seems
>> like the wrong tool for an atomic decrement.
> 
> I can use {read,write}_atomic() instead, but there's been a tendency
> to phase out the usage of those functions.  On x86 this should be a
> plain `addl` or `subl` instruction (the non-locked version of
> atomic_{inc,dec}()).

If remote CPUs are to access the variable, is a non-locked access okay
in the first place?

Furthermore, iirc the compiler won't ever emit a plain ADD or SUB (or
INC/DEC) when accessing volatile memory (which ACCESS_ONCE() enforces).
It'll favor simple insns (MOV plus arithmetic) in that case. Of course
use of {read,write}_atomic() would also "guarantee" a 3-insn sequence
(MOV-SUB-MOV or MOV-DEC-MOV).

Jan

Re: [PATCH v2 1/3] xen/rcu: introduce the concept of RCU epoch

[Alejandro Vallejo]

On Tue Sep 29, 2026 at 4:52 PM CEST, Jan Beulich wrote:
> On 29.09.2026 16:20, Roger Pau Monné wrote:
> > On Mon, Sep 28, 2026 at 10:24:04AM +0200, Alejandro Vallejo wrote:
> >> On Fri Sep 25, 2026 at 6:30 PM CEST, Roger Pau Monne wrote:
> >>>  static inline void rcu_quiesce_disable(void)
> >>>  {
> >>> +    unsigned int cpu = smp_processor_id();
> >>> +
> >>>      preempt_disable();
> >>> -    this_cpu(rcu_lock_cnt)++;
> >>> -    barrier();
> >>> +    if ( !ACCESS_ONCE(per_cpu(rcu_lock_cnt, cpu))++ )
> >>> +        ACCESS_ONCE(per_cpu(rcu_lock_epoch, cpu)) = ACCESS_ONCE(rcu_epoch);
> >>
> >> I'm not terribly convinced about using ACCESS_ONCE() for atomic
> >> accesses. In particular...
> >>
> >>> +    smp_mb();
> >>>  }
> >>>  
> >>>  static inline void rcu_quiesce_enable(void)
> >>>  {
> >>> -    barrier();
> >>> -    this_cpu(rcu_lock_cnt)--;
> >>> +
> >>> +    smp_mb();
> >>> +    ACCESS_ONCE(this_cpu(rcu_lock_cnt))--;
> >>
> >> ... this variable is to be read by remote CPUs and autodecrement seems
> >> like the wrong tool for an atomic decrement.
> > 
> > I can use {read,write}_atomic() instead, but there's been a tendency
> > to phase out the usage of those functions.  On x86 this should be a
> > plain `addl` or `subl` instruction (the non-locked version of
> > atomic_{inc,dec}()).

I don't think the compiler can merge volatile reads and writes. It can
shatter them (thus my fear), but they are meant to be distinct because
each might have different side effects. Jan already mentioned this in
other words.

You don't need an atomic decrement though, just an atomic write. Note
IRQs restore it on exit and other CPUs don't write to it.

  unsigned int *lock_cnt = &this_cpu(rcu_lock_cnt);
  write_atomic(lock_cnt, *lock_cnt - 1);

and this yields a plain mov, through a static inline stub (plus the
mov from the read access + dec)

The problem with ACCESS_ONCE() is that the write isn't guaranteed to be
atomic.

>
> If remote CPUs are to access the variable, is a non-locked access okay
> in the first place?

You use locked access when you need sequential consistency or an atomic
RMW, and that's not it here. There's a single writer (currrent CPU) and
IRQs don't play a part because the restore the previous value before
returning. So it's race-free. All we need to ensure is that remote CPUs
never see a partial write.

Another take would be to subsume the prior smp_mb() into the decrement.
Something like:

  arch_fetch_add(&this_cpu(rcu_lock_cnt), RCU_LOCK_CNT_MAX);

This would make the decrement itself atomic (through a locked insn), but
ensure sequential consistency as well, thus removing the need for the
prior barrier. It's trickery through twos complement to behave like a unit
decrement.

My point isn't so much the "how" (should be obvious by now there's many
ways), but to not use ACCESS_ONCE() where an atomic read or write was
meant. Not necessarily a locked instruction. Just atomic.

Cheers,
Alejandro

Re: [PATCH v2 1/3] xen/rcu: introduce the concept of RCU epoch

[Jan Beulich]

On 30.09.2026 00:14, Alejandro Vallejo wrote:
> On Tue Sep 29, 2026 at 4:52 PM CEST, Jan Beulich wrote:
>> On 29.09.2026 16:20, Roger Pau Monné wrote:
>>> On Mon, Sep 28, 2026 at 10:24:04AM +0200, Alejandro Vallejo wrote:
>>>> On Fri Sep 25, 2026 at 6:30 PM CEST, Roger Pau Monne wrote:
>>>>>  static inline void rcu_quiesce_disable(void)
>>>>>  {
>>>>> +    unsigned int cpu = smp_processor_id();
>>>>> +
>>>>>      preempt_disable();
>>>>> -    this_cpu(rcu_lock_cnt)++;
>>>>> -    barrier();
>>>>> +    if ( !ACCESS_ONCE(per_cpu(rcu_lock_cnt, cpu))++ )
>>>>> +        ACCESS_ONCE(per_cpu(rcu_lock_epoch, cpu)) = ACCESS_ONCE(rcu_epoch);
>>>>
>>>> I'm not terribly convinced about using ACCESS_ONCE() for atomic
>>>> accesses. In particular...
>>>>
>>>>> +    smp_mb();
>>>>>  }
>>>>>  
>>>>>  static inline void rcu_quiesce_enable(void)
>>>>>  {
>>>>> -    barrier();
>>>>> -    this_cpu(rcu_lock_cnt)--;
>>>>> +
>>>>> +    smp_mb();
>>>>> +    ACCESS_ONCE(this_cpu(rcu_lock_cnt))--;
>>>>
>>>> ... this variable is to be read by remote CPUs and autodecrement seems
>>>> like the wrong tool for an atomic decrement.
>>>
>>> I can use {read,write}_atomic() instead, but there's been a tendency
>>> to phase out the usage of those functions.  On x86 this should be a
>>> plain `addl` or `subl` instruction (the non-locked version of
>>> atomic_{inc,dec}()).
> 
> I don't think the compiler can merge volatile reads and writes. It can
> shatter them (thus my fear), but they are meant to be distinct because
> each might have different side effects. Jan already mentioned this in
> other words.
> 
> You don't need an atomic decrement though, just an atomic write. Note
> IRQs restore it on exit and other CPUs don't write to it.
> 
>   unsigned int *lock_cnt = &this_cpu(rcu_lock_cnt);
>   write_atomic(lock_cnt, *lock_cnt - 1);
> 
> and this yields a plain mov, through a static inline stub (plus the
> mov from the read access + dec)
> 
> The problem with ACCESS_ONCE() is that the write isn't guaranteed to be
> atomic.

It may not be properly written down anywhere, but at least the compilers
we use are known to not tear aligned accesses up to a machine word in
width.

Jan

Re: [PATCH v2 1/3] xen/rcu: introduce the concept of RCU epoch

[Roger Pau Monné]

On Wed, Sep 30, 2026 at 07:58:55AM +0200, Jan Beulich wrote:
> On 30.09.2026 00:14, Alejandro Vallejo wrote:
> > On Tue Sep 29, 2026 at 4:52 PM CEST, Jan Beulich wrote:
> >> On 29.09.2026 16:20, Roger Pau Monné wrote:
> >>> On Mon, Sep 28, 2026 at 10:24:04AM +0200, Alejandro Vallejo wrote:
> >>>> On Fri Sep 25, 2026 at 6:30 PM CEST, Roger Pau Monne wrote:
> >>>>>  static inline void rcu_quiesce_disable(void)
> >>>>>  {
> >>>>> +    unsigned int cpu = smp_processor_id();
> >>>>> +
> >>>>>      preempt_disable();
> >>>>> -    this_cpu(rcu_lock_cnt)++;
> >>>>> -    barrier();
> >>>>> +    if ( !ACCESS_ONCE(per_cpu(rcu_lock_cnt, cpu))++ )
> >>>>> +        ACCESS_ONCE(per_cpu(rcu_lock_epoch, cpu)) = ACCESS_ONCE(rcu_epoch);
> >>>>
> >>>> I'm not terribly convinced about using ACCESS_ONCE() for atomic
> >>>> accesses. In particular...
> >>>>
> >>>>> +    smp_mb();
> >>>>>  }
> >>>>>  
> >>>>>  static inline void rcu_quiesce_enable(void)
> >>>>>  {
> >>>>> -    barrier();
> >>>>> -    this_cpu(rcu_lock_cnt)--;
> >>>>> +
> >>>>> +    smp_mb();
> >>>>> +    ACCESS_ONCE(this_cpu(rcu_lock_cnt))--;
> >>>>
> >>>> ... this variable is to be read by remote CPUs and autodecrement seems
> >>>> like the wrong tool for an atomic decrement.
> >>>
> >>> I can use {read,write}_atomic() instead, but there's been a tendency
> >>> to phase out the usage of those functions.  On x86 this should be a
> >>> plain `addl` or `subl` instruction (the non-locked version of
> >>> atomic_{inc,dec}()).
> > 
> > I don't think the compiler can merge volatile reads and writes. It can
> > shatter them (thus my fear), but they are meant to be distinct because
> > each might have different side effects. Jan already mentioned this in
> > other words.
> > 
> > You don't need an atomic decrement though, just an atomic write. Note
> > IRQs restore it on exit and other CPUs don't write to it.
> > 
> >   unsigned int *lock_cnt = &this_cpu(rcu_lock_cnt);
> >   write_atomic(lock_cnt, *lock_cnt - 1);
> > 
> > and this yields a plain mov, through a static inline stub (plus the
> > mov from the read access + dec)
> > 
> > The problem with ACCESS_ONCE() is that the write isn't guaranteed to be
> > atomic.
> 
> It may not be properly written down anywhere, but at least the compilers
> we use are known to not tear aligned accesses up to a machine word in
> width.

Then I possibly don't need the ACCESS_ONCE() at all, as I was using it
to guarantee unshattered accesses.  However this should be written
somewhere, maybe in docs/process/coding-best-practices.pandoc?

Thanks, Roger.

Re: [PATCH v2 1/3] xen/rcu: introduce the concept of RCU epoch

[Jan Beulich]

On 30.09.2026 10:09, Roger Pau Monné wrote:
> On Wed, Sep 30, 2026 at 07:58:55AM +0200, Jan Beulich wrote:
>> On 30.09.2026 00:14, Alejandro Vallejo wrote:
>>> On Tue Sep 29, 2026 at 4:52 PM CEST, Jan Beulich wrote:
>>>> On 29.09.2026 16:20, Roger Pau Monné wrote:
>>>>> On Mon, Sep 28, 2026 at 10:24:04AM +0200, Alejandro Vallejo wrote:
>>>>>> On Fri Sep 25, 2026 at 6:30 PM CEST, Roger Pau Monne wrote:
>>>>>>>  static inline void rcu_quiesce_disable(void)
>>>>>>>  {
>>>>>>> +    unsigned int cpu = smp_processor_id();
>>>>>>> +
>>>>>>>      preempt_disable();
>>>>>>> -    this_cpu(rcu_lock_cnt)++;
>>>>>>> -    barrier();
>>>>>>> +    if ( !ACCESS_ONCE(per_cpu(rcu_lock_cnt, cpu))++ )
>>>>>>> +        ACCESS_ONCE(per_cpu(rcu_lock_epoch, cpu)) = ACCESS_ONCE(rcu_epoch);
>>>>>>
>>>>>> I'm not terribly convinced about using ACCESS_ONCE() for atomic
>>>>>> accesses. In particular...
>>>>>>
>>>>>>> +    smp_mb();
>>>>>>>  }
>>>>>>>  
>>>>>>>  static inline void rcu_quiesce_enable(void)
>>>>>>>  {
>>>>>>> -    barrier();
>>>>>>> -    this_cpu(rcu_lock_cnt)--;
>>>>>>> +
>>>>>>> +    smp_mb();
>>>>>>> +    ACCESS_ONCE(this_cpu(rcu_lock_cnt))--;
>>>>>>
>>>>>> ... this variable is to be read by remote CPUs and autodecrement seems
>>>>>> like the wrong tool for an atomic decrement.
>>>>>
>>>>> I can use {read,write}_atomic() instead, but there's been a tendency
>>>>> to phase out the usage of those functions.  On x86 this should be a
>>>>> plain `addl` or `subl` instruction (the non-locked version of
>>>>> atomic_{inc,dec}()).
>>>
>>> I don't think the compiler can merge volatile reads and writes. It can
>>> shatter them (thus my fear), but they are meant to be distinct because
>>> each might have different side effects. Jan already mentioned this in
>>> other words.
>>>
>>> You don't need an atomic decrement though, just an atomic write. Note
>>> IRQs restore it on exit and other CPUs don't write to it.
>>>
>>>   unsigned int *lock_cnt = &this_cpu(rcu_lock_cnt);
>>>   write_atomic(lock_cnt, *lock_cnt - 1);
>>>
>>> and this yields a plain mov, through a static inline stub (plus the
>>> mov from the read access + dec)
>>>
>>> The problem with ACCESS_ONCE() is that the write isn't guaranteed to be
>>> atomic.
>>
>> It may not be properly written down anywhere, but at least the compilers
>> we use are known to not tear aligned accesses up to a machine word in
>> width.
> 
> Then I possibly don't need the ACCESS_ONCE() at all, as I was using it
> to guarantee unshattered accesses.  However this should be written
> somewhere, maybe in docs/process/coding-best-practices.pandoc?

To write it down in any of our docs, we'd first need a proper reference
to somewhere. "Known to" isn't quite sufficient imo to actually nail
things down. There's still the possibility of compiler bugs, and there's
also the possibility that volatile accesses have better guarantees than
non-volatile ones. (Recall that in prior discussions it was actually said
that we may need to make far more use of ACCESS_ONCE(), which wouldn't be
warranted if collectively we were certain aligned accesses cannot be
torn.)

Jan

Re: [PATCH v2 1/3] xen/rcu: introduce the concept of RCU epoch

[Roger Pau Monné]

On Wed, Sep 30, 2026 at 10:34:02AM +0200, Jan Beulich wrote:
> On 30.09.2026 10:09, Roger Pau Monné wrote:
> > On Wed, Sep 30, 2026 at 07:58:55AM +0200, Jan Beulich wrote:
> >> On 30.09.2026 00:14, Alejandro Vallejo wrote:
> >>> On Tue Sep 29, 2026 at 4:52 PM CEST, Jan Beulich wrote:
> >>>> On 29.09.2026 16:20, Roger Pau Monné wrote:
> >>>>> On Mon, Sep 28, 2026 at 10:24:04AM +0200, Alejandro Vallejo wrote:
> >>>>>> On Fri Sep 25, 2026 at 6:30 PM CEST, Roger Pau Monne wrote:
> >>>>>>>  static inline void rcu_quiesce_disable(void)
> >>>>>>>  {
> >>>>>>> +    unsigned int cpu = smp_processor_id();
> >>>>>>> +
> >>>>>>>      preempt_disable();
> >>>>>>> -    this_cpu(rcu_lock_cnt)++;
> >>>>>>> -    barrier();
> >>>>>>> +    if ( !ACCESS_ONCE(per_cpu(rcu_lock_cnt, cpu))++ )
> >>>>>>> +        ACCESS_ONCE(per_cpu(rcu_lock_epoch, cpu)) = ACCESS_ONCE(rcu_epoch);
> >>>>>>
> >>>>>> I'm not terribly convinced about using ACCESS_ONCE() for atomic
> >>>>>> accesses. In particular...
> >>>>>>
> >>>>>>> +    smp_mb();
> >>>>>>>  }
> >>>>>>>  
> >>>>>>>  static inline void rcu_quiesce_enable(void)
> >>>>>>>  {
> >>>>>>> -    barrier();
> >>>>>>> -    this_cpu(rcu_lock_cnt)--;
> >>>>>>> +

Just noticed there's an unwanted newline being added here.

> >>>>>>> +    smp_mb();
> >>>>>>> +    ACCESS_ONCE(this_cpu(rcu_lock_cnt))--;
> >>>>>>
> >>>>>> ... this variable is to be read by remote CPUs and autodecrement seems
> >>>>>> like the wrong tool for an atomic decrement.
> >>>>>
> >>>>> I can use {read,write}_atomic() instead, but there's been a tendency
> >>>>> to phase out the usage of those functions.  On x86 this should be a
> >>>>> plain `addl` or `subl` instruction (the non-locked version of
> >>>>> atomic_{inc,dec}()).
> >>>
> >>> I don't think the compiler can merge volatile reads and writes. It can
> >>> shatter them (thus my fear), but they are meant to be distinct because
> >>> each might have different side effects. Jan already mentioned this in
> >>> other words.
> >>>
> >>> You don't need an atomic decrement though, just an atomic write. Note
> >>> IRQs restore it on exit and other CPUs don't write to it.
> >>>
> >>>   unsigned int *lock_cnt = &this_cpu(rcu_lock_cnt);
> >>>   write_atomic(lock_cnt, *lock_cnt - 1);
> >>>
> >>> and this yields a plain mov, through a static inline stub (plus the
> >>> mov from the read access + dec)
> >>>
> >>> The problem with ACCESS_ONCE() is that the write isn't guaranteed to be
> >>> atomic.
> >>
> >> It may not be properly written down anywhere, but at least the compilers
> >> we use are known to not tear aligned accesses up to a machine word in
> >> width.
> > 
> > Then I possibly don't need the ACCESS_ONCE() at all, as I was using it
> > to guarantee unshattered accesses.  However this should be written
> > somewhere, maybe in docs/process/coding-best-practices.pandoc?
> 
> To write it down in any of our docs, we'd first need a proper reference
> to somewhere. "Known to" isn't quite sufficient imo to actually nail
> things down.

Fair enough, I also assume compilers would never like to write
this down anywhere in the documentation.

> There's still the possibility of compiler bugs,

Compiler bugs could apply to anything that a compiler generates, and
hence is not likely relevant to mention for each compiler feature or
behavior we rely upon.

> and there's
> also the possibility that volatile accesses have better guarantees than
> non-volatile ones. (Recall that in prior discussions it was actually said
> that we may need to make far more use of ACCESS_ONCE(), which wouldn't be
> warranted if collectively we were certain aligned accesses cannot be
> torn.)

OK, so are we in agreement that ACCESS_ONCE(this_cpu(rcu_lock_cnt))--
ensures accesses to the variable are always non-torn?  Splitting
the access as a RMW is fine, as long as the accesses are not torn.

Thanks, Roger.

Re: [PATCH v2 1/3] xen/rcu: introduce the concept of RCU epoch

[Jan Beulich]

On 30.09.2026 12:07, Roger Pau Monné wrote:
> On Wed, Sep 30, 2026 at 10:34:02AM +0200, Jan Beulich wrote:
>> On 30.09.2026 10:09, Roger Pau Monné wrote:
>>> On Wed, Sep 30, 2026 at 07:58:55AM +0200, Jan Beulich wrote:
>>>> On 30.09.2026 00:14, Alejandro Vallejo wrote:
>>>>> On Tue Sep 29, 2026 at 4:52 PM CEST, Jan Beulich wrote:
>>>>>> On 29.09.2026 16:20, Roger Pau Monné wrote:
>>>>>>> On Mon, Sep 28, 2026 at 10:24:04AM +0200, Alejandro Vallejo wrote:
>>>>>>>> On Fri Sep 25, 2026 at 6:30 PM CEST, Roger Pau Monne wrote:
>>>>>>>>> +    ACCESS_ONCE(this_cpu(rcu_lock_cnt))--;
>>>>>>>>
>>>>>>>> ... this variable is to be read by remote CPUs and autodecrement seems
>>>>>>>> like the wrong tool for an atomic decrement.
>>>>>>>
>>>>>>> I can use {read,write}_atomic() instead, but there's been a tendency
>>>>>>> to phase out the usage of those functions.  On x86 this should be a
>>>>>>> plain `addl` or `subl` instruction (the non-locked version of
>>>>>>> atomic_{inc,dec}()).
>>>>>
>>>>> I don't think the compiler can merge volatile reads and writes. It can
>>>>> shatter them (thus my fear), but they are meant to be distinct because
>>>>> each might have different side effects. Jan already mentioned this in
>>>>> other words.
>>>>>
>>>>> You don't need an atomic decrement though, just an atomic write. Note
>>>>> IRQs restore it on exit and other CPUs don't write to it.
>>>>>
>>>>>   unsigned int *lock_cnt = &this_cpu(rcu_lock_cnt);
>>>>>   write_atomic(lock_cnt, *lock_cnt - 1);
>>>>>
>>>>> and this yields a plain mov, through a static inline stub (plus the
>>>>> mov from the read access + dec)
>>>>>
>>>>> The problem with ACCESS_ONCE() is that the write isn't guaranteed to be
>>>>> atomic.
>>>>
>>>> It may not be properly written down anywhere, but at least the compilers
>>>> we use are known to not tear aligned accesses up to a machine word in
>>>> width.
>>>
>>> Then I possibly don't need the ACCESS_ONCE() at all, as I was using it
>>> to guarantee unshattered accesses.  However this should be written
>>> somewhere, maybe in docs/process/coding-best-practices.pandoc?
>>
>> To write it down in any of our docs, we'd first need a proper reference
>> to somewhere. "Known to" isn't quite sufficient imo to actually nail
>> things down.
> 
> Fair enough, I also assume compilers would never like to write
> this down anywhere in the documentation.
> 
>> There's still the possibility of compiler bugs,
> 
> Compiler bugs could apply to anything that a compiler generates, and
> hence is not likely relevant to mention for each compiler feature or
> behavior we rely upon.
> 
>> and there's
>> also the possibility that volatile accesses have better guarantees than
>> non-volatile ones. (Recall that in prior discussions it was actually said
>> that we may need to make far more use of ACCESS_ONCE(), which wouldn't be
>> warranted if collectively we were certain aligned accesses cannot be
>> torn.)
> 
> OK, so are we in agreement that ACCESS_ONCE(this_cpu(rcu_lock_cnt))--
> ensures accesses to the variable are always non-torn?

Yes, to the best of our knowledge.

>  Splitting
> the access as a RMW is fine, as long as the accesses are not torn.

Oh, okay. Then the earlier reference to atomic_{inc,dec}() was somewhat
misleading.

Jan

Re: [PATCH v2 1/3] xen/rcu: introduce the concept of RCU epoch

[Alejandro Vallejo]

On Wed Sep 30, 2026 at 12:07 PM CEST, Roger Pau Monné wrote:
> On Wed, Sep 30, 2026 at 10:34:02AM +0200, Jan Beulich wrote:
> > On 30.09.2026 10:09, Roger Pau Monné wrote:
> > > On Wed, Sep 30, 2026 at 07:58:55AM +0200, Jan Beulich wrote:
> > >> On 30.09.2026 00:14, Alejandro Vallejo wrote:
> > >>> On Tue Sep 29, 2026 at 4:52 PM CEST, Jan Beulich wrote:
> > >>>> On 29.09.2026 16:20, Roger Pau Monné wrote:
> > >>>>> On Mon, Sep 28, 2026 at 10:24:04AM +0200, Alejandro Vallejo wrote:
> > >>>>>> On Fri Sep 25, 2026 at 6:30 PM CEST, Roger Pau Monne wrote:
> > >>>>>>>  static inline void rcu_quiesce_disable(void)
> > >>>>>>>  {
> > >>>>>>> +    unsigned int cpu = smp_processor_id();
> > >>>>>>> +
> > >>>>>>>      preempt_disable();
> > >>>>>>> -    this_cpu(rcu_lock_cnt)++;
> > >>>>>>> -    barrier();
> > >>>>>>> +    if ( !ACCESS_ONCE(per_cpu(rcu_lock_cnt, cpu))++ )
> > >>>>>>> +        ACCESS_ONCE(per_cpu(rcu_lock_epoch, cpu)) = ACCESS_ONCE(rcu_epoch);
> > >>>>>>
> > >>>>>> I'm not terribly convinced about using ACCESS_ONCE() for atomic
> > >>>>>> accesses. In particular...
> > >>>>>>
> > >>>>>>> +    smp_mb();
> > >>>>>>>  }
> > >>>>>>>  
> > >>>>>>>  static inline void rcu_quiesce_enable(void)
> > >>>>>>>  {
> > >>>>>>> -    barrier();
> > >>>>>>> -    this_cpu(rcu_lock_cnt)--;
> > >>>>>>> +
>
> Just noticed there's an unwanted newline being added here.
>
> > >>>>>>> +    smp_mb();
> > >>>>>>> +    ACCESS_ONCE(this_cpu(rcu_lock_cnt))--;
> > >>>>>>
> > >>>>>> ... this variable is to be read by remote CPUs and autodecrement seems
> > >>>>>> like the wrong tool for an atomic decrement.
> > >>>>>
> > >>>>> I can use {read,write}_atomic() instead, but there's been a tendency
> > >>>>> to phase out the usage of those functions.  On x86 this should be a
> > >>>>> plain `addl` or `subl` instruction (the non-locked version of
> > >>>>> atomic_{inc,dec}()).
> > >>>
> > >>> I don't think the compiler can merge volatile reads and writes. It can
> > >>> shatter them (thus my fear), but they are meant to be distinct because
> > >>> each might have different side effects. Jan already mentioned this in
> > >>> other words.
> > >>>
> > >>> You don't need an atomic decrement though, just an atomic write. Note
> > >>> IRQs restore it on exit and other CPUs don't write to it.
> > >>>
> > >>>   unsigned int *lock_cnt = &this_cpu(rcu_lock_cnt);
> > >>>   write_atomic(lock_cnt, *lock_cnt - 1);
> > >>>
> > >>> and this yields a plain mov, through a static inline stub (plus the
> > >>> mov from the read access + dec)
> > >>>
> > >>> The problem with ACCESS_ONCE() is that the write isn't guaranteed to be
> > >>> atomic.
> > >>
> > >> It may not be properly written down anywhere, but at least the compilers
> > >> we use are known to not tear aligned accesses up to a machine word in
> > >> width.
> > > 
> > > Then I possibly don't need the ACCESS_ONCE() at all, as I was using it
> > > to guarantee unshattered accesses.  However this should be written
> > > somewhere, maybe in docs/process/coding-best-practices.pandoc?
> > 
> > To write it down in any of our docs, we'd first need a proper reference
> > to somewhere. "Known to" isn't quite sufficient imo to actually nail
> > things down.
>
> Fair enough, I also assume compilers would never like to write
> this down anywhere in the documentation.
>
> > There's still the possibility of compiler bugs,
>
> Compiler bugs could apply to anything that a compiler generates, and
> hence is not likely relevant to mention for each compiler feature or
> behavior we rely upon.
>
> > and there's
> > also the possibility that volatile accesses have better guarantees than
> > non-volatile ones. (Recall that in prior discussions it was actually said
> > that we may need to make far more use of ACCESS_ONCE(), which wouldn't be
> > warranted if collectively we were certain aligned accesses cannot be
> > torn.)
>
> OK, so are we in agreement that ACCESS_ONCE(this_cpu(rcu_lock_cnt))--
> ensures accesses to the variable are always non-torn?  Splitting
> the access as a RMW is fine, as long as the accesses are not torn.

... so long as the variable is no wider than the machine word size. It
(may) shatter should rcu_lock_cnt be 64bits and this be run in arm32.
Which, granted, is not the case today so long as unsigned int is 32bits,
but we're using unsigned int because there's an assumption that width
doesn't matter for it.

I still think ACCESS_ONCE() is the wrong tool when atomic operations are
needed, but I will push no further on it for now. I do think this will
compile fine on current targets.

Cheers,
Alejandro

>
> Thanks, Roger.

Re: [PATCH v2 1/3] xen/rcu: introduce the concept of RCU epoch

[Roger Pau Monné]

On Wed, Sep 30, 2026 at 12:14:15AM +0200, Alejandro Vallejo wrote:
> On Tue Sep 29, 2026 at 4:52 PM CEST, Jan Beulich wrote:
> > On 29.09.2026 16:20, Roger Pau Monné wrote:
> > > On Mon, Sep 28, 2026 at 10:24:04AM +0200, Alejandro Vallejo wrote:
> > >> On Fri Sep 25, 2026 at 6:30 PM CEST, Roger Pau Monne wrote:
> > >>>  static inline void rcu_quiesce_disable(void)
> > >>>  {
> > >>> +    unsigned int cpu = smp_processor_id();
> > >>> +
> > >>>      preempt_disable();
> > >>> -    this_cpu(rcu_lock_cnt)++;
> > >>> -    barrier();
> > >>> +    if ( !ACCESS_ONCE(per_cpu(rcu_lock_cnt, cpu))++ )
> > >>> +        ACCESS_ONCE(per_cpu(rcu_lock_epoch, cpu)) = ACCESS_ONCE(rcu_epoch);
> > >>
> > >> I'm not terribly convinced about using ACCESS_ONCE() for atomic
> > >> accesses. In particular...
> > >>
> > >>> +    smp_mb();
> > >>>  }
> > >>>  
> > >>>  static inline void rcu_quiesce_enable(void)
> > >>>  {
> > >>> -    barrier();
> > >>> -    this_cpu(rcu_lock_cnt)--;
> > >>> +
> > >>> +    smp_mb();
> > >>> +    ACCESS_ONCE(this_cpu(rcu_lock_cnt))--;
> > >>
> > >> ... this variable is to be read by remote CPUs and autodecrement seems
> > >> like the wrong tool for an atomic decrement.
> > > 
> > > I can use {read,write}_atomic() instead, but there's been a tendency
> > > to phase out the usage of those functions.  On x86 this should be a
> > > plain `addl` or `subl` instruction (the non-locked version of
> > > atomic_{inc,dec}()).
> 
> I don't think the compiler can merge volatile reads and writes. It can
> shatter them (thus my fear), but they are meant to be distinct because
> each might have different side effects. Jan already mentioned this in
> other words.
> 
> You don't need an atomic decrement though, just an atomic write. Note
> IRQs restore it on exit and other CPUs don't write to it.
> 
>   unsigned int *lock_cnt = &this_cpu(rcu_lock_cnt);
>   write_atomic(lock_cnt, *lock_cnt - 1);
> 
> and this yields a plain mov, through a static inline stub (plus the
> mov from the read access + dec)
> 
> The problem with ACCESS_ONCE() is that the write isn't guaranteed to be
> atomic.

It has always been blurry to me using ACCESS_ONCE() to signal
unshattered accesses, but we seem to rely on this property in other
places already (which doesn't mean is correct).  See
ffa_negotiate_version() for example.

This is all kind of moot if we guarantee that all supported compilers
will never tear accesses up to machine word sizes.

> >
> > If remote CPUs are to access the variable, is a non-locked access okay
> > in the first place?
> 
> You use locked access when you need sequential consistency or an atomic
> RMW, and that's not it here. There's a single writer (currrent CPU) and
> IRQs don't play a part because the restore the previous value before
> returning. So it's race-free. All we need to ensure is that remote CPUs
> never see a partial write.
> 
> Another take would be to subsume the prior smp_mb() into the decrement.
> Something like:
> 
>   arch_fetch_add(&this_cpu(rcu_lock_cnt), RCU_LOCK_CNT_MAX);

Kind of a nit (as I've already attempted this approach), but
arch_fetch_and_add() is not usable as-is here.  We we need to consume
the returned value, otherwise the compiler complains.

./arch/x86/include/asm/system.h:189:6: error: value computed is not used [-Werror=unused-value]

Could do:

unsigned int prev = arch_fetch_and_add(&this_cpu(rcu_lock_cnt), -1);
ASSERT(prev > 0);

> This would make the decrement itself atomic (through a locked insn), but
> ensure sequential consistency as well, thus removing the need for the
> prior barrier. It's trickery through twos complement to behave like a unit
> decrement.
> 
> My point isn't so much the "how" (should be obvious by now there's many
> ways), but to not use ACCESS_ONCE() where an atomic read or write was
> meant. Not necessarily a locked instruction. Just atomic.

I don't have a strong opinion of using smp_mb() vs
arch_fetch_and_add().  I think smp_mb() expresses the intention
better, and avoids explicitly using a locked instruction, which is not
required here.

Thanks, Roger.

Re: [PATCH v2 1/3] xen/rcu: introduce the concept of RCU epoch

[Alejandro Vallejo]

On Wed Sep 30, 2026 at 10:32 AM CEST, Roger Pau Monné wrote:
> On Wed, Sep 30, 2026 at 12:14:15AM +0200, Alejandro Vallejo wrote:
> > On Tue Sep 29, 2026 at 4:52 PM CEST, Jan Beulich wrote:
> > > On 29.09.2026 16:20, Roger Pau Monné wrote:
> > > > On Mon, Sep 28, 2026 at 10:24:04AM +0200, Alejandro Vallejo wrote:
> > > >> On Fri Sep 25, 2026 at 6:30 PM CEST, Roger Pau Monne wrote:
> > > >>>  static inline void rcu_quiesce_disable(void)
> > > >>>  {
> > > >>> +    unsigned int cpu = smp_processor_id();
> > > >>> +
> > > >>>      preempt_disable();
> > > >>> -    this_cpu(rcu_lock_cnt)++;
> > > >>> -    barrier();
> > > >>> +    if ( !ACCESS_ONCE(per_cpu(rcu_lock_cnt, cpu))++ )
> > > >>> +        ACCESS_ONCE(per_cpu(rcu_lock_epoch, cpu)) = ACCESS_ONCE(rcu_epoch);
> > > >>
> > > >> I'm not terribly convinced about using ACCESS_ONCE() for atomic
> > > >> accesses. In particular...
> > > >>
> > > >>> +    smp_mb();
> > > >>>  }
> > > >>>  
> > > >>>  static inline void rcu_quiesce_enable(void)
> > > >>>  {
> > > >>> -    barrier();
> > > >>> -    this_cpu(rcu_lock_cnt)--;
> > > >>> +
> > > >>> +    smp_mb();
> > > >>> +    ACCESS_ONCE(this_cpu(rcu_lock_cnt))--;
> > > >>
> > > >> ... this variable is to be read by remote CPUs and autodecrement seems
> > > >> like the wrong tool for an atomic decrement.
> > > > 
> > > > I can use {read,write}_atomic() instead, but there's been a tendency
> > > > to phase out the usage of those functions.  On x86 this should be a
> > > > plain `addl` or `subl` instruction (the non-locked version of
> > > > atomic_{inc,dec}()).
> > 
> > I don't think the compiler can merge volatile reads and writes. It can
> > shatter them (thus my fear), but they are meant to be distinct because
> > each might have different side effects. Jan already mentioned this in
> > other words.
> > 
> > You don't need an atomic decrement though, just an atomic write. Note
> > IRQs restore it on exit and other CPUs don't write to it.
> > 
> >   unsigned int *lock_cnt = &this_cpu(rcu_lock_cnt);
> >   write_atomic(lock_cnt, *lock_cnt - 1);
> > 
> > and this yields a plain mov, through a static inline stub (plus the
> > mov from the read access + dec)
> > 
> > The problem with ACCESS_ONCE() is that the write isn't guaranteed to be
> > atomic.
>
> It has always been blurry to me using ACCESS_ONCE() to signal
> unshattered accesses, but we seem to rely on this property in other
> places already (which doesn't mean is correct).  See
> ffa_negotiate_version() for example.
>
> This is all kind of moot if we guarantee that all supported compilers
> will never tear accesses up to machine word sizes.
>
> > >
> > > If remote CPUs are to access the variable, is a non-locked access okay
> > > in the first place?
> > 
> > You use locked access when you need sequential consistency or an atomic
> > RMW, and that's not it here. There's a single writer (currrent CPU) and
> > IRQs don't play a part because the restore the previous value before
> > returning. So it's race-free. All we need to ensure is that remote CPUs
> > never see a partial write.
> > 
> > Another take would be to subsume the prior smp_mb() into the decrement.
> > Something like:
> > 
> >   arch_fetch_add(&this_cpu(rcu_lock_cnt), RCU_LOCK_CNT_MAX);
>
> Kind of a nit (as I've already attempted this approach), but
> arch_fetch_and_add() is not usable as-is here.  We we need to consume
> the returned value, otherwise the compiler complains.
>
> ./arch/x86/include/asm/system.h:189:6: error: value computed is not used [-Werror=unused-value]

(void)arch_fetch_and_add() would shut up the warning, but...

>
> Could do:
>
> unsigned int prev = arch_fetch_and_add(&this_cpu(rcu_lock_cnt), -1);
> ASSERT(prev > 0);
>
> > This would make the decrement itself atomic (through a locked insn), but
> > ensure sequential consistency as well, thus removing the need for the
> > prior barrier. It's trickery through twos complement to behave like a unit
> > decrement.
> > 
> > My point isn't so much the "how" (should be obvious by now there's many
> > ways), but to not use ACCESS_ONCE() where an atomic read or write was
> > meant. Not necessarily a locked instruction. Just atomic.
>
> I don't have a strong opinion of using smp_mb() vs
> arch_fetch_and_add().  I think smp_mb() expresses the intention
> better, and avoids explicitly using a locked instruction, which is not
> required here.

... I agree. My point was rather on ACCESS_ONCE() avoidance. Let that be
for now, it's clearly meat for another barbequeue.

Cheers,
Alejandro

>
> Thanks, Roger.

[PATCH v2 2/3] xen/rcu: simplify RCU implementation

[Roger Pau Monne]

The current implementation has two shortcomings for certain Xen usages:

 * When using the null scheduler it's possible for a CPU to never enter Xen
   context.  A CPU not entering Xen context can block other CPUs from
   executing RCU callbacks, as there will be no quiescent state observed if
   the CPU doesn't enter Xen context.

 * If a certain amount of callbacks are pending, RCU will try to force a
   quiescent state, by sending an IPI to remote CPUs.  This causes unwanted
   interference.

Keep track of the RCU epoch when a callback was added, and only execute it
once all CPUs are either outside of RCU critical regions, or any CPUs
inside of RCU critical regions have entered such past the epoch when the
callback was queued.  Knowing whether a CPU is inside a RCU critical region
is done based on the CPU rcu_lock_cnt value.

This removes the grace/quiescent period state tracking entirely: the
control block, the per-CPU quiescent bookkeeping, the batch lists and the
forced-quiescent-state IPI all go away.  As a result no RCU-driven IPIs are
sent to remote CPUs, and callback progress no longer depends on every CPU
passing through Xen context.

Signed-off-by: Roger Pau Monné <roger@xenproject.org>
---
The maximum batch of callbacks processed is limited to 10, this is bit
arbitrary, but I think matches what the current logic attempts does.

There's a bit more pruning to do regarding the usage of grace and
quiesce in comments or functions names - I've done that as a followup
change to reduce the size of this change.

I think the manual raising of an RCU softirq is not really needed when
the queue exceeds a certain length - the softirq loop will already
attempt RCU callback processing if possible.
---
Changes since v1:
 - Rework rcu_pending() so it reports whether there are callback ready to
   be pruned, instead of simply reporting whether there are any callbacks
   pending.  This matches the previous implementation.
 - Force a barrier around entry/exit to each critical region, instead of
   doing so only in the first and last entry/exit respectively.  This
   makes the logic more obviously correct.
---
 xen/common/rcupdate.c      | 493 +++++++++++--------------------------
 xen/include/xen/rcupdate.h |  14 +-
 2 files changed, 146 insertions(+), 361 deletions(-)

diff --git a/xen/common/rcupdate.c b/xen/common/rcupdate.c
index babaabbbeefe..79d25c958f0a 100644
--- a/xen/common/rcupdate.c
+++ b/xen/common/rcupdate.c
@@ -35,6 +35,7 @@
 #include <xen/cpu.h>
 #include <xen/init.h>
 #include <xen/kernel.h>
+#include <xen/list_sort.h>
 #include <xen/param.h>
 #include <xen/percpu.h>
 #include <xen/rcupdate.h>
@@ -55,74 +56,31 @@ DEFINE_PER_CPU(unsigned int, rcu_lock_epoch);
 /* Current RCU epoch, bumped every time a new callback is queued. */
 unsigned int rcu_epoch;
 
-/* Global control variables for rcupdate callback mechanism. */
-static struct rcu_ctrlblk {
-    long cur;           /* Current batch number.                      */
-    long completed;     /* Number of the last completed batch         */
-    int  next_pending;  /* Is the next batch already waiting?         */
-
-    spinlock_t  lock __cacheline_aligned;
-    cpumask_t   cpumask; /* CPUs that need to switch in order ... */
-    cpumask_t   idle_cpumask; /* ... unless they are already idle */
-    /* for current batch to proceed.        */
-} __cacheline_aligned rcu_ctrlblk = {
-    .cur = -300,
-    .completed = -300,
-    .lock = SPIN_LOCK_UNLOCKED,
-};
-
-/*
- * Per-CPU data for Read-Copy Update.
- * nxtlist - new callbacks are added here
- * curlist - current batch for which quiescent cycle started if any
- */
+/* Per-CPU data for Read-Copy Update. */
 struct rcu_data {
-    /* 1) quiescent state handling : */
-    long quiescbatch;    /* Batch # for grace period */
-    int  qs_pending;     /* core waits for quiesc state */
-
-    /* 2) batch handling */
-    long            batch;            /* Batch # for current RCU batch */
-    struct rcu_head *nxtlist;
-    struct rcu_head **nxttail;
-    long            qlen;             /* # of queued callbacks */
-    struct rcu_head *curlist;
-    struct rcu_head **curtail;
-    struct rcu_head *donelist;
-    struct rcu_head **donetail;
-    long            blimit;           /* Upper limit on a processed batch */
-    int cpu;
-    long            last_rs_qlen;     /* qlen during the last resched */
-
-    /* 3) idle CPUs handling */
+    /*
+     * List of pending callbacks, sorted by ascending epoch.  Use a threshold
+     * value to raise an RCU softirq if the queue exceeds a given length.
+     */
+    struct list_head pending;
+    unsigned int nr;
+#define RCU_QUEUE_THRESHOLD 100
+
+    /* Keep track of CPUs blocking first RCU callback execution. */
+    cpumask_t rcu_blocking;
+
+    /* Idle CPU handling */
     struct timer idle_timer;
     bool idle_timer_active;
 
-    bool            process_callbacks;
+    /* Barrier handling. */
     bool            barrier_active;
 };
 
 /*
- * If a CPU with RCU callbacks queued goes idle, when the grace period is
- * not finished yet, how can we make sure that the callbacks will eventually
- * be executed? In Linux (2.6.21, the first "tickless idle" Linux kernel),
- * the periodic timer tick would not be stopped for such CPU. Here in Xen,
- * we (may) don't even have a periodic timer tick, so we need to use a
- * special purpose timer.
- *
- * Such timer:
- * 1) is armed only when a CPU with an RCU callback(s) queued goes idle
- *    before the end of the current grace period (_not_ for any CPUs that
- *    go idle!);
- * 2) when it fires, it is only re-armed if the grace period is still
- *    running;
- * 3) it is stopped immediately, if the CPU wakes up from idle and
- *    resumes 'normal' execution.
- *
- * About how far in the future the timer should be programmed each time,
- * it's hard to tell (guess!!). Since this mimics Linux's periodic timer
- * tick, take values used there as an indication. In Linux 2.6.21, tick
- * period can be 10ms, 4ms, 3.33ms or 1ms.
+ * If a CPU with RCU callbacks queued goes idle before the callbacks can be
+ * drained use a timer to ensure the CPU is woken up to process the remaining
+ * callback queue.
  *
  * By default, we use 10ms, to enable at least some power saving on the
  * CPU that is going idle. The user can change this, via a boot time
@@ -137,21 +95,18 @@ static s_time_t __read_mostly idle_timer_period;
 /*
  * Increment and decrement values for the idle timer handler. The algorithm
  * works as follows:
- * - if the timer actually fires, and it finds out that the grace period isn't
- *   over yet, we add IDLE_TIMER_PERIOD_INCR to the timer's period;
- * - if the timer actually fires and it finds the grace period over, we
- *   subtract IDLE_TIMER_PERIOD_DECR from the timer's period.
+ * - If the timer actually fires, and it finds out there are CPUs still in RCU
+ *   critical regions, we add IDLE_TIMER_PERIOD_INCR to the timer's period.
+ *   Note this is not very accurate, as the CPUs in those RCU critical regions
+ *   might not be holding back the execution of the local callbacks.
+ * - If the timer actually fires and it finds no CPUs in critical RCU regions,
+ *   we subtract IDLE_TIMER_PERIOD_DECR from the timer's period.
  */
 #define IDLE_TIMER_PERIOD_INCR    MILLISECS(10)
 #define IDLE_TIMER_PERIOD_DECR    MICROSECS(100)
 
 static DEFINE_PER_CPU(struct rcu_data, rcu_data);
 
-static int blimit = 10;
-static int qhimark = 10000;
-static int qlowmark = 100;
-static int rsinterval = 1000;
-
 /*
  * rcu_barrier() handling:
  * Two counters are used to synchronize rcu_barrier() work:
@@ -246,35 +201,12 @@ void rcu_barrier(void)
     put_cpu_maps();
 }
 
-/* Is batch a before batch b ? */
-static inline int rcu_batch_before(long a, long b)
-{
-    return (a - b) < 0;
-}
-
-static void force_quiescent_state(struct rcu_data *rdp,
-                                  struct rcu_ctrlblk *rcp)
-{
-    cpumask_t cpumask;
-    raise_softirq(RCU_SOFTIRQ);
-    if (unlikely(rdp->qlen - rdp->last_rs_qlen > rsinterval)) {
-        rdp->last_rs_qlen = rdp->qlen;
-        /*
-         * Don't send IPI to itself. With irqs disabled,
-         * rdp->cpu is the current cpu.
-         */
-        cpumask_andnot(&cpumask, &rcp->cpumask, cpumask_of(rdp->cpu));
-        cpumask_raise_softirq(&cpumask, RCU_SOFTIRQ);
-    }
-}
-
 /**
  * call_rcu - Queue an RCU callback for invocation after a grace period.
  * @head: structure to be used for queueing the RCU updates.
  * @func: actual update function to be invoked after the grace period
  *
- * The update function will be invoked some time after a full grace
- * period elapses, in other words after all currently executing RCU
+ * The update function will be invoked after all currently executing RCU
  * read-side critical sections have completed.  RCU read-side critical
  * sections are delimited by rcu_read_lock() and rcu_read_unlock(),
  * and may be nested.
@@ -283,205 +215,88 @@ void call_rcu(struct rcu_head *head,
               void (*func)(struct rcu_head *rcu))
 {
     unsigned long flags;
-    struct rcu_data *rdp;
+    struct rcu_data *rdp = &this_cpu(rcu_data);
 
     head->func = func;
-    head->next = NULL;
     local_irq_save(flags);
     head->added = arch_fetch_and_add(&rcu_epoch, 1);
-    rdp = &this_cpu(rcu_data);
-    *rdp->nxttail = head;
-    rdp->nxttail = &head->next;
-    if (unlikely(++rdp->qlen > qhimark)) {
-        rdp->blimit = INT_MAX;
-        force_quiescent_state(rdp, &rcu_ctrlblk);
-    }
-    local_irq_restore(flags);
-}
-
-/*
- * Invoke the completed RCU callbacks. They are expected to be in
- * a per-cpu list.
- */
-static void rcu_do_batch(struct rcu_data *rdp)
-{
-    struct rcu_head *next, *list;
-    int count = 0;
-
-    list = rdp->donelist;
-    while (list) {
-        next = rdp->donelist = list->next;
-        list->func(list);
-        list = next;
-        rdp->qlen--;
-        if (++count >= rdp->blimit)
-            break;
-    }
-    if (rdp->blimit == INT_MAX && rdp->qlen <= qlowmark)
-        rdp->blimit = blimit;
-    if (!rdp->donelist)
-        rdp->donetail = &rdp->donelist;
-    else
-    {
-        rdp->process_callbacks = true;
-        raise_softirq(RCU_SOFTIRQ);
-    }
-}
-
-/*
- * Grace period handling:
- * The grace period handling consists out of two steps:
- * - A new grace period is started.
- *   This is done by rcu_start_batch. The start is not broadcasted to
- *   all cpus, they must pick this up by comparing rcp->cur with
- *   rdp->quiescbatch. All cpus are recorded  in the
- *   rcu_ctrlblk.cpumask bitmap.
- * - All cpus must go through a quiescent state.
- *   Since the start of the grace period is not broadcasted, at least two
- *   calls to rcu_check_quiescent_state are required:
- *   The first call just notices that a new grace period is running. The
- *   following calls check if there was a quiescent state since the beginning
- *   of the grace period. If so, it updates rcu_ctrlblk.cpumask. If
- *   the bitmap is empty, then the grace period is completed.
- *   rcu_check_quiescent_state calls rcu_start_batch(0) to start the next grace
- *   period (if necessary).
- */
-/*
- * Register a new batch of callbacks, and start it up if there is currently no
- * active batch and the batch to be registered has not already occurred.
- * Caller must hold rcu_ctrlblk.lock.
- */
-static void rcu_start_batch(struct rcu_ctrlblk *rcp)
-{
-    if (rcp->next_pending &&
-        rcp->completed == rcp->cur) {
-        rcp->next_pending = 0;
+    list_add_tail(&head->list, &rdp->pending);
+    if ( ++rdp->nr > RCU_QUEUE_THRESHOLD )
         /*
-         * next_pending == 0 must be visible in
-         * __rcu_process_callbacks() before it can see new value of cur.
+         * Raise a softirq to attempt to force draining the queue, albeit
+         * there's no guarantee.
          */
-        smp_wmb();
-        rcp->cur++;
-
-       /*
-        * Make sure the increment of rcp->cur is visible so, even if a
-        * CPU that is about to go idle, is captured inside rcp->cpumask,
-        * rcu_pending() will return false, which then means cpu_quiet()
-        * will be invoked, before the CPU would actually enter idle.
-        *
-        * This barrier is paired with the one in rcu_idle_enter().
-        */
-        smp_mb();
-        cpumask_andnot(&rcp->cpumask, &cpu_online_map, &rcp->idle_cpumask);
-    }
-}
-
-/*
- * cpu went through a quiescent state since the beginning of the grace period.
- * Clear it from the cpu mask and complete the grace period if it was the last
- * cpu. Start another grace period if someone has further entries pending
- */
-static void cpu_quiet(int cpu, struct rcu_ctrlblk *rcp)
-{
-    cpumask_clear_cpu(cpu, &rcp->cpumask);
-    if (cpumask_empty(&rcp->cpumask)) {
-        /* batch completed ! */
-        rcp->completed = rcp->cur;
-        rcu_start_batch(rcp);
-    }
+        raise_softirq(RCU_SOFTIRQ);
+    local_irq_restore(flags);
 }
 
-/*
- * Check if the cpu has gone through a quiescent state (say context
- * switch). If so and if it already hasn't done so in this RCU
- * quiescent cycle, then indicate that it has done so.
- */
-static void rcu_check_quiescent_state(struct rcu_ctrlblk *rcp,
-                                      struct rcu_data *rdp)
+#define RCU_MAX_BATCH 10
+static void cf_check rcu_process_callbacks(void)
 {
-    if (rdp->quiescbatch != rcp->cur) {
-        /* start new grace period: */
-        rdp->qs_pending = 1;
-        rdp->quiescbatch = rcp->cur;
-        return;
-    }
-
-    /* Grace period already completed for this cpu?
-     * qs_pending is checked instead of the actual bitmap to avoid
-     * cacheline trashing.
-     */
-    if (!rdp->qs_pending)
-        return;
+    static DEFINE_PER_CPU(cpumask_t, rcu_scratch);
+    cpumask_t *in_rcu = &this_cpu(rcu_scratch);
+    struct rcu_data *rdp = &this_cpu(rcu_data);
+    unsigned long flags;
+    unsigned int queued = 0, cpu;
+    LIST_HEAD(expired);
+    struct rcu_head *rcu;
 
-    rdp->qs_pending = 0;
+    /* Clear any existing blocking mask - we will redo it if needed. */
+    cpumask_clear(&rdp->rcu_blocking);
 
-    spin_lock(&rcp->lock);
     /*
-     * rdp->quiescbatch/rcp->cur and the cpu bitmap can come out of sync
-     * during cpu startup. Ignore the quiescent state.
+     * Populate a cpumask with any CPUs inside RCU critical regions.  Note that
+     * CPUs entering past this point are of no interest, they will certainly
+     * use an epoch past any queued callbacks here.
      */
-    if (likely(rdp->quiescbatch == rcp->cur))
-        cpu_quiet(rdp->cpu, rcp);
-
-    spin_unlock(&rcp->lock);
-}
-
-
-/*
- * This does the RCU processing work from softirq context. 
- */
-static void __rcu_process_callbacks(struct rcu_ctrlblk *rcp,
-                                    struct rcu_data *rdp)
-{
-    if (rdp->curlist && !rcu_batch_before(rcp->completed, rdp->batch)) {
-        *rdp->donetail = rdp->curlist;
-        rdp->donetail = rdp->curtail;
-        rdp->curlist = NULL;
-        rdp->curtail = &rdp->curlist;
-    }
-
-    local_irq_disable();
-    if (rdp->nxtlist && !rdp->curlist) {
-        rdp->curlist = rdp->nxtlist;
-        rdp->curtail = rdp->nxttail;
-        rdp->nxtlist = NULL;
-        rdp->nxttail = &rdp->nxtlist;
-        local_irq_enable();
+    cpumask_clear(in_rcu);
+    for_each_online_cpu ( cpu )
+        if ( ACCESS_ONCE(per_cpu(rcu_lock_cnt, cpu)) )
+            __cpumask_set_cpu(cpu, in_rcu);
 
+    local_irq_save(flags);
+    while ( queued < RCU_MAX_BATCH &&
+            (rcu = list_first_entry_or_null(&rdp->pending, struct rcu_head,
+                                            list)) )
+    {
         /*
-         * start the next batch of callbacks
+         * Fetching rcu_lock_epoch out of order is not a concern here: in the
+         * worst case it's going to result in an older more restrictive epoch
+         * being checked against.  Note the adding of a callback issues a
+         * arch_fetch_and_add() which is a barrier on itself, and guarantees
+         * remote changes to the CPU mask to be visible here.
          */
-
-        /* determine batch number */
-        rdp->batch = rcp->cur + 1;
-        /* see the comment and corresponding wmb() in
-         * the rcu_start_batch()
-         */
-        smp_rmb();
-
-        if (!rcp->next_pending) {
-            /* and start it/schedule start if it's a new batch */
-            spin_lock(&rcp->lock);
-            rcp->next_pending = 1;
-            rcu_start_batch(rcp);
-            spin_unlock(&rcp->lock);
-        }
-    } else {
-        local_irq_enable();
+        for_each_cpu ( cpu, in_rcu )
+            if ( (int)(ACCESS_ONCE(per_cpu(rcu_lock_epoch, cpu)) -
+                       rcu->added) <= 0 )
+                /* Record CPUs blocking the execution of the callback. */
+                __cpumask_set_cpu(cpu, &rdp->rcu_blocking);
+
+        if ( !cpumask_empty(&rdp->rcu_blocking) )
+            /*
+             * Callbacks are sorted, exit loop as soon as we find one that
+             * can't be processed yet.
+             */
+            goto process;
+
+        list_del(&rcu->list);
+        list_add_tail(&rcu->list, &expired);
+        ASSERT(rdp->nr);
+        rdp->nr--;
+        queued++;
     }
-    rcu_check_quiescent_state(rcp, rdp);
-    if (rdp->donelist)
-        rcu_do_batch(rdp);
-}
 
-static void cf_check rcu_process_callbacks(void)
-{
-    struct rcu_data *rdp = &this_cpu(rcu_data);
+    if ( queued == RCU_MAX_BATCH && rdp->nr )
+        /* There's more work to do, yield and raise a softirq to come back. */
+        raise_softirq(RCU_SOFTIRQ);
 
-    if ( rdp->process_callbacks )
+ process:
+    local_irq_restore(flags);
+
+    while ( (rcu = list_first_entry_or_null(&expired, struct rcu_head, list)) )
     {
-        rdp->process_callbacks = false;
-        __rcu_process_callbacks(&rcu_ctrlblk, rdp);
+        list_del(&rcu->list);
+        rcu->func(rcu);
     }
 
     if ( atomic_read(&cpu_count) && !rdp->barrier_active )
@@ -492,33 +307,26 @@ static void cf_check rcu_process_callbacks(void)
     }
 }
 
-static int __rcu_pending(struct rcu_ctrlblk *rcp, struct rcu_data *rdp)
+bool rcu_pending(unsigned int cpu)
 {
-    /* This cpu has pending rcu entries and the grace period
-     * for them has completed.
-     */
-    if (rdp->curlist && !rcu_batch_before(rcp->completed, rdp->batch))
-        return 1;
+    struct rcu_data *rdp = &per_cpu(rcu_data, cpu);
+    const struct rcu_head *rcu =
+        list_first_entry_or_null(&rdp->pending, struct rcu_head, list);
+    unsigned int rcpu;
 
-    /* This cpu has no pending entries, but there are new entries */
-    if (!rdp->curlist && rdp->nxtlist)
-        return 1;
+    /* Modifying rcu_blocking per-CPU mask is only safe if done locally. */
+    ASSERT(smp_processor_id() == cpu);
 
-    /* This cpu has finished callbacks to invoke */
-    if (rdp->donelist)
-        return 1;
+    if ( !rcu )
+        return false;
 
-    /* The rcu core waits for a quiescent state from the cpu */
-    if (rdp->quiescbatch != rcp->cur || rdp->qs_pending)
-        return 1;
+    for_each_cpu ( rcpu, &rdp->rcu_blocking )
+        if ( (int)(ACCESS_ONCE(per_cpu(rcu_lock_epoch, rcpu)) -
+                   rcu->added) > 0 )
+            /* Record CPUs blocking the execution of the callback. */
+            __cpumask_clear_cpu(rcpu, &rdp->rcu_blocking);
 
-    /* nothing to do */
-    return 0;
-}
-
-bool rcu_pending(unsigned int cpu)
-{
-    return !!__rcu_pending(&rcu_ctrlblk, &per_cpu(rcu_data, cpu));
+    return cpumask_empty(&rdp->rcu_blocking);
 }
 
 /*
@@ -529,15 +337,13 @@ bool rcu_pending(unsigned int cpu)
  */
 bool rcu_needs_cpu(unsigned int cpu)
 {
-    struct rcu_data *rdp = &per_cpu(rcu_data, cpu);
-
-    return (rdp->curlist && !rdp->idle_timer_active) || rcu_pending(cpu);
+    return !!per_cpu(rcu_data, cpu).nr;
 }
 
 /*
  * Timer for making sure the CPU where a callback is queued does
  * periodically poke rcu_pending(), so that it will invoke the callback
- * not too late after the end of the grace period.
+ * not too late.
  */
 static void rcu_idle_timer_start(void)
 {
@@ -545,10 +351,9 @@ static void rcu_idle_timer_start(void)
 
     /*
      * Note that we don't check rcu_pending() here. In fact, we don't want
-     * the timer armed on CPUs that are in the process of quiescing while
-     * going idle, unless they really are the ones with a queued callback.
+     * the timer armed on CPUs that don't have pending callbacks.
      */
-    if (likely(!rdp->curlist))
+    if (likely(!rdp->nr))
         return;
 
     set_timer(&rdp->idle_timer, NOW() + idle_timer_period);
@@ -587,7 +392,7 @@ static void cf_check rcu_idle_timer_handler(void* data)
 {
     perfc_incr(rcu_idle_timer);
 
-    if ( !cpumask_empty(&rcu_ctrlblk.cpumask) )
+    if ( this_cpu(rcu_data).nr )
         idle_timer_period = min(idle_timer_period + IDLE_TIMER_PERIOD_INCR,
                                 IDLE_TIMER_PERIOD_MAX);
     else
@@ -597,55 +402,51 @@ static void cf_check rcu_idle_timer_handler(void* data)
 
 void rcu_check_callbacks(unsigned int cpu)
 {
-    struct rcu_data *rdp = &this_cpu(rcu_data);
-
-    rdp->process_callbacks = true;
     raise_softirq(RCU_SOFTIRQ);
 }
 
-static void rcu_move_batch(struct rcu_data *this_rdp, struct rcu_head *list,
-                           struct rcu_head **tail)
+/* Sorting functions for RCU list concatenation when a CPU goes offline. */
+static int cf_check cmp_rcu(void *priv, struct list_head *a,
+                            struct list_head *b)
 {
-    local_irq_disable();
-    *this_rdp->nxttail = list;
-    if (list)
-        this_rdp->nxttail = tail;
-    local_irq_enable();
+    const struct rcu_head *l = container_of(a, struct rcu_head, list),
+                          *r = container_of(b, struct rcu_head, list);
+
+    return (int)(l->added - r->added);
 }
 
 static void rcu_offline_cpu(struct rcu_data *this_rdp,
-                            struct rcu_ctrlblk *rcp, struct rcu_data *rdp)
+                            struct rcu_data *rdp)
 {
     kill_timer(&rdp->idle_timer);
 
-    /* If the cpu going offline owns the grace period we can block
-     * indefinitely waiting for it, so flush it here.
-     */
-    spin_lock(&rcp->lock);
-    if (rcp->cur != rcp->completed)
-        cpu_quiet(rdp->cpu, rcp);
-    spin_unlock(&rcp->lock);
-
-    rcu_move_batch(this_rdp, rdp->donelist, rdp->donetail);
-    rcu_move_batch(this_rdp, rdp->curlist, rdp->curtail);
-    rcu_move_batch(this_rdp, rdp->nxtlist, rdp->nxttail);
+    if ( !rdp->nr )
+        return;
 
+    /*
+     * Append pending callbacks to the current CPU.  By the time this is
+     * executed the CPU going offline cannot be in any RCU critical section or
+     * queue any more RCU work.
+     */
     local_irq_disable();
-    this_rdp->qlen += rdp->qlen;
+    list_splice(&rdp->pending, &this_rdp->pending);
+    this_rdp->nr += rdp->nr;
+    list_sort(NULL, &this_rdp->pending, cmp_rcu);
+    /* Clear blocking CPU mask: the top callback might have changed. */
+    cpumask_clear(&this_rdp->rcu_blocking);
     local_irq_enable();
+
+    if ( this_rdp->nr > RCU_QUEUE_THRESHOLD )
+        raise_softirq(RCU_SOFTIRQ);
+
+    INIT_LIST_HEAD(&rdp->pending);
+    rdp->nr = 0;
 }
 
-static void rcu_init_percpu_data(int cpu, struct rcu_ctrlblk *rcp,
-                                 struct rcu_data *rdp)
+static void rcu_init_percpu_data(int cpu, struct rcu_data *rdp)
 {
     memset(rdp, 0, sizeof(*rdp));
-    rdp->curtail = &rdp->curlist;
-    rdp->nxttail = &rdp->nxtlist;
-    rdp->donetail = &rdp->donelist;
-    rdp->quiescbatch = rcp->completed;
-    rdp->qs_pending = 0;
-    rdp->cpu = cpu;
-    rdp->blimit = blimit;
+    INIT_LIST_HEAD(&rdp->pending);
     init_timer(&rdp->idle_timer, rcu_idle_timer_handler, rdp, cpu);
 }
 
@@ -658,11 +459,11 @@ static int cf_check cpu_callback(
     switch ( action )
     {
     case CPU_UP_PREPARE:
-        rcu_init_percpu_data(cpu, &rcu_ctrlblk, rdp);
+        rcu_init_percpu_data(cpu, rdp);
         break;
     case CPU_UP_CANCELED:
     case CPU_DEAD:
-        rcu_offline_cpu(&this_cpu(rcu_data), &rcu_ctrlblk, rdp);
+        rcu_offline_cpu(&this_cpu(rcu_data), rdp);
         break;
     default:
         break;
@@ -693,36 +494,18 @@ void __init rcu_init(void)
     }
     idle_timer_period = MILLISECS(idle_timer_period_ms);
 
-    cpumask_clear(&rcu_ctrlblk.idle_cpumask);
     cpu_callback(&cpu_nfb, CPU_UP_PREPARE, cpu);
     register_cpu_notifier(&cpu_nfb);
     open_softirq(RCU_SOFTIRQ, rcu_process_callbacks);
 }
 
-/*
- * The CPU is becoming idle, so no more read side critical
- * sections, and one more step toward grace period.
- */
+/* The CPU is becoming idle, ensure pending RCU work will get processed. */
 void rcu_idle_enter(unsigned int cpu)
 {
-    ASSERT(!cpumask_test_cpu(cpu, &rcu_ctrlblk.idle_cpumask));
-    cpumask_set_cpu(cpu, &rcu_ctrlblk.idle_cpumask);
-    /*
-     * If some other CPU is starting a new grace period, we'll notice that
-     * by seeing a new value in rcp->cur (different than our quiescbatch).
-     * That will force us all the way until cpu_quiet(), clearing our bit
-     * in rcp->cpumask, even in case we managed to get in there.
-     *
-     * Se the comment before cpumask_andnot() in  rcu_start_batch().
-     */
-    smp_mb();
-
     rcu_idle_timer_start();
 }
 
 void rcu_idle_exit(unsigned int cpu)
 {
     rcu_idle_timer_stop();
-    ASSERT(cpumask_test_cpu(cpu, &rcu_ctrlblk.idle_cpumask));
-    cpumask_clear_cpu(cpu, &rcu_ctrlblk.idle_cpumask);
 }
diff --git a/xen/include/xen/rcupdate.h b/xen/include/xen/rcupdate.h
index 5846e8c169d7..312551d5b089 100644
--- a/xen/include/xen/rcupdate.h
+++ b/xen/include/xen/rcupdate.h
@@ -35,6 +35,7 @@
 #include <xen/spinlock.h>
 #include <xen/cpumask.h>
 #include <xen/lib.h>
+#include <xen/list.h>
 #include <xen/percpu.h>
 #include <xen/preempt.h>
 
@@ -57,7 +58,6 @@ static inline void rcu_quiesce_disable(void)
 
 static inline void rcu_quiesce_enable(void)
 {
-
     smp_mb();
     ACCESS_ONCE(this_cpu(rcu_lock_cnt))--;
     preempt_enable();
@@ -70,19 +70,21 @@ static inline bool rcu_quiesce_allowed(void)
 
 /**
  * struct rcu_head - callback structure for use with RCU
- * @next: next update requests in a list
+ * @list: list anchor.
  * @func: actual update function to call after the grace period.
+ * @added: epoch when the callback was added.
  */
 struct rcu_head {
-    struct rcu_head *next;
+    struct list_head list;
     void (*func)(struct rcu_head *head);
     unsigned int added;
 };
 
-#define RCU_HEAD_INIT   { .next = NULL, .func = NULL }
-#define RCU_HEAD(head) struct rcu_head head = RCU_HEAD_INIT
+#define RCU_HEAD_INIT(head)   { .list = LIST_HEAD_INIT((head).list), \
+                                .func = NULL }
+#define RCU_HEAD(head) struct rcu_head head = RCU_HEAD_INIT(head)
 #define INIT_RCU_HEAD(ptr) do { \
-       (ptr)->next = NULL; (ptr)->func = NULL; \
+    INIT_LIST_HEAD(&(ptr)->list); (ptr)->func = NULL; \
 } while (0)
 
 
-- 
2.55.0

Re: [PATCH v2 2/3] xen/rcu: simplify RCU implementation

[Alejandro Vallejo]

On Fri Sep 25, 2026 at 6:30 PM CEST, Roger Pau Monne wrote:
> The current implementation has two shortcomings for certain Xen usages:
>
>  * When using the null scheduler it's possible for a CPU to never enter Xen
>    context.  A CPU not entering Xen context can block other CPUs from
>    executing RCU callbacks, as there will be no quiescent state observed if
>    the CPU doesn't enter Xen context.
>
>  * If a certain amount of callbacks are pending, RCU will try to force a
>    quiescent state, by sending an IPI to remote CPUs.  This causes unwanted
>    interference.
>
> Keep track of the RCU epoch when a callback was added, and only execute it
> once all CPUs are either outside of RCU critical regions, or any CPUs
> inside of RCU critical regions have entered such past the epoch when the
> callback was queued.  Knowing whether a CPU is inside a RCU critical region
> is done based on the CPU rcu_lock_cnt value.
>
> This removes the grace/quiescent period state tracking entirely: the
> control block, the per-CPU quiescent bookkeeping, the batch lists and the
> forced-quiescent-state IPI all go away.  As a result no RCU-driven IPIs are
> sent to remote CPUs, and callback progress no longer depends on every CPU
> passing through Xen context.
>
> Signed-off-by: Roger Pau Monné <roger@xenproject.org>

All review I already did on this was on the applied patch. I find the
diff a bit too unwieldy. Perhaps it might be good to add the new logic
on top first, and then remove the prior logic.

Something like:

#define RCU_USES_EPOCH 1

#ifdef RCU_USES_EPOCH
<old rcu specifics>
#else
<new rcu specifics>
#endif

... then have a last patch that drops old RCU code. Structs, older
functions, etc can stay until then, because they'll be removed later.
It's just a means of shrinking the diff in order to make it tractable.

My .02, anyway. Perhaps you already tried and it's terrible anyhow.

Cheers,
Alejandro

Re: [PATCH v2 2/3] xen/rcu: simplify RCU implementation

[Alejandro Vallejo]

On Fri Sep 25, 2026 at 6:30 PM CEST, Roger Pau Monne wrote:
> The current implementation has two shortcomings for certain Xen usages:
>
>  * When using the null scheduler it's possible for a CPU to never enter Xen
>    context.  A CPU not entering Xen context can block other CPUs from
>    executing RCU callbacks, as there will be no quiescent state observed if
>    the CPU doesn't enter Xen context.
>
>  * If a certain amount of callbacks are pending, RCU will try to force a
>    quiescent state, by sending an IPI to remote CPUs.  This causes unwanted
>    interference.
>
> Keep track of the RCU epoch when a callback was added, and only execute it
> once all CPUs are either outside of RCU critical regions, or any CPUs
> inside of RCU critical regions have entered such past the epoch when the
> callback was queued.  Knowing whether a CPU is inside a RCU critical region
> is done based on the CPU rcu_lock_cnt value.
>
> This removes the grace/quiescent period state tracking entirely: the
> control block, the per-CPU quiescent bookkeeping, the batch lists and the
> forced-quiescent-state IPI all go away.  As a result no RCU-driven IPIs are
> sent to remote CPUs, and callback progress no longer depends on every CPU
> passing through Xen context.
>
> Signed-off-by: Roger Pau Monné <roger@xenproject.org>
> ---
> The maximum batch of callbacks processed is limited to 10, this is bit
> arbitrary, but I think matches what the current logic attempts does.
>
> There's a bit more pruning to do regarding the usage of grace and
> quiesce in comments or functions names - I've done that as a followup
> change to reduce the size of this change.
>
> I think the manual raising of an RCU softirq is not really needed when
> the queue exceeds a certain length - the softirq loop will already
> attempt RCU callback processing if possible.
> ---
> Changes since v1:
>  - Rework rcu_pending() so it reports whether there are callback ready to
>    be pruned, instead of simply reporting whether there are any callbacks
>    pending.  This matches the previous implementation.
>  - Force a barrier around entry/exit to each critical region, instead of
>    doing so only in the first and last entry/exit respectively.  This
>    makes the logic more obviously correct.
> ---
>  xen/common/rcupdate.c      | 493 +++++++++++--------------------------
>  xen/include/xen/rcupdate.h |  14 +-
>  2 files changed, 146 insertions(+), 361 deletions(-)
>
> diff --git a/xen/common/rcupdate.c b/xen/common/rcupdate.c
> index babaabbbeefe..79d25c958f0a 100644
> --- a/xen/common/rcupdate.c
> +++ b/xen/common/rcupdate.c
> @@ -35,6 +35,7 @@
>  #include <xen/cpu.h>
>  #include <xen/init.h>
>  #include <xen/kernel.h>
> +#include <xen/list_sort.h>
>  #include <xen/param.h>
>  #include <xen/percpu.h>
>  #include <xen/rcupdate.h>
> @@ -55,74 +56,31 @@ DEFINE_PER_CPU(unsigned int, rcu_lock_epoch);
>  /* Current RCU epoch, bumped every time a new callback is queued. */
>  unsigned int rcu_epoch;
>  
> -/* Global control variables for rcupdate callback mechanism. */
> -static struct rcu_ctrlblk {
> -    long cur;           /* Current batch number.                      */
> -    long completed;     /* Number of the last completed batch         */
> -    int  next_pending;  /* Is the next batch already waiting?         */
> -
> -    spinlock_t  lock __cacheline_aligned;
> -    cpumask_t   cpumask; /* CPUs that need to switch in order ... */
> -    cpumask_t   idle_cpumask; /* ... unless they are already idle */
> -    /* for current batch to proceed.        */
> -} __cacheline_aligned rcu_ctrlblk = {
> -    .cur = -300,
> -    .completed = -300,
> -    .lock = SPIN_LOCK_UNLOCKED,
> -};
> -
> -/*
> - * Per-CPU data for Read-Copy Update.
> - * nxtlist - new callbacks are added here
> - * curlist - current batch for which quiescent cycle started if any
> - */
> +/* Per-CPU data for Read-Copy Update. */
>  struct rcu_data {
> -    /* 1) quiescent state handling : */
> -    long quiescbatch;    /* Batch # for grace period */
> -    int  qs_pending;     /* core waits for quiesc state */
> -
> -    /* 2) batch handling */
> -    long            batch;            /* Batch # for current RCU batch */
> -    struct rcu_head *nxtlist;
> -    struct rcu_head **nxttail;
> -    long            qlen;             /* # of queued callbacks */
> -    struct rcu_head *curlist;
> -    struct rcu_head **curtail;
> -    struct rcu_head *donelist;
> -    struct rcu_head **donetail;
> -    long            blimit;           /* Upper limit on a processed batch */
> -    int cpu;
> -    long            last_rs_qlen;     /* qlen during the last resched */
> -
> -    /* 3) idle CPUs handling */
> +    /*
> +     * List of pending callbacks, sorted by ascending epoch.  Use a threshold
> +     * value to raise an RCU softirq if the queue exceeds a given length.
> +     */
> +    struct list_head pending;
> +    unsigned int nr;
> +#define RCU_QUEUE_THRESHOLD 100
> +
> +    /* Keep track of CPUs blocking first RCU callback execution. */
> +    cpumask_t rcu_blocking;
> +
> +    /* Idle CPU handling */
>      struct timer idle_timer;
>      bool idle_timer_active;
>  
> -    bool            process_callbacks;
> +    /* Barrier handling. */
>      bool            barrier_active;
>  };
>  
>  /*
> - * If a CPU with RCU callbacks queued goes idle, when the grace period is
> - * not finished yet, how can we make sure that the callbacks will eventually
> - * be executed? In Linux (2.6.21, the first "tickless idle" Linux kernel),
> - * the periodic timer tick would not be stopped for such CPU. Here in Xen,
> - * we (may) don't even have a periodic timer tick, so we need to use a
> - * special purpose timer.
> - *
> - * Such timer:
> - * 1) is armed only when a CPU with an RCU callback(s) queued goes idle
> - *    before the end of the current grace period (_not_ for any CPUs that
> - *    go idle!);
> - * 2) when it fires, it is only re-armed if the grace period is still
> - *    running;
> - * 3) it is stopped immediately, if the CPU wakes up from idle and
> - *    resumes 'normal' execution.
> - *
> - * About how far in the future the timer should be programmed each time,
> - * it's hard to tell (guess!!). Since this mimics Linux's periodic timer
> - * tick, take values used there as an indication. In Linux 2.6.21, tick
> - * period can be 10ms, 4ms, 3.33ms or 1ms.
> + * If a CPU with RCU callbacks queued goes idle before the callbacks can be
> + * drained use a timer to ensure the CPU is woken up to process the remaining
> + * callback queue.
>   *
>   * By default, we use 10ms, to enable at least some power saving on the
>   * CPU that is going idle. The user can change this, via a boot time
> @@ -137,21 +95,18 @@ static s_time_t __read_mostly idle_timer_period;
>  /*
>   * Increment and decrement values for the idle timer handler. The algorithm
>   * works as follows:
> - * - if the timer actually fires, and it finds out that the grace period isn't
> - *   over yet, we add IDLE_TIMER_PERIOD_INCR to the timer's period;
> - * - if the timer actually fires and it finds the grace period over, we
> - *   subtract IDLE_TIMER_PERIOD_DECR from the timer's period.
> + * - If the timer actually fires, and it finds out there are CPUs still in RCU
> + *   critical regions, we add IDLE_TIMER_PERIOD_INCR to the timer's period.
> + *   Note this is not very accurate, as the CPUs in those RCU critical regions
> + *   might not be holding back the execution of the local callbacks.
> + * - If the timer actually fires and it finds no CPUs in critical RCU regions,
> + *   we subtract IDLE_TIMER_PERIOD_DECR from the timer's period.
>   */
>  #define IDLE_TIMER_PERIOD_INCR    MILLISECS(10)
>  #define IDLE_TIMER_PERIOD_DECR    MICROSECS(100)
>  
>  static DEFINE_PER_CPU(struct rcu_data, rcu_data);
>  
> -static int blimit = 10;
> -static int qhimark = 10000;
> -static int qlowmark = 100;
> -static int rsinterval = 1000;
> -
>  /*
>   * rcu_barrier() handling:
>   * Two counters are used to synchronize rcu_barrier() work:
> @@ -246,35 +201,12 @@ void rcu_barrier(void)
>      put_cpu_maps();
>  }
>  
> -/* Is batch a before batch b ? */
> -static inline int rcu_batch_before(long a, long b)
> -{
> -    return (a - b) < 0;
> -}
> -
> -static void force_quiescent_state(struct rcu_data *rdp,
> -                                  struct rcu_ctrlblk *rcp)
> -{
> -    cpumask_t cpumask;
> -    raise_softirq(RCU_SOFTIRQ);
> -    if (unlikely(rdp->qlen - rdp->last_rs_qlen > rsinterval)) {
> -        rdp->last_rs_qlen = rdp->qlen;
> -        /*
> -         * Don't send IPI to itself. With irqs disabled,
> -         * rdp->cpu is the current cpu.
> -         */
> -        cpumask_andnot(&cpumask, &rcp->cpumask, cpumask_of(rdp->cpu));
> -        cpumask_raise_softirq(&cpumask, RCU_SOFTIRQ);
> -    }
> -}
> -
>  /**
>   * call_rcu - Queue an RCU callback for invocation after a grace period.
>   * @head: structure to be used for queueing the RCU updates.
>   * @func: actual update function to be invoked after the grace period
>   *
> - * The update function will be invoked some time after a full grace
> - * period elapses, in other words after all currently executing RCU
> + * The update function will be invoked after all currently executing RCU
>   * read-side critical sections have completed.  RCU read-side critical
>   * sections are delimited by rcu_read_lock() and rcu_read_unlock(),
>   * and may be nested.
> @@ -283,205 +215,88 @@ void call_rcu(struct rcu_head *head,
>                void (*func)(struct rcu_head *rcu))
>  {
>      unsigned long flags;
> -    struct rcu_data *rdp;
> +    struct rcu_data *rdp = &this_cpu(rcu_data);
>  
>      head->func = func;
> -    head->next = NULL;
>      local_irq_save(flags);
>      head->added = arch_fetch_and_add(&rcu_epoch, 1);
> -    rdp = &this_cpu(rcu_data);
> -    *rdp->nxttail = head;
> -    rdp->nxttail = &head->next;
> -    if (unlikely(++rdp->qlen > qhimark)) {
> -        rdp->blimit = INT_MAX;
> -        force_quiescent_state(rdp, &rcu_ctrlblk);
> -    }
> -    local_irq_restore(flags);
> -}
> -
> -/*
> - * Invoke the completed RCU callbacks. They are expected to be in
> - * a per-cpu list.
> - */
> -static void rcu_do_batch(struct rcu_data *rdp)
> -{
> -    struct rcu_head *next, *list;
> -    int count = 0;
> -
> -    list = rdp->donelist;
> -    while (list) {
> -        next = rdp->donelist = list->next;
> -        list->func(list);
> -        list = next;
> -        rdp->qlen--;
> -        if (++count >= rdp->blimit)
> -            break;
> -    }
> -    if (rdp->blimit == INT_MAX && rdp->qlen <= qlowmark)
> -        rdp->blimit = blimit;
> -    if (!rdp->donelist)
> -        rdp->donetail = &rdp->donelist;
> -    else
> -    {
> -        rdp->process_callbacks = true;
> -        raise_softirq(RCU_SOFTIRQ);
> -    }
> -}
> -
> -/*
> - * Grace period handling:
> - * The grace period handling consists out of two steps:
> - * - A new grace period is started.
> - *   This is done by rcu_start_batch. The start is not broadcasted to
> - *   all cpus, they must pick this up by comparing rcp->cur with
> - *   rdp->quiescbatch. All cpus are recorded  in the
> - *   rcu_ctrlblk.cpumask bitmap.
> - * - All cpus must go through a quiescent state.
> - *   Since the start of the grace period is not broadcasted, at least two
> - *   calls to rcu_check_quiescent_state are required:
> - *   The first call just notices that a new grace period is running. The
> - *   following calls check if there was a quiescent state since the beginning
> - *   of the grace period. If so, it updates rcu_ctrlblk.cpumask. If
> - *   the bitmap is empty, then the grace period is completed.
> - *   rcu_check_quiescent_state calls rcu_start_batch(0) to start the next grace
> - *   period (if necessary).
> - */
> -/*
> - * Register a new batch of callbacks, and start it up if there is currently no
> - * active batch and the batch to be registered has not already occurred.
> - * Caller must hold rcu_ctrlblk.lock.
> - */
> -static void rcu_start_batch(struct rcu_ctrlblk *rcp)
> -{
> -    if (rcp->next_pending &&
> -        rcp->completed == rcp->cur) {
> -        rcp->next_pending = 0;
> +    list_add_tail(&head->list, &rdp->pending);
> +    if ( ++rdp->nr > RCU_QUEUE_THRESHOLD )
>          /*
> -         * next_pending == 0 must be visible in
> -         * __rcu_process_callbacks() before it can see new value of cur.
> +         * Raise a softirq to attempt to force draining the queue, albeit
> +         * there's no guarantee.
>           */
> -        smp_wmb();
> -        rcp->cur++;
> -
> -       /*
> -        * Make sure the increment of rcp->cur is visible so, even if a
> -        * CPU that is about to go idle, is captured inside rcp->cpumask,
> -        * rcu_pending() will return false, which then means cpu_quiet()
> -        * will be invoked, before the CPU would actually enter idle.
> -        *
> -        * This barrier is paired with the one in rcu_idle_enter().
> -        */
> -        smp_mb();
> -        cpumask_andnot(&rcp->cpumask, &cpu_online_map, &rcp->idle_cpumask);
> -    }
> -}
> -
> -/*
> - * cpu went through a quiescent state since the beginning of the grace period.
> - * Clear it from the cpu mask and complete the grace period if it was the last
> - * cpu. Start another grace period if someone has further entries pending
> - */
> -static void cpu_quiet(int cpu, struct rcu_ctrlblk *rcp)
> -{
> -    cpumask_clear_cpu(cpu, &rcp->cpumask);
> -    if (cpumask_empty(&rcp->cpumask)) {
> -        /* batch completed ! */
> -        rcp->completed = rcp->cur;
> -        rcu_start_batch(rcp);
> -    }
> +        raise_softirq(RCU_SOFTIRQ);
> +    local_irq_restore(flags);
>  }
>  
> -/*
> - * Check if the cpu has gone through a quiescent state (say context
> - * switch). If so and if it already hasn't done so in this RCU
> - * quiescent cycle, then indicate that it has done so.
> - */
> -static void rcu_check_quiescent_state(struct rcu_ctrlblk *rcp,
> -                                      struct rcu_data *rdp)
> +#define RCU_MAX_BATCH 10
> +static void cf_check rcu_process_callbacks(void)
>  {
> -    if (rdp->quiescbatch != rcp->cur) {
> -        /* start new grace period: */
> -        rdp->qs_pending = 1;
> -        rdp->quiescbatch = rcp->cur;
> -        return;
> -    }
> -
> -    /* Grace period already completed for this cpu?
> -     * qs_pending is checked instead of the actual bitmap to avoid
> -     * cacheline trashing.
> -     */
> -    if (!rdp->qs_pending)
> -        return;
> +    static DEFINE_PER_CPU(cpumask_t, rcu_scratch);
> +    cpumask_t *in_rcu = &this_cpu(rcu_scratch);
> +    struct rcu_data *rdp = &this_cpu(rcu_data);
> +    unsigned long flags;
> +    unsigned int queued = 0, cpu;
> +    LIST_HEAD(expired);
> +    struct rcu_head *rcu;
>  
> -    rdp->qs_pending = 0;
> +    /* Clear any existing blocking mask - we will redo it if needed. */
> +    cpumask_clear(&rdp->rcu_blocking);
>  
> -    spin_lock(&rcp->lock);
>      /*
> -     * rdp->quiescbatch/rcp->cur and the cpu bitmap can come out of sync
> -     * during cpu startup. Ignore the quiescent state.
> +     * Populate a cpumask with any CPUs inside RCU critical regions.  Note that
> +     * CPUs entering past this point are of no interest, they will certainly
> +     * use an epoch past any queued callbacks here.
>       */
> -    if (likely(rdp->quiescbatch == rcp->cur))
> -        cpu_quiet(rdp->cpu, rcp);
> -
> -    spin_unlock(&rcp->lock);
> -}
> -
> -
> -/*
> - * This does the RCU processing work from softirq context. 
> - */
> -static void __rcu_process_callbacks(struct rcu_ctrlblk *rcp,
> -                                    struct rcu_data *rdp)
> -{
> -    if (rdp->curlist && !rcu_batch_before(rcp->completed, rdp->batch)) {
> -        *rdp->donetail = rdp->curlist;
> -        rdp->donetail = rdp->curtail;
> -        rdp->curlist = NULL;
> -        rdp->curtail = &rdp->curlist;
> -    }
> -
> -    local_irq_disable();
> -    if (rdp->nxtlist && !rdp->curlist) {
> -        rdp->curlist = rdp->nxtlist;
> -        rdp->curtail = rdp->nxttail;
> -        rdp->nxtlist = NULL;
> -        rdp->nxttail = &rdp->nxtlist;
> -        local_irq_enable();
> +    cpumask_clear(in_rcu);
> +    for_each_online_cpu ( cpu )
> +        if ( ACCESS_ONCE(per_cpu(rcu_lock_cnt, cpu)) )
> +            __cpumask_set_cpu(cpu, in_rcu);
>  
> +    local_irq_save(flags);
> +    while ( queued < RCU_MAX_BATCH &&
> +            (rcu = list_first_entry_or_null(&rdp->pending, struct rcu_head,
> +                                            list)) )
> +    {
>          /*
> -         * start the next batch of callbacks
> +         * Fetching rcu_lock_epoch out of order is not a concern here: in the
> +         * worst case it's going to result in an older more restrictive epoch
> +         * being checked against.  Note the adding of a callback issues a
> +         * arch_fetch_and_add() which is a barrier on itself, and guarantees
> +         * remote changes to the CPU mask to be visible here.
>           */
> -
> -        /* determine batch number */
> -        rdp->batch = rcp->cur + 1;
> -        /* see the comment and corresponding wmb() in
> -         * the rcu_start_batch()
> -         */
> -        smp_rmb();
> -
> -        if (!rcp->next_pending) {
> -            /* and start it/schedule start if it's a new batch */
> -            spin_lock(&rcp->lock);
> -            rcp->next_pending = 1;
> -            rcu_start_batch(rcp);
> -            spin_unlock(&rcp->lock);
> -        }
> -    } else {
> -        local_irq_enable();
> +        for_each_cpu ( cpu, in_rcu )
> +            if ( (int)(ACCESS_ONCE(per_cpu(rcu_lock_epoch, cpu)) -
> +                       rcu->added) <= 0 )
> +                /* Record CPUs blocking the execution of the callback. */
> +                __cpumask_set_cpu(cpu, &rdp->rcu_blocking);
> +
> +        if ( !cpumask_empty(&rdp->rcu_blocking) )
> +            /*
> +             * Callbacks are sorted, exit loop as soon as we find one that
> +             * can't be processed yet.
> +             */
> +            goto process;
> +
> +        list_del(&rcu->list);
> +        list_add_tail(&rcu->list, &expired);
> +        ASSERT(rdp->nr);
> +        rdp->nr--;
> +        queued++;
>      }
> -    rcu_check_quiescent_state(rcp, rdp);
> -    if (rdp->donelist)
> -        rcu_do_batch(rdp);
> -}
>  
> -static void cf_check rcu_process_callbacks(void)
> -{
> -    struct rcu_data *rdp = &this_cpu(rcu_data);
> +    if ( queued == RCU_MAX_BATCH && rdp->nr )
> +        /* There's more work to do, yield and raise a softirq to come back. */
> +        raise_softirq(RCU_SOFTIRQ);
>  
> -    if ( rdp->process_callbacks )
> + process:
> +    local_irq_restore(flags);
> +
> +    while ( (rcu = list_first_entry_or_null(&expired, struct rcu_head, list)) )
>      {
> -        rdp->process_callbacks = false;
> -        __rcu_process_callbacks(&rcu_ctrlblk, rdp);
> +        list_del(&rcu->list);
> +        rcu->func(rcu);
>      }
>  
>      if ( atomic_read(&cpu_count) && !rdp->barrier_active )
> @@ -492,33 +307,26 @@ static void cf_check rcu_process_callbacks(void)
>      }
>  }
>  
> -static int __rcu_pending(struct rcu_ctrlblk *rcp, struct rcu_data *rdp)
> +bool rcu_pending(unsigned int cpu)
>  {
> -    /* This cpu has pending rcu entries and the grace period
> -     * for them has completed.
> -     */
> -    if (rdp->curlist && !rcu_batch_before(rcp->completed, rdp->batch))
> -        return 1;
> +    struct rcu_data *rdp = &per_cpu(rcu_data, cpu);
> +    const struct rcu_head *rcu =
> +        list_first_entry_or_null(&rdp->pending, struct rcu_head, list);
> +    unsigned int rcpu;
>  
> -    /* This cpu has no pending entries, but there are new entries */
> -    if (!rdp->curlist && rdp->nxtlist)
> -        return 1;
> +    /* Modifying rcu_blocking per-CPU mask is only safe if done locally. */
> +    ASSERT(smp_processor_id() == cpu);
>  
> -    /* This cpu has finished callbacks to invoke */
> -    if (rdp->donelist)
> -        return 1;
> +    if ( !rcu )
> +        return false;
>  
> -    /* The rcu core waits for a quiescent state from the cpu */
> -    if (rdp->quiescbatch != rcp->cur || rdp->qs_pending)
> -        return 1;
> +    for_each_cpu ( rcpu, &rdp->rcu_blocking )
> +        if ( (int)(ACCESS_ONCE(per_cpu(rcu_lock_epoch, rcpu)) -
> +                   rcu->added) > 0 )
> +            /* Record CPUs blocking the execution of the callback. */
> +            __cpumask_clear_cpu(rcpu, &rdp->rcu_blocking);
>  
> -    /* nothing to do */
> -    return 0;
> -}
> -
> -bool rcu_pending(unsigned int cpu)
> -{
> -    return !!__rcu_pending(&rcu_ctrlblk, &per_cpu(rcu_data, cpu));
> +    return cpumask_empty(&rdp->rcu_blocking);
>  }
>  
>  /*
> @@ -529,15 +337,13 @@ bool rcu_pending(unsigned int cpu)
>   */
>  bool rcu_needs_cpu(unsigned int cpu)
>  {
> -    struct rcu_data *rdp = &per_cpu(rcu_data, cpu);
> -
> -    return (rdp->curlist && !rdp->idle_timer_active) || rcu_pending(cpu);
> +    return !!per_cpu(rcu_data, cpu).nr;

Local reads are fine to be relaxed with IRQs off because they can't
change. But remote reads and reads with IRQs on are subject to data
races.

This read should be atomic. And the writes should be atomic too because
the field is read from remote CPUs.

>  }
>  
>  /*
>   * Timer for making sure the CPU where a callback is queued does
>   * periodically poke rcu_pending(), so that it will invoke the callback
> - * not too late after the end of the grace period.
> + * not too late.
>   */
>  static void rcu_idle_timer_start(void)
>  {
> @@ -545,10 +351,9 @@ static void rcu_idle_timer_start(void)
>  
>      /*
>       * Note that we don't check rcu_pending() here. In fact, we don't want
> -     * the timer armed on CPUs that are in the process of quiescing while
> -     * going idle, unless they really are the ones with a queued callback.
> +     * the timer armed on CPUs that don't have pending callbacks.
>       */
> -    if (likely(!rdp->curlist))
> +    if (likely(!rdp->nr))
>          return;
>  
>      set_timer(&rdp->idle_timer, NOW() + idle_timer_period);
> @@ -587,7 +392,7 @@ static void cf_check rcu_idle_timer_handler(void* data)
>  {
>      perfc_incr(rcu_idle_timer);
>  
> -    if ( !cpumask_empty(&rcu_ctrlblk.cpumask) )
> +    if ( this_cpu(rcu_data).nr )
>          idle_timer_period = min(idle_timer_period + IDLE_TIMER_PERIOD_INCR,
>                                  IDLE_TIMER_PERIOD_MAX);
>      else
> @@ -597,55 +402,51 @@ static void cf_check rcu_idle_timer_handler(void* data)
>  
>  void rcu_check_callbacks(unsigned int cpu)
>  {
> -    struct rcu_data *rdp = &this_cpu(rcu_data);
> -
> -    rdp->process_callbacks = true;
>      raise_softirq(RCU_SOFTIRQ);
>  }
>  
> -static void rcu_move_batch(struct rcu_data *this_rdp, struct rcu_head *list,
> -                           struct rcu_head **tail)
> +/* Sorting functions for RCU list concatenation when a CPU goes offline. */
> +static int cf_check cmp_rcu(void *priv, struct list_head *a,
> +                            struct list_head *b)
>  {
> -    local_irq_disable();
> -    *this_rdp->nxttail = list;
> -    if (list)
> -        this_rdp->nxttail = tail;
> -    local_irq_enable();
> +    const struct rcu_head *l = container_of(a, struct rcu_head, list),
> +                          *r = container_of(b, struct rcu_head, list);
> +
> +    return (int)(l->added - r->added);
>  }
>  
>  static void rcu_offline_cpu(struct rcu_data *this_rdp,
> -                            struct rcu_ctrlblk *rcp, struct rcu_data *rdp)
> +                            struct rcu_data *rdp)
>  {
>      kill_timer(&rdp->idle_timer);
>  
> -    /* If the cpu going offline owns the grace period we can block
> -     * indefinitely waiting for it, so flush it here.
> -     */
> -    spin_lock(&rcp->lock);
> -    if (rcp->cur != rcp->completed)
> -        cpu_quiet(rdp->cpu, rcp);
> -    spin_unlock(&rcp->lock);
> -
> -    rcu_move_batch(this_rdp, rdp->donelist, rdp->donetail);
> -    rcu_move_batch(this_rdp, rdp->curlist, rdp->curtail);
> -    rcu_move_batch(this_rdp, rdp->nxtlist, rdp->nxttail);
> +    if ( !rdp->nr )
> +        return;
>  
> +    /*
> +     * Append pending callbacks to the current CPU.  By the time this is
> +     * executed the CPU going offline cannot be in any RCU critical section or
> +     * queue any more RCU work.
> +     */
>      local_irq_disable();
> -    this_rdp->qlen += rdp->qlen;
> +    list_splice(&rdp->pending, &this_rdp->pending);
> +    this_rdp->nr += rdp->nr;
> +    list_sort(NULL, &this_rdp->pending, cmp_rcu);
> +    /* Clear blocking CPU mask: the top callback might have changed. */
> +    cpumask_clear(&this_rdp->rcu_blocking);
>      local_irq_enable();
> +
> +    if ( this_rdp->nr > RCU_QUEUE_THRESHOLD )
> +        raise_softirq(RCU_SOFTIRQ);
> +
> +    INIT_LIST_HEAD(&rdp->pending);
> +    rdp->nr = 0;
>  }
>  
> -static void rcu_init_percpu_data(int cpu, struct rcu_ctrlblk *rcp,
> -                                 struct rcu_data *rdp)
> +static void rcu_init_percpu_data(int cpu, struct rcu_data *rdp)
>  {
>      memset(rdp, 0, sizeof(*rdp));
> -    rdp->curtail = &rdp->curlist;
> -    rdp->nxttail = &rdp->nxtlist;
> -    rdp->donetail = &rdp->donelist;
> -    rdp->quiescbatch = rcp->completed;
> -    rdp->qs_pending = 0;
> -    rdp->cpu = cpu;
> -    rdp->blimit = blimit;
> +    INIT_LIST_HEAD(&rdp->pending);
>      init_timer(&rdp->idle_timer, rcu_idle_timer_handler, rdp, cpu);
>  }
>  
> @@ -658,11 +459,11 @@ static int cf_check cpu_callback(
>      switch ( action )
>      {
>      case CPU_UP_PREPARE:
> -        rcu_init_percpu_data(cpu, &rcu_ctrlblk, rdp);
> +        rcu_init_percpu_data(cpu, rdp);
>          break;
>      case CPU_UP_CANCELED:
>      case CPU_DEAD:
> -        rcu_offline_cpu(&this_cpu(rcu_data), &rcu_ctrlblk, rdp);
> +        rcu_offline_cpu(&this_cpu(rcu_data), rdp);
>          break;
>      default:
>          break;
> @@ -693,36 +494,18 @@ void __init rcu_init(void)
>      }
>      idle_timer_period = MILLISECS(idle_timer_period_ms);
>  
> -    cpumask_clear(&rcu_ctrlblk.idle_cpumask);
>      cpu_callback(&cpu_nfb, CPU_UP_PREPARE, cpu);
>      register_cpu_notifier(&cpu_nfb);
>      open_softirq(RCU_SOFTIRQ, rcu_process_callbacks);
>  }
>  
> -/*
> - * The CPU is becoming idle, so no more read side critical
> - * sections, and one more step toward grace period.
> - */
> +/* The CPU is becoming idle, ensure pending RCU work will get processed. */
>  void rcu_idle_enter(unsigned int cpu)
>  {
> -    ASSERT(!cpumask_test_cpu(cpu, &rcu_ctrlblk.idle_cpumask));
> -    cpumask_set_cpu(cpu, &rcu_ctrlblk.idle_cpumask);
> -    /*
> -     * If some other CPU is starting a new grace period, we'll notice that
> -     * by seeing a new value in rcp->cur (different than our quiescbatch).
> -     * That will force us all the way until cpu_quiet(), clearing our bit
> -     * in rcp->cpumask, even in case we managed to get in there.
> -     *
> -     * Se the comment before cpumask_andnot() in  rcu_start_batch().
> -     */
> -    smp_mb();
> -
>      rcu_idle_timer_start();
>  }
>  
>  void rcu_idle_exit(unsigned int cpu)
>  {
>      rcu_idle_timer_stop();
> -    ASSERT(cpumask_test_cpu(cpu, &rcu_ctrlblk.idle_cpumask));
> -    cpumask_clear_cpu(cpu, &rcu_ctrlblk.idle_cpumask);
>  }
> diff --git a/xen/include/xen/rcupdate.h b/xen/include/xen/rcupdate.h
> index 5846e8c169d7..312551d5b089 100644
> --- a/xen/include/xen/rcupdate.h
> +++ b/xen/include/xen/rcupdate.h
> @@ -35,6 +35,7 @@
>  #include <xen/spinlock.h>
>  #include <xen/cpumask.h>
>  #include <xen/lib.h>
> +#include <xen/list.h>
>  #include <xen/percpu.h>
>  #include <xen/preempt.h>
>  
> @@ -57,7 +58,6 @@ static inline void rcu_quiesce_disable(void)
>  
>  static inline void rcu_quiesce_enable(void)
>  {
> -
>      smp_mb();
>      ACCESS_ONCE(this_cpu(rcu_lock_cnt))--;
>      preempt_enable();
> @@ -70,19 +70,21 @@ static inline bool rcu_quiesce_allowed(void)
>  
>  /**
>   * struct rcu_head - callback structure for use with RCU
> - * @next: next update requests in a list
> + * @list: list anchor.
>   * @func: actual update function to call after the grace period.
> + * @added: epoch when the callback was added.
>   */
>  struct rcu_head {
> -    struct rcu_head *next;
> +    struct list_head list;
>      void (*func)(struct rcu_head *head);
>      unsigned int added;
>  };
>  
> -#define RCU_HEAD_INIT   { .next = NULL, .func = NULL }
> -#define RCU_HEAD(head) struct rcu_head head = RCU_HEAD_INIT
> +#define RCU_HEAD_INIT(head)   { .list = LIST_HEAD_INIT((head).list), \
> +                                .func = NULL }
> +#define RCU_HEAD(head) struct rcu_head head = RCU_HEAD_INIT(head)
>  #define INIT_RCU_HEAD(ptr) do { \
> -       (ptr)->next = NULL; (ptr)->func = NULL; \
> +    INIT_LIST_HEAD(&(ptr)->list); (ptr)->func = NULL; \
>  } while (0)
>  
>  

Re: [PATCH v2 2/3] xen/rcu: simplify RCU implementation

[Roger Pau Monné]

On Mon, Sep 28, 2026 at 11:05:10AM +0200, Alejandro Vallejo wrote:
> On Fri Sep 25, 2026 at 6:30 PM CEST, Roger Pau Monne wrote:
> >  bool rcu_needs_cpu(unsigned int cpu)
> >  {
> > -    struct rcu_data *rdp = &per_cpu(rcu_data, cpu);
> > -
> > -    return (rdp->curlist && !rdp->idle_timer_active) || rcu_pending(cpu);
> > +    return !!per_cpu(rcu_data, cpu).nr;
> 
> Local reads are fine to be relaxed with IRQs off because they can't
> change. But remote reads and reads with IRQs on are subject to data
> races.
> 
> This read should be atomic. And the writes should be atomic too because
> the field is read from remote CPUs.

Sorry, this is a leftover from a further change I wanted to do.
rcu_needs_cpu() is never called against remote CPUs (as is
rcu_pending()), so I wanted to get rid of the cpu parameter and
instead use this_cpu(rcu_data).nr.  Then decided it was too much
churn (and extra logic to fetch the current cpu repeatedly) and forgot
to add the ACCESS_ONCE().

I will add the ACCESS_ONCE() and keep the function interface as-is.

Thanks, Roger.

[PATCH v2 3/3] xen/rcu: remove rcu_quiesce_{disable,enable}()

[Roger Pau Monne]

Fold the code into its callers, having them as separate function just makes
the code harder to read.  Rename rcu_quiesce_allowed() as the RCU
implementation no longer has the concept of quiesce periods.

No functional change intended.

Signed-off-by: Roger Pau Monné <roger@xenproject.org>
---
Changes since v2:
 - New in this version.
---
 xen/common/softirq.c       |  4 ++--
 xen/include/xen/rcupdate.h | 32 +++++++++++---------------------
 2 files changed, 13 insertions(+), 23 deletions(-)

diff --git a/xen/common/softirq.c b/xen/common/softirq.c
index dc3aabce3330..39e1a9844afa 100644
--- a/xen/common/softirq.c
+++ b/xen/common/softirq.c
@@ -31,7 +31,7 @@ static void __do_softirq(unsigned long ignore_mask)
     unsigned long pending;
     bool rcu_allowed = !(ignore_mask & (1UL << RCU_SOFTIRQ));
 
-    ASSERT(!rcu_allowed || rcu_quiesce_allowed());
+    ASSERT(!rcu_allowed || rcu_processing_allowed());
 
     for ( ; ; )
     {
@@ -61,7 +61,7 @@ void process_pending_softirqs(void)
                                 (1UL << SCHED_SLAVE_SOFTIRQ);
 
     /* Block RCU processing in case of rcu_read_lock() held. */
-    if ( !rcu_quiesce_allowed() )
+    if ( !rcu_processing_allowed() )
         ignore_mask |= 1UL << RCU_SOFTIRQ;
 
     ASSERT(!in_irq() && local_irq_is_enabled());
diff --git a/xen/include/xen/rcupdate.h b/xen/include/xen/rcupdate.h
index 312551d5b089..cb16d7f79ae8 100644
--- a/xen/include/xen/rcupdate.h
+++ b/xen/include/xen/rcupdate.h
@@ -46,24 +46,7 @@ DECLARE_PER_CPU(unsigned int, rcu_lock_epoch);
 
 extern unsigned int rcu_epoch;
 
-static inline void rcu_quiesce_disable(void)
-{
-    unsigned int cpu = smp_processor_id();
-
-    preempt_disable();
-    if ( !ACCESS_ONCE(per_cpu(rcu_lock_cnt, cpu))++ )
-        ACCESS_ONCE(per_cpu(rcu_lock_epoch, cpu)) = ACCESS_ONCE(rcu_epoch);
-    smp_mb();
-}
-
-static inline void rcu_quiesce_enable(void)
-{
-    smp_mb();
-    ACCESS_ONCE(this_cpu(rcu_lock_cnt))--;
-    preempt_enable();
-}
-
-static inline bool rcu_quiesce_allowed(void)
+static inline bool rcu_processing_allowed(void)
 {
     return !this_cpu(rcu_lock_cnt);
 }
@@ -129,7 +112,12 @@ typedef struct _rcu_read_lock rcu_read_lock_t;
  */
 static inline void rcu_read_lock(rcu_read_lock_t *lock)
 {
-    rcu_quiesce_disable();
+    unsigned int cpu = smp_processor_id();
+
+    preempt_disable();
+    if ( !ACCESS_ONCE(per_cpu(rcu_lock_cnt, cpu))++ )
+        ACCESS_ONCE(per_cpu(rcu_lock_epoch, cpu)) = ACCESS_ONCE(rcu_epoch);
+    smp_mb();
 }
 
 /**
@@ -139,8 +127,10 @@ static inline void rcu_read_lock(rcu_read_lock_t *lock)
  */
 static inline void rcu_read_unlock(rcu_read_lock_t *lock)
 {
-    ASSERT(!rcu_quiesce_allowed());
-    rcu_quiesce_enable();
+    ASSERT(!rcu_processing_allowed());
+    smp_mb();
+    ACCESS_ONCE(this_cpu(rcu_lock_cnt))--;
+    preempt_enable();
 }
 
 /*
-- 
2.55.0