Re: [PATCH v6 6/9] rv: Add tlob hybrid automaton monitor

[Gabriele Monaco <gmonaco@redhat.com>]

On Fri, 2026-08-21 at 00:45 +0800, wen.yang@linux.dev wrote:
> From: Wen Yang <wen.yang@linux.dev>
> 
> +
> +Description
> +-----------
> +
> +The tlob monitor tracks per-task elapsed wall-clock time (CLOCK_MONOTONIC,
> +spanning running, waiting, and sleeping states) and reports a violation when
> +the monitored task exceeds a configurable per-invocation budget threshold.
> +
> +The monitor implements a four-state hybrid automaton with a single clock
> +environment variable ``clk_elapsed``.  The clock invariant
> +``clk_elapsed < BUDGET_NS()`` is active in the ``running``, ``waiting``, and
> +``sleeping`` states (``stopped`` has no invariant, hence no timer); when it
> +is violated the HA timer fires and the framework emits ``error_env_tlob``
> +then calls ``da_monitor_reset()`` automatically::
> +
> +                                           | (initial)
> +                                           v
> +           +--------------+           +----------+
> +           |   running    | --------> |  stopped |
> +        |->+--------------+ <-------- +----------+
> +   switch_in  preempt  sleep

This triggered my OCD ;)
Please fix the arrow waiting -> running:

+           +--------------+           +----------+
+        +->|   running    | --------> |  stopped |
+        |  +--------------+ <-------- +----------+
+   switch_in  preempt  sleep

> +        |        |       |
> +        |        |       |
> +        |        v       v
> +        +---------+  +---------+
> +        | waiting |  | sleeping|
> +        +---------+  +---------+
> +             ^            v
> +             |   wakeup   |
> +             |            |
> +             +------------+
> +
> +  A fourth state, ``stopped``, has no clock invariant (hence no timer).
> +  ``running`` reaches it on ``stop`` (``tlob_stop_task()``, window ended,
> +  per-task state parked rather than freed) and returns to ``running`` on
> +  ``start`` (``tlob_start_task()`` restarting the same task's parked
> +  window).

So you define a pseudo-state "parked" that is essentially stopped but
after monitoring started (we allocated) and before the task exits (we
deallocate), is that right?
It looks kind of like an implementation detail rather than something
related to your model: there isn't any parked state in the model and you
don't need one.
If you really want the concept of parked to explain how you handle
allocation, perhaps you could make it clear in the code only.
For instance (if I got it right) when describing the
tlob_task_state->stopping you could say: tasks with this flag set are
"parked" until deallocation.

> +
> +  Key transitions:
> +    running  --(sleep)------> sleeping   (task blocks waiting for a resource)
> +    running  --(preempt)----> waiting    (task preempted, back in runqueue)
> +    sleeping --(wakeup)-----> waiting    (resource available, enters
> runqueue)
> +    waiting  --(switch_in)--> running    (scheduler picks task, back on CPU)
> +    running  --(stop)-------> stopped    (tlob_stop_task(): window ended,
> parked)
> +    stopped  --(start)------> running    (tlob_start_task(): window
> restarted)
> +
> +  ``tlob_start_task()`` calls ``da_handle_start_run_event(task->pid, ws,
> start_tlob)``.
> +  The ``start_tlob`` edge goes ``stopped`` -> ``running`` for both a fresh
> +  allocation (the initial state is ``stopped``) and a parked window's
> restart;
> +  there is no ``start`` self-loop on ``running`` (a running task's START is
> +  rejected with ``-EALREADY``).  The transition triggers
> ``ha_setup_invariants()``,
> +  which anchors ``clk_elapsed`` and arms the budget timer automatically.
> +  ``tlob_stop_task()`` cancels the HA timer synchronously
> +  via ``ha_cancel_timer_sync()``, then dispatches the ``stop_tlob`` event
> +  (running -> stopped) instead of resetting the monitor: the per-task state
> +  is parked, not freed, so a later ``tlob_start_task()`` call for the same
> +  task can restart it without reallocating.  Final teardown (task exit,
> +  uprobe unbind, or monitor disable) is what actually calls
> +  ``da_monitor_reset()`` and frees the state.

All allocation or broadly implementation details don't belong here. I
believe it's already clear from the model, but you may still stress that a
task can start another measuring window after the previous was stopped.

Being general about implementation in your documentation saves you some
headache while keeping that in sync (and I believe AIs make this problem
worse, by the way).

...
> +Kernel API
> +----------
> +
> +``tlob_start_task`` and ``tlob_stop_task`` are the implementation-level
> +functions called by the uprobe entry/exit handlers; the interface is
> +driven from userspace.
> +
> +.. kernel-doc:: kernel/trace/rv/monitors/tlob/tlob.c
> +   :functions: tlob_start_task tlob_stop_task

I remember mentioning this, there's no real kernel API, those functions
aren't exported nor meant to be called by anyone besides the model. I
would remove this section altogether.

...
> +++ b/kernel/trace/rv/monitors/tlob/Kconfig
> @@ -0,0 +1,12 @@
> +# SPDX-License-Identifier: GPL-2.0-only
> +#
> +config RV_MON_TLOB
> +	bool "tlob monitor"
> +	depends on RV && UPROBES && HIGH_RES_TIMERS
> +	select HA_MON_EVENTS_ID
> +	select RV_UPROBE
> +	help
> +	  Enable the tlob (task latency over budget) hybrid-automaton RV
> +	  monitor.  tlob tracks per-task elapsed wall-clock time across a
> +	  user-delimited code section and emits error_env_tlob when the

emits a violation when...

> +	  elapsed time exceeds a configurable per-invocation budget.
> diff --git a/kernel/trace/rv/monitors/tlob/tlob.c
> b/kernel/trace/rv/monitors/tlob/tlob.c
> new file mode 100644
> index 000000000000..08b1bee884cc
> --- /dev/null
> +++ b/kernel/trace/rv/monitors/tlob/tlob.c
...
> +struct tlob_task_state {
> +	struct task_struct	*task;		/* via get_task_struct */
> +	u64			threshold_ns;	/* budget in nanoseconds */
> +
> +	/*
> +	 * Per-window: 1 = this window ended (stop or timer expiry).  Blocks
> +	 * timer re-arm in ha_setup_invariants(); cleared on restart.
> +	 */
> +	atomic_t		stopping;
> +
> +	/*
> +	 * Per-task, one-shot: final teardown has claimed this slot; never
> +	 * reset (a window can end and restart, the task cannot).  atomic_t
> +	 * so cmpxchg is well-defined on every arch.
> +	 */
> +	atomic_t		destroying;
> +
> +	bool			budget_exceeded;
> +
> +	/*
> +	 * Opaque owner: the binding that started this task.  Set once on
> +	 * fresh allocation (NULL for callers with no binding), cleared by
> +	 * tlob_unbind_reap() for an active task whose binding is removed.
> +	 * Immutable elsewhere.  Protected by tlob_ws_lock.
> +	 */
> +	void			*binding;

Does this really need to be opaque? You're casting it anyway so I don't
see why it can't be a struct tlob_uprobe_binding * to begin with.

> +	/*
> +	 * Linked into binding->started_list for the whole lifetime (not just
> +	 * while parked) so unbind reaping finds parked and active tasks.
> +	 * Protected by tlob_ws_lock.
> +	 */
> +	struct list_head	started_node;
> +
> +	/* Serialises accs_ns[]; held briefly (hardirq-safe). */
> +	raw_spinlock_t		entry_lock;
> +	u64			accs_ns[TLOB_ACC_MAX]; /* per-state elapsed
> ns */
> +	ktime_t			last_ts;
> +
> +	struct rcu_head		rcu;
> +};
...
> +
> +/*
> + * Unlink ws from its binding's started_list before returning it to the pool.
> + * ws->binding is left stale: the next tlob_ws_alloc() memsets it, and the
> + * restart path checks destroying first.  Idempotent (list_del_init no-op).
> + */
> +static inline void tlob_detach_from_binding(struct tlob_task_state *ws)
> +{
> +	if (!ws->binding)
> +		return;

Should you access also the binding field under the lock?
And perhaps be set to NULL also here?

> +	guard(spinlock)(&tlob_ws_lock);
> +	list_del_init(&ws->started_node);
> +}
...
> +
> +/**
> + * tlob_start_task - begin monitoring @task with budget @threshold_ns ns.
> + * @task:         Task to monitor; may be current or another task.
> + * @threshold_ns: Budget in ns, in [1000, TLOB_MAX_THRESHOLD_NS].
> + * @binding:      Opaque owner, recorded on fresh allocation and checked for
> + *                an exact match on restart; NULL for callers that never
> + *                restart a parked window.
> + *
> + * Allocates a fresh entry if @task has none, or restarts a parked entry in
> + * place when @binding matches (see tlob.dot: "start" fires from both
> + * running and stopped).
> + *
> + * Returns 0, -ENODEV, -ERANGE, -EALREADY, -ESRCH, or -ENOSPC (fresh start
> + * past pool capacity).
> + */
> +static int tlob_start_task(struct task_struct *task, u64 threshold_ns, void
> *binding)
> +{
> +	struct tlob_task_state *ws;
> +
> +	if (!da_monitor_enabled())
> +		return -ENODEV;
> +
> +	if (threshold_ns < TLOB_MIN_THRESHOLD_NS ||
> +	    threshold_ns > TLOB_MAX_THRESHOLD_NS)
> +		return -ERANGE;
> +
> +	/* Serialise duplicate-check + pool-slot claim; see tlob_ws_lock. */
> +	guard(spinlock)(&tlob_ws_lock);
> +
> +	/*
> +	 * da_get_target_by_id() uses hash_for_each_possible_rcu(), which
> +	 * requires an RCU read-side critical section.
> +	 */
> +	scoped_guard(rcu) {
> +		ws = da_get_target_by_id(task->pid);
> +		if (ws) {
> +			if (!atomic_read(&ws->stopping))
> +				return -EALREADY;
> +			if (atomic_read(&ws->destroying))
> +				return -ESRCH;
> +			/*
> +			 * Exact match only.  An orphaned parked ws (binding
> +			 * cleared while active, then parked) is not adopted:
> +			 * that would need re-linking into the new binding's
> +			 * started_list.  Accepted gap; the slot is reclaimed
> +			 * at task exit.
> +			 */
> +			if (ws->binding != binding)
> +				return -EALREADY;
> +
> +			/* Restart in place: same slot, hash entry, task ref,
> list node. */
> +			ws->threshold_ns = threshold_ns;
> +			WRITE_ONCE(ws->budget_exceeded, false);
> +			memset(ws->accs_ns, 0, sizeof(ws->accs_ns));
> +			ws->last_ts = ktime_get();
> +
> +			/*
> +			 * Keep stopping set: __tlob_acc() gates out sched
> +			 * events until ha_setup_invariants() clears it after
> +			 * the state is running_tlob.  Clearing here would
> let
> +			 * events hit stopped_tlob (INVALID transitions).
> +			 */
> +
> +			/* Only failure here: monitor disabled since the
> check above. */
> +			if (!da_handle_start_run_event(task->pid, ws,
> start_tlob))
> +				return -ENODEV;
> +			return 0;
> +		}
> +	}
> +
> +	ws = tlob_ws_alloc();
> +	if (!ws)
> +		return -ENOSPC;
> +
> +	ws->task = task;
> +	get_task_struct(task);
> +	ws->threshold_ns = threshold_ns;
> +	ws->last_ts = ktime_get();
> +	raw_spin_lock_init(&ws->entry_lock);
> +	ws->binding = binding;
> +	if (binding)
> +		list_add_tail(&ws->started_node,
> +			      &((struct tlob_uprobe_binding *)binding)-
> >started_list);

Why do you add to the list and then remove if start failed? Is the
binding needed when the model does it's job?
Cannot you just do all that after only if the start passed?

Also, can binding really be NULL ?

> +
> +	/* Dispatch failed (pool exhausted or monitor disabled): unwind the
> slot. */
> +	if (!da_handle_start_run_event(task->pid, ws, start_tlob)) {
> +		if (binding)
> +			list_del_init(&ws->started_node);
> +		/* stopping=1 short-circuits the reset hook; destroy before
> freeing ws. */
> +		atomic_set(&ws->stopping, 1);
> +		da_destroy_storage(task->pid);
> +		put_task_struct(task);
> +		tlob_ws_direct_return(ws);
> +		return -ENOSPC;
> +	}
> +
> +	return 0;
> +}
> +
> +/**
> + * tlob_stop_task - end the current monitoring window for @task.
> + * @task: Task to stop.
> + * @binding: Opaque owner; must match ws->binding to end a normal (uprobe)
> + *           window.  NULL (task exit) skips the check.
> + *
> + * Ends the window (dispatches "stop") but does NOT free the entry: it stays
> + * parked so a later tlob_start_task() can restart it.  Call
> + * tlob_destroy_task() once @task will never restart.
> + *
> + * cmpxchg on stopping (0->1) under RCU claims ownership; the winner cancels
> + * the timer synchronously.
> + *
> + * Returns 0, -EOVERFLOW (budget exceeded), -ESRCH (not monitored),
> + * -EAGAIN (window already ended), or -EALREADY (owned by another binding).
> + */
> +static int tlob_stop_task(struct task_struct *task, void *binding)
> +{
> +	struct ha_monitor *ha_mon;
> +	struct tlob_task_state *ws;
> +	bool budget_exceeded;
> +
> +	scoped_guard(rcu) {
> +		ha_mon = ha_get_monitor(task->pid, NULL);
> +		if (!ha_mon)
> +			return -ESRCH;
> +
> +		ws = ha_get_target(ha_mon);
> +		if (WARN_ON_ONCE(!ws))
> +			return -ESRCH;
> +
> +		/* Only the binding that opened the window may end it; NULL
> +		 * (task exit) skips the check.  Symmetric with the restart
> +		 * check in tlob_start_task(). */

The format of this comment is wrong (break the line after /* on
multi-line comments).

> +		if (binding && ws->binding != binding)
> +			return -EALREADY;
> +
> +		/* cmpxchg (0->1) claims the window under RCU; _release pairs
> +		 * with the acquire in ha_setup_invariants(). */

Same here and probably somewhere else, please check around.

> +		if (atomic_cmpxchg_release(&ws->stopping, 0, 1) != 0)
> +			return -EAGAIN;
> +
> +		/*
> +		 * ws may be destroyed concurrently (unbind -> call_rcu), so
> +		 * keep its access under RCU; dispatch re-looks-up under RCU.
> +		 */

This is the correct format.
Although I wonder: if we need a multi-line comment on each line, aren't
we perhaps overdoing it? cmpxchg (0->1) is documented at the function
level, it's probably enough to leave it there.

Also this specific comment has little to do with the lines that come
after. Prefer function level documentation where possible.
If a function is very large (and you're convinced that's fine),
you can document some non-trivial steps as brief as possible.

> +		ha_cancel_timer_sync(ha_mon);
> +		budget_exceeded = READ_ONCE(ws->budget_exceeded);
> +	}
> +
> +	/* running -> stopped: no reset or destroy, the entry stays parked.
> */
> +	da_handle_event(task->pid, NULL, stop_tlob);
> +
> +	return budget_exceeded ? -EOVERFLOW : 0;
> +}
> +
> +/*
> + * tlob_destroy_task - final teardown for @task's entry: frees the pool slot,

Double line break after the : and continue with the long description.
It's fine not writing a full-blown kernel-doc, but you can do better
here (exactly like you do in tlob_unbind_reap).

> + * drops the task_struct ref, removes the hash entry, whether active or
> parked.
> + * Idempotent via the destroying cmpxchg (same pattern as
> tlob_extra_cleanup()).
> + * Callers must end the window first (see handle_sched_process_exit()).
> + */
> +static void tlob_destroy_task(struct task_struct *task)
> +{
> +	struct ha_monitor *ha_mon;
> +	struct tlob_task_state *ws;
> +
> +	scoped_guard(rcu) {
> +		ha_mon = ha_get_monitor(task->pid, NULL);
> +		if (!ha_mon)
> +			return;
> +		ws = ha_get_target(ha_mon);
> +		if (WARN_ON_ONCE(!ws))
> +			return;
> +		if (atomic_cmpxchg_release(&ws->destroying, 0, 1) != 0)
> +			return;
> +	}
> +
> +	tlob_detach_from_binding(ws);
> +
> +	/* Force the window ended: @task may never have reached STOP or a
> timer. */
> +	atomic_set(&ws->stopping, 1);
> +	ha_cancel_timer_sync(ha_mon);
> +
> +	scoped_guard(rcu) {
> +		da_monitor_reset(&ha_mon->da_mon);
> +	}
> +	da_destroy_storage(task->pid);
> +
> +	put_task_struct(ws->task);
> +	call_rcu(&ws->rcu, tlob_ws_return_cb);
> +}
> +
> +static int tlob_uprobe_entry_handler(struct uprobe_consumer *self,
> +				     struct pt_regs *regs, __u64 *data)
> +{
> +	struct tlob_uprobe_binding *b =
> +		container_of(self, struct tlob_uprobe_binding,
> start_probe.uc);
> +
> +	tlob_start_task(current, b->threshold_ns, b);
> +	return 0;
> +}
> +
> +static int tlob_uprobe_stop_handler(struct uprobe_consumer *self,
> +				    struct pt_regs *regs, __u64 *data)
> +{
> +	struct tlob_uprobe_binding *b =
> +		container_of(self, struct tlob_uprobe_binding,
> stop_probe.uc);
> +
> +	tlob_stop_task(current, b);
> +	return 0;
> +}
> +
> +/*
> + * Register start + stop entry uprobes for a binding.
> + * Called with tlob_uprobe_mutex held.
> + */
> +static int tlob_add_uprobe(u64 threshold_ns, const char *binpath,
> +			   loff_t offset_start, loff_t offset_stop)
> +{
> +	struct tlob_uprobe_binding *tmp_b;
> +	char pathbuf[TLOB_MAX_PATH];
> +	struct inode *inode;
> +	struct path path __free(path_put) = {};
> +	char *canon;
> +	int ret;
> +
> +	if (binpath[0] != '/')
> +		return -EINVAL;
> +
> +	struct tlob_uprobe_binding *b __free(kfree) = kzalloc_obj(*b,
> GFP_KERNEL);
> +	if (!b)
> +		return -ENOMEM;
> +
> +	b->threshold_ns = threshold_ns;
> +	b->offset_start = offset_start;
> +	b->offset_stop  = offset_stop;
> +	INIT_LIST_HEAD(&b->started_list);
> +
> +	ret = kern_path(binpath, LOOKUP_FOLLOW, &path);
> +	if (ret)
> +		return ret;
> +
> +	if (!d_is_reg(path.dentry))
> +		return -EINVAL;
> +
> +	inode = d_real_inode(path.dentry);
> +
> +	/* Reject duplicate start offset for the same binary inode. */
> +	list_for_each_entry(tmp_b, &tlob_uprobe_list, list) {
> +		if (tmp_b->offset_start == offset_start &&
> +		    rv_uprobe_is_registered(&tmp_b->start_probe) &&
> +		    d_real_inode(tmp_b->start_probe.path.dentry) == inode)
> +			return -EEXIST;
> +	}
> +
> +	canon = d_path(&path, pathbuf, sizeof(pathbuf));
> +	if (IS_ERR(canon))
> +		return PTR_ERR(canon);
> +	strscpy(b->binpath, canon, sizeof(b->binpath));
> +
> +	b->start_probe.uc.handler = tlob_uprobe_entry_handler;
> +	ret = rv_uprobe_register(b->binpath, offset_start, &b->start_probe);
> +	if (ret)
> +		return ret;
> +
> +	b->stop_probe.uc.handler = tlob_uprobe_stop_handler;
> +	ret = rv_uprobe_register(b->binpath, offset_stop, &b->stop_probe);
> +	if (ret) {
> +		rv_uprobe_unregister(&b->start_probe);
> +		return ret;
> +	}
> +
> +	/* NOT "b = no_free_ptr(b)": the re-assignment would free the live
> node. */

This comment feels like an AI tried the wrong way to use no_free_ptr and
left it not to make the same mistake again, we don't need it.

> +	list_add_tail(&no_free_ptr(b)->list, &tlob_uprobe_list);
> +	return 0;
> +}
> +
> +/*
> + * tlob_unbind_reap - detach every task @b started, destroy the parked ones.
> + *
> + * Caller must have unregistered @b's uprobes and called rv_uprobe_sync():
> + * no start/stop can then be in flight for @b, so started_list is safe to
> + * walk.  Active tasks are detached (binding cleared) and left running,
> + * matching unbind behaviour today; parked tasks are destroyed, or their
> + * pool slot leaks until the task next exits.
> + */
> +static void tlob_unbind_reap(struct tlob_uprobe_binding *b)
> +{
> +	struct tlob_task_state *ws, *tmp;
> +	LIST_HEAD(to_destroy);
> +
> +	scoped_guard(spinlock, &tlob_ws_lock) {
> +		list_for_each_entry_safe(ws, tmp, &b->started_list,
> started_node) {
> +			list_del_init(&ws->started_node);
> +			ws->binding = NULL;
> +			if (atomic_read(&ws->stopping))
> +				list_add_tail(&ws->started_node,
> &to_destroy);
> +		}
> +	}
> +
> +	list_for_each_entry_safe(ws, tmp, &to_destroy, started_node) {
> +		list_del_init(&ws->started_node);
> +		tlob_destroy_task(ws->task);
> +	}
> +}
> +
> +static int tlob_remove_uprobe_by_key(loff_t offset_start, const char
> *binpath)
> +{
> +	struct tlob_uprobe_binding *b, *tmp;
> +	struct path remove_path;
> +	struct inode *inode;
> +	int ret;
> +
> +	ret = kern_path(binpath, LOOKUP_FOLLOW, &remove_path);
> +	if (ret)
> +		return ret;
> +
> +	inode = d_real_inode(remove_path.dentry);
> +
> +	ret = -ENOENT;
> +	list_for_each_entry_safe(b, tmp, &tlob_uprobe_list, list) {
> +		if (b->offset_start != offset_start)
> +			continue;
> +		if (d_real_inode(b->start_probe.path.dentry) != inode)
> +			continue;
> +		list_del(&b->list);
> +		/*
> +		 * rv_uprobe_sync() may sleep; list_del() already made the
> +		 * binding invisible to new readers.
> +		 */
> +		rv_uprobe_unregister_nosync(&b->start_probe);
> +		rv_uprobe_unregister_nosync(&b->stop_probe);
> +		rv_uprobe_sync();
> +		tlob_unbind_reap(b);
> +		path_put(&b->start_probe.path);
> +		path_put(&b->stop_probe.path);
> +		kfree(b);
> +		ret = 0;
> +		break;
> +	}
> +
> +	path_put(&remove_path);

Just for consistency I would use __free(path_put) also for this. But you
don't have to if you prefer this way.

> +	return ret;
> +}
...
> +/*
> + * Parse "p PATH:OFFSET_START OFFSET_STOP threshold=NS".
> + * PATH may contain ':'; the last ':' separates path from offset.
> + * Returns 0, -EINVAL, or -ERANGE.
> + */
> +VISIBLE_IF_KUNIT int tlob_parse_uprobe_line(char *buf, u64 *thr_out,
> +					    char **path_out,
> +					    loff_t *start_out, loff_t
> *stop_out)

These VISIBLE_IF_KUNIT stuff are left from the previous implementation
and removed from the KUnit patch, they shouldn't be here.

> +{
> +	unsigned long long thr = 0, stop_val = 0;
> +	long long start_val;
> +	char *p, *path_token, *token, *colon;
> +	bool got_stop = false, got_thr = false;
> +	int n;
> +
> +	/* Must start with "p " */
> +	if (buf[0] != 'p' || buf[1] != ' ')
> +		return -EINVAL;
> +
> +	p = buf + 2;
> +	while (*p == ' ')
> +		p++;
> +
> +	/* First space-delimited token is PATH:OFFSET_START */
> +	path_token = strsep(&p, " \t");
> +	if (!path_token || !*path_token)
> +		return -EINVAL;
> +
> +	/* Split at last ':' to handle paths that contain ':'. */
> +	colon = strrchr(path_token, ':');
> +	if (!colon || colon - path_token < 2)
> +		return -EINVAL;
> +	*colon = '\0';
> +
> +	if (path_token[0] != '/')
> +		return -EINVAL;
> +
> +	n = 0;
> +	if (sscanf(colon + 1, "%lli%n", &start_val, &n) != 1 || n == 0)
> +		return -EINVAL;
> +	if (start_val < 0)
> +		return -EINVAL;
> +
> +	/* Remaining tokens: OFFSET_STOP threshold=NS */
> +	while (p && (token = strsep(&p, " \t")) != NULL) {
> +		if (!*token)
> +			continue;
> +		if (strncmp(token, "threshold=", 10) == 0) {
> +			if (kstrtoull(token + 10, 0, &thr))
> +				return -EINVAL;
> +			if (thr < TLOB_MIN_THRESHOLD_NS || thr >
> TLOB_MAX_THRESHOLD_NS)
> +				return -ERANGE;
> +			got_thr = true;
> +		} else if (!got_stop) {
> +			long long sv;
> +
> +			n = 0;
> +			if (sscanf(token, "%lli%n", &sv, &n) != 1 || n == 0)
> +				return -EINVAL;
> +			if (sv < 0)
> +				return -EINVAL;
> +			stop_val = (unsigned long long)sv;
> +			got_stop = true;
> +		} else {
> +			return -EINVAL;
> +		}
> +	}
> +
> +	if (!got_stop || !got_thr)
> +		return -EINVAL;
> +	if (start_val == (long long)stop_val)
> +		return -EINVAL;
> +
> +	*thr_out   = thr;
> +	*path_out  = path_token;
> +	*start_out = (loff_t)start_val;
> +	*stop_out  = (loff_t)stop_val;
> +	return 0;
> +}
> +EXPORT_SYMBOL_IF_KUNIT(tlob_parse_uprobe_line);

Same with these.

> +
> +/*
> + * Parse "-PATH:OFFSET_START" (ftrace uprobe_events removal convention).
> + */
> +VISIBLE_IF_KUNIT int tlob_parse_remove_line(char *buf, char **path_out,
> +					    loff_t *start_out)

And here.

> +{
> +	char *binpath, *colon;
> +	long long off;
> +	int n = 0;
> +
> +	if (buf[0] != '-')
> +		return -EINVAL;
> +	binpath = buf + 1;
> +	if (binpath[0] != '/')
> +		return -EINVAL;
> +	colon = strrchr(binpath, ':');
> +	if (!colon || colon - binpath < 2)
> +		return -EINVAL;
> +	*colon = '\0';
> +	if (sscanf(colon + 1, "%lli%n", &off, &n) != 1 || n == 0)
> +		return -EINVAL;
> +	if (off < 0)
> +		return -EINVAL;
> +	*path_out  = binpath;
> +	*start_out = (loff_t)off;
> +	return 0;
> +}
> +EXPORT_SYMBOL_IF_KUNIT(tlob_parse_remove_line);

And here.

> +
> +static int tlob_create_or_delete_uprobe(char *buf)
> +{
> +	loff_t offset_start, offset_stop;
> +	u64 threshold_ns;
> +	char *binpath;
> +	int ret;
> +
> +	if (buf[0] == '-') {
> +		ret = tlob_parse_remove_line(buf, &binpath, &offset_start);
> +		if (ret)
> +			return ret;
> +		mutex_lock(&tlob_uprobe_mutex);
> +		ret = tlob_remove_uprobe_by_key(offset_start, binpath);
> +		mutex_unlock(&tlob_uprobe_mutex);

It's probably more readable if you take this locks inside the functions.

I'd just put a guard (not scoped) from the first point where it seems
needed and let it be (e.g. just before list_for_each_entry_safe).
You don't need to be overly precise at the cost of readability since
this isn't a hot path.

> +		return ret;
> +	}
> +	ret = tlob_parse_uprobe_line(buf, &threshold_ns, &binpath,
> +				     &offset_start, &offset_stop);
> +	if (ret)
> +		return ret;
> +	mutex_lock(&tlob_uprobe_mutex);
> +	ret = tlob_add_uprobe(threshold_ns, binpath, offset_start,
> offset_stop);
> +	mutex_unlock(&tlob_uprobe_mutex);

Same here, you can guard-lock before list_for_each_entry.

> +	return ret;
> +}

Implementation looks good otherwise and seems to work as far as I could
test.

Thanks,
Gabriele