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