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