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