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