==> src/dispatch.c <==1/* jobs.rs's Node, the dispatcher: on_results, dispatch (loss, expiry, deadlines, completion, the
2 * next leases, stealing), cancel. */
3#include "internal.h"
4#include "mesh/abi/peer_frames.h"
5#include "mesh/jobs/copies.h"
6
7#include <math.h>
8#include <stdlib.h>
9#include <string.h>
10
11static void picks_free(void *v) { mj_picks_free(v); }
12
13static uint64_t sat_sub(uint64_t a, uint64_t b) { return a > b ? a - b : 0; }
14
15/* Rust's `f64 as u64`: saturating, NaN to 0. */
16static uint64_t f2u(double x) {
17 if (x != x || x <= 0.0) return 0;
18 if (x >= 18446744073709551615.0) return UINT64_MAX;
19 return (uint64_t)x;
20}
21
22static void free_lease(mj_lease *x) {
23 free(x->program);
24 mj_ranges_free(&x->out);
25}
26
27/* ---- results ----------------------------------------------------------------------------------- */
28
29typedef struct {
30 mj_job_id job;
31 uint64_t s, e, lease;
32} run4;
33
34void mj_node_on_results(mj_node *n, uint32_t from, mj_item_res *items, size_t count, uint64_t now, uint64_t unix_ms, mj_outs *out) {
35 (void)unix_ms;
36 uint64_t dead = n->cfg.dead_ms;
37 uint32_t me = n->cfg.me;
38 mj_lead *l = n->lead;
39 run4 *runs = NULL;
40 size_t nruns = 0;
41 mj_mpmap counted = {0}; /* program -> count, member unused (0) */
42 for (size_t i = 0; i < count; i++) {
43 mj_item_res *r = &items[i];
44 mj_job *j = mj_table_get(&l->working, r->job);
45 if (!j || !mj_job_active(j)) continue;
46 /* A payload result is committed only while this leader holds its blob: else it waits here, and the
47 * blob is fetched from its holders (a result resent to a new leader). */
48 if (r->status == MESH_JOBS_RS_OK && r->has_blob) {
49 if (!mj_job_settled(j, r->item) && !mj_node_has_blob(n, r->job, &r->blob.hash)) {
50 uint64_t key[5];
51 mj_key5(key, r->job, &r->blob.hash);
52 mj_parked_list *v = momap_put(l->parked, key, NULL);
53 bool dup = false;
54 for (size_t k = 0; k < v->n; k++)
55 if (v->v[k].from == from && v->v[k].r.lease == r->lease && v->v[k].r.item == r->item) dup = true;
56 if (v->n < MESH_JOBS_PARKED_MAX && !dup) {
57 if (v->n == v->cap) {
58 v->cap = v->cap ? 2 * v->cap : 4;
59 v->v = mxrealloc(v->v, v->cap * sizeof *v->v);
60 }
61 v->v[v->n++] = (mj_parked){from, mj_item_res_clone(r)};
62 }
63 uint64_t *t = momap_get(l->fetch_at, key);
64 if (!t || sat_sub(now, *t) >= MESH_JOBS_FETCH_RESEND_MS) {
65 *(uint64_t *)momap_put(l->fetch_at, key, NULL) = now > 1 ? now : 1;
66 mj_outs_push(out, mj_out_fetch_blob(r->job, r->blob.hash, r->blob.holders));
67 }
68 continue;
69 }
70 r->blob.holders |= (uint32_t)1 << me;
71 }
72 mj_mode mode = j->spec.mode;
73 uint32_t attempts = j->spec.attempts;
74 /* The program that ran it (a multi-program job's rates are per program). */
75 mj_lease *x = momap_get1(l->disp.leases, r->lease);
76 char *program;
77 if (x) program = mstrdup(x->program);
78 else {
79 const char *p;
80 size_t len = mj_spec_first_program(&j->spec, &p);
81 program = mstrndup(p, len);
82 }
83 /* Was it this lease's (and still outstanding)? */
84 bool mine = false;
85 if (x && x->member == from) {
86 bool had = mj_ranges_take(&x->out, r->item);
87 if (had) {
88 mj_mp *rt = mj_mpmap_get(&l->disp.rates, from, program);
89 double rate = rt ? rt->rate.ips : 0.0;
90 x->deadline = now + mj_lease_time(mj_ranges_len(&x->out), rate) * 3 + dead;
91 }
92 if (mj_ranges_is_empty(&x->out)) {
93 mj_lease gone;
94 momap_take(l->disp.leases, &r->lease, &gone);
95 bool last = true;
96 uint64_t id;
97 momap_iter cur11_;
98 momap_iter_init(&cur11_, l->disp.leases);
99 for (const mj_lease *y = momap_iter_next1(&cur11_, &id); y; y = momap_iter_next1(&cur11_, &id))
100 if (y->member == from && strcmp(y->program, gone.program) == 0) last = false;
101 mj_rate_lease_done(&mj_mpmap_put(&l->disp.rates, from, gone.program)->rate, gone.size, gone.issued, now, last);
102 free_lease(&gone);
103 }
104 mine = had;
105 }
106 if (r->status != MESH_JOBS_RS_RETURNED) mj_mpmap_put(&counted, 0, program)->until++;
107 free(program);
108 j = mj_table_get(&l->working, r->job);
109 switch (r->status) {
110 case MESH_JOBS_RS_OK: {
111 if (mj_job_settled(j, r->item)) {
112 /* Exactly once: the first result stays; another attempt's is a duplicate (a re-send of the
113 * same is not). (An item whose run was folded away has no lease on record: not counted.) */
114 uint64_t lease;
115 if (mj_job_lease_of(j, r->item, &lease) && lease != r->lease) {
116 bool fresh;
117 uint64_t *d = momap_put1(l->dupes, r->job, &fresh);
118 if (fresh) *d = j->dupes;
119 (*d)++;
120 }
121 continue;
122 }
123 if (!mj_ranges_contains(&j->spec.items, r->item)) continue;
124 bool keep = mode == MJ_MODE_COLLECT ? true : mode == MJ_MODE_NONZERO ? r->result != 0 : false;
125 if (keep) {
126 mj_res res = {r->lease, from, r->result, r->out, r->out_len, r->has_blob, r->blob};
127 mj_op op = {.kind = MJ_OP_DONE, .job = r->job, .item = r->item, .res = res};
128 r->out = NULL;
129 r->out_len = 0;
130 mj_table_apply(&l->working, &op);
131 mj_ops_push(&l->ops, op);
132 } else {
133 mj_op op = mj_op_done_run(r->job, r->item, r->item + 1, r->lease, from);
134 mj_table_apply(&l->working, &op);
135 if (nruns && runs[nruns - 1].job == r->job && runs[nruns - 1].lease == r->lease && runs[nruns - 1].e == r->item)
136 runs[nruns - 1].e = r->item + 1;
137 else {
138 runs = mxrealloc(runs, (nruns + 1) * sizeof *runs);
139 runs[nruns++] = (run4){r->job, r->item, r->item + 1, r->lease};
140 }
141 }
142 break;
143 }
144 case MESH_JOBS_RS_FAIL: {
145 if (mj_job_settled(j, r->item) || !mine) continue;
146 const uint32_t *f = momap_get1(j->fails, r->item);
147 uint32_t tries = (f ? *f : 0) + 1;
148 char *why = mj_utf8_lossy(r->out, r->out_len);
149 mj_op op = mj_op_fails(r->job, r->item, tries);
150 mj_table_apply(&l->working, &op);
151 mj_ops_push(&l->ops, op);
152 if (tries >= attempts) {
153 char *w = mfmt("system %u: %s", from, why);
154 mj_op op2 = mj_op_fail(r->job, r->item, tries, w);
155 free(w);
156 mj_table_apply(&l->working, &op2);
157 mj_ops_push(&l->ops, op2);
158 } else {
159 mj_queue *q = momap_get1(l->disp.queues, r->job);
160 if (q) mj_ranges_add(&q->retry, r->item);
161 }
162 free(why);
163 break;
164 }
165 default:
166 if (mine && !mj_job_settled(j, r->item)) {
167 mj_queue *q = momap_get1(l->disp.queues, r->job);
168 if (q) mj_ranges_add(&q->retry, r->item);
169 }
170 }
171 }
172 for (size_t i = 0; i < nruns; i++) mj_ops_push(&l->ops, mj_op_done_run(runs[i].job, runs[i].s, runs[i].e, runs[i].lease, from));
173 free(runs);
174 for (size_t i = 0; i < counted.n; i++) mj_rate_count(&mj_mpmap_put(&l->disp.rates, from, counted.v[i].program)->rate, counted.v[i].until, now);
175 for (size_t i = 0; i < counted.n; i++) free(counted.v[i].program);
176 free(counted.v);
177 for (size_t i = 0; i < count; i++) mj_item_res_free(&items[i]);
178 free(items);
179}
180
181/* ---- cancel ------------------------------------------------------------------------------------ */
182
183typedef struct {
184 uint32_t to;
185 mj_frame f;
186} send1;
187
188typedef struct {
189 send1 *v;
190 size_t n;
191} sends;
192
193static void push_send(sends *s, uint32_t to, mj_frame f) {
194 s->v = mxrealloc(s->v, (s->n + 1) * sizeof *s->v);
195 s->v[s->n++] = (send1){to, f};
196}
197
198static void cancel_inner(mj_lead *l, mj_job_id id, const char *why, uint32_t state, uint64_t unix_ms, sends *s) {
199 mj_op a = mj_op_cancel(id, why), b = mj_op_finish(id, unix_ms, state);
200 mj_table_apply(&l->working, &a);
201 mj_ops_push(&l->ops, a);
202 mj_table_apply(&l->working, &b);
203 mj_ops_push(&l->ops, b);
204 momap_del1(l->disp.queues, id);
205 uint64_t lease;
206 for (mj_lease *x = momap_first(l->disp.leases, &lease); x; x = momap_next(l->disp.leases, &lease, &lease)) {
207 if (x->job != id) continue;
208 uint32_t m = x->member;
209 momap_del1(l->disp.leases, lease);
210 push_send(s, m, mj_frame_cut(l->lid, lease, 0));
211 }
212}
213
214/* Ends job id (cancel or deadline): its leases cut (sent as they are, to itself too), its queue dropped. */
215void mj_lead_cancel(mj_lead *l, mj_job_id id, const char *why, uint32_t state, uint64_t unix_ms, mj_outs *out) {
216 sends s = {0};
217 cancel_inner(l, id, why, state, unix_ms, &s);
218 for (size_t i = 0; i < s.n; i++) mj_outs_push(out, mj_out_send(s.v[i].to, s.v[i].f));
219 free(s.v);
220}
221
222/* ---- dispatch ---------------------------------------------------------------------------------- */
223
224static const mj_pick *pick_of(const momap *picks, mj_job_id job, uint32_t m) {
225 const mj_picks *p = momap_get1(picks, job);
226 return p ? mj_picks_ok(p, m) : NULL;
227}
228
229static double best_of(const momap *picks, uint32_t m) {
230 double best = -1.7976931348623157e308;
231 uint64_t job;
232 momap_iter cur12_;
233 momap_iter_init(&cur12_, picks);
234 for (const mj_picks *p = momap_iter_next1(&cur12_, &job); p; p = momap_iter_next1(&cur12_, &job)) {
235 for (size_t i = 0; i < p->n; i++)
236 if (p->v[i].member == m) {
237 if (p->v[i].ok) best = fmax(best, p->v[i].score.total);
238 break;
239 }
240 }
241 return best;
242}
243
244/* Rust's f64::total_cmp, for the orderings the dispatcher sorts by. */
245static int total_cmp(double a, double b) {
246 int64_t x, y;
247 memcpy(&x, &a, 8);
248 memcpy(&y, &b, 8);
249 x ^= (int64_t)((uint64_t)(x >> 63) >> 1);
250 y ^= (int64_t)((uint64_t)(y >> 63) >> 1);
251 return x < y ? -1 : x > y ? 1 : 0;
252}
253
254void mj_node_dispatch(mj_node *n, uint64_t now, uint64_t unix_ms, mj_outs *out) {
255 uint32_t me = n->cfg.me;
256 uint32_t alive = n->view.alive | (uint32_t)1 << me;
257 uint64_t dead = n->cfg.dead_ms;
258 sends s = {0};
259 mj_lead *l = n->lead;
260 uint64_t lid = l->lid;
261 uint64_t id;
262 /* A member lost: its leases go back (no try counted: the member's fault, not the item's). */
263 for (mj_lease *x = momap_first(l->disp.leases, &id); x; x = momap_next(l->disp.leases, &id, &id)) {
264 if (alive >> x->member & 1) continue;
265 mj_lease g;
266 momap_take(l->disp.leases, &id, &g);
267 char *w = mj_word(g.job);
268 mj_outs_push(out, mj_out_log(mfmt("jobs: system %u lost: lease %llx of job %s: %llu item(s) re-queued", g.member, (unsigned long long)id, w,
269 (unsigned long long)mj_ranges_len(&g.out))));
270 free(w);
271 mj_queue *q = momap_get1(l->disp.queues, g.job);
272 if (q) mj_ranges_union(&q->retry, &g.out);
273 free_lease(&g);
274 }
275 for (uint32_t k = 0; k < MJ_MEMBERS; k++)
276 if (l->disp.progs.has[k] && !(alive >> k & 1)) mj_member_progs_remove(&l->disp.progs, k);
277 /* Expired: a try counted for each item (a hanging item fails after `attempts`). */
278 for (mj_lease *x = momap_first(l->disp.leases, &id); x; x = momap_next(l->disp.leases, &id, &id)) {
279 if (!(now > x->deadline)) continue;
280 mj_lease g;
281 momap_take(l->disp.leases, &id, &g);
282 uint64_t hold = MESH_JOBS_HOLDOFF_MS > 4 * dead ? MESH_JOBS_HOLDOFF_MS : 4 * dead;
283 char *w = mj_word(g.job);
284 mj_outs_push(out, mj_out_log(mfmt("jobs: lease %llx of job %s on system %u expired: %llu item(s) re-queued; no new %s lease there for %llu s",
285 (unsigned long long)id, w, g.member, (unsigned long long)mj_ranges_len(&g.out), g.program,
286 (unsigned long long)(hold / 1000))));
287 free(w);
288 mj_mpmap_put(&l->disp.holdoff, g.member, g.program)->until = now + hold;
289 push_send(&s, g.member, mj_frame_cut(lid, id, 0));
290 mj_job *j = mj_table_get(&l->working, g.job);
291 if (!j) {
292 free_lease(&g);
293 continue;
294 }
295 uint32_t attempts = j->spec.attempts;
296 if (mj_ranges_len(&g.out) > MESH_JOBS_EXPIRY_TRIES_MAX) {
297 /* A huge lease (an endless job's): back to the queue whole, no tries counted per item. */
298 mj_queue *q = momap_get1(l->disp.queues, g.job);
299 if (q) mj_ranges_union(&q->retry, &g.out);
300 free_lease(&g);
301 continue;
302 }
303 for (size_t r = 0; r < g.out.n; r++)
304 for (uint64_t item = g.out.v[r][0]; item < g.out.v[r][1]; item++) {
305 const uint32_t *f = momap_get1(mj_table_get(&l->working, g.job)->fails, item);
306 uint32_t tries = (f ? *f : 0) + 1;
307 mj_op a = mj_op_fails(g.job, item, tries);
308 mj_table_apply(&l->working, &a);
309 mj_ops_push(&l->ops, a);
310 if (tries >= attempts) {
311 char *why = mfmt("lease expired on system %u", g.member);
312 mj_op b = mj_op_fail(g.job, item, tries, why);
313 free(why);
314 mj_table_apply(&l->working, &b);
315 mj_ops_push(&l->ops, b);
316 } else {
317 mj_queue *q = momap_get1(l->disp.queues, g.job);
318 if (q) mj_ranges_add(&q->retry, item);
319 }
320 }
321 free_lease(&g);
322 }
323 /* Deadlines. */
324 for (mj_job *j = mj_table_first(&l->working, &id); j; j = mj_table_next(&l->working, &id)) {
325 if (!(mj_job_active(j) && j->spec.deadline_ms > 0 && unix_ms > j->spec.deadline_ms)) continue;
326 char *w = mj_word(id);
327 mj_outs_push(out, mj_out_log(mfmt("jobs: job %s passed its deadline", w)));
328 free(w);
329 cancel_inner(l, id, "deadline", MESH_JOBS_FIN_DEADLINE, unix_ms, &s);
330 }
331 /* Complete: nothing queued, nothing leased, every item done or failed. */
332 for (mj_job *j = mj_table_first(&l->working, &id); j; j = mj_table_next(&l->working, &id)) {
333 if (!(mj_job_active(j) && mj_job_complete(j))) continue;
334 bool leased = false;
335 uint64_t lk;
336 momap_iter cur13_;
337 momap_iter_init(&cur13_, l->disp.leases);
338 for (const mj_lease *x = momap_iter_next1(&cur13_, &lk); x; x = momap_iter_next1(&cur13_, &lk))
339 if (x->job == id) leased = true;
340 if (leased) continue;
341 mj_op op = mj_op_finish(id, unix_ms, MESH_JOBS_FIN_DONE);
342 mj_table_apply(&l->working, &op);
343 mj_ops_push(&l->ops, op);
344 momap_del1(l->disp.queues, id);
345 char *w = mj_word(id);
346 mj_outs_push(out, mj_out_log(mfmt("jobs: job %s done", w)));
347 free(w);
348 }
349 /* Who may take each job's leases (F4's filter and score), best first. */
350 momap *picks = momap_new(1, sizeof(mj_picks), picks_free, NULL);
351 for (mj_job *j = mj_table_first(&l->working, &id); j; j = mj_table_next(&l->working, &id))
352 if (mj_job_active(j)) *(mj_picks *)momap_put1(picks, id, NULL) = mj_select(&j->spec, alive, &l->disp.progs, n->sched);
353 /* The next leases: each member that runs a job program and has less than half a lease left, the
354 * best-scored first (a short queue goes to the best members). */
355 uint32_t members[MJ_MEMBERS];
356 size_t nm = 0;
357 for (uint32_t k = 0; k < MJ_MEMBERS; k++)
358 if (l->disp.progs.has[k] && (alive >> k & 1)) members[nm++] = k;
359 double best[MJ_MEMBERS];
360 for (size_t i = 0; i < nm; i++) best[i] = best_of(picks, members[i]);
361 for (size_t i = 1; i < nm; i++) { /* stable: best descending, then id */
362 uint32_t m = members[i];
363 double b = best[i];
364 size_t j = i;
365 while (j > 0) {
366 int c = total_cmp(b, best[j - 1]);
367 if (c > 0 || (c == 0 && m < members[j - 1])) {
368 members[j] = members[j - 1];
369 best[j] = best[j - 1];
370 j--;
371 } else break;
372 }
373 members[j] = m;
374 best[j] = b;
375 }
376 for (size_t mi = 0; mi < nm; mi++) {
377 uint32_t m = members[mi];
378 size_t np = l->disp.progs.n[m];
379 mj_prog *progs = mj_progs_clone(l->disp.progs.v[m], np);
380 for (size_t pi = 0; pi < np; pi++) {
381 mj_prog *p = &progs[pi];
382 mj_mp *h = mj_mpmap_get(&l->disp.holdoff, m, p->name);
383 if (h && now < h->until) continue;
384 uint64_t outstanding = 0;
385 size_t held = 0;
386 uint64_t lk;
387 momap_iter cur14_;
388 momap_iter_init(&cur14_, l->disp.leases);
389 for (const mj_lease *x = momap_iter_next1(&cur14_, &lk); x; x = momap_iter_next1(&cur14_, &lk))
390 if (x->member == m && strcmp(x->program, p->name) == 0) {
391 outstanding += mj_ranges_len(&x->out);
392 held++;
393 }
394 /* Pipelined: the running lease and the next, never more (a queue of leases would expire at its back). */
395 if (held >= 2) continue;
396 mj_mp *rt = mj_mpmap_get(&l->disp.rates, m, p->name);
397 double rate = rt ? rt->rate.ips : 0.0;
398 uint64_t slow = rt ? rt->rate.target : 0;
399 /* At least two batches of tickets in flight, and LEASE_MS at the measured rate or what slow start
400 * reached, whichever is more. Slow start never runs past 4 x LEASE_MS at the measured rate. */
401 uint64_t floor_ = 2 * (uint64_t)p->cap > 2 ? 2 * (uint64_t)p->cap : 2;
402 if (rate > 0.0) {
403 uint64_t cap4 = f2u(4.0 * rate * (double)MESH_JOBS_LEASE_MS / 1000.0);
404 if (cap4 < slow) slow = cap4;
405 }
406 uint64_t at_rate = f2u(rate * (double)MESH_JOBS_LEASE_MS / 1000.0);
407 if (at_rate > MESH_JOBS_LEASE_MAX) at_rate = MESH_JOBS_LEASE_MAX;
408 uint64_t target = floor_ > slow ? floor_ : slow;
409 if (at_rate > target) target = at_rate;
410 if (outstanding * 2 > target) continue;
411 /* The job: highest priority first, then the oldest, with items to lease. */
412 mj_job_id chosen = 0;
413 bool found = false;
414 int64_t best_prio = 0;
415 for (mj_job *j = mj_table_first(&l->working, &id); j; j = mj_table_next(&l->working, &id)) {
416 if (!mj_job_active(j) || !mj_spec_runs(&j->spec, p->name)) continue;
417 const mj_queue *q = momap_get1(l->disp.queues, id);
418 if (!q || (mj_ranges_is_empty(&q->fresh) && mj_ranges_is_empty(&q->retry))) continue;
419 if (!pick_of(picks, id, m)) continue;
420 int64_t prio = -(int64_t)j->spec.priority;
421 if (!found || prio < best_prio) {
422 found = true;
423 best_prio = prio;
424 chosen = id;
425 }
426 }
427 if (found) {
428 mj_job *j = mj_table_get(&l->working, chosen);
429 const mj_picks *pk = momap_get1(picks, chosen);
430 uint64_t k = 0;
431 for (size_t i = 0; i < pk->n; i++) k += pk->v[i].ok;
432 mj_queue *q = momap_get1(l->disp.queues, chosen);
433 uint64_t a = mj_ranges_len(&q->fresh), b = mj_ranges_len(&q->retry);
434 uint64_t left = a > UINT64_MAX - b ? UINT64_MAX : a + b;
435 const mj_pick *pp = pick_of(picks, chosen, m);
436 uint64_t shrunk = pp && pp->busy ? (floor_ < target ? floor_ : target) : target;
437 uint64_t d = 2 * (k > 1 ? k : 1);
438 uint64_t share = left / d + (left % d != 0);
439 if (share < 1) share = 1;
440 uint64_t size = shrunk < share ? shrunk : share;
441 if (size < 1) size = 1;
442 /* Re-queued items first, then fresh ones; never an item done meanwhile. */
443 mj_ranges items = mj_ranges_take_front(&q->retry, size);
444 if (mj_ranges_len(&items) < size) {
445 mj_ranges more = mj_ranges_take_front(&q->fresh, size - mj_ranges_len(&items));
446 mj_ranges_union(&items, &more);
447 mj_ranges_free(&more);
448 }
449 mj_ranges_subtract(&items, &j->done);
450 uint64_t fx;
451 momap_iter cur15_;
452 momap_iter_init(&cur15_, j->failed);
453 for (const void *v = momap_iter_next1(&cur15_, &fx); v; v = momap_iter_next1(&cur15_, &fx)) mj_ranges_take(&items, fx);
454 if (mj_ranges_is_empty(&items)) {
455 mj_ranges_free(&items);
456 continue;
457 }
458 if (outstanding == 0) mj_rate_busy(&mj_mpmap_put(&l->disp.rates, m, p->name)->rate, now);
459 l->disp.next_lease++;
460 uint64_t lease = l->disp.next_lease;
461 uint64_t cnt = mj_ranges_len(&items);
462 /* Its deadline counts the lease queued before it there. */
463 mj_lease *x = momap_put1(l->disp.leases, lease, NULL);
464 *x = (mj_lease){lease, chosen, m, mstrdup(p->name), mj_ranges_clone(&items), cnt, now,
465 now + 3 * mj_lease_time(cnt + outstanding, rate) + dead};
466 mj_lease_msg lm = {lease, chosen, mstrdup(p->name), j->spec.op, mj_setup_clone(j->spec.setup, j->spec.nsetup), j->spec.nsetup, items};
467 push_send(&s, m, mj_frame_lease(lid, lm));
468 } else if (outstanding == 0) {
469 /* Steal: the tail of the lease of this program with the most time left (a lease on a member no
470 * longer eligible, cordoned or busy, first), for a job this member may run. */
471 double mine_rate = rate;
472 bool have_v = false, vgone = false;
473 uint64_t vid = 0;
474 double vsecs = 0, vr = 0;
475 momap_iter cur16_;
476 momap_iter_init(&cur16_, l->disp.leases);
477 for (const mj_lease *x = momap_iter_next1(&cur16_, &lk); x; x = momap_iter_next1(&cur16_, &lk)) {
478 if (x->member == m || mj_ranges_len(&x->out) < 2) continue;
479 const mj_job *xj = mj_table_get(&l->working, x->job);
480 if (!xj || !mj_spec_runs(&xj->spec, p->name)) continue;
481 const mj_pick *mp = pick_of(picks, x->job, m);
482 if (!mp || mp->busy) continue;
483 /* Its member's rate, or this lease's own progress if that is slower. */
484 mj_mp *xr = mj_mpmap_get(&l->disp.rates, x->member, x->program);
485 double r = xr ? xr->rate.ips : 0.0;
486 uint64_t dt = sat_sub(now, x->issued);
487 double lr = dt >= 100 ? (double)(x->size - mj_ranges_len(&x->out)) * 1000.0 / (double)dt : r;
488 r = r > 0.0 ? fmin(r, lr) : lr;
489 bool gone = pick_of(picks, x->job, x->member) == NULL;
490 double secs = (double)mj_ranges_len(&x->out) / fmax(r, 1e-3);
491 /* max_by: the last of equals wins. */
492 if (!have_v || (gone != vgone ? gone > vgone : total_cmp(secs, vsecs) >= 0)) {
493 have_v = true;
494 vid = lk;
495 vsecs = secs;
496 vr = r;
497 vgone = gone;
498 }
499 }
500 if (have_v) {
501 if (!vgone && vsecs * 1000.0 < 2.0 * (double)MESH_JOBS_LEASE_MS) continue;
502 mj_lease *x = momap_get1(l->disp.leases, vid);
503 uint64_t cnt = mj_ranges_len(&x->out);
504 double share = mine_rate > 0.0 && vr > 0.0 ? vr / (vr + mine_rate) : 0.5;
505 uint64_t keep = f2u(ceil((double)cnt * share));
506 if (keep < 1) keep = 1;
507 if (keep > cnt - 1) keep = cnt - 1;
508 uint64_t cut;
509 if (!mj_ranges_nth(&x->out, keep, &cut)) continue;
510 mj_ranges moved = mj_ranges_clone(&x->out);
511 mj_ranges_remove(&moved, 0, cut);
512 mj_ranges_remove(&x->out, cut, UINT64_MAX);
513 char *w = mj_word(x->job);
514 mj_outs_push(out, mj_out_log(mfmt("jobs: system %u idle: steals %llu item(s) of lease %llx (job %s) from system %u%s", m,
515 (unsigned long long)mj_ranges_len(&moved), (unsigned long long)vid, w, x->member,
516 vgone ? " (no longer eligible there)" : "")));
517 free(w);
518 push_send(&s, x->member, mj_frame_cut(lid, vid, cut));
519 mj_queue *q = momap_get1(l->disp.queues, x->job);
520 if (q) mj_ranges_union(&q->retry, &moved);
521 mj_ranges_free(&moved);
522 }
523 }
524 }
525 mj_progs_free(progs, np);
526 }
527 momap_free(l->disp.picks);
528 l->disp.picks = picks;
529 for (size_t i = 0; i < s.n; i++) mj_node_send(n, s.v[i].to, s.v[i].f, out);
530 free(s.v);
531}