1#include "thread/workqueue/tests/test_internal.h"
2
3TEST_GROUP_DECLARE(workqueue, .intensity_desc = {
4 .curve = SCALE_PIECEWISE_LOG,
5 .unit = "items",
6 });
7
8static atomic_bool workqueue_ran = false;
9static _Atomic uint32_t workqueue_times = 0;
10static void workqueue_fn(void *arg, void *unused) {
11 (void) arg, (void) unused;
12 atomic_store(&workqueue_ran, true);
13 atomic_fetch_add(&workqueue_times, 1);
14}
15
16TEST_DECLARE_UNIT(workqueue, fast_oneshot, TEST_INTENSITY(32, 256, 4096)) {
17 atomic_store(&workqueue_ran, false);
18 atomic_store(&workqueue_times, 0);
19
20 uint64_t tsc = rdtsc();
21 uint64_t times = ctx->intensity_val ? ctx->intensity_val : 256;
22
23 for (uint64_t i = 0; i < times; i++) {
24 enum workqueue_error err =
25 workqueue_add_fast_oneshot(func: workqueue_fn, WORK_ARGS(NULL, NULL));
26 (void) err;
27 }
28
29 uint64_t total = rdtsc() - tsc;
30 sleep_spin_ms(msec: 50);
31
32 while (!atomic_load(&workqueue_ran))
33 cpu_relax();
34
35 char *msg = kmalloc(100, ALLOC_FLAGS_ZERO);
36 TEST_ASSERT_NONNULL(msg);
37 snprintf(buffer: msg, buffer_len: 100, format: "Took %lu clock cycles to add to event pool %lu times",
38 total, times);
39 test_info(msg);
40 kfree(msg);
41
42 TEST_ASSERT(atomic_load(&workqueue_ran));
43
44 msg = kmalloc(100, ALLOC_FLAGS_ZERO);
45 TEST_ASSERT_NONNULL(msg);
46 snprintf(buffer: msg, buffer_len: 100,
47 format: "Event pool ran %u times, tests should've had it run %lu times",
48 atomic_load(&workqueue_times), times);
49 test_info(msg);
50 kfree(msg);
51
52 return TEST_SUCCESS;
53}
54
55#define WQ_2_THREADS 2
56
57static _Atomic uint32_t times_2 = 0;
58static size_t wq_2_items_per_thread = 2048;
59
60static void wq_test_2(void *a, void *b) {
61 (void) a, (void) b;
62 atomic_fetch_add(&times_2, 1);
63 for (uint64_t i = 0; i < 500; i++)
64 cpu_relax();
65}
66
67static struct workqueue *wq = NULL;
68static _Atomic uint32_t threads_left = WQ_2_THREADS;
69
70static void enqueue_thread(void *) {
71 for (size_t i = 0; i < wq_2_items_per_thread; i++) {
72 for (uint64_t j = 0; j < 500; j++)
73 cpu_relax();
74
75 workqueue_enqueue_oneshot(queue: wq, func: wq_test_2, WORK_ARGS(NULL, wq));
76 scheduler_yield();
77 }
78 atomic_fetch_sub(&threads_left, 1);
79}
80
81TEST_DECLARE_UNIT(workqueue, concurrent_enqueue_scaling,
82 TEST_INTENSITY(512, 4096, 32768)) {
83 size_t total_items = ctx->intensity_val ? ctx->intensity_val : 4096;
84 wq_2_items_per_thread = total_items / WQ_2_THREADS;
85 atomic_store(&times_2, 0);
86 atomic_store(&threads_left, WQ_2_THREADS);
87
88 struct cpu_mask mask;
89 alloc_or_die(cpu_mask_init(&mask, global.core_count));
90
91 cpu_mask_set_all(&mask);
92
93 struct workqueue_attributes attrs = {
94 .capacity = total_items,
95 .flags = WORKQUEUE_FLAG_AUTO_SPAWN | WORKQUEUE_FLAG_ON_DEMAND,
96 .spawn_delay = 1,
97 .idle_check.max = 10000,
98 .idle_check.min = 2000,
99 .min_workers = 2,
100 .max_workers = 64,
101 .worker_cpu_mask = mask,
102 };
103
104 wq = workqueue_create(NULL, attrs: &attrs);
105
106 struct thread *enqueuers[WQ_2_THREADS];
107 for (size_t i = 0; i < WQ_2_THREADS; i++) {
108 test_info("spawning workqueue enqueue threads");
109 enqueuers[i] = thread_spawn_joinable(name: "workqueue_enqueue_thread",
110 entry: enqueue_thread, NULL);
111 }
112
113 test_info("waiting for enqueue threads");
114 for (size_t i = 0; i < WQ_2_THREADS; i++) {
115 if (enqueuers[i])
116 thread_join(t: enqueuers[i]);
117 }
118
119 TEST_ASSERT_EQ(atomic_load(&threads_left), 0);
120
121 uint64_t workers = wq->num_workers;
122
123 char *msg = kmalloc(100, ALLOC_FLAGS_ZERO);
124 if (msg) {
125 snprintf(buffer: msg, buffer_len: 100, format: "There are %lu workers", workers);
126 test_info(msg);
127 kfree(msg);
128 }
129
130 test_info("destroy");
131 workqueue_destroy(queue: wq);
132 return TEST_SUCCESS;
133}
134