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
32static bool dequeue_oneshot_task(struct workqueue *queue, struct work *out) {
33 uint64_t pos;
34 struct work *t;
35
36 while (true) {
37 pos = atomic_load_explicit(&queue->tail, memory_order_relaxed);
38 t = &queue->oneshot_works[pos % queue->attrs.capacity];
39 uint64_t seq = atomic_load_explicit(&t->seq, memory_order_acquire);
40 int64_t diff = (int64_t) seq - (int64_t) (pos + 1);
41
42 if (diff == 0) {
43 if (atomic_compare_exchange_weak_explicit(
44 &queue->tail, &pos, pos + 1, memory_order_acq_rel,
45 memory_order_relaxed)) {
46
47 *out = *t;
48 atomic_store_explicit(&t->seq, pos + queue->attrs.capacity,
49 memory_order_release);
50
51 atomic_fetch_sub(&queue->num_tasks, 1);
52
53 return true;
54 }
55
56 continue;
57 } else if (diff < 0) {
58 return false;
59 }
60 }
61}
62
63enum workqueue_error workqueue_enqueue_oneshot(struct workqueue *queue,
64 work_function func,
65 struct work_args args) {
66 if (!workqueue_usable(q: queue))
67 return WORKQUEUE_ERROR_UNUSABLE;
68
69 uint64_t pos;
70 struct work *t;
71
72 while (true) {
73 pos = atomic_load_explicit(&queue->head, memory_order_relaxed);
74 t = &queue->oneshot_works[pos % queue->attrs.capacity];
75 uint64_t seq = atomic_load_explicit(&t->seq, memory_order_acquire);
76 int64_t diff = (int64_t) seq - (int64_t) pos;
77
78 if (diff == 0) {
79 if (atomic_compare_exchange_weak_explicit(
80 &queue->head, &pos, pos + 1, memory_order_acq_rel,
81 memory_order_relaxed)) {
82
83 t->func = func;
84 t->args = args;
85
86 atomic_fetch_add(&queue->num_tasks, 1);
87
88 atomic_store_explicit(&t->seq, pos + 1, memory_order_release);
89
90 return signal_worker(queue);
91 }
92 } else if (diff < 0) {
93 return WORKQUEUE_ERROR_FULL;
94 }
95 }
96}
97
98int32_t workqueue_dequeue_task(struct workqueue *queue, struct work **out,
99 struct work *oneshot_task) {
100 if (dequeue_oneshot_task(queue, out: oneshot_task))
101 return DEQUEUE_FROM_ONESHOT_CODE;
102
103 enum irql irql = spin_lock_irq_disable(lock: &queue->work_lock);
104 struct list_head *lh = list_pop_front(head: &queue->works);
105 spin_unlock(lock: &queue->work_lock, old: irql);
106
107 if (lh) {
108 struct work *work = work_from_worklist_node(lh);
109 *out = work;
110 atomic_store(&work->enqueued, false);
111 atomic_fetch_sub(&queue->num_tasks, 1);
112 return DEQUEUE_FROM_REGULAR_CODE;
113 }
114
115 return 0;
116}
117
118enum workqueue_error workqueue_enqueue(struct workqueue *queue,
119 struct work *work) {
120 enum irql irql = spin_lock_irq_disable(lock: &queue->work_lock);
121 if (atomic_exchange(&work->active, true)) {
122 spin_unlock(lock: &queue->work_lock, old: irql);
123 return WORKQUEUE_ERROR_WORK_EXECUTING;
124 }
125
126 list_add_tail(new: &work->list_node, head: &queue->works);
127
128 atomic_store(&work->enqueued, true);
129
130 spin_unlock(lock: &queue->work_lock, old: irql);
131
132 atomic_fetch_add(&queue->num_tasks, 1);
133
134 return signal_worker(queue);
135}
136