1/* Scheduler load balancing policy */
2#include <mem/numa.h>
3#include <smp/domain.h>
4
5#include "internal.h"
6
7#define SCHEDULER_REMOTE_NODE_SCALE_NUMERATOR 1
8#define SCHEDULER_REMOTE_NODE_SCALE_DENOMINATOR 5
9#define IDLE_LONG_ENOUGH 10 /* if the other core is idle for 10ms */
10#define IDLE_MIN_MIGRATABLE 3 /* and we have 3 migratable */
11
12static inline bool can_push_thread(size_t caller, struct thread *t) {
13 return scheduler_can_take_thread(core: caller, target: t) && t != thread_get_current();
14}
15
16/* fraction = remote_scale * (1 / (1 + dist))
17 *
18 * using integer math:
19 *
20 * to_migrate = (count * remote_scale_num) / ((1 + dist) * remote_scale_den);
21 */
22
23/* it's OK if this races, we are just counting threads */
24static size_t migratable_in_tree(size_t caller, struct rbt *rbt) {
25 struct rbt_node *rb;
26 size_t agg = 0;
27 rbt_for_each(rb, rbt) {
28 struct thread *t = thread_from_rq_rbt_node(rb);
29 if (can_push_thread(caller, t))
30 agg++;
31 }
32 return agg;
33}
34
35static size_t migratable_in_list(size_t caller, struct list_head *tq) {
36 struct list_node *ln;
37 size_t agg = 0;
38 list_for_each(ln, tq) {
39 struct thread *t = thread_from_rq_list_node(ln);
40 if (can_push_thread(caller, t))
41 agg++;
42 }
43 return agg;
44}
45
46static void
47scheduler_count_migratable_threads(struct scheduler *caller,
48 struct scheduler *s,
49 size_t agg[THREAD_PRIO_CLASS_COUNT]) {
50 SPINLOCK_ASSERT_HELD(&caller->lock);
51 SPINLOCK_ASSERT_HELD(&s->lock);
52 /* count urgent, count realtime, count timesharing, count background */
53 size_t c = caller->core_id;
54 agg[THREAD_PRIO_CLASS_URGENT] = migratable_in_list(caller: c, tq: &s->urgent_threads);
55 agg[THREAD_PRIO_CLASS_REALTIME] = migratable_in_list(caller: c, tq: &s->rt_threads);
56 agg[THREAD_PRIO_CLASS_BACKGROUND] = migratable_in_list(caller: c, tq: &s->bg_threads);
57 agg[THREAD_PRIO_CLASS_TIMESHARE] = migratable_in_tree(caller: c, rbt: &s->completed_rbt);
58 agg[THREAD_PRIO_CLASS_TIMESHARE] += migratable_in_tree(caller: c, rbt: &s->thread_rbt);
59}
60
61/* this is a fun trick... if there is no NUMA, the `associated_node` will be
62 * NULL, and this will always return true. if there is NUMA, then this will
63 * actually do the comparison, so we can use this function the same
64 * way in either setting! :) */
65static inline bool cores_in_same_numa_node(struct core *a, struct core *b) {
66 return a->domain->associated_node == b->domain->associated_node;
67}
68
69static void move_ts_thread_raw(struct scheduler *dest, struct scheduler *source,
70 struct rbt *tree, struct thread *thread) {
71 rbt_delete(tree, z: &thread->rq_tree_node);
72 scheduler_decrement_thread_count(sched: source, t: thread);
73
74 thread->perceived_prio_class = THREAD_PRIO_CLASS_TIMESHARE;
75 rbt_insert(tree: &dest->thread_rbt, new_node: &thread->rq_tree_node);
76 scheduler_increment_thread_count(sched: dest, t: thread);
77 thread_set_runqueue(t: thread, s: dest);
78 thread_post_migrate(t: thread, old_cpu: source->core_id, new_cpu: dest->core_id);
79}
80
81static size_t migrate_from_tree(struct scheduler *to,
82 struct scheduler *from_sched, struct rbt *from,
83 size_t target) {
84 size_t migrated = 0;
85 struct rbt_node *rb;
86
87 /* the idea is that we should ideally check "every other thread" to migrate
88 * them over. however, we will sometimes encounter threads that cannot be
89 * migrated. if we detect during iteration that there is a next thread and
90 * it cannot be migrated, we will opt to migrate the current thread instead,
91 * even if it slightly breaks our "every other thread" rule. */
92 bool prev_migrated = false;
93 struct rbt_node *tmp;
94
95 rbt_for_each_safe(rb, tmp, from) {
96 if (migrated >= target)
97 break;
98
99 struct thread *t = thread_from_rq_rbt_node(rb);
100
101 /* we are on a thread we will give priority to migrating */
102 if (!prev_migrated) {
103 if (can_push_thread(caller: to->core_id, t)) {
104 move_ts_thread_raw(dest: to, source: from_sched, tree: from, thread: t);
105 prev_migrated = true;
106 migrated++;
107 }
108
109 } else {
110 prev_migrated = false;
111 }
112 }
113
114 return migrated;
115}
116
117/* migrate `num_threads` threads, and return how many threads we migrated. In
118 * the event that we are migrating from the timesharing threads, every OTHER
119 * thread will be migrated until `num_threads` is reached. we do this to
120 * prevent stealing either all the high prio threads, all the low prio threads,
121 * or all the middlers. */
122static size_t migrate_from_prio_class(struct scheduler *to,
123 struct scheduler *from,
124 enum thread_prio_class class,
125 size_t nthreads) {
126 if (!nthreads)
127 return 0;
128
129 size_t migrated = 0;
130 if (class != THREAD_PRIO_CLASS_TIMESHARE) {
131 struct list_head *from_queue =
132 scheduler_get_this_thread_queue(sched: from, prio: class);
133 struct list_head *to_queue = scheduler_get_this_thread_queue(sched: to, prio: class);
134
135 struct list_node *ln, *tmp;
136 list_for_each_safe(ln, tmp, from_queue) {
137 if (migrated >= nthreads)
138 break;
139
140 struct thread *t = thread_from_rq_list_node(ln);
141 if (can_push_thread(caller: to->core_id, t)) {
142
143 list_node_delete_init(n: ln);
144 scheduler_decrement_thread_count(sched: from, t);
145
146 list_head_push_back(h: to_queue, n: ln);
147 scheduler_increment_thread_count(sched: to, t);
148
149 migrated++;
150 thread_set_runqueue(t, s: to);
151 thread_post_migrate(t, old_cpu: from->core_id, new_cpu: to->core_id);
152 }
153 }
154 } else {
155 /* migrating timesharing threads. first try to migrate threads that have
156 * not ran this period, skipping every other thread, and then try and
157 * migrate the completed threads */
158 size_t took = migrate_from_tree(to, from_sched: from, from: &from->thread_rbt, target: nthreads);
159 migrated += took;
160 nthreads -= took;
161
162 migrated += migrate_from_tree(to, from_sched: from, from: &from->completed_rbt, target: nthreads);
163 }
164
165 return migrated;
166}
167
168size_t scheduler_try_push_to_idle_core(struct scheduler *sched) {
169 if (!global.scheduler_domains[TOPOLOGY_LEVEL_MACHINE])
170 return 0;
171
172 int32_t other_cpu_id = scheduler_push_target(from: global.cores[sched->core_id]);
173 if (other_cpu_id < 0)
174 return 0;
175
176 struct core *this_core = global.cores[sched->core_id];
177 struct core *other_core = global.cores[other_cpu_id];
178
179 struct scheduler *other = global.schedulers[other_cpu_id];
180
181 /* we trylock to avoid spinning as this is called from the hotpath and can
182 * also have a circular dependency deadlock/starvation chance... */
183 if (!spin_trylock_raw(&other->lock))
184 return 0;
185
186 /* from the OTHER core's perspective, how many threads can we migrate? */
187 size_t migratable[THREAD_PRIO_CLASS_COUNT];
188 scheduler_count_migratable_threads(caller: other, s: sched, agg: migratable);
189
190 size_t total_migratable = 0;
191 for (size_t i = 0; i < THREAD_PRIO_CLASS_COUNT; i++)
192 total_migratable += migratable[i];
193
194 size_t migrated = 0;
195 if (!total_migratable)
196 goto out;
197
198 /* if we are in the same NUMA node, migrate half of our threads
199 * from each priority class so we are near identical */
200 if (cores_in_same_numa_node(a: this_core, b: other_core)) {
201 for (size_t i = 0; i < THREAD_PRIO_CLASS_COUNT; i++) {
202 size_t to_migrate = migratable[i] / 2;
203 migrated += migrate_from_prio_class(to: other, from: sched, class: i, nthreads: to_migrate);
204 }
205 } else {
206 /* remote node */
207 size_t numa_id = other_core->domain->id;
208 size_t dist = this_core->domain->associated_node->rel_dists[numa_id];
209
210 if (dist == 0)
211 dist = 1; /* shouldn't happen here */
212
213 /* TODO: tune */
214 const size_t remote_scale_num = 1; /* numerator of scale, tuneable */
215 const size_t remote_scale_den = 5; /* denominator of scale, tuneable */
216
217 size_t dist_factor = (1 + dist) * remote_scale_den;
218
219 for (size_t i = 0; i < THREAD_PRIO_CLASS_COUNT; i++) {
220 size_t count = migratable[i];
221 if (count == 0)
222 continue;
223
224 size_t to_migrate = (count * remote_scale_num) / dist_factor;
225
226 /* Enforce minimum if remote move is allowed */
227 if (to_migrate == 0 && count > 0)
228 to_migrate = 1;
229
230 time_ms_t idle_entry =
231 atomic_load_rlx(&other->idle_thread_data.last_entry_ms);
232 time_ms_t idle_for = time_get_ms() - idle_entry;
233 if (to_migrate == 0 && count >= IDLE_MIN_MIGRATABLE &&
234 idle_for >= IDLE_LONG_ENOUGH) {
235 to_migrate = 1;
236 }
237
238 migrated += migrate_from_prio_class(to: other, from: sched, class: i, nthreads: to_migrate);
239 }
240 }
241
242 /* if the other core is in the same node as us, we push half of our threads
243 * over there. otherwise, we push (1 / distance * scale) of our threads */
244
245out:
246 if (migrated) {
247 scheduler_debug_validate(sched);
248 scheduler_debug_validate(sched: other);
249 }
250
251 spin_unlock_raw(&other->lock);
252
253 if (migrated)
254 scheduler_defer_resched(target: other);
255
256 return migrated;
257}
258