Something went wrong. Try again.
Linux kernel mirror (for testing) git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295// SPDX-License-Identifier: GPL-2.0
/* * Test module for stress and performance analysis of workqueue. * * Benchmarks queue_work() throughput on an unbound workqueue to measure * pool->lock contention under different affinity scope configurations * (e.g., cache vs cache_shard). * * The affinity scope is changed between runs via the workqueue's sysfs * affinity_scope attribute (WQ_SYSFS). * * Copyright (c) 2026 Meta Platforms, Inc. and affiliates * Copyright (c) 2026 Breno Leitao <leitao@debian.org> * */#include <linux/init.h>#include <linux/kernel.h>#include <linux/module.h>#include <linux/workqueue.h>#include <linux/kthread.h>#include <linux/moduleparam.h>#include <linux/completion.h>#include <linux/atomic.h>#include <linux/slab.h>#include <linux/ktime.h>#include <linux/cpumask.h>#include <linux/sched.h>#include <linux/sort.h>#include <linux/fs.h>
#define WQ_NAME "bench_wq"#define SCOPE_PATH "/sys/bus/workqueue/devices/" WQ_NAME "/affinity_scope"
static int nr_threads;module_param(nr_threads, int, 0444);MODULE_PARM_DESC(nr_threads, "Number of threads to spawn (default: 0 = num_online_cpus())");
static int wq_items = 50000;module_param(wq_items, int, 0444);MODULE_PARM_DESC(wq_items, "Number of work items each thread queues (default: 50000)");
static struct workqueue_struct *bench_wq;static atomic_t threads_done;static DECLARE_COMPLETION(start_comp);static DECLARE_COMPLETION(all_done_comp);
struct thread_ctx { struct completion work_done; struct work_struct work; u64 *latencies; int cpu; int items;};
static void bench_work_fn(struct work_struct *work){ struct thread_ctx *ctx = container_of(work, struct thread_ctx, work);
complete(&ctx->work_done);}
static int bench_kthread_fn(void *data){ struct thread_ctx *ctx = data; ktime_t t_start, t_end; int i;
/* Wait for all threads to be ready */ wait_for_completion(&start_comp);
if (kthread_should_stop()) return 0;
for (i = 0; i < ctx->items; i++) { reinit_completion(&ctx->work_done); INIT_WORK(&ctx->work, bench_work_fn);
t_start = ktime_get(); queue_work(bench_wq, &ctx->work); t_end = ktime_get();
ctx->latencies[i] = ktime_to_ns(ktime_sub(t_end, t_start)); wait_for_completion(&ctx->work_done); }
if (atomic_dec_and_test(&threads_done)) complete(&all_done_comp);
/* * Wait for kthread_stop() so the module text isn't freed * while we're still executing. */ while (!kthread_should_stop()) schedule();
return 0;}
static int cmp_u64(const void *a, const void *b){ u64 va = *(const u64 *)a; u64 vb = *(const u64 *)b;
if (va < vb) return -1; if (va > vb) return 1; return 0;}
static int __init set_affn_scope(const char *scope){ struct file *f; loff_t pos = 0; ssize_t ret;
f = filp_open(SCOPE_PATH, O_WRONLY, 0); if (IS_ERR(f)) { pr_err("test_workqueue: open %s failed: %ld\n", SCOPE_PATH, PTR_ERR(f)); return PTR_ERR(f); }
ret = kernel_write(f, scope, strlen(scope), &pos); filp_close(f, NULL);
if (ret < 0) { pr_err("test_workqueue: write '%s' failed: %zd\n", scope, ret); return ret; }
return 0;}
static int __init run_bench(int n_threads, const char *scope, const char *label){ struct task_struct **tasks; unsigned long total_items; struct thread_ctx *ctxs; u64 *all_latencies; ktime_t start, end; int cpu, i, j, ret; s64 elapsed_us;
ret = set_affn_scope(scope); if (ret) return ret;
ctxs = kcalloc(n_threads, sizeof(*ctxs), GFP_KERNEL); if (!ctxs) return -ENOMEM;
tasks = kcalloc(n_threads, sizeof(*tasks), GFP_KERNEL); if (!tasks) { kfree(ctxs); return -ENOMEM; }
total_items = (unsigned long)n_threads * wq_items; all_latencies = kvmalloc_array(total_items, sizeof(u64), GFP_KERNEL); if (!all_latencies) { kfree(tasks); kfree(ctxs); return -ENOMEM; }
/* Allocate per-thread latency arrays */ for (i = 0; i < n_threads; i++) { ctxs[i].latencies = kvmalloc_array(wq_items, sizeof(u64), GFP_KERNEL); if (!ctxs[i].latencies) { while (--i >= 0) kvfree(ctxs[i].latencies); kvfree(all_latencies); kfree(tasks); kfree(ctxs); return -ENOMEM; } }
atomic_set(&threads_done, n_threads); reinit_completion(&all_done_comp); reinit_completion(&start_comp);
/* Create kthreads, each bound to a different online CPU */ i = 0; for_each_online_cpu(cpu) { if (i >= n_threads) break;
ctxs[i].cpu = cpu; ctxs[i].items = wq_items; init_completion(&ctxs[i].work_done);
tasks[i] = kthread_create(bench_kthread_fn, &ctxs[i], "wq_bench/%d", cpu); if (IS_ERR(tasks[i])) { ret = PTR_ERR(tasks[i]); pr_err("test_workqueue: failed to create kthread %d: %d\n", i, ret); /* Unblock threads waiting on start_comp before stopping them */ complete_all(&start_comp); while (--i >= 0) kthread_stop(tasks[i]); goto out_free; }
kthread_bind(tasks[i], cpu); wake_up_process(tasks[i]); i++; }
/* Start timing and release all threads */ start = ktime_get(); complete_all(&start_comp);
/* Wait for all threads to finish the benchmark */ wait_for_completion(&all_done_comp);
/* Drain any remaining work */ flush_workqueue(bench_wq);
/* Ensure all kthreads have fully exited before module memory is freed */ for (i = 0; i < n_threads; i++) kthread_stop(tasks[i]);
end = ktime_get(); elapsed_us = ktime_us_delta(end, start);
/* Merge all per-thread latencies and sort for percentile calculation */ j = 0; for (i = 0; i < n_threads; i++) { memcpy(&all_latencies[j], ctxs[i].latencies, wq_items * sizeof(u64)); j += wq_items; }
sort(all_latencies, total_items, sizeof(u64), cmp_u64, NULL);
pr_info("test_workqueue: %-16s %llu items/sec\tp50=%llu\tp90=%llu\tp95=%llu ns\n", label, elapsed_us ? div_u64(total_items * 1000000ULL, elapsed_us) : 0, all_latencies[total_items * 50 / 100], all_latencies[total_items * 90 / 100], all_latencies[total_items * 95 / 100]);
ret = 0;out_free: for (i = 0; i < n_threads; i++) kvfree(ctxs[i].latencies); kvfree(all_latencies); kfree(tasks); kfree(ctxs);
return ret;}
static const char * const bench_scopes[] = { "cpu", "smt", "cache_shard", "cache", "numa", "system",};
static int __init test_workqueue_init(void){ int n_threads = min(nr_threads ?: num_online_cpus(), num_online_cpus()); int i;
if (wq_items <= 0) { pr_err("test_workqueue: wq_items must be > 0\n"); return -EINVAL; }
bench_wq = alloc_workqueue(WQ_NAME, WQ_UNBOUND | WQ_SYSFS, 0); if (!bench_wq) return -ENOMEM;
pr_info("test_workqueue: running %d threads, %d items/thread\n", n_threads, wq_items);
for (i = 0; i < ARRAY_SIZE(bench_scopes); i++) run_bench(n_threads, bench_scopes[i], bench_scopes[i]);
destroy_workqueue(bench_wq);
/* Return -EAGAIN so the module doesn't stay loaded after the benchmark */ return -EAGAIN;}
module_init(test_workqueue_init);MODULE_AUTHOR("Breno Leitao <leitao@debian.org>");MODULE_DESCRIPTION("Stress/performance benchmark for workqueue subsystem");MODULE_LICENSE("GPL");