1/* Job items (plan F5): leases queued per slot, each job's setup first, items published as OP tickets. */
2#include "driver_int.h"
3
4#include <stdlib.h>
5#include <string.h>
6
7/* Job item tickets in flight per slot: two per node keep every node fed (Metal: `fanout`, one command buffer). */
8uint32_t drv_job_cap(const mesh_manifest *m, uint32_t exec) {
9    if (exec == MESH_BACKEND_METAL) return m->fanout > 1 ? m->fanout : 1;
10    uint32_t c = 2 * m->nodes;
11    if (c < 2) c = 2;
12    if (c > MESH_PROG_RING / 2) c = MESH_PROG_RING / 2;
13    return c;
14}
15
16void drv_lease_free(dlease *l) {
17    free(l->setup);
18    mj_ranges_free(&l->items);
19    memset(l, 0, sizeof *l);
20}
21
22/* An item's outcome to the jobs thread. */
23void drv_job_done(mesh_driver *d, uint64_t lease, uint64_t job, uint64_t item, uint32_t status, uint32_t result, const char *out) {
24    mesh_job_event e = mesh_job_event_new(MESH_JEV_DONE);
25    e.res.lease = lease;
26    e.res.job = job;
27    e.res.item = item;
28    e.res.status = status;
29    e.res.result = result;
30    e.res.out_len = strlen(out);
31    e.res.out = mesh_xmalloc(e.res.out_len ? e.res.out_len : 1);
32    memcpy(e.res.out, out, e.res.out_len);
33    if (mesh_chan_send(d->sh->jobs, &e) != MESH_CHAN_OK) mesh_job_event_free(&e);
34}
35
36/* Every item of a lease (at most a million per range, as the Rust) reported with status and why. */
37void drv_jobs_returned(mesh_driver *d, const dlease *l, uint32_t status, const char *why) {
38    for (size_t r = 0; r < l->items.n; r++) {
39        uint64_t a = l->items.v[r][0], b = l->items.v[r][1];
40        uint64_t end = b < a + 1000000 ? b : a + 1000000;
41        for (uint64_t item = a; item < end; item++) drv_job_done(d, l->lease, l->job, item, status, 0, why);
42    }
43}
44
45/* Slot s's next job items: its leases in order, each job's setup first (nothing of another job in flight),
46 * the item's 2^32-window base in the view's `job_base`, at most job_cap tickets in flight. */
47void drv_publish_jobs(mesh_driver *d, uint32_t s) {
48    mesh_mailbox mb = d->sh->mb;
49    dslot *x = d->slots[s];
50    uint32_t cap = drv_job_cap(&x->image.manifest, x->image.exec);
51    djobslot *j = &d->jobw[s];
52    dlease *failed = NULL;
53    char **why = NULL;
54    size_t nfailed = 0;
55    while (j->n) {
56        dlease *front = &j->q[0];
57        uint64_t item;
58        if (!mj_ranges_first(&front->items, &item)) {
59            drv_lease_free(front);
60            memmove(j->q, j->q + 1, (j->n - 1) * sizeof *j->q);
61            j->n--;
62            continue;
63        }
64        uint64_t job = front->job, window = item & ~0xffffffffull;
65        if (!j->has_ctx || j->ctx_job != job || j->ctx_window != window) {
66            /* Another job or window: what is in flight completes first. */
67            if (j->inflight > 0 || j->setup.busy) break;
68            if (!j->has_ctx || j->ctx_job != job) {
69                mesh_setup_step st = mesh_setup_state_step(&j->setup, job, front->setup, front->nsetup);
70                if (st.k == MESH_SETUP_WAIT) break;
71                if (st.k == MESH_SETUP_PUBLISH) {
72                    /* One setup task at a time: the next is published when this one completes. */
73                    mesh_kind k = mesh_kind_of(MESH_TK_JOB_SETUP);
74                    k.job = job;
75                    k.op = st.op;
76                    k.arg = st.arg;
77                    if (mesh_pring_can_publish(&x->ring) && mesh_pring_publish(&x->ring, mb, st.op, st.arg, k, NULL)) mesh_setup_state_published(&j->setup);
78                    break;
79                }
80                if (st.k == MESH_SETUP_FAILED) {
81                    /* The setup failed: so does every queued item of the job here (a try each). */
82                    char *e = mesh_xstrdup(st.err);
83                    size_t k = 0;
84                    for (size_t i = 0; i < j->n; i++) {
85                        if (j->q[i].job == job) {
86                            failed = mesh_xrealloc(failed, (nfailed + 1) * sizeof *failed);
87                            why = mesh_xrealloc(why, (nfailed + 1) * sizeof *why);
88                            failed[nfailed] = j->q[i];
89                            why[nfailed] = mesh_xstrdup(e);
90                            nfailed++;
91                        } else j->q[k++] = j->q[i];
92                    }
93                    j->n = k;
94                    free(e);
95                    mesh_setup_state_reset(&j->setup);
96                    continue;
97                }
98            }
99            if (x->image.manifest.has_job_base) mesh_mailbox_set64(mb, (size_t)x->view_off + x->image.manifest.job_base, window);
100            j->has_ctx = true;
101            j->ctx_job = job;
102            j->ctx_window = window;
103        }
104        if (j->inflight >= cap || !mesh_pring_can_publish(&x->ring)) break;
105        mj_ranges_take(&front->items, item);
106        mesh_kind k = mesh_kind_of(MESH_TK_JOB);
107        k.lease = front->lease;
108        k.job = job;
109        k.item = item;
110        if (!mesh_pring_publish(&x->ring, mb, front->op, (uint32_t)item, k, NULL)) { mj_ranges_add(&front->items, item); break; }
111        j->inflight++;
112    }
113    for (size_t i = 0; i < nfailed; i++) {
114        drv_jobs_returned(d, &failed[i], MESH_JOBS_RS_FAIL, why[i]);
115        drv_lease_free(&failed[i]);
116        free(why[i]);
117    }
118    free(failed);
119    free(why);
120}