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