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