1#include <mem/alloc.h>
2#include <sch/sched.h>
3#include <thread/daemon.h>
4
5#include <string.h>
6
7static void daemon_putref(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 enforce(DAEMON_FLAG_TEST(d, DAEMON_FLAG_HAS_WORKQUEUE));
13 workqueue_destroy(queue: d->workqueue);
14 }
15
16 if (d->name)
17 vfree(d->name);
18
19 vfree(d);
20 }
21}
22
23static void daemon_getref(struct daemon *d) {
24 enforce(refcount_inc_not_zero(&d->refcount), STALE_REFERENCE);
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_head_push_front(h: &daemon->timesharing_threads, n: &thread->list_node);
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_node_delete_init(n: &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_dec(&daemon->attrs.timesharing_threads);
82 } else {
83 daemon->background_thread = NULL;
84 set_bg_present(daemon, false);
85 }
86
87 vfree(self);
88
89 daemon_putref(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 (atomic_load_acq(&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_inc(&daemon->attrs.idle_timesharing_threads);
119
120 if (self->background)
121 semaphore_wait(s: &daemon->bg_sem);
122 else
123 semaphore_wait(s: &daemon->ts_sem);
124
125 atomic_dec(&daemon->attrs.idle_timesharing_threads);
126
127 if (atomic_load_acq(&daemon->state) == DAEMON_STATE_DESTROYING)
128 daemon_thread_exit(daemon, self);
129}
130
131static void daemon_main(void *a) {
132 cc_ignore(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 unreachable("daemon loop exited");
157}
158
159static struct daemon_thread *daemon_thread_create(struct daemon *daemon) {
160 struct daemon_thread *thread = vnew(thread, ALLOC_ZERO);
161 if (!thread)
162 return NULL;
163
164 thread->daemon = daemon;
165 list_node_init(n: &thread->list_node);
166
167 struct thread *t = thread_create(
168 ("daemon_%s_thread", daemon->name ? daemon->name : "unnamed"),
169 daemon_main, .private = thread,
170 .allowed_cpus = daemon->attrs.thread_cpu_mask);
171 if (!t) {
172 vfree(thread);
173 return NULL;
174 }
175
176 thread->thread = t;
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_getref(d: daemon);
195 struct daemon_thread *t = create(daemon);
196 if (!t) {
197 refcount_put_not_last(rc: &daemon->refcount);
198 return NULL;
199 }
200
201 if (!t->background)
202 atomic_inc(&daemon->attrs.timesharing_threads);
203
204 thread_enqueue(t: t->thread);
205
206 return t;
207}
208
209static void daemon_thread_destroy_unsafe(struct daemon_thread *dt) {
210 thread_free(t: dt->thread);
211 vfree(dt);
212}
213
214struct daemon *daemon_create_full(struct daemon_attributes *attrs,
215 struct daemon_work *timesharing_work,
216 struct daemon_work *background_work,
217 struct workqueue_attributes *wq_attrs,
218 const char *fmt, ...) {
219 va_list args;
220 va_start(args, fmt);
221
222 struct daemon *daemon = vnew(daemon, ALLOC_ZERO);
223 struct daemon_thread *dt = NULL, *bg = NULL;
224
225 if (!daemon)
226 goto err;
227
228 daemon->attrs = *attrs;
229
230 enforce(!cpu_mask_empty(&attrs->thread_cpu_mask),
231 "please set a valid CPU mask");
232 enforce(attrs->min_timesharing_threads <= attrs->max_timesharing_threads);
233
234 if (!DAEMON_FLAG_TEST(daemon, DAEMON_FLAG_NO_TS_THREADS))
235 enforce(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, 0, fmt, args_copy) + 1;
242 va_end(args_copy);
243
244 char *name = valloc(needed, ALLOC_ZERO);
245 if (!name)
246 goto err;
247
248 va_copy(args_copy, args);
249 vsnprintf(name, needed, fmt, args_copy);
250 va_end(args_copy);
251 daemon->name = name;
252 }
253
254 atomic_init(&daemon->attrs.background_thread_present, false);
255 atomic_init(&daemon->attrs.idle_timesharing_threads, 0);
256 atomic_init(&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 list_head_init(h: &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 ("workqueue_daemon_%s", daemon->name ? daemon->name : ""),
271 wq_attrs);
272
273 if (!wq)
274 goto err;
275
276 daemon->workqueue = wq;
277 }
278
279 if (!DAEMON_FLAG_TEST(daemon, DAEMON_FLAG_NO_TS_THREADS) &&
280 timesharing_work) {
281 for (size_t i = 0; i < attrs->min_timesharing_threads; i++) {
282 dt = daemon_thread_spawn(daemon, create: daemon_thread_create);
283 if (!dt)
284 goto err;
285
286 daemon_list_add(daemon, thread: dt);
287 }
288 }
289
290 if (background_work) {
291 bg = daemon_thread_spawn(daemon, create: daemon_thread_create_bg);
292 if (!bg)
293 goto err;
294
295 daemon->background_thread = bg;
296 set_bg_present(daemon, true);
297 }
298
299 atomic_store_rlx(&daemon->state, DAEMON_STATE_ACTIVE);
300
301 va_end(args);
302 return daemon;
303
304err:
305 if (dt)
306 daemon_thread_destroy_unsafe(dt);
307
308 if (daemon && daemon->name)
309 vfree(daemon->name);
310
311 if (daemon)
312 vfree(daemon);
313
314 va_end(args);
315 return NULL;
316}
317
318static void boost_bg_thread_to_ts(struct daemon *daemon) {
319 if (daemon->background_thread)
320 thread_set_timesharing(t: daemon->background_thread->thread);
321}
322
323/* Assume that all daemons must have daemon works
324 * finish executing before they can be safely destroyed */
325
326/* TODO: Actually use a refcount here */
327void daemon_destroy(struct daemon *daemon) {
328 /* Make all the threads go sleep on the semaphore */
329 atomic_store_release(&daemon->state, DAEMON_STATE_DESTROYING);
330
331 boost_bg_thread_to_ts(daemon);
332
333 /* Wait for all currently running threads to exit or sleep */
334 while (total_ts_workers(daemon) > idle_ts_workers(daemon))
335 scheduler_yield();
336
337 /* Wakes up everyone, and they'll read daemon->state == DESTROYING */
338 semaphore_postn(s: &daemon->ts_sem, n: total_ts_workers(daemon));
339
340 /* handle if existing, post once */
341 if (bg_present(daemon))
342 semaphore_post(s: &daemon->bg_sem);
343
344 /* Initial reference gone */
345 daemon_putref(d: daemon);
346}
347
348struct daemon_thread *daemon_spawn_worker(struct daemon *daemon) {
349 struct daemon_thread *dt =
350 daemon_thread_spawn(daemon, create: daemon_thread_create);
351 if (!dt)
352 return NULL;
353
354 daemon_list_add(daemon, thread: dt);
355 return dt;
356}
357
358enum workqueue_error daemon_submit_work(struct daemon *daemon,
359 struct work *work) {
360 enforce(daemon->workqueue &&
361 DAEMON_FLAG_TEST(daemon, DAEMON_FLAG_HAS_WORKQUEUE));
362 return workqueue_enqueue(queue: daemon->workqueue, work);
363}
364
365void daemon_wake_background_worker(struct daemon *daemon) {
366 semaphore_post(s: &daemon->bg_sem);
367}
368
369void daemon_wake_timesharing_worker(struct daemon *daemon) {
370 if (daemon_needs_spawn_worker(d: daemon))
371 daemon_spawn_worker(daemon);
372
373 semaphore_post(s: &daemon->ts_sem);
374}
375
376void daemon_wake_all_idle_timesharing_workers(struct daemon *daemon) {
377 if (daemon_needs_spawn_worker(d: daemon))
378 daemon_spawn_worker(daemon);
379
380 semaphore_postn(s: &daemon->ts_sem, n: idle_ts_workers(daemon));
381}
382
383void daemon_print(struct daemon *daemon) {
384 struct daemon_attributes *attrs = &daemon->attrs;
385 printf(format: "struct daemon \"%s\" = {\n", daemon->name ? daemon->name : "NULL");
386 printf(format: " .attrs = {\n");
387 printf(format: " .max_timesharing_threads = %zu\n",
388 attrs->max_timesharing_threads);
389 printf(format: " .idle_timesharing_threads = %zu\n",
390 atomic_load_rlx(&attrs->idle_timesharing_threads));
391 printf(format: " .timesharing_threads = %zu\n",
392 atomic_load_rlx(&attrs->timesharing_threads));
393 printf(format: " .flags = 0b%b\n", attrs->flags);
394 printf(format: " }\n");
395 printf(format: " .state = %s\n",
396 daemon_state_str(atomic_load_rlx(&daemon->state)));
397 printf(format: "}\n");
398}
399