1#include <sch/sched.h>
2#include <thread/thread.h>
3
4#include "internal.h"
5#include <mem/alloc.h>
6
7static_assert(WORKQUEUE_DEFAULT_MAX_IDLE_CHECK / 4 >
8 WORKQUEUE_DEFAULT_MIN_IDLE_CHECK);
9
10static time_ms_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_acq_rel(&p->spawner_flag_internal) == 0;
38}
39
40static void release_spawner(struct workqueue *p) {
41 atomic_flag_clear_release(&p->spawner_flag_internal);
42}
43
44static void worker_init(struct workqueue *queue, struct worker *w,
45 struct thread *t) {
46 list_node_init(n: &w->list_node);
47 w->thread = t;
48 w->present = true;
49 w->workqueue = queue;
50 w->timeout_ran = true;
51 queue->last_spawn_attempt = time_get_ms();
52
53 t->private = w;
54
55 atomic_inc(&queue->num_workers);
56}
57
58static struct thread *workqueue_worker_thread_create(struct workqueue *queue) {
59 return worker_create(mask: queue->attrs.worker_cpu_mask,
60 niceness: queue->attrs.worker_niceness);
61}
62
63static void workqueue_enqueue_thread(struct workqueue *queue,
64 struct thread *t) {
65 if (WORKQUEUE_FLAG_TEST(queue, WORKQUEUE_FLAG_PERMANENT)) {
66 thread_enqueue_on_core(t, core_id: queue->core);
67 } else {
68 thread_enqueue(t);
69 }
70}
71
72static struct worker *workqueue_worker_create(struct workqueue *queue) {
73 if (queue->attrs.flags & WORKQUEUE_FLAG_STATIC_WORKERS) {
74 enum irql irql = spin_lock_high(&queue->worker_array_lock);
75 struct worker *ret = NULL;
76 for (size_t i = 0; i < queue->attrs.max_workers; i++) {
77 if (queue->worker_array[i].thread == NULL) {
78 ret = &queue->worker_array[i];
79 goto out;
80 }
81 }
82
83 out:
84 spin_unlock(&queue->worker_array_lock, irql);
85 return ret;
86 } else {
87 struct worker *w = vnew(w, ALLOC_ZERO);
88 return w;
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 vfree(w);
122
123 release_spawner(p: queue);
124 return false;
125}
126
127bool workqueue_should_spawn_worker(struct workqueue *queue) {
128 time_ms_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 struct thread *ret =
159 thread_create("workqueue_worker", worker_main, .allowed_cpus = mask);
160 if (!ret)
161 return NULL;
162
163 ret->niceness = niceness;
164
165 return ret;
166}
167