1#include <sch/sched.h>
2#include <thread/thread.h>
3
4#include "internal.h"
5#include <mem/alloc.h>
6
7_Static_assert(WORKQUEUE_DEFAULT_MAX_IDLE_CHECK / 4 >
8 WORKQUEUE_DEFAULT_MIN_IDLE_CHECK,
9 "");
10
11static time_ms_t get_inactivity_timeout(struct workqueue *queue) {
12 uint32_t num_workers = atomic_load(&queue->num_workers);
13 size_t min = queue->attrs.idle_check.min;
14 size_t max = queue->attrs.idle_check.max;
15
16 if (num_workers <= (queue->attrs.max_workers / 8))
17 return max;
18
19 if (num_workers <= (queue->attrs.max_workers / 4))
20 return max / 2;
21
22 if (num_workers <= (queue->attrs.max_workers / 2))
23 return max / 4;
24
25 return min;
26}
27
28void workqueue_link_thread_and_worker(struct worker *worker,
29 struct thread *thread) {
30 worker->present = true;
31 worker->timeout_ran = true;
32 worker->thread = thread;
33
34 thread->private = worker;
35}
36
37static bool claim_spawner(struct workqueue *p) {
38 return atomic_flag_test_and_set_explicit(&p->spawner_flag_internal,
39 memory_order_acq_rel) == 0;
40}
41
42static void release_spawner(struct workqueue *p) {
43 atomic_flag_clear_explicit(&p->spawner_flag_internal, memory_order_release);
44}
45
46static void worker_init(struct workqueue *queue, struct worker *w,
47 struct thread *t) {
48 INIT_LIST_HEAD(list: &w->list_node);
49 w->thread = t;
50 w->present = true;
51 w->workqueue = queue;
52 w->timeout_ran = true;
53 queue->last_spawn_attempt = time_get_ms();
54
55 t->private = w;
56
57 atomic_fetch_add(&queue->num_workers, 1);
58}
59
60static struct thread *workqueue_worker_thread_create(struct workqueue *queue) {
61 return worker_create(mask: queue->attrs.worker_cpu_mask,
62 niceness: queue->attrs.worker_niceness);
63}
64
65static void workqueue_enqueue_thread(struct workqueue *queue,
66 struct thread *t) {
67 if (WORKQUEUE_FLAG_TEST(queue, WORKQUEUE_FLAG_PERMANENT)) {
68 thread_enqueue_on_core(t, core_id: queue->core);
69 } else {
70 thread_enqueue(t);
71 }
72}
73
74struct worker *workqueue_worker_create(struct workqueue *queue) {
75 if (queue->attrs.flags & WORKQUEUE_FLAG_STATIC_WORKERS) {
76 enum irql irql = spin_lock_irq_disable(&queue->worker_array_lock);
77 struct worker *ret = NULL;
78 for (size_t i = 0; i < queue->attrs.max_workers; i++) {
79 if (queue->worker_array[i].thread == NULL) {
80 ret = &queue->worker_array[i];
81 goto out;
82 }
83 }
84
85 out:
86 spin_unlock(&queue->worker_array_lock, irql);
87 return ret;
88 } else {
89 return kmalloc(sizeof(struct worker), ALLOC_FLAGS_ZERO);
90 }
91}
92
93static void workqueue_init_new_worker(struct workqueue *queue, struct worker *w,
94 struct thread *t) {
95 w->inactivity_check_period = get_inactivity_timeout(queue);
96
97 worker_init(queue, w, t);
98 workqueue_add_worker(wq: queue, wker: w);
99 workqueue_enqueue_thread(queue, t);
100}
101
102/* This is only for non-request based worker thread spawning */
103bool workqueue_spawn_worker_internal(struct workqueue *queue) {
104 if (!claim_spawner(p: queue))
105 return false;
106
107 struct worker *w = workqueue_worker_create(queue);
108 if (!w)
109 goto fail;
110
111 struct thread *t = workqueue_worker_thread_create(queue);
112 if (!t)
113 goto fail;
114
115 workqueue_init_new_worker(queue, w, t);
116
117 release_spawner(p: queue);
118 return true;
119
120fail:
121 if (w && !(queue->attrs.flags & WORKQUEUE_FLAG_STATIC_WORKERS))
122 kfree(w);
123
124 release_spawner(p: queue);
125 return false;
126}
127
128bool workqueue_should_spawn_worker(struct workqueue *queue) {
129 time_ms_t now = time_get_ms();
130 if (now - queue->last_spawn_attempt <= queue->attrs.spawn_delay)
131 return false;
132
133 bool no_idle = atomic_load(&queue->idle_workers) == 0;
134 bool work_pending = !workqueue_empty(queue);
135
136 bool under_limit =
137 atomic_load(&queue->num_workers) < queue->attrs.max_workers;
138
139 /* Permanent workqueues are per-core and spawning
140 * extra threads on them doesn't help */
141 bool non_permanent = !WORKQUEUE_FLAG_TEST(queue, WORKQUEUE_FLAG_PERMANENT);
142
143 return no_idle && work_pending && under_limit && non_permanent;
144}
145
146bool workqueue_try_spawn_worker(struct workqueue *queue) {
147 if (!workqueue_should_spawn_worker(queue))
148 return false;
149
150 if (irq_in_interrupt()) {
151 workqueue_set_needs_spawn(queue, true);
152 return true;
153 }
154
155 return workqueue_spawn_worker_internal(queue);
156}
157
158struct thread *worker_create(struct cpu_mask mask, nice_t niceness) {
159 struct thread *ret = thread_create_custom_stack(
160 name: "workqueue_worker", entry_point: worker_main, NULL, THREAD_STACK_SIZE);
161 if (!ret)
162 return NULL;
163
164 ret->niceness = niceness;
165 ret->allowed_cpus = mask;
166
167 return ret;
168}
169