1#include <compiler.h>
2#include <kassert.h>
3#include <mem/alloc.h>
4#include <mem/alloc_or_die.h>
5#include <sch/sched.h>
6#include <smp/domain.h>
7#include <stdarg.h>
8#include <string.h>
9#include <sync/condvar.h>
10#include <sync/spinlock.h>
11#include <thread/thread.h>
12#include <thread/workqueue.h>
13
14#include "internal.h"
15
16LOCK_CHK_CLASS_DECLARE_LOCAL(workqueue_irq);
17LOCK_CHK_CLASS_DECLARE_LOCAL(workqueue_disp);
18
19enum workqueue_error workqueue_add_oneshot(work_function func,
20 struct work_args args) {
21 struct workqueue *queue = workqueue_get_least_loaded();
22 return workqueue_enqueue_oneshot(queue, func, args);
23}
24
25enum workqueue_error workqueue_add_remote_oneshot(work_function func,
26 struct work_args args) {
27 struct workqueue *queue = workqueue_get_least_loaded_remote();
28 return workqueue_enqueue_oneshot(queue, func, args);
29}
30
31enum workqueue_error workqueue_add_local_oneshot(work_function func,
32 struct work_args args) {
33 struct workqueue *queue = global.workqueues[smp_id_raw()];
34 return workqueue_enqueue_oneshot(queue, func, args);
35}
36
37static struct workqueue *find_optimal_domain_wq(void) {
38 struct core *pos;
39
40 struct workqueue *optimal =
41 global.workqueues[(smp_id_raw() + 1) % global.core_count];
42
43 struct workqueue *local = global.workqueues[smp_id_raw()];
44
45 size_t least_loaded = WORKQUEUE_NUM_WORKS(optimal);
46
47 domain_for_each_core_local(TOPC_NONE, pos) {
48 struct workqueue *queue = global.workqueues[pos->id];
49 size_t load = WORKQUEUE_NUM_WORKS(queue);
50
51 if (load < least_loaded && queue != local) {
52 least_loaded = load;
53 optimal = queue;
54 }
55 }
56
57 return optimal;
58}
59
60enum workqueue_error workqueue_add_fast_oneshot(work_function func,
61 struct work_args args) {
62 struct workqueue *optimal = find_optimal_domain_wq();
63 return workqueue_enqueue_oneshot(queue: optimal, func, args);
64}
65
66enum workqueue_error workqueue_add_fast(struct work *work) {
67 struct workqueue *optimal = find_optimal_domain_wq();
68 return workqueue_enqueue(queue: optimal, work);
69}
70
71enum workqueue_error workqueue_add(struct work *work) {
72 struct workqueue *queue = workqueue_get_least_loaded();
73 return workqueue_enqueue(queue, work);
74}
75
76enum workqueue_error workqueue_add_local(struct work *work) {
77 struct workqueue *queue = global.workqueues[smp_id_raw()];
78 return workqueue_enqueue(queue, work);
79}
80
81enum workqueue_error workqueue_add_remote(struct work *work) {
82 struct workqueue *queue = workqueue_get_least_loaded_remote();
83 return workqueue_enqueue(queue, work);
84}
85
86void work_execute(struct work *task) {
87 kassert(task);
88 task->func(task->args.arg1, task->args.arg2);
89 atomic_exchange(&task->active, false);
90}
91
92struct workqueue *workqueue_create_internal(struct workqueue_attributes *attrs,
93 const char *fmt, va_list args) {
94 bool permanent = attrs->flags & WORKQUEUE_FLAG_PERMANENT;
95
96 /* Permanent workqueues are moved after initialization so
97 * their structs are aligned up a page so that they can be
98 * properly moved without overlapping with each other */
99 size_t size = permanent ? sizeof(struct workqueue)
100 : PAGE_ALIGN_UP(sizeof(struct workqueue));
101
102 struct workqueue *wq = kmalloc(size, ALLOC_FLAGS_ZERO);
103 if (!wq)
104 goto err;
105
106 if (attrs->flags & WORKQUEUE_FLAG_ISR_SAFE) {
107 spinlock_init_chk(&wq->lock, LOCK_CHK_CLASS(workqueue_irq),
108 LOCK_CHKD_FULL);
109 } else {
110 spinlock_init_chk(&wq->lock, LOCK_CHK_CLASS(workqueue_disp),
111 LOCK_CHKD_FULL);
112 }
113 spinlock_init(&wq->worker_array_lock);
114 spinlock_init(&wq->worker_lock);
115 spinlock_init(&wq->work_lock);
116
117 if (cpu_mask_empty(m: &attrs->worker_cpu_mask))
118 panic("please set a CPU mask before creating the workqueue");
119
120 wq->attrs = *attrs;
121 condvar_init(cv: &wq->queue_cv, irq_disable: attrs->flags & WORKQUEUE_FLAG_ISR_SAFE
122 ? CONDVAR_INIT_IRQ_DISABLE
123 : CONDVAR_INIT_NORMAL);
124 kassert(THREAD_NICENESS_VALID(attrs->worker_niceness));
125
126 size = sizeof(struct work) * attrs->capacity;
127 if (permanent)
128 size = PAGE_ALIGN_UP(size);
129
130 wq->oneshot_works = kmalloc(size, ALLOC_FLAGS_ZERO);
131 if (!wq->oneshot_works)
132 goto err;
133
134 if (attrs->flags & WORKQUEUE_FLAG_STATIC_WORKERS) {
135 wq->worker_array = kmalloc(sizeof(struct worker) * attrs->max_workers,
136 ALLOC_FLAGS_ZERO);
137 if (!wq->worker_array)
138 goto err;
139 }
140
141 if (attrs->flags & WORKQUEUE_FLAG_NAMED) {
142 kassert(fmt);
143
144 va_list args_copy;
145 va_copy(args_copy, args);
146 size_t needed = vsnprintf(NULL, buffer_len: 0, format: fmt, args: args_copy) + 1;
147 va_end(args_copy);
148
149 wq->name = kmalloc(needed, ALLOC_FLAGS_ZERO);
150 if (!wq->name)
151 goto err;
152
153 va_copy(args_copy, args);
154 vsnprintf(buffer: wq->name, buffer_len: needed, format: fmt, args: args_copy);
155 va_end(args_copy);
156 }
157
158 INIT_LIST_HEAD(list: &wq->workers);
159 INIT_LIST_HEAD(list: &wq->works);
160
161 for (uint64_t i = 0; i < attrs->capacity; i++)
162 atomic_store_explicit(&wq->oneshot_works[i].seq, i,
163 memory_order_relaxed);
164
165 refcount_init(rc: &wq->refcount, val: 1);
166 wq->state = WORKQUEUE_STATE_ACTIVE;
167
168 return wq;
169
170err:
171 if (wq) {
172 kfree(wq->worker_array);
173 kfree(wq->request);
174 kfree(wq->name);
175 kfree(wq->oneshot_works);
176 }
177
178 kfree(wq);
179
180 return NULL;
181}
182
183struct workqueue *workqueue_create(const char *fmt,
184 struct workqueue_attributes *attrs, ...) {
185 if (attrs->min_workers == 0)
186 attrs->min_workers = 1;
187
188 va_list args;
189 va_start(args, attrs);
190
191 struct workqueue *ret = workqueue_create_internal(attrs, fmt, args);
192
193 va_end(args);
194
195 if (ret)
196 for (size_t i = 0; i < attrs->min_workers; i++)
197 workqueue_spawn_permanent_worker(queue: ret);
198
199 return ret;
200}
201
202struct workqueue *workqueue_create_default(const char *fmt, ...) {
203 struct cpu_mask cmask;
204 if (!cpu_mask_init(m: &cmask, nbits: global.core_count))
205 return NULL;
206
207 cpu_mask_set_all(&cmask);
208 struct workqueue_attributes attrs = {
209 .capacity = WORKQUEUE_DEFAULT_CAPACITY,
210 .idle_check = WORKQUEUE_DEFAULT_IDLE_CHECK,
211 .max_workers = WORKQUEUE_DEFAULT_MAX_WORKERS,
212 .min_workers = 1,
213 .spawn_delay = WORKQUEUE_DEFAULT_SPAWN_DELAY,
214 .worker_cpu_mask = cmask,
215 .worker_niceness = 0,
216 .flags = WORKQUEUE_FLAG_DEFAULTS,
217 };
218
219 va_list args;
220 va_start(args, fmt);
221
222 struct workqueue *ret = workqueue_create_internal(attrs: &attrs, fmt, args);
223
224 va_end(args);
225 if (ret)
226 workqueue_spawn_permanent_worker(queue: ret);
227
228 return ret;
229}
230
231static void mark_worker_exit(struct thread *t) {
232 if (t) {
233 struct worker *worker = t->private;
234 worker->next_action = WORKER_NEXT_ACTION_EXIT;
235 }
236}
237
238void workqueue_free(struct workqueue *wq) {
239 kassert(atomic_load(&wq->refcount) == 0);
240 WORKQUEUE_STATE_SET(wq, WORKQUEUE_STATE_DEAD);
241 kfree(wq->oneshot_works);
242 kfree(wq->request);
243 kfree(wq);
244}
245
246/* Give all threads the exit signal and clean up the structs */
247void workqueue_destroy(struct workqueue *queue) {
248 kassert(queue);
249
250 WORKQUEUE_STATE_SET(queue, WORKQUEUE_STATE_DESTROYING);
251 atomic_store(&queue->ignore_timeouts, true);
252
253 thread_apply_cpu_penalty(t: thread_get_current());
254 while (workqueue_workers(wq: queue) > workqueue_idlers(wq: queue)) {
255 scheduler_yield();
256 }
257
258 /* All workers now idle */
259 condvar_broadcast_callback(cv: &queue->queue_cv, cb: mark_worker_exit);
260
261 while (workqueue_workers(wq: queue) > 0) {
262 thread_apply_cpu_penalty(t: thread_get_current());
263 scheduler_yield();
264 condvar_broadcast_callback(cv: &queue->queue_cv, cb: mark_worker_exit);
265 }
266
267 workqueue_put(queue);
268}
269
270void workqueue_kick(struct workqueue *queue) {
271 condvar_signal(cv: &queue->queue_cv);
272}
273
274struct worker *workqueue_spawn_permanent_worker(struct workqueue *queue) {
275 struct thread *thread = worker_create(mask: queue->attrs.worker_cpu_mask,
276 niceness: queue->attrs.worker_niceness);
277
278 if (!thread)
279 return NULL;
280
281 struct worker *worker = kmalloc(sizeof(struct worker), ALLOC_FLAGS_ZERO);
282 if (!worker)
283 return NULL;
284
285 INIT_LIST_HEAD(list: &worker->list_node);
286
287 worker->is_permanent = true;
288 worker->inactivity_check_period = queue->attrs.idle_check.max;
289 worker->workqueue = queue;
290
291 workqueue_link_thread_and_worker(worker, thread);
292
293 thread_enqueue(t: thread);
294
295 workqueue_add_worker(wq: queue, wker: worker);
296 atomic_fetch_add(&queue->num_workers, 1);
297
298 return worker;
299}
300
301void workqueues_permanent_init(void) {
302 int64_t num_workqueues = global.core_count;
303 global.workqueues = alloc_or_die(
304 kmalloc(sizeof(struct workqueue *) * num_workqueues, ALLOC_FLAGS_ZERO));
305
306 for (int64_t i = 0; i < num_workqueues; i++) {
307
308 struct cpu_mask mask;
309 if (!cpu_mask_init(m: &mask, nbits: global.core_count))
310 panic("Failed to initialize CPU mask");
311
312 cpu_mask_set(m: &mask, cpu: i);
313
314 struct workqueue_attributes attrs = {
315 .capacity = WORKQUEUE_DEFAULT_CAPACITY,
316 .max_workers = WORKQUEUE_DEFAULT_MAX_WORKERS,
317 .spawn_delay = WORKQUEUE_DEFAULT_SPAWN_DELAY,
318 .idle_check = WORKQUEUE_DEFAULT_IDLE_CHECK,
319 .flags = WORKQUEUE_FLAG_PERMANENT | WORKQUEUE_FLAG_AUTO_SPAWN |
320 WORKQUEUE_FLAG_NO_WORKER_GC,
321 .worker_cpu_mask = mask,
322 };
323
324 global.workqueues[i] = workqueue_create_internal(
325 attrs: &attrs, /* fmt = */ NULL, /* args = */ NULL);
326 if (!global.workqueues[i])
327 panic("Failed to spawn permanent workqueue");
328
329 global.workqueues[i]->core = i;
330
331 if (!workqueue_spawn_permanent_worker(queue: global.workqueues[i]))
332 panic("Failed to spawn initial worker on workqueue %u", i);
333 }
334}
335
336struct work *work_init(struct work *work, work_function fn,
337 struct work_args args) {
338 work->args = args;
339 work->active = false;
340 work->enqueued = false;
341 work->seq = 0;
342 work->func = fn;
343 INIT_LIST_HEAD(list: &work->list_node);
344 return work;
345}
346
347struct work *work_create(work_function fn, struct work_args args) {
348 struct work *work = kmalloc(sizeof(struct work), ALLOC_FLAGS_ZERO);
349 if (!work)
350 return NULL;
351
352 return work_init(work, fn, args);
353}
354