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