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