1#include <mem/alloc.h>
2#include <sch/sched.h>
3#include <thread/daemon.h>
4
5#include <string.h>
6
7static struct daemon_thread *current_daemon_thread(void) {
8 return thread_get_current()->private;
9}
10
11static bool mark_daemon_thread_executing(struct daemon_thread *thread,
12 bool state) {
13 return atomic_exchange(&thread->executing_work, state);
14}
15
16static size_t total_ts_workers(struct daemon *daemon) {
17 return atomic_load(&daemon->attrs.timesharing_threads);
18}
19
20static size_t idle_ts_workers(struct daemon *daemon) {
21 return atomic_load(&daemon->attrs.idle_timesharing_threads);
22}
23
24static size_t max_ts_workers(struct daemon *daemon) {
25 return daemon->attrs.max_timesharing_threads;
26}
27
28static bool bg_present(struct daemon *daemon) {
29 return atomic_load(&daemon->attrs.background_thread_present);
30}
31
32static bool set_bg_present(struct daemon *daemon, bool state) {
33 return atomic_exchange(&daemon->attrs.background_thread_present, state);
34}
35
36static bool ts_busy(struct daemon *d) {
37 /* No idle workers */
38 return idle_ts_workers(daemon: d) == 0 && total_ts_workers(daemon: d) > 0;
39}
40
41static void daemon_list_add(struct daemon *daemon,
42 struct daemon_thread *thread) {
43 enum irql irql = spin_lock(lock: &daemon->lock);
44 list_add(new: &thread->list_node, head: &daemon->timesharing_threads);
45 spin_unlock(lock: &daemon->lock, old: irql);
46}
47
48static void daemon_list_del(struct daemon *daemon,
49 struct daemon_thread *thread) {
50 enum irql irql = spin_lock(lock: &daemon->lock);
51 list_del_init(entry: &thread->list_node);
52 spin_unlock(lock: &daemon->lock, old: irql);
53}
54
55static void daemon_thread_exit(struct daemon *daemon,
56 struct daemon_thread *self) {
57 bool background = self->background;
58
59 if (!background) {
60 daemon_list_del(daemon, thread: self);
61 atomic_fetch_sub(&daemon->attrs.timesharing_threads, 1);
62 } else {
63 daemon->background_thread = NULL;
64 set_bg_present(daemon, false);
65 }
66
67 kfree(self);
68 thread_exit();
69}
70
71static bool daemon_needs_spawn_worker(struct daemon *d) {
72 bool want_spawn = ts_busy(d) && DAEMON_FLAG_TEST(d, DAEMON_FLAG_AUTO_SPAWN);
73 bool allowed_to_spawn = total_ts_workers(daemon: d) < max_ts_workers(daemon: d);
74 return want_spawn && allowed_to_spawn;
75}
76
77static struct daemon_work *work_on_thread(struct daemon_thread *thread) {
78 return thread->background ? thread->daemon->background_work
79 : thread->daemon->timesharing_work;
80}
81
82static void daemon_work_execute(struct daemon_work *w,
83 struct daemon_thread *self) {
84 mark_daemon_thread_executing(thread: self, true);
85
86 self->command = w->function(w->args.arg1, w->args.arg2);
87
88 mark_daemon_thread_executing(thread: self, false);
89
90 /* Exit if we are destroying */
91 if (self->daemon->state == DAEMON_STATE_DESTROYING)
92 self->command = DAEMON_THREAD_COMMAND_EXIT;
93}
94
95static void daemon_wait(struct daemon *daemon, struct daemon_thread *self) {
96 atomic_fetch_add(&daemon->attrs.idle_timesharing_threads, 1);
97
98 if (self->background)
99 semaphore_wait(s: &daemon->bg_sem);
100 else
101 semaphore_wait(s: &daemon->ts_sem);
102
103 atomic_fetch_sub(&daemon->attrs.idle_timesharing_threads, 1);
104
105 if (daemon->state == DAEMON_STATE_DESTROYING)
106 daemon_thread_exit(daemon, self);
107}
108
109void daemon_main(void *a) {
110 (void) a;
111
112 struct daemon_thread *self = current_daemon_thread();
113 struct daemon *daemon = self->daemon;
114 bool timesharing = !self->background;
115
116 if (timesharing)
117 atomic_fetch_add(&daemon->attrs.timesharing_threads, 1);
118
119 struct daemon_work *work = work_on_thread(thread: self);
120
121 while (true) {
122 daemon_wait(daemon, self);
123
124 start_execute:
125
126 daemon_work_execute(w: work, self);
127
128 switch (self->command) {
129 case DAEMON_THREAD_COMMAND_SLEEP: break; /* Go wait on the semaphore */
130 case DAEMON_THREAD_COMMAND_RESTART: goto start_execute;
131 case DAEMON_THREAD_COMMAND_EXIT:
132 daemon_thread_exit(daemon, self);
133 break;
134 default:
135 panic("Unknown daemon thread command with value %u", self->command);
136 }
137 }
138
139 panic("Daemon thread should not be able to exit the loop");
140}
141
142struct daemon_thread *daemon_thread_create(struct daemon *daemon) {
143 struct daemon_thread *thread =
144 kmalloc(sizeof(struct daemon_thread), ALLOC_FLAGS_ZERO);
145 if (!thread)
146 return NULL;
147
148 thread->daemon = daemon;
149 INIT_LIST_HEAD(list: &thread->list_node);
150
151 struct thread *t = thread_create(name: "daemon_%s_thread", entry_point: daemon_main, NULL,
152 daemon->name ? daemon->name : "unnamed");
153 if (!t) {
154 kfree(thread);
155 return NULL;
156 }
157
158 thread->thread = t;
159 t->private = thread;
160
161 return thread;
162}
163
164static struct daemon_thread *daemon_thread_create_bg(struct daemon *daemon) {
165 struct daemon_thread *ret = daemon_thread_create(daemon);
166 if (!ret)
167 return NULL;
168
169 ret->background = true;
170 thread_set_background(t: ret->thread);
171 return ret;
172}
173
174static struct daemon_thread *
175daemon_thread_spawn(struct daemon *daemon,
176 struct daemon_thread *(*create)(struct daemon *) ) {
177 struct daemon_thread *t = create(daemon);
178 if (!t)
179 return NULL;
180
181 t->thread->allowed_cpus = daemon->attrs.thread_cpu_mask;
182 thread_enqueue(t: t->thread);
183
184 return t;
185}
186
187void daemon_thread_destroy_unsafe(struct daemon_thread *dt) {
188 thread_free(t: dt->thread);
189 kfree(dt);
190}
191
192struct daemon *daemon_create(const char *fmt, struct daemon_attributes *attrs,
193 struct daemon_work *timesharing_work,
194 struct daemon_work *background_work,
195 struct workqueue_attributes *wq_attrs, ...) {
196 va_list args;
197 va_start(args, wq_attrs);
198
199 struct daemon *daemon = kmalloc(sizeof(struct daemon), ALLOC_FLAGS_ZERO);
200 struct daemon_thread *dt = NULL, *bg = NULL;
201
202 if (!daemon)
203 goto err;
204
205 daemon->attrs = *attrs;
206
207 kassert(attrs->thread_cpu_mask.nbits != 0, "please set a valid CPU mask");
208 kassert(attrs->min_timesharing_threads <= attrs->max_timesharing_threads);
209
210 if (!DAEMON_FLAG_TEST(daemon, DAEMON_FLAG_NO_TS_THREADS))
211 kassert(attrs->min_timesharing_threads,
212 "needs min timesharing threads");
213
214 if (DAEMON_FLAG_TEST(daemon, DAEMON_FLAG_HAS_NAME)) {
215 va_list args_copy;
216 va_copy(args_copy, args);
217 int needed = vsnprintf(NULL, buffer_len: 0, format: fmt, args: args_copy) + 1;
218 va_end(args_copy);
219
220 char *name = kmalloc(needed, ALLOC_FLAGS_ZERO);
221 if (!name)
222 goto err;
223
224 va_copy(args_copy, args);
225 vsnprintf(buffer: name, buffer_len: needed, format: fmt, args: args_copy);
226 va_end(args_copy);
227 daemon->name = name;
228 }
229
230 daemon->attrs.background_thread_present = false;
231 daemon->attrs.idle_timesharing_threads = 0;
232 daemon->attrs.timesharing_threads = 0;
233
234 spinlock_init(lock: &daemon->lock);
235 semaphore_init(s: &daemon->bg_sem, value: 0, SEMAPHORE_INIT_NORMAL);
236 semaphore_init(s: &daemon->ts_sem, value: 0, SEMAPHORE_INIT_NORMAL);
237
238 INIT_LIST_HEAD(list: &daemon->timesharing_threads);
239 daemon->timesharing_work = timesharing_work;
240 daemon->background_work = background_work;
241
242 if (DAEMON_FLAG_TEST(daemon, DAEMON_FLAG_HAS_WORKQUEUE)) {
243 wq_attrs->flags |= WORKQUEUE_FLAG_NAMED;
244 struct workqueue *wq = workqueue_create(
245 fmt: "workqueue_daemon_%s", attrs: wq_attrs, daemon->name ? daemon->name : "");
246
247 if (!wq)
248 goto err;
249
250 daemon->workqueue = wq;
251 }
252
253 if (!DAEMON_FLAG_TEST(daemon, DAEMON_FLAG_NO_TS_THREADS) &&
254 timesharing_work) {
255 for (size_t i = 0; i < attrs->min_timesharing_threads; i++) {
256 dt = daemon_thread_spawn(daemon, create: daemon_thread_create);
257 if (!dt)
258 goto err;
259
260 daemon_list_add(daemon, thread: dt);
261 }
262 }
263
264 if (background_work) {
265 bg = daemon_thread_spawn(daemon, create: daemon_thread_create_bg);
266 if (!bg)
267 goto err;
268
269 daemon->background_thread = bg;
270 set_bg_present(daemon, true);
271 }
272
273 daemon->state = DAEMON_STATE_ACTIVE;
274
275 va_end(args);
276 return daemon;
277
278err:
279 if (dt)
280 daemon_thread_destroy_unsafe(dt);
281
282 if (daemon && daemon->name)
283 kfree(daemon->name);
284
285 if (daemon)
286 kfree(daemon);
287
288 va_end(args);
289 return NULL;
290}
291
292static void boost_bg_thread_to_ts(struct daemon *daemon) {
293 if (daemon->background_thread)
294 thread_set_timesharing(t: daemon->background_thread->thread);
295}
296
297/* Assume that all daemons must have daemon works
298 * finish executing before they can be safely destroyed */
299void daemon_destroy(struct daemon *daemon) {
300 /* Make all the threads go sleep on the semaphore */
301 daemon->state = DAEMON_STATE_DESTROYING;
302
303 boost_bg_thread_to_ts(daemon);
304
305 /* Wait for all currently running threads to exit or sleep */
306 while (total_ts_workers(daemon) > idle_ts_workers(daemon))
307 scheduler_yield();
308
309 /* Great, now we send the signal and wake everyone */
310 while (total_ts_workers(daemon) > 0) {
311 semaphore_post(s: &daemon->ts_sem);
312 scheduler_yield();
313 }
314
315 kassert(!total_ts_workers(daemon));
316 /* All timesharing threads should be gone now.
317 * Time to handle the background thread. */
318 while (bg_present(daemon)) {
319 semaphore_post(s: &daemon->bg_sem);
320 scheduler_yield();
321 }
322
323 daemon->state = DAEMON_STATE_DEAD;
324
325 /* Ok now all threads are gone */
326 if (daemon->workqueue) {
327 kassert(DAEMON_FLAG_TEST(daemon, DAEMON_FLAG_HAS_WORKQUEUE));
328 workqueue_destroy(queue: daemon->workqueue);
329 }
330
331 if (daemon->name)
332 kfree(daemon->name);
333
334 kfree(daemon);
335}
336
337struct daemon_thread *daemon_spawn_worker(struct daemon *daemon) {
338 struct daemon_thread *dt =
339 daemon_thread_spawn(daemon, create: daemon_thread_create);
340
341 daemon_list_add(daemon, thread: dt);
342 return dt;
343}
344
345enum workqueue_error daemon_submit_oneshot_work(struct daemon *daemon,
346 work_function function,
347 struct work_args args) {
348 kassert(daemon->workqueue &&
349 DAEMON_FLAG_TEST(daemon, DAEMON_FLAG_HAS_WORKQUEUE));
350 return workqueue_enqueue_oneshot(queue: daemon->workqueue, func: function, args);
351}
352
353enum workqueue_error daemon_submit_work(struct daemon *daemon,
354 struct work *work) {
355 kassert(daemon->workqueue &&
356 DAEMON_FLAG_TEST(daemon, DAEMON_FLAG_HAS_WORKQUEUE));
357 return workqueue_enqueue(queue: daemon->workqueue, work);
358}
359
360void daemon_wake_background_worker(struct daemon *daemon) {
361 semaphore_post(s: &daemon->bg_sem);
362}
363
364void daemon_wake_timesharing_worker(struct daemon *daemon) {
365 if (daemon_needs_spawn_worker(d: daemon))
366 daemon_spawn_worker(daemon);
367
368 semaphore_post(s: &daemon->ts_sem);
369}
370
371void daemon_wake_all_idle_timesharing_workers(struct daemon *daemon) {
372 if (daemon_needs_spawn_worker(d: daemon))
373 daemon_spawn_worker(daemon);
374
375 semaphore_postn(s: &daemon->ts_sem, n: idle_ts_workers(daemon));
376}
377
378void daemon_print(struct daemon *daemon) {
379 struct daemon_attributes *attrs = &daemon->attrs;
380 printf(format: "struct daemon \"%s\" = {\n", daemon->name ? daemon->name : "NULL");
381 printf(format: " .attrs = {\n");
382 printf(format: " .max_timesharing_threads = %u\n",
383 attrs->max_timesharing_threads);
384 printf(format: " .idle_timesharing_threads = %u\n",
385 attrs->idle_timesharing_threads);
386 printf(format: " .timesharing_threads = %u\n",
387 attrs->timesharing_threads);
388 printf(format: " .flags = 0b%b\n", attrs->flags);
389 printf(format: " }\n");
390 printf(format: " .state = %s\n", daemon_state_str(s: daemon->state));
391 printf(format: "}\n");
392}
393