1#include <atomic.h>
2#include <enforce.h>
3#include <mem/alloc.h>
4#include <mem/nofail.h>
5#include <sch/sched.h>
6#include <smp/core.h>
7#include <stdint.h>
8#include <sync/once_latch.h>
9#include <thread/dpc.h>
10
11static struct dpc *dpc_steal_queue(struct dpc_queue *dq) {
12 struct dpc *list = atomic_xchg_acq(&dq->head, NULL);
13
14 if (!list)
15 return NULL;
16
17 struct dpc *rev = NULL;
18
19 while (list) {
20 struct dpc *next = atomic_load_rlx(&list->next);
21 atomic_store_rlx(&list->next, rev);
22 rev = list;
23 list = next;
24 }
25
26 return rev;
27}
28
29static void dpc_execute_all_in_queue(struct dpc_queue *dq) {
30 while (true) {
31 struct dpc *it = dpc_steal_queue(dq);
32 if (!it)
33 break;
34
35 while (it) {
36 struct dpc *next = atomic_load_rlx(&it->next);
37 atomic_store_release(&it->enqueued, false);
38 struct dpc_context dc = {.dpc = it};
39 it->func(&dc, it->a, it->b);
40 it = next;
41 }
42 }
43}
44
45void dpc_drain_local(void) {
46 struct core *me = smp_core(cond: TOPC_IRQL);
47 if (atomic_load_rlx(&me->in_resched))
48 return;
49
50 /* Recursion guard */
51 if (atomic_exchange(&me->executing_dpcs, true))
52 return;
53
54 enforce(irql_get() == IRQL_DISPATCH_LEVEL, INVALID_CONTEXT);
55
56 size_t cpu = me->id;
57 struct dpc_cpu *dc = &global.dpc_data[cpu];
58
59again:
60 do {
61 dpc_execute_all_in_queue(dq: &dc->queue);
62 } while (atomic_load_rlx(&dc->queue.head) != NULL);
63
64 atomic_store(&me->executing_dpcs, false);
65
66 /* If an IPI lands here, we need to go do it again,
67 * so we don't lose a DPC */
68 if (atomic_load(&dc->queue.head) != NULL &&
69 !atomic_exchange(&me->executing_dpcs, true))
70 goto again;
71}
72
73void dpc_run_local(void) {
74 enum irql irql = irql_raise(new_level: IRQL_DISPATCH_LEVEL);
75 dpc_drain_local();
76
77 /* The raise is only used to satisfy the IRQL requirement here,
78 * and the matching lower shouldn't reschedule and recurse,
79 * as scheduler_yield() will lower its IRQL too */
80 irql_lower_no_resched(old_level: irql);
81}
82
83void dpc_run_dpcs_from_irq(void) {
84 dpc_run_local();
85}
86
87static void dpc_queue_enqueue(struct dpc_queue *dq, struct dpc *d) {
88 while (true) {
89 struct dpc *old_head = atomic_load_acq(&dq->head);
90 atomic_store_rlx(&d->next, old_head);
91 if (atomic_cas_weak(&dq->head, &old_head, d, mo_release, mo_rlx)) {
92 break;
93 }
94 cpu_pause();
95 }
96}
97
98bool dpc_enqueue_on_cpu(size_t cpu, struct dpc *cc_nonnull_ptr d) {
99 enforce_precond(d);
100
101 if (atomic_xchg_acq_rel(&d->enqueued, true))
102 return false;
103
104 struct dpc_cpu *dc = &global.dpc_data[cpu];
105
106 /* Clear next pointer then push via CAS loop */
107 atomic_store_rlx(&d->next, NULL);
108
109 struct dpc_queue *dq = &dc->queue;
110 dpc_queue_enqueue(dq, d);
111
112 scheduler_force_run_dpcs(cpu);
113
114 return true;
115}
116
117bool dpc_enqueue_local(struct dpc *d) {
118 return dpc_enqueue_on_cpu(cpu: smp_id_raw(), d);
119}
120
121void dpc_init_percpu(void) {
122 global.dpc_data =
123 nofail_vnews(global.dpc_data, global.core_count, ALLOC_ZERO);
124 size_t i;
125 for_each_cpu_id(i) {
126 atomic_store_rlx(&global.dpc_data[i].queue.head, NULL);
127 }
128}
129
130struct dpc *dpc_init(struct dpc *d, dpc_func_t fn, void *a, void *b) {
131 d->func = fn;
132 d->a = a;
133 d->b = b;
134 atomic_store_rlx(&d->next, NULL);
135 atomic_store_rlx(&d->enqueued, false);
136 return d;
137}
138
139/* DPC creation helpers */
140struct dpc *dpc_create(dpc_func_t fn, void *a, void *b) {
141 struct dpc *d = vnew(d);
142 if (!d)
143 return NULL;
144
145 return dpc_init(d, fn, a, b);
146}
147
148static void dpc_fanout_wrapper(struct dpc_context *dc, void *fn, void *latch) {
149 cc_ignore(dc);
150 any_fn_to(any_fn_from_data(fn), dpc_fanout_fn_t)();
151 once_latch_count_down(latch: (struct once_latch *) latch);
152}
153
154void dpc_fanout(struct dpc *cc_nonnull_ptr storage, struct cpu_mask cpus,
155 dpc_fanout_fn_t cc_nonnull_ptr fn) {
156 enforce_precond(storage);
157 enforce_precond(fn);
158
159 size_t count = cpu_mask_popcount(m: &cpus);
160 if (!count)
161 return;
162
163 struct once_latch latch;
164 once_latch_init(latch: &latch, count);
165
166 size_t slot = 0;
167 size_t cpu;
168 cpu_mask_for_each(cpu, cpus) {
169 dpc_init(d: &storage[slot], fn: dpc_fanout_wrapper,
170 any_fn_to_data(any_fn_from(fn)), b: &latch);
171 dpc_enqueue_on_cpu(cpu, d: &storage[slot]);
172 slot++;
173 }
174
175 enforce(slot == count);
176 once_latch_spin_wait(latch: &latch);
177}
178