@@ -39,7 +39,7 @@ static struct rb_root uprobes_tree = RB_ROOT;
*/
#define no_uprobe_events() RB_EMPTY_ROOT(&uprobes_tree)
-static DEFINE_RWLOCK(uprobes_treelock); /* serialize rbtree access */
+DEFINE_STATIC_PERCPU_RWSEM(uprobes_treelock); /* serialize rbtree access */
#define UPROBES_HASH_SZ 13
/* serialize uprobe->pending_list */
@@ -684,7 +684,7 @@ static void __put_uprobe(struct uprobe *uprobe, bool tree_locked)
bool destroy;
if (!tree_locked)
- write_lock(&uprobes_treelock);
+ percpu_down_write(&uprobes_treelock);
/*
* We might race with find_uprobe()->__get_uprobe() executed
* from inside read-locked uprobes_treelock, which can bump
@@ -708,7 +708,7 @@ static void __put_uprobe(struct uprobe *uprobe, bool tree_locked)
if (destroy && uprobe_is_active(uprobe))
rb_erase(&uprobe->rb_node, &uprobes_tree);
if (!tree_locked)
- write_unlock(&uprobes_treelock);
+ percpu_up_write(&uprobes_treelock);
/*
* Beyond here we don't need RCU protection, we are either the
@@ -816,9 +816,9 @@ static struct uprobe *find_uprobe(struct inode *inode, loff_t offset)
{
struct uprobe *uprobe;
- read_lock(&uprobes_treelock);
+ percpu_down_read(&uprobes_treelock);
uprobe = __find_uprobe(inode, offset);
- read_unlock(&uprobes_treelock);
+ percpu_up_read(&uprobes_treelock);
return uprobe;
}
@@ -1205,7 +1205,7 @@ void uprobe_unregister_batch(struct inode *inode, int cnt, uprobe_consumer_fn ge
up_write(&uprobe->register_rwsem);
}
- write_lock(&uprobes_treelock);
+ percpu_down_write(&uprobes_treelock);
for (i = 0; i < cnt; i++) {
uc = get_uprobe_consumer(i, ctx);
uprobe = uc->uprobe;
@@ -1216,7 +1216,7 @@ void uprobe_unregister_batch(struct inode *inode, int cnt, uprobe_consumer_fn ge
__put_uprobe(uprobe, true);
uc->uprobe = NULL;
}
- write_unlock(&uprobes_treelock);
+ percpu_up_write(&uprobes_treelock);
}
static struct uprobe_consumer *uprobe_consumer_identity(size_t idx, void *ctx)
@@ -1321,7 +1321,7 @@ int uprobe_register_batch(struct inode *inode, int cnt,
}
ret = 0;
- write_lock(&uprobes_treelock);
+ percpu_down_write(&uprobes_treelock);
for (i = 0; i < cnt; i++) {
struct uprobe *cur_uprobe;
@@ -1344,7 +1344,7 @@ int uprobe_register_batch(struct inode *inode, int cnt,
}
}
unlock_treelock:
- write_unlock(&uprobes_treelock);
+ percpu_up_write(&uprobes_treelock);
if (ret)
goto cleanup_uprobes;
@@ -1376,7 +1376,7 @@ int uprobe_register_batch(struct inode *inode, int cnt,
}
cleanup_uprobes:
/* put all the successfully allocated/reused uprobes */
- write_lock(&uprobes_treelock);
+ percpu_down_write(&uprobes_treelock);
for (i = 0; i < cnt; i++) {
uc = get_uprobe_consumer(i, ctx);
@@ -1384,7 +1384,7 @@ int uprobe_register_batch(struct inode *inode, int cnt,
__put_uprobe(uc->uprobe, true);
uc->uprobe = NULL;
}
- write_unlock(&uprobes_treelock);
+ percpu_up_write(&uprobes_treelock);
return ret;
}
@@ -1492,7 +1492,7 @@ static void build_probe_list(struct inode *inode,
min = vaddr_to_offset(vma, start);
max = min + (end - start) - 1;
- read_lock(&uprobes_treelock);
+ percpu_down_read(&uprobes_treelock);
n = find_node_in_range(inode, min, max);
if (n) {
for (t = n; t; t = rb_prev(t)) {
@@ -1510,7 +1510,7 @@ static void build_probe_list(struct inode *inode,
list_add(&u->pending_list, head);
}
}
- read_unlock(&uprobes_treelock);
+ percpu_up_read(&uprobes_treelock);
}
/* @vma contains reference counter, not the probed instruction. */
@@ -1601,9 +1601,9 @@ vma_has_uprobes(struct vm_area_struct *vma, unsigned long start, unsigned long e
min = vaddr_to_offset(vma, start);
max = min + (end - start) - 1;
- read_lock(&uprobes_treelock);
+ percpu_down_read(&uprobes_treelock);
n = find_node_in_range(inode, min, max);
- read_unlock(&uprobes_treelock);
+ percpu_up_read(&uprobes_treelock);
return !!n;
}
With all the batch uprobe APIs work we are now finally ready to reap the benefits. Switch uprobes_treelock from reader-writer spinlock to a much more efficient and scalable per-CPU RW semaphore. Benchmarks and numbers time. I've used BPF selftests' bench tool, trig-uprobe-nop benchmark specifically, to see how uprobe total throughput scales with number of competing threads (mapped to individual CPUs). Here are results: # threads BEFORE (mln/s) AFTER (mln/s) --------- -------------- ------------- 1 3.131 3.140 2 3.394 3.601 3 3.630 3.960 4 3.317 3.551 5 3.448 3.464 6 3.345 3.283 7 3.469 3.444 8 3.182 3.258 9 3.138 3.139 10 2.999 3.212 11 2.903 3.183 12 2.802 3.027 13 2.792 3.027 14 2.695 3.086 15 2.822 2.965 16 2.679 2.939 17 2.622 2.888 18 2.628 2.914 19 2.702 2.836 20 2.561 2.837 One can see that per-CPU RW semaphore-based implementation scales better with number of CPUs (especially that single CPU throughput is basically the same). Note, scalability is still limited by register_rwsem and this will hopefully be address in follow up patch set(s). Signed-off-by: Andrii Nakryiko <andrii@kernel.org> --- kernel/events/uprobes.c | 30 +++++++++++++++--------------- 1 file changed, 15 insertions(+), 15 deletions(-)