[PATCH v4 1/8] rv/da: introduce DA_MON_ALLOCATION_STRATEGY

wen.yang@linux.dev posted 8 patches 1 month ago
[PATCH v4 1/8] rv/da: introduce DA_MON_ALLOCATION_STRATEGY
Posted by wen.yang@linux.dev 1 month ago
From: Wen Yang <wen.yang@linux.dev>

Per-object DA storage allocation is currently limited to kmalloc on
demand. Add a compile-time selector so monitors can choose among three
strategies:

  DA_ALLOC_AUTO   (default) - kmalloc per object on the monitor path
  DA_ALLOC_POOL             - pre-allocated fixed-size llist pool;
                              selected by defining DA_MON_POOL_SIZE
  DA_ALLOC_MANUAL           - caller pre-inserts storage; framework
                              only links the target field

The pool strategy uses a lock-free llist (cmpxchg, no spinlock) so
pool release is safe from RCU callback context without acquiring a
lock. Moving allocation before the measurement window also prevents
kmalloc latency.

nomiss is updated to DA_ALLOC_MANUAL.

Suggested-by: Gabriele Monaco <gmonaco@redhat.com>
Signed-off-by: Wen Yang <wen.yang@linux.dev>
---
 include/rv/da_monitor.h                  | 247 +++++++++++++++++++----
 include/rv/ha_monitor.h                  |   6 +
 kernel/trace/rv/monitors/nomiss/nomiss.c |   6 +-
 3 files changed, 221 insertions(+), 38 deletions(-)

diff --git a/include/rv/da_monitor.h b/include/rv/da_monitor.h
index 34b8fba9ecd4..9c9acc123e3b 100644
--- a/include/rv/da_monitor.h
+++ b/include/rv/da_monitor.h
@@ -14,7 +14,56 @@
 #ifndef _RV_DA_MONITOR_H
 #define _RV_DA_MONITOR_H
 
+/*
+ * Allocation strategies for RV_MON_PER_OBJ monitors.
+ *
+ * Select the strategy with a single define before including this header:
+ *
+ *   #define DA_MON_POOL_SIZE N          - pool mode; N pre-allocated slots.
+ *                                         Implies DA_ALLOC_POOL automatically.
+ *   #define DA_MON_ALLOCATION_STRATEGY \
+ *           DA_ALLOC_MANUAL             - manual mode (see below).
+ *   (neither)                           - auto mode (default).
+ *
+ * Do not define both DA_MON_POOL_SIZE and DA_MON_ALLOCATION_STRATEGY.
+ *
+ * DA_ALLOC_AUTO   - lock-free kmalloc on the hot path; unbounded capacity.
+ * DA_ALLOC_POOL   - pre-allocated fixed-size pool; set by defining DA_MON_POOL_SIZE.
+ * DA_ALLOC_MANUAL - caller inserts storage before da_handle_start_event();
+ *                   the framework only links the target field.
+ */
+#define DA_ALLOC_AUTO   0
+#define DA_ALLOC_POOL   1
+#define DA_ALLOC_MANUAL 2
+
+#ifdef DA_MON_POOL_SIZE
+#ifdef DA_MON_ALLOCATION_STRATEGY
+#error "Define only one of DA_MON_POOL_SIZE or DA_MON_ALLOCATION_STRATEGY"
+#endif
+#if DA_MON_POOL_SIZE == 0
+#error "DA_MON_POOL_SIZE must be non-zero"
+#endif
+#define DA_MON_ALLOCATION_STRATEGY DA_ALLOC_POOL
+#endif
+
+#ifndef DA_MON_ALLOCATION_STRATEGY
+#define DA_MON_ALLOCATION_STRATEGY DA_ALLOC_AUTO
+#endif
+
+/*
+ * Provide a zero default so da_monitor_init() can reference
+ * DA_MON_POOL_SIZE in a plain C if() without an #if guard; the
+ * compiler eliminates the dead branch.
+ */
+#ifndef DA_MON_POOL_SIZE
+#if DA_MON_ALLOCATION_STRATEGY == DA_ALLOC_POOL
+#error "DA_ALLOC_POOL requires DA_MON_POOL_SIZE to be defined and non-zero"
+#endif
+#define DA_MON_POOL_SIZE 0
+#endif
+
 #include <rv/automata.h>
+#include <linux/llist.h>
 #include <linux/rv.h>
 #include <linux/stringify.h>
 #include <linux/bug.h>
@@ -66,6 +115,16 @@ static struct rv_monitor rv_this;
 #define da_monitor_sync_hook()
 #endif
 
+/*
+ * Per-object teardown hook, called after da_monitor_reset_all() +
+ * da_monitor_sync_hook() and before hash_del_rcu() for each entry.
+ * All HA timer callbacks have completed at this point.
+ * Define before including this header.  Default: no-op.
+ */
+#ifndef da_extra_cleanup
+#define da_extra_cleanup(da_mon)
+#endif
+
 /*
  * Type for the target id, default to int but can be overridden.
  * A long type can work as hash table key (PER_OBJ) but will be downgraded to
@@ -404,6 +463,12 @@ struct da_monitor_storage {
 	union rv_task_monitor rv;
 	struct hlist_node node;
 	struct rcu_head rcu;
+	/*
+	 * Mutually exclusive with rcu: rcu is live during the RCU callback
+	 * flight; free_node when the slot is in da_pool_free_list.
+	 * Present in all monitors to avoid #if-gating the pool helpers.
+	 */
+	struct llist_node free_node;
 };
 
 #ifndef DA_MONITOR_HT_BITS
@@ -495,18 +560,6 @@ static inline da_id_type da_get_id(struct da_monitor *da_mon)
 	return container_of(da_mon, struct da_monitor_storage, rv.da_mon)->id;
 }
 
-/*
- * da_create_or_get - create the per-object storage if not already there
- *
- * This needs a lookup so should be guarded by RCU, the condition is checked
- * directly in da_create_storage()
- */
-static inline void da_create_or_get(da_id_type id, monitor_target target)
-{
-	guard(rcu)();
-	da_create_storage(id, target, da_get_monitor(id, target));
-}
-
 /*
  * da_fill_empty_storage - store the target in a pre-allocated storage
  *
@@ -537,15 +590,79 @@ static inline monitor_target da_get_target_by_id(da_id_type id)
 	return mon_storage->target;
 }
 
+/*
+ * Lock-free llist (cmpxchg) rather than kmem_cache/mempool: on
+ * PREEMPT_RT spinlock_t becomes a sleeping lock, which is forbidden
+ * in the rcuc kthread context where RCU callbacks run.
+ *
+ * Multiple producers (any context, any CPU) call llist_add; a single
+ * consumer (llist_del_first, serialised by the monitor's start lock)
+ * needs no additional synchronisation.
+ *
+ * Per-TU statics: each PER_OBJ monitor gets its own pool instance;
+ * da_pool_storage and da_pool_free_list are NULL/empty and the pool
+ * paths are dead code for non-pool monitors.
+ */
+static struct da_monitor_storage *da_pool_storage;
+static LLIST_HEAD(da_pool_free_list);
+
+static void da_pool_return_cb(struct rcu_head *head)
+{
+	struct da_monitor_storage *ms =
+		container_of(head, struct da_monitor_storage, rcu);
+
+	llist_add(&ms->free_node, &da_pool_free_list);
+}
+
+/*
+ * da_create_pool_storage - pop a free pool slot and insert it into the hash.
+ *
+ * Returns the new da_monitor, or NULL if the pool is exhausted.  Finding
+ * an existing entry for the same id fires WARN_ON_ONCE (double-start bug).
+ *
+ * Caller must hold an RCU read-side CS and the monitor's serialisation lock.
+ */
+static inline struct da_monitor *
+da_create_pool_storage(da_id_type id, monitor_target target,
+		       struct da_monitor *da_mon)
+{
+	struct da_monitor_storage *mon_storage, *existing;
+	struct llist_node *node;
+
+	if (da_mon)
+		return da_mon;
+
+	node = llist_del_first(&da_pool_free_list);
+	if (!node)
+		return NULL;
+	mon_storage = llist_entry(node, struct da_monitor_storage, free_node);
+
+	mon_storage->id = id;
+	mon_storage->target = target;
+
+	/*
+	 * The caller's serialization lock ensures llist_del_first() is
+	 * single-consumer, so no concurrent start for the same id is possible.
+	 * Reaching here indicates a programming error (double-start for the
+	 * same pid).
+	 */
+	existing = __da_get_mon_storage(id);
+	if (WARN_ON_ONCE(existing)) {
+		llist_add(&mon_storage->free_node, &da_pool_free_list);
+		return NULL;
+	}
+	hash_add_rcu(da_monitor_ht, &mon_storage->node, id);
+	return &mon_storage->rv.da_mon;
+}
+
 /*
  * da_destroy_storage - destroy the per-object storage
  *
- * The caller is responsible to synchronise writers, either with locks or
- * implicitly. For instance, if da_destroy_storage is called at sched_exit and
- * da_create_storage can never occur after that, it's safe to call this without
- * locks.
- * This function includes an RCU read-side critical section to synchronise
- * against da_monitor_destroy().
+ * Pool mode: removes from hash and returns the slot via call_rcu().
+ * Kmalloc mode: removes from hash and frees via kfree_rcu().
+ *
+ * Includes an RCU read-side critical section to synchronise against
+ * da_monitor_destroy().
  */
 static inline void da_destroy_storage(da_id_type id)
 {
@@ -558,7 +675,10 @@ static inline void da_destroy_storage(da_id_type id)
 		return;
 	da_monitor_reset_hook(&mon_storage->rv.da_mon);
 	hash_del_rcu(&mon_storage->node);
-	kfree_rcu(mon_storage, rcu);
+	if (DA_MON_ALLOCATION_STRATEGY == DA_ALLOC_POOL)
+		call_rcu(&mon_storage->rcu, da_pool_return_cb);
+	else
+		kfree_rcu(mon_storage, rcu);
 }
 
 static void __da_monitor_reset_all(void (*reset)(struct da_monitor *))
@@ -581,41 +701,98 @@ static inline void da_monitor_reset_state_all(void)
 	__da_monitor_reset_all(da_monitor_reset_state);
 }
 
+/* Not part of the public API; called only by da_monitor_init(). */
+static inline int __da_monitor_init_pool(unsigned int prealloc_count)
+{
+	unsigned int i;
+
+	da_pool_storage = kcalloc(prealloc_count, sizeof(*da_pool_storage),
+				  GFP_KERNEL);
+	if (!da_pool_storage)
+		return -ENOMEM;
+
+	for (i = 0; i < prealloc_count; i++)
+		llist_add(&da_pool_storage[i].free_node, &da_pool_free_list);
+	return 0;
+}
+
+/*
+ * da_monitor_init - initialise the per-object monitor
+ */
 static inline int da_monitor_init(void)
 {
 	hash_init(da_monitor_ht);
+	if (DA_MON_ALLOCATION_STRATEGY == DA_ALLOC_POOL)
+		return __da_monitor_init_pool(DA_MON_POOL_SIZE);
 	return 0;
 }
 
+/*
+ * da_monitor_destroy - tear down the per-object monitor
+ *
+ * tracepoint_synchronize_unregister() flushes all in-flight tracepoint
+ * handlers and performs synchronize_rcu(), so no RCU reader holds a
+ * reference to any da_monitor_storage after it returns.
+ * da_monitor_reset_all() disables monitoring; combined with
+ * da_monitor_sync_hook() (synchronize_rcu() for HA), no timer callback
+ * can fire or be re-armed after this sequence.
+ *
+ * Pool mode: remaining hash entries are returned to the free list
+ * directly (no call_rcu needed -- see above).  rcu_barrier() drains any
+ * da_pool_return_cb() callbacks queued by earlier da_destroy_storage()
+ * calls before the backing array is freed.
+ */
 static inline void da_monitor_destroy(void)
 {
-	struct da_monitor_storage *mon_storage;
+	struct da_monitor_storage *ms;
 	struct hlist_node *tmp;
 	int bkt;
 
 	tracepoint_synchronize_unregister();
 	da_monitor_reset_all();
 	da_monitor_sync_hook();
-	/*
-	 * This function is called after all probes are disabled and no longer
-	 * pending, we can safely assume no concurrent user.
-	 */
-	hash_for_each_safe(da_monitor_ht, bkt, tmp, mon_storage, node) {
-		hash_del_rcu(&mon_storage->node);
-		kfree(mon_storage);
+
+	hash_for_each_safe(da_monitor_ht, bkt, tmp, ms, node) {
+		da_extra_cleanup(&ms->rv.da_mon);
+		hash_del_rcu(&ms->node);
+		/* No RCU readers remain; skip the grace period. */
+		if (DA_MON_ALLOCATION_STRATEGY == DA_ALLOC_POOL) {
+			llist_add(&ms->free_node, &da_pool_free_list);
+		} else {
+			kfree(ms);
+		}
+	}
+
+	if (DA_MON_ALLOCATION_STRATEGY == DA_ALLOC_POOL) {
+		/*
+		 * Flush any da_pool_return_cb callbacks queued by
+		 * da_destroy_storage() during normal monitor operation.
+		 * After rcu_barrier(), no callback can reference pool slots;
+		 * the backing array is safe to free.
+		 */
+		rcu_barrier();
+		init_llist_head(&da_pool_free_list);
+		kfree(da_pool_storage);
+		da_pool_storage = NULL;
 	}
 }
 
 /*
- * Allow the per-object monitors to run allocation manually, necessary if the
- * start condition is in a context problematic for allocation (e.g. scheduling).
- * In such case, if the storage was pre-allocated without a target, set it now.
+ * da_prepare_storage - allocate or retrieve storage for a monitoring session
+ *
+ * Dispatches to the strategy selected by DA_MON_ALLOCATION_STRATEGY.
+ * Caller must hold an RCU read-side CS.
  */
-#ifdef DA_SKIP_AUTO_ALLOC
-#define da_prepare_storage da_fill_empty_storage
-#else
-#define da_prepare_storage da_create_storage
-#endif /* DA_SKIP_AUTO_ALLOC */
+static inline struct da_monitor *
+da_prepare_storage(da_id_type id, monitor_target target,
+		   struct da_monitor *da_mon)
+{
+	if (DA_MON_ALLOCATION_STRATEGY == DA_ALLOC_POOL)
+		return da_create_pool_storage(id, target, da_mon);
+	if (DA_MON_ALLOCATION_STRATEGY == DA_ALLOC_MANUAL)
+		return da_fill_empty_storage(id, target, da_mon);
+	return da_create_storage(id, target, da_mon);
+}
 
 #endif /* RV_MON_TYPE */
 
diff --git a/include/rv/ha_monitor.h b/include/rv/ha_monitor.h
index 28d3c74cabfc..83199f90afe8 100644
--- a/include/rv/ha_monitor.h
+++ b/include/rv/ha_monitor.h
@@ -365,6 +365,12 @@ static inline bool ha_check_invariant_ns(struct ha_monitor *ha_mon,
 }
 /*
  * ha_invariant_passed_ns - prepare the invariant and return the time since reset
+ *
+ * If the env has not been initialised yet (first entry into a state with an
+ * invariant), anchor the guard clock at the current time so that the full
+ * budget is available from this point.  This preserves the documented
+ * guard->invariant ordering: ha_set_invariant_ns() is always preceded by a
+ * valid guard representation in env_store.
  */
 static inline u64 ha_invariant_passed_ns(struct ha_monitor *ha_mon, enum envs env,
 				   u64 expire, u64 time_ns)
diff --git a/kernel/trace/rv/monitors/nomiss/nomiss.c b/kernel/trace/rv/monitors/nomiss/nomiss.c
index 8ead8783c29f..ac4d334e757f 100644
--- a/kernel/trace/rv/monitors/nomiss/nomiss.c
+++ b/kernel/trace/rv/monitors/nomiss/nomiss.c
@@ -17,8 +17,8 @@
 
 #define RV_MON_TYPE RV_MON_PER_OBJ
 #define HA_TIMER_TYPE HA_TIMER_WHEEL
-/* The start condition is on sched_switch, it's dangerous to allocate there */
-#define DA_SKIP_AUTO_ALLOC
+/* Allocate storage in sched_setscheduler; sched_switch is too hot to alloc. */
+#define DA_MON_ALLOCATION_STRATEGY DA_ALLOC_MANUAL
 typedef struct sched_dl_entity *monitor_target;
 #include "nomiss.h"
 #include <rv/ha_monitor.h>
@@ -214,7 +214,7 @@ static void handle_sys_enter(void *data, struct pt_regs *regs, long id)
 	if (p->policy == SCHED_DEADLINE)
 		da_reset(EXPAND_ID_TASK(p));
 	else if (new_policy == SCHED_DEADLINE)
-		da_create_or_get(EXPAND_ID_TASK(p));
+		da_create_empty_storage(get_entity_id(&p->dl, task_cpu(p), DL_TASK));
 }
 
 static void handle_sched_wakeup(void *data, struct task_struct *tsk)
-- 
2.25.1
Re: [PATCH v4 1/8] rv/da: introduce DA_MON_ALLOCATION_STRATEGY
Posted by Gabriele Monaco 2 weeks, 5 days ago
On Wed, 2026-07-08 at 23:38 +0800, wen.yang@linux.dev wrote:
> From: Wen Yang <wen.yang@linux.dev>
> diff --git a/include/rv/ha_monitor.h b/include/rv/ha_monitor.h
> index 28d3c74cabfc..83199f90afe8 100644
> --- a/include/rv/ha_monitor.h
> +++ b/include/rv/ha_monitor.h
> @@ -365,6 +365,12 @@ static inline bool ha_check_invariant_ns(struct
> ha_monitor *ha_mon,
>  }
>  /*
>   * ha_invariant_passed_ns - prepare the invariant and return the time since
> reset
> + *
> + * If the env has not been initialised yet (first entry into a state with an
> + * invariant), anchor the guard clock at the current time so that the full
> + * budget is available from this point.  This preserves the documented
> + * guard->invariant ordering: ha_set_invariant_ns() is always preceded by a
> + * valid guard representation in env_store.
>   */
>  static inline u64 ha_invariant_passed_ns(struct ha_monitor *ha_mon, enum envs
> env,
>  				   u64 expire, u64 time_ns)

This hunk belongs to a later patch, doesn't it?
Re: [PATCH v4 1/8] rv/da: introduce DA_MON_ALLOCATION_STRATEGY
Posted by Gabriele Monaco 3 weeks, 1 day ago
On Wed, 2026-07-08 at 23:38 +0800, wen.yang@linux.dev wrote:
> From: Wen Yang <wen.yang@linux.dev>
> 
> Per-object DA storage allocation is currently limited to kmalloc on
> demand. Add a compile-time selector so monitors can choose among three
> strategies:
> 
>   DA_ALLOC_AUTO   (default) - kmalloc per object on the monitor path
>   DA_ALLOC_POOL             - pre-allocated fixed-size llist pool;
>                               selected by defining DA_MON_POOL_SIZE
>   DA_ALLOC_MANUAL           - caller pre-inserts storage; framework
>                               only links the target field
> 
> The pool strategy uses a lock-free llist (cmpxchg, no spinlock) so
> pool release is safe from RCU callback context without acquiring a
> lock. Moving allocation before the measurement window also prevents
> kmalloc latency.

Measurement window here is tlob's, remember RV isn't itself a measurement
tool (yet, perhaps).
And isn't this also happening with other methods? We're trying to do
allocation when the monitor starts (so before this measurement window).

I'm a bit puzzled since you're mentioning it many times, when have we
done /allocations/ from RCU callbacks?
We surely do deallocations (kfree_rcu) but allocations are at most in RCU
read-side critical sections and it's perfectly fine to take sleeping
spinlocks there (that's a special kind of sleep under PREEMPT_RT).
Besides I'm not quite sure spinlocks are that bad in RCU callbacks
either (kfree surely takes them).
I'm not sure what you mean here but I don't think deallocation was ever
a problem, was it?

> nomiss is updated to DA_ALLOC_MANUAL.
> 
> Suggested-by: Gabriele Monaco <gmonaco@redhat.com>
> Signed-off-by: Wen Yang <wen.yang@linux.dev>
> ---
>  include/rv/da_monitor.h                  | 247 +++++++++++++++++++----
>  include/rv/ha_monitor.h                  |   6 +
>  kernel/trace/rv/monitors/nomiss/nomiss.c |   6 +-
>  3 files changed, 221 insertions(+), 38 deletions(-)
> 
> diff --git a/include/rv/da_monitor.h b/include/rv/da_monitor.h
> index 34b8fba9ecd4..9c9acc123e3b 100644
> --- a/include/rv/da_monitor.h
> +++ b/include/rv/da_monitor.h
> @@ -14,7 +14,56 @@
>  #ifndef _RV_DA_MONITOR_H
>  #define _RV_DA_MONITOR_H
>  
> +/*
> + * Allocation strategies for RV_MON_PER_OBJ monitors.
> + *
> + * Select the strategy with a single define before including this header:
> + *
> + *   #define DA_MON_POOL_SIZE N          - pool mode; N pre-allocated slots.
> + *                                         Implies DA_ALLOC_POOL
> automatically.
> + *   #define DA_MON_ALLOCATION_STRATEGY \
> + *           DA_ALLOC_MANUAL             - manual mode (see below).
> + *   (neither)                           - auto mode (default).
> + *
> + * Do not define both DA_MON_POOL_SIZE and DA_MON_ALLOCATION_STRATEGY.
> + *
> + * DA_ALLOC_AUTO   - lock-free kmalloc on the hot path; unbounded capacity.
> + * DA_ALLOC_POOL   - pre-allocated fixed-size pool; set by defining
> DA_MON_POOL_SIZE.
> + * DA_ALLOC_MANUAL - caller inserts storage before da_handle_start_event();
> + *                   the framework only links the target field.
> + */
> +#define DA_ALLOC_AUTO   0
> +#define DA_ALLOC_POOL   1
> +#define DA_ALLOC_MANUAL 2
> +
> +#ifdef DA_MON_POOL_SIZE
> +#ifdef DA_MON_ALLOCATION_STRATEGY
> +#error "Define only one of DA_MON_POOL_SIZE or DA_MON_ALLOCATION_STRATEGY"
> +#endif
> +#if DA_MON_POOL_SIZE == 0
> +#error "DA_MON_POOL_SIZE must be non-zero"
> +#endif
> +#define DA_MON_ALLOCATION_STRATEGY DA_ALLOC_POOL
> +#endif

Longer ifdefs should have comments to make them readable, like

  #endif /* DA_MON_POOL_SIZE */

> +
> +#ifndef DA_MON_ALLOCATION_STRATEGY
> +#define DA_MON_ALLOCATION_STRATEGY DA_ALLOC_AUTO
> +#endif
> +
> +/*
> + * Provide a zero default so da_monitor_init() can reference
> + * DA_MON_POOL_SIZE in a plain C if() without an #if guard; the
> + * compiler eliminates the dead branch.
> + */
> +#ifndef DA_MON_POOL_SIZE
> +#if DA_MON_ALLOCATION_STRATEGY == DA_ALLOC_POOL
> +#error "DA_ALLOC_POOL requires DA_MON_POOL_SIZE to be defined and non-zero"
> +#endif
> +#define DA_MON_POOL_SIZE 0
> +#endif

Same here, better to have a comment.

> +
>  #include <rv/automata.h>
> +#include <linux/llist.h>
>  #include <linux/rv.h>
>  #include <linux/stringify.h>
>  #include <linux/bug.h>
> @@ -66,6 +115,16 @@ static struct rv_monitor rv_this;
>  #define da_monitor_sync_hook()
>  #endif
>  
> +/*
> + * Per-object teardown hook, called after da_monitor_reset_all() +
> + * da_monitor_sync_hook() and before hash_del_rcu() for each entry.
> + * All HA timer callbacks have completed at this point.
> + * Define before including this header.  Default: no-op.
> + */
> +#ifndef da_extra_cleanup
> +#define da_extra_cleanup(da_mon)
> +#endif
> +
>  /*
>   * Type for the target id, default to int but can be overridden.
>   * A long type can work as hash table key (PER_OBJ) but will be downgraded to
> @@ -404,6 +463,12 @@ struct da_monitor_storage {
>  	union rv_task_monitor rv;
>  	struct hlist_node node;
>  	struct rcu_head rcu;
> +	/*
> +	 * Mutually exclusive with rcu: rcu is live during the RCU callback
> +	 * flight; free_node when the slot is in da_pool_free_list.
> +	 * Present in all monitors to avoid #if-gating the pool helpers.
> +	 */

I really don't understand much more about it by this comment, perhaps
drop it here and make the separate usages clearer later?

By the way, if they are /really/ mutually exclusive and you want to save
space, why not having them in an anonymous union?

> +	struct llist_node free_node;
>  };
>  
>  #ifndef DA_MONITOR_HT_BITS
> @@ -495,18 +560,6 @@ static inline da_id_type da_get_id(struct da_monitor
> *da_mon)
>  	return container_of(da_mon, struct da_monitor_storage, rv.da_mon)-
> >id;
>  }
>  
> -/*
> - * da_create_or_get - create the per-object storage if not already there
> - *
> - * This needs a lookup so should be guarded by RCU, the condition is checked
> - * directly in da_create_storage()
> - */
> -static inline void da_create_or_get(da_id_type id, monitor_target target)
> -{
> -	guard(rcu)();
> -	da_create_storage(id, target, da_get_monitor(id, target));
> -}
> -
>  /*
>   * da_fill_empty_storage - store the target in a pre-allocated storage
>   *
> @@ -537,15 +590,79 @@ static inline monitor_target
> da_get_target_by_id(da_id_type id)
>  	return mon_storage->target;
>  }
>  
> +/*
> + * Lock-free llist (cmpxchg) rather than kmem_cache/mempool: on
> + * PREEMPT_RT spinlock_t becomes a sleeping lock, which is forbidden
> + * in the rcuc kthread context where RCU callbacks run.

This comment kind of implies we were using a kmem_cache, it's great for
a changelog and helped me understand why you're doing this, but doesn't
belong to the final version as is.

> + *
> + * Multiple producers (any context, any CPU) call llist_add; a single
> + * consumer (llist_del_first, serialised by the monitor's start lock)

Which monitor's start lock? There is no such a thing defined anywhere,
maybe you wanted to say that monitors using this allocation scheme MUST
lock during their start event. This by the way needs to be a global lock
among all instances of the monitor (as you're indeed doing in tlob).

With that in mind, I don't really see how this is better than the
original mempool: you still need to lock. There's nothing wrong in
freeing stuff from RCU callbacks, that's what they're for.

> + * needs no additional synchronisation.
> + *
> + * Per-TU statics: each PER_OBJ monitor gets its own pool instance;
> + * da_pool_storage and da_pool_free_list are NULL/empty and the pool
> + * paths are dead code for non-pool monitors.
> + */

> +static struct da_monitor_storage *da_pool_storage;
> +static LLIST_HEAD(da_pool_free_list);

...

> +++ b/kernel/trace/rv/monitors/nomiss/nomiss.c
> @@ -17,8 +17,8 @@
>  
>  #define RV_MON_TYPE RV_MON_PER_OBJ
>  #define HA_TIMER_TYPE HA_TIMER_WHEEL
> -/* The start condition is on sched_switch, it's dangerous to allocate there
> */
> -#define DA_SKIP_AUTO_ALLOC
> +/* Allocate storage in sched_setscheduler; sched_switch is too hot to alloc.
> */
> +#define DA_MON_ALLOCATION_STRATEGY DA_ALLOC_MANUAL
>  typedef struct sched_dl_entity *monitor_target;
>  #include "nomiss.h"
>  #include <rv/ha_monitor.h>
> @@ -214,7 +214,7 @@ static void handle_sys_enter(void *data, struct pt_regs
> *regs, long id)
>  	if (p->policy == SCHED_DEADLINE)
>  		da_reset(EXPAND_ID_TASK(p));
>  	else if (new_policy == SCHED_DEADLINE)
> -		da_create_or_get(EXPAND_ID_TASK(p));
> +		da_create_empty_storage(get_entity_id(&p->dl, task_cpu(p),
> DL_TASK));

I'm starting to doubt this is the right thing to do. We do have the
target (p) and that function doesn't check if the id already has a
storage (which shouldn't happen but well, doesn't hurt checking).

This simplification is probably just not worth it, and doesn't look
related to the rest of the change.

Thanks,
Gabriele