1#include "internal.h"
2#include <thread/thread.h>
3
4static void signal_callback(struct thread *t) {
5 if (t) {
6 ((struct worker *) (t->private))->next_action = WORKER_NEXT_ACTION_RUN;
7 }
8}
9
10static enum workqueue_error signal_worker(struct workqueue *queue) {
11 struct thread *woke =
12 condvar_signal_callback(cv: &queue->queue_cv, cb: signal_callback);
13
14 enum workqueue_error ret = WORKQUEUE_ERROR_OK;
15
16 /* No worker was woken up because all are busy */
17 if (!woke) {
18 if (!WORKQUEUE_FLAG_TEST(queue, WORKQUEUE_FLAG_AUTO_SPAWN)) {
19 ret = WORKQUEUE_ERROR_NEED_NEW_WORKER;
20 } else if (workqueue_current_worker_count(q: queue) ==
21 queue->attrs.max_workers) {
22 ret = WORKQUEUE_ERROR_NEED_NEW_WQ;
23 }
24 }
25
26 if (WORKQUEUE_FLAG_TEST(queue, WORKQUEUE_FLAG_AUTO_SPAWN))
27 workqueue_try_spawn_worker(queue);
28
29 return ret;
30}
31
32bool workqueue_dequeue_task(struct workqueue *queue, struct work **out) {
33 enum irql irql = spin_lock_high(&queue->work_lock);
34 struct list_node *lh = list_head_pop_front(h: &queue->works);
35 spin_unlock(&queue->work_lock, irql);
36
37 if (lh) {
38 struct work *work = work_from_worklist_node(lh);
39 *out = work;
40 atomic_store(&work->enqueued, false);
41 atomic_dec(&queue->num_tasks);
42 return true;
43 }
44
45 return false;
46}
47
48enum workqueue_error workqueue_enqueue(struct workqueue *queue,
49 struct work *work) {
50 enum irql irql = spin_lock_high(&queue->work_lock);
51 if (atomic_exchange(&work->active, true)) {
52 spin_unlock(&queue->work_lock, irql);
53 return WORKQUEUE_ERROR_WORK_EXECUTING;
54 }
55
56 list_head_push_back(h: &queue->works, n: &work->list_node);
57
58 atomic_store(&work->enqueued, true);
59
60 spin_unlock(&queue->work_lock, irql);
61
62 atomic_inc(&queue->num_tasks);
63
64 return signal_worker(queue);
65}
66