1#include <sch/sched.h>
2#include <string.h>
3#include <thread/thread.h>
4
5#include "internal.h"
6
7static enum wake_reason worker_wait(struct workqueue *wq, struct worker *w,
8 enum irql irql, enum irql *out) {
9 enum wake_reason sig;
10
11 atomic_fetch_add(&wq->idle_workers, 1);
12
13 /* Do not garbage collect workers, just wait... */
14 if (wq->attrs.flags & WORKQUEUE_FLAG_NO_WORKER_GC) {
15 sig = condvar_wait(cv: &wq->queue_cv, lock: &wq->lock, irql, out);
16 } else {
17 if (w->timeout_ran && !w->is_permanent) {
18 sig = condvar_wait_timeout(cv: &wq->queue_cv, lock: &wq->lock,
19 timeout_ms: w->inactivity_check_period, irql, out);
20 w->timeout_ran = false;
21 } else {
22 sig = condvar_wait(cv: &wq->queue_cv, lock: &wq->lock, irql, out);
23 }
24 }
25
26 atomic_fetch_sub(&wq->idle_workers, 1);
27
28 if (sig == WAKE_REASON_TIMEOUT && !ignore_timeouts(q: wq)) {
29 w->timeout_ran = true;
30 if (!w->idle) {
31 w->idle = true;
32 w->start_idle = time_get_ms();
33 }
34 }
35
36 return sig;
37}
38
39static inline bool worker_should_exit(const struct worker *worker,
40 enum wake_reason signal) {
41 if (worker->next_action == WORKER_NEXT_ACTION_EXIT)
42 return true;
43
44 const time_t timeout = worker->inactivity_check_period;
45
46 /* We don't mark `idle` if timeouts are to be ignored */
47 if (!worker->is_permanent && worker->idle && signal == WAKE_REASON_TIMEOUT)
48 if (time_get_ms() - worker->start_idle >= timeout)
49 return true;
50
51 return false;
52}
53
54static void worker_reset(struct worker *worker) {
55 memset(worker, 0, sizeof(struct worker));
56}
57
58static void worker_destroy(struct workqueue *queue, struct worker *worker) {
59 if (queue->attrs.flags & WORKQUEUE_FLAG_STATIC_WORKERS) {
60 enum irql irql = spin_lock_irq_disable(lock: &queue->worker_array_lock);
61
62 bool found = false;
63
64 for (size_t i = 0; i < queue->attrs.max_workers; i++) {
65 struct worker *maybe = &queue->worker_array[i];
66 if (maybe == worker) {
67 found = true;
68 worker_reset(worker);
69 break;
70 }
71 }
72
73 if (!found)
74 panic("Potential corrupted worker %p in STATIC_WORKERS "
75 "workqueue",
76 worker);
77
78 spin_unlock(lock: &queue->worker_array_lock, old: irql);
79 } else {
80 kfree(worker);
81 }
82}
83
84static void worker_exit(struct workqueue *queue, struct worker *worker,
85 enum irql irql) {
86 worker->present = false;
87 worker->idle = false;
88 worker->should_exit = true;
89
90 worker->thread = NULL;
91
92 workqueue_remove_worker(wq: queue, worker);
93 atomic_fetch_sub(&queue->num_workers, 1);
94
95 spin_unlock(lock: &queue->lock, old: irql);
96
97 worker_destroy(queue, worker);
98
99 workqueue_put(queue);
100
101 thread_exit();
102}
103
104void worker_main(void *unused) {
105 (void) unused;
106
107 struct worker *w = thread_get_current()->private;
108 struct workqueue *queue = w->workqueue;
109 kassert(w);
110
111 kassert(workqueue_get(queue));
112
113 while (true) {
114
115 struct work *task = NULL;
116 struct work oneshot_task = {0};
117 int32_t dequeue = workqueue_dequeue_task(queue, out: &task, oneshot_out: &oneshot_task);
118 if (dequeue > 0) {
119 w->last_active = time_get_ms();
120 w->idle = false;
121
122 if (dequeue == DEQUEUE_FROM_ONESHOT_CODE) {
123 work_execute(task: &oneshot_task);
124 } else {
125 work_execute(task);
126 }
127
128 continue;
129 }
130
131 enum irql irql = workqueue_lock(workqueue: queue);
132
133 while (workqueue_empty(queue)) {
134 if (workqueue_needs_spawn(queue)) {
135 workqueue_set_needs_spawn(queue, false);
136 if (WORKQUEUE_FLAG_TEST(queue, WORKQUEUE_FLAG_AUTO_SPAWN))
137 workqueue_spawn_worker_internal(queue);
138 }
139
140 enum wake_reason signal = worker_wait(wq: queue, w, irql, out: &irql);
141
142 if (worker_should_exit(worker: w, signal))
143 worker_exit(queue, worker: w, irql);
144 }
145
146 spin_unlock(lock: &queue->lock, old: irql);
147 }
148}
149