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