1#include "thread/workqueue/tests/test_internal.h"
2
3TEST_GROUP_DEFINE(workqueue, .intensity_desc = {
4 .curve = SCALE_PIECEWISE_LOG,
5 .unit = "items",
6 });
7
8#define WQ_2_THREADS 2
9
10static atomic_uint32_t times_2 = 0;
11static size_t wq_2_items_per_thread = 512;
12
13static void wq_test_2(void *a, void *b) {
14 cc_ignore(a, b);
15
16 atomic_inc(&times_2);
17 for (uint64_t i = 0; i < 500; i++)
18 cpu_pause();
19}
20
21static struct workqueue *wq = NULL;
22static atomic_uint32_t threads_left = WQ_2_THREADS;
23
24static void enqueue_thread(void *arg) {
25 cc_ignore(arg);
26 struct work works[wq_2_items_per_thread];
27
28 for (size_t i = 0; i < wq_2_items_per_thread; i++) {
29 for (uint64_t j = 0; j < 500; j++)
30 cpu_pause();
31
32 work_init(work: &works[i], fn: wq_test_2, WORK_ARGS(NULL, wq));
33 workqueue_enqueue(queue: wq, work: &works[i]);
34 scheduler_yield();
35 }
36
37 for (size_t i = 0; i < wq_2_items_per_thread; i++)
38 while (work_active(work: &works[i]))
39 cpu_pause();
40
41 atomic_dec(&threads_left);
42}
43
44TEST_DEFINE_UNIT(workqueue, mt_enqueue_scaling,
45 TEST_INTENSITY(512, 4096, 32768)) {
46 size_t total_items = ctx->intensity_val ? ctx->intensity_val : 4096;
47 wq_2_items_per_thread = total_items / WQ_2_THREADS;
48 atomic_store(&times_2, 0);
49 atomic_store(&threads_left, WQ_2_THREADS);
50
51 struct cpu_mask mask;
52 cpu_mask_init(m: &mask, nbits: global.core_count);
53
54 cpu_mask_set_all(&mask);
55
56 struct workqueue_attributes attrs = {
57 .flags = WORKQUEUE_FLAG_AUTO_SPAWN | WORKQUEUE_FLAG_ON_DEMAND,
58 .spawn_delay = 1,
59 .idle_check.max = 10000,
60 .idle_check.min = 2000,
61 .min_workers = 2,
62 .max_workers = 64,
63 .worker_cpu_mask = mask,
64 };
65
66 wq = workqueue_create(NULL, &attrs);
67
68 struct thread *enqueuers[WQ_2_THREADS];
69 for (size_t i = 0; i < WQ_2_THREADS; i++) {
70 test_info("spawning workqueue enqueue threads");
71
72 /* Big stack */
73 enqueuers[i] = thread_spawn("workqueue_enqueue_thread", enqueue_thread,
74 .stack_pages = 32, .joinable = true);
75 }
76
77 test_info("waiting for enqueue threads");
78 for (size_t i = 0; i < WQ_2_THREADS; i++) {
79 if (enqueuers[i])
80 thread_join(t: enqueuers[i]);
81 }
82
83 TEST_ASSERT_EQ(atomic_load(&threads_left), 0);
84
85 uint64_t workers = atomic_load_rlx(&wq->num_workers);
86
87 char *msg = valloc(100, ALLOC_ZERO);
88 if (msg) {
89 snprintf(msg, 100, "There are %lu workers", workers);
90 test_info("%s", msg);
91 vfree(msg);
92 }
93
94 test_info("destroy");
95 workqueue_destroy(queue: wq);
96 return TEST_SUCCESS;
97}
98