1#include <mem/alloc.h>
2#include <sch/sched.h>
3#include <smp/core.h>
4#include <smp/percpu.h>
5#include <stdatomic.h>
6#include <structures/locked_list.h>
7#include <sync/rcu.h>
8#include <sync/semaphore.h>
9#include <sync/spinlock.h>
10#include <thread/thread.h>
11#include <thread/workqueue.h>
12
13#include "sch/internal.h" /* for tick_enabled */
14
15static struct rcu_buckets rcu_buckets;
16
17static uint64_t rcu_advance_gp() {
18 return atomic_fetch_add(&global.rcu_gen, 1) + 1;
19}
20
21static inline bool thread_rcu_not_reached_target(struct thread *t,
22 uint64_t target) {
23 uint32_t nesting =
24 atomic_load_explicit(&t->rcu_nesting, memory_order_seq_cst);
25
26 if (nesting != 0) {
27 uint64_t start_gen =
28 atomic_load_explicit(&t->rcu_start_gen, memory_order_acquire);
29 return start_gen != 0 && start_gen < target;
30 } else {
31 uint64_t qgen =
32 atomic_load_explicit(&t->rcu_quiescent_gen, memory_order_acquire);
33 return qgen != UINT64_MAX && qgen < target;
34 }
35}
36
37static uint64_t rcu_read_global_gen(void) {
38 return atomic_load_explicit(&global.rcu_gen, memory_order_acquire);
39}
40
41/* power-of-two ring capacity */
42void rcu_worker_notify() {
43 semaphore_post(s: &rcu_buckets.sem);
44}
45
46void rcu_synchronize(void) {
47 uint64_t target = rcu_advance_gp();
48 rcu_worker_notify();
49
50 while (true) {
51 bool all_done = true;
52
53 enum irql irql = spin_lock_irq_disable(lock: &global.thread_list.lock);
54 struct thread *t, *tmp;
55 list_for_each_entry_safe(t, tmp, &global.thread_list.list,
56 thread_list) {
57 if (thread_rcu_not_reached_target(t, target)) {
58 all_done = false;
59 break;
60 }
61 }
62 spin_unlock(lock: &global.thread_list.lock, old: irql);
63
64 if (all_done)
65 break;
66
67 scheduler_yield();
68 }
69}
70
71void rcu_defer(struct rcu_cb *cb, rcu_fn func, void *arg) {
72 cb->fn = func;
73 cb->arg = arg;
74
75 enum irql irql = irql_raise(new_level: IRQL_DISPATCH_LEVEL);
76
77 uint64_t gen = rcu_advance_gp();
78 cb->target_gen = gen;
79 cb->enqueued_waiting_on_gen = gen - 1;
80 INIT_LIST_HEAD(list: &cb->list);
81
82 size_t bucket = (gen) & (RCU_BUCKETS - 1);
83 struct rcu_buckets *buckets = &rcu_buckets;
84
85 enum irql lirql = spin_lock(lock: &buckets->buckets[bucket].lock);
86 list_add_tail(new: &cb->list, head: &buckets->buckets[bucket].list);
87 spin_unlock(lock: &buckets->buckets[bucket].lock, old: lirql);
88
89 semaphore_post(s: &buckets->sem);
90
91 irql_lower(old_level: irql);
92}
93
94void rcu_read_lock(void) {
95 struct thread *t = thread_get_current();
96 uint32_t old =
97 atomic_fetch_add_explicit(&t->rcu_nesting, 1, memory_order_relaxed);
98 if (old == 0) {
99 uint64_t gen;
100 do {
101 gen = rcu_read_global_gen();
102 atomic_store_explicit(&t->rcu_start_gen, gen, memory_order_release);
103 } while (gen != rcu_read_global_gen());
104 atomic_thread_fence(memory_order_seq_cst);
105 }
106}
107
108void rcu_read_unlock(void) {
109 struct thread *t = thread_get_current();
110 uint32_t old =
111 atomic_fetch_sub_explicit(&t->rcu_nesting, 1, memory_order_seq_cst);
112 if (old == 0)
113 panic("RCU nesting underflow");
114
115 if (old == 1) {
116 uint64_t start_gen =
117 atomic_load_explicit(&t->rcu_start_gen, memory_order_acquire);
118 atomic_store_explicit(&t->rcu_quiescent_gen, start_gen - 1,
119 memory_order_release);
120
121 atomic_store_explicit(&t->rcu_start_gen, 0, memory_order_release);
122 }
123}
124
125/* internal: detach and return head under lock for a given bucket */
126static void rcu_detach_bucket(struct rcu_buckets *bkts, size_t bucket,
127 struct list_head *lh) {
128 enum irql irql = spin_lock(lock: &bkts->buckets[bucket].lock);
129
130 INIT_LIST_HEAD(list: lh);
131 if (!list_empty(head: &bkts->buckets[bucket].list)) {
132 list_splice_init(src: &bkts->buckets[bucket].list, dst: lh);
133 }
134
135 spin_unlock(lock: &bkts->buckets[bucket].lock, old: irql);
136}
137
138static bool thread_list_has_pending_readers(uint64_t gen) {
139 bool pending = false;
140
141 enum irql irql = spin_lock_irq_disable(lock: &global.thread_list.lock);
142 struct thread *t, *tmp;
143 list_for_each_entry_safe(t, tmp, &global.thread_list.list, thread_list) {
144 if (thread_rcu_not_reached_target(t, target: gen)) {
145 pending = true;
146 break;
147 }
148 }
149 spin_unlock(lock: &global.thread_list.lock, old: irql);
150
151 return pending;
152}
153
154static void rcu_exec_callbacks(struct rcu_buckets *buckets, uint64_t target) {
155 /* Now safe to run callbacks for older bucket (generation target-1) */
156 size_t mask = RCU_BUCKETS - 1;
157 size_t old_bucket = (target - 1) & mask;
158
159 struct list_head detached;
160 rcu_detach_bucket(bkts: buckets, bucket: old_bucket, lh: &detached);
161
162 if (list_empty(head: &detached))
163 return;
164
165 /* Per-bucket lists for callbacks that don't belong to (target-1) */
166 struct list_head tmp[RCU_BUCKETS];
167 for (size_t i = 0; i < RCU_BUCKETS; ++i)
168 INIT_LIST_HEAD(list: &tmp[i]);
169
170 struct rcu_cb *iter, *tmpn;
171 list_for_each_entry_safe(iter, tmpn, &detached, list) {
172 list_del_init(entry: &iter->list);
173
174 if (!thread_list_has_pending_readers(gen: iter->target_gen)) {
175 iter->gen_when_called = iter->target_gen;
176 iter->fn(iter, iter->arg);
177 } else {
178 size_t idx = iter->target_gen & mask;
179 list_add_tail(new: &iter->list, head: &tmp[idx]);
180 }
181 }
182
183 /* Reinsert the postponed callbacks into their corresponding buckets. */
184 for (size_t i = 0; i < RCU_BUCKETS; ++i) {
185 if (list_empty(head: &tmp[i]))
186 continue;
187
188 enum irql lirql = spin_lock(lock: &buckets->buckets[i].lock);
189 /* Splice the tmp list into the real bucket list (tail) */
190 list_splice_tail_init(list: &tmp[i], head: &buckets->buckets[i].list);
191 spin_unlock(lock: &buckets->buckets[i].lock, old: lirql);
192 }
193}
194
195static void rcu_gp_worker(void *unused) {
196 (void) unused;
197 struct semaphore *rcu_sem = &(rcu_buckets).sem;
198 struct rcu_buckets *buckets = &(rcu_buckets);
199 while (true) {
200 semaphore_wait(s: rcu_sem);
201
202 /* new GP */
203 uint64_t target = rcu_advance_gp();
204
205 while (true) {
206 bool everybody_ok = true;
207
208 enum irql irql = spin_lock_irq_disable(lock: &global.thread_list.lock);
209 struct thread *t, *tmp;
210 list_for_each_entry_safe(t, tmp, &global.thread_list.list,
211 thread_list) {
212 if (thread_rcu_not_reached_target(t, target)) {
213 everybody_ok = false;
214 break;
215 }
216 }
217 spin_unlock(lock: &global.thread_list.lock, old: irql);
218
219 if (everybody_ok)
220 break;
221
222 /* yield so readers can make progress */
223 scheduler_yield();
224 }
225
226 rcu_exec_callbacks(buckets, target);
227 }
228}
229
230void rcu_init(void) {
231 global.rcu_gen = 1;
232 semaphore_init(s: &rcu_buckets.sem, value: 0, SEMAPHORE_INIT_NORMAL);
233 for (size_t i = 0; i < RCU_BUCKETS; i++) {
234 INIT_LIST_HEAD(list: &rcu_buckets.buckets[i].list);
235 spinlock_init(lock: &rcu_buckets.buckets[i].lock);
236 }
237
238 /* create worker thread */
239 thread_spawn(name: "rcu_gp_worker", entry: rcu_gp_worker, NULL);
240}
241