include/thread/workqueue.h View source View on GitHubstruct work_args {
void *arg1;
void *arg2;
};
struct deferred_event {
size_t timer;
timestamp_t timestamp_ms;
work_function callback;
struct work_args args;
struct deferred_event *next;
};
struct work {
work_function func;
struct work_args args;
struct list_head list_node;
atomic_bool enqueued;
atomic_bool active;
_Atomic uint64_t seq;
};
struct worker {
struct thread *thread;
struct workqueue *workqueue;
time_t last_active;
time_t inactivity_check_period;
time_t start_idle;
bool timeout_ran : 1;
bool should_exit : 1;
bool is_permanent : 1;
bool present : 1;
bool idle : 1;
enum worker_next_action next_action;
struct list_head list_node;
};
struct workqueue_stats {
uint64_t total_tasks_added;
uint64_t total_tasks_executed;
uint64_t total_workers_spawned;
uint64_t total_worker_exits;
uint64_t max_queue_length;
uint64_t current_queue_length;
uint64_t total_spawn_attempts;
uint64_t total_spawn_failures;
uint64_t num_idle_workers;
uint64_t num_active_workers;
};
struct workqueue_attributes {
size_t min_workers;
size_t max_workers;
size_t capacity;
time_t spawn_delay;
nice_t worker_niceness;
struct {
uint64_t min;
uint64_t max;
} idle_check;
enum workqueue_flags flags;
struct cpu_mask worker_cpu_mask;
};
struct workqueue {
char *name;
atomic_bool ignore_timeouts;
struct spinlock work_lock;
struct spinlock worker_lock;
struct spinlock worker_array_lock;
struct spinlock lock;
struct condvar queue_cv;
struct work *oneshot_works;
struct list_head workers;
struct list_head works;
struct worker *worker_array;
_Atomic uint64_t head;
_Atomic uint64_t tail;
atomic_bool spawn_pending;
_Atomic uint32_t num_tasks;
_Atomic uint32_t num_workers;
_Atomic uint32_t idle_workers;
cpu_id_t core;
time_t last_spawn_attempt;
atomic_flag spawner_flag_internal;
struct workqueue_attributes attrs;
_Atomic enum workqueue_state state;
struct thread_request *request;
refcount_t refcount;
};
enum worker_next_action {
WORKER_NEXT_ACTION_RUN,
WORKER_NEXT_ACTION_EXIT,
};
enum workqueue_flags : uint16_t {
WORKQUEUE_FLAG_PERMANENT = 1 << 1,
WORKQUEUE_FLAG_AUTO_SPAWN = 1 << 2,
WORKQUEUE_FLAG_NAMED = 1 << 3,
WORKQUEUE_FLAG_STATIC_WORKERS = 1 << 4,
WORKQUEUE_FLAG_NO_WORKER_GC = 1 << 5,
WORKQUEUE_FLAG_ISR_SAFE = 1 << 6,
WORKQUEUE_FLAG_NO_AUTO_SPAWN = 0,
WORKQUEUE_FLAG_ON_DEMAND = 0,
WORKQUEUE_FLAG_NAMELESS = 0,
WORKQUEUE_FLAG_NON_STATIC_WORKERS = 0,
WORKQUEUE_FLAG_WORKER_GC = 0,
WORKQUEUE_FLAG_NON_ISR_SAFE = 0,
WORKQUEUE_FLAG_DEFAULTS = WORKQUEUE_FLAG_AUTO_SPAWN | WORKQUEUE_FLAG_NAMED,
};
enum workqueue_state : uint16_t {
WORKQUEUE_STATE_DEAD,
WORKQUEUE_STATE_DESTROYING,
WORKQUEUE_STATE_ACTIVE,
};
enum workqueue_error : int32_t {
WORKQUEUE_ERROR_NEED_NEW_WORKER = 4,
WORKQUEUE_ERROR_NEED_NEW_WQ = 3,
WORKQUEUE_ERROR_OK = 0,
WORKQUEUE_ERROR_FULL = -1,
WORKQUEUE_ERROR_WLIST_EXECUTING = -2,
WORKQUEUE_ERROR_UNUSABLE = -3,
WORKQUEUE_ERROR_WORK_EXECUTING = -4,
};
typedef void (*work_function)(void * arg, void * arg2);
void defer_init(void);
bool defer_enqueue(work_function func, struct work_args args, uint64_t delay_ms);
void workqueues_permanent_init(void);
struct workqueue * workqueue_create(const char *fmt, struct workqueue_attributes *attrs);
struct workqueue * workqueue_create_default(const char *fmt);
struct work * work_create(work_function func, struct work_args args);
struct work * work_init(struct work *work, work_function fn, struct work_args args);
void workqueue_free(struct workqueue *queue);
enum workqueue_error workqueue_enqueue_oneshot(struct workqueue *queue, work_function func, struct work_args args);
enum workqueue_error workqueue_enqueue(struct workqueue *queue, struct work *work);
enum workqueue_error __warn_unused_result workqueue_add_oneshot(work_function func, struct work_args args);
enum workqueue_error __warn_unused_result workqueue_add_remote_oneshot(work_function func, struct work_args args);
enum workqueue_error __warn_unused_result workqueue_add_local_oneshot(work_function func, struct work_args args);
enum workqueue_error __warn_unused_result workqueue_add_fast_oneshot(work_function func, struct work_args args);
enum workqueue_error __warn_unused_result workqueue_add(struct work *work);
enum workqueue_error __warn_unused_result workqueue_add_remote(struct work *work);
enum workqueue_error __warn_unused_result workqueue_add_local(struct work *work);
enum workqueue_error __warn_unused_result workqueue_add_fast(struct work *work);
void work_execute(struct work *task);
bool workqueue_should_spawn_worker(struct workqueue *queue);
void workqueue_kick(struct workqueue *queue);
void workqueue_destroy(struct workqueue *queue);
void worker_main(void * *);
bool work_active(struct work *work);
#define WORK_ARGS(a, b) ((struct work_args) {.arg1 = a, .arg2 = b})
#define WORKQUEUE_FLAG_SET(q, f) (q->attrs.flags |= f)
#define WORKQUEUE_FLAG_UNSET(q, f) (q->attrs.flags &= ~f)
#define WORKQUEUE_FLAG_TEST(q, f) (q->attrs.flags & f)
#define WORKQUEUE_STATE_SET(q, s) (atomic_store(&q->state, s))
#define WORKQUEUE_STATE_GET(q) (atomic_load(&q->state))
#define WORKQUEUE_DEFAULT_CAPACITY 512
#define WORKQUEUE_DEFAULT_MAX_WORKERS 16
#define WORKQUEUE_DEFAULT_SPAWN_DELAY 150
#define WORKQUEUE_DEFAULT_MIN_IDLE_CHECK SECONDS_TO_MS(2)
#define WORKQUEUE_DEFAULT_MAX_IDLE_CHECK SECONDS_TO_MS(40)
#define WORKQUEUE_DEFAULT_IDLE_CHECK \
{.max = WORKQUEUE_DEFAULT_MAX_IDLE_CHECK, \
.min = WORKQUEUE_DEFAULT_MIN_IDLE_CHECK}