1/* @title: Workqueues */
2#pragma once
3#include <atomic.h>
4#include <smp/topology.h>
5#include <stdbool.h>
6#include <stdint.h>
7#include <structures/list.h>
8#include <sync/condvar.h>
9#include <sync/spinlock.h>
10#include <time/timer.h>
11#include <types/refcount.h>
12#include <types/types.h>
13
14typedef void (*work_function)(void *arg, void *arg2);
15
16struct work_args {
17 void *arg1;
18 void *arg2;
19};
20#define WORK_ARGS(a, b) ((struct work_args) {.arg1 = a, .arg2 = b})
21
22struct work {
23 work_function func;
24 struct work_args args;
25
26 struct list_node list_node;
27
28 atomic_bool enqueued;
29 atomic_bool active;
30};
31
32struct delayed_work {
33 struct work work;
34 struct timer timer;
35 struct workqueue *wq;
36};
37
38enum worker_next_action {
39 WORKER_NEXT_ACTION_RUN,
40 WORKER_NEXT_ACTION_EXIT,
41};
42
43struct worker {
44 struct thread *thread; /* Assoc. thread. */
45 struct workqueue *workqueue; /* Assoc. wq */
46
47 time_ms_t last_active; /* Monotonic starting time of most
48 * recent work execution */
49
50 time_ms_t inactivity_check_period; /* How much time in between
51 * timeout GC events */
52
53 time_ms_t start_idle; /* Monotonic starting time of
54 * most recent idle */
55
56 /* Internal flags */
57 bool timeout_ran : 1;
58 bool should_exit : 1;
59 bool is_permanent : 1;
60 bool present : 1;
61 bool idle : 1;
62
63 enum worker_next_action next_action;
64
65 struct list_node list_node;
66};
67
68/* TODO: Get in profiling.h and put these under there */
69#ifdef TESTS
70struct workqueue_stats {
71 uint64_t total_tasks_added; /* Total # of tasks submitted to the queue */
72 uint64_t total_tasks_executed; /* Number of tasks successfully executed */
73 uint64_t total_workers_spawned; /* Total worker threads spawned */
74 uint64_t total_worker_exits; /* Total workers that exited */
75 uint64_t max_queue_length; /* Max observed length of the task queue */
76 uint64_t current_queue_length; /* Current length of the queue */
77 uint64_t total_spawn_attempts; /* # times spawn_worker was attempted */
78 uint64_t total_spawn_failures; /* Number of times spawn_worker failed */
79 uint64_t num_idle_workers; /* Snapshot of current idle workers */
80 uint64_t num_active_workers; /* Snapshot of current active workers */
81};
82#endif
83
84enum workqueue_flags : uint16_t {
85 WORKQUEUE_FLAG_PERMANENT = 1 << 1, /* Inverse: On-demand
86 *
87 * Permanent workqueues are attached
88 * to each core and are always Active
89 * workqueues with on-demand
90 * worker spawning */
91
92 WORKQUEUE_FLAG_AUTO_SPAWN = 1 << 2, /* Inverse: No auto spawn
93 *
94 * This flag allows workqueues
95 * with multiple workers to
96 * spawn workers automatically if they
97 * detect that workers are busy.
98 *
99 * Otherwise, that doesn't happen, and
100 * workers are manually spawned */
101
102 WORKQUEUE_FLAG_NAMED = 1 << 3, /* Has name - will honor (fmt, ...) */
103
104 WORKQUEUE_FLAG_STATIC_WORKERS = 1 << 4, /* `struct worker` will be
105 * statically allocated
106 * during workqueue creation.
107 *
108 * This allows allocators to
109 * safely use workqueues that
110 * dynamically spawn threads,
111 * but shouldn't be used everywhere
112 * because it can waste memory */
113
114 WORKQUEUE_FLAG_NO_WORKER_GC = 1 << 5, /* Do not timeout workers */
115
116 WORKQUEUE_FLAG_ISR_SAFE = 1 << 6,
117
118 WORKQUEUE_FLAG_NO_AUTO_SPAWN = 0, /* Do not auto spawn workers */
119 WORKQUEUE_FLAG_ON_DEMAND = 0, /* Inverse of a permanent workqueue */
120 WORKQUEUE_FLAG_NAMELESS = 0,
121 WORKQUEUE_FLAG_NON_STATIC_WORKERS = 0,
122 WORKQUEUE_FLAG_WORKER_GC = 0,
123 WORKQUEUE_FLAG_NON_ISR_SAFE = 0,
124
125 WORKQUEUE_FLAG_DEFAULTS = WORKQUEUE_FLAG_AUTO_SPAWN | WORKQUEUE_FLAG_NAMED,
126};
127#define WORKQUEUE_FLAG_SET(q, f) (q->attrs.flags |= f)
128#define WORKQUEUE_FLAG_UNSET(q, f) (q->attrs.flags &= ~f)
129#define WORKQUEUE_FLAG_TEST(q, f) (q->attrs.flags & f)
130
131enum workqueue_state : uint16_t {
132 WORKQUEUE_STATE_DEAD, /* Gone, about to be freed */
133 WORKQUEUE_STATE_DESTROYING, /* Destroying - Do not spawn threads */
134 WORKQUEUE_STATE_ACTIVE, /* Active */
135};
136
137#define WORKQUEUE_STATE_SET(q, s) (atomic_store(&q->state, s))
138#define WORKQUEUE_STATE_GET(q) (atomic_load(&q->state))
139
140struct workqueue_attributes {
141 size_t min_workers; /* If set to 0, this field will be treated as a "1" */
142 size_t max_workers;
143 time_ms_t spawn_delay;
144 nice_t worker_niceness;
145 struct {
146 uint64_t min;
147 uint64_t max;
148 } idle_check;
149
150 enum workqueue_flags flags;
151 struct cpu_mask worker_cpu_mask;
152};
153
154#define WORKQUEUE_DEFAULT_MAX_WORKERS 16
155#define WORKQUEUE_DEFAULT_SPAWN_DELAY 150
156#define WORKQUEUE_DEFAULT_MIN_IDLE_CHECK SECONDS_TO_MS(2)
157#define WORKQUEUE_DEFAULT_MAX_IDLE_CHECK SECONDS_TO_MS(40)
158#define WORKQUEUE_DEFAULT_IDLE_CHECK \
159 {.max = WORKQUEUE_DEFAULT_MAX_IDLE_CHECK, \
160 .min = WORKQUEUE_DEFAULT_MIN_IDLE_CHECK}
161
162struct workqueue {
163 char *name;
164
165 atomic_bool ignore_timeouts;
166
167 struct spinlock work_lock; /* For works */
168 struct spinlock worker_lock; /* For worker list */
169 struct spinlock worker_array_lock; /* For worker array */
170 struct spinlock lock; /* For condvar */
171
172 struct condvar queue_cv;
173
174 struct list_head workers;
175 struct list_head works;
176 struct worker *worker_array; /* if STATIC_WORKER is needed */
177
178 atomic_uint64_t head;
179 atomic_uint64_t tail;
180
181 atomic_bool spawn_pending; /* Some enqueue wants us to spawn a worker */
182 atomic_uint32_t num_tasks; /* How many tasks do we have in the ringbuf */
183
184 atomic_uint32_t num_workers; /* Current # workers */
185 atomic_uint32_t idle_workers; /* # idle */
186
187 cpu_id_t core;
188 time_ms_t last_spawn_attempt;
189
190 atomic_flag spawner_flag_internal;
191
192 struct workqueue_attributes attrs;
193
194#ifdef TESTS
195 struct workqueue_stats stats;
196#endif
197
198 atomic(enum workqueue_state) state; /* Atomic to avoid
199 * race where stale
200 * state is seen */
201 struct thread_request *request;
202 refcount_t refcount;
203};
204
205/* Positive values are success with a message,
206 * zero is success with nothing special.
207 *
208 * Negative values are errors */
209enum workqueue_error : int32_t {
210 WORKQUEUE_ERROR_NEED_NEW_WORKER = 4, /* For manual worker spawn */
211 WORKQUEUE_ERROR_NEED_NEW_WQ = 3, /* All worker slots filled */
212 WORKQUEUE_ERROR_OK = 0, /* No message */
213 WORKQUEUE_ERROR_FULL = -1, /* Full ringbuffer */
214 WORKQUEUE_ERROR_WLIST_EXECUTING = -2, /* Worklist executing */
215 WORKQUEUE_ERROR_UNUSABLE = -3, /* Being destroyed, etc. */
216 WORKQUEUE_ERROR_WORK_EXECUTING = -4,
217};
218
219void delayed_work_init(struct delayed_work *dwork, work_function fn,
220 struct work_args args);
221bool delayed_work_schedule(struct delayed_work *dwork, time_ms_t delay_ms);
222bool delayed_work_schedule_on(struct workqueue *wq, struct delayed_work *dwork,
223 time_ms_t delay_ms);
224bool delayed_work_cancel(struct delayed_work *dwork);
225bool delayed_work_cancel_sync(struct delayed_work *dwork);
226
227void workqueues_permanent_init(void);
228
229struct workqueue *workqueue_create_internal(struct workqueue_attributes *attrs,
230 const char *fmt, ...)
231 cc_printf_like(2, 3);
232
233#define workqueue_create(name, attrs) \
234 workqueue_create_internal((attrs), PP_UNPAREN(name))
235struct workqueue *workqueue_create_default(const char *fmt, ...)
236 cc_printf_like(1, 2);
237struct work *work_create(work_function func, struct work_args args);
238struct work *work_init(struct work *work, work_function fn,
239 struct work_args args);
240
241void workqueue_free(struct workqueue *queue);
242
243enum workqueue_error workqueue_enqueue(struct workqueue *queue,
244 struct work *work);
245
246enum workqueue_error workqueue_add(struct work *work) cc_warn_unused_result;
247
248enum workqueue_error workqueue_add_remote(struct work *work)
249 cc_warn_unused_result;
250
251enum workqueue_error workqueue_add_local(struct work *work)
252 cc_warn_unused_result;
253
254enum workqueue_error workqueue_add_fast(struct work *work)
255 cc_warn_unused_result;
256
257void work_execute(struct work *cc_nonnull_ptr task);
258bool workqueue_should_spawn_worker(struct workqueue *queue);
259
260void workqueue_kick(struct workqueue *queue);
261void workqueue_destroy(struct workqueue *cc_nonnull_ptr queue);
262
263void worker_main(void *);
264
265static inline bool work_active(struct work *work) {
266 return atomic_load(&work->active);
267}
268