1/* load.rs Prog::sample, in the Rust's float widths. */
2#include "mesh/load_prog.h"
3#include "mesh/buf.h"
4#include "mesh/cpuuse.h"
5
6#include <stdlib.h>
7#include <string.h>
8
9mesh_nl mesh_nl_read(mesh_mailbox mb, uint32_t s, uint32_t n, uint32_t *max_gap_ns, uint32_t *win) {
10    size_t r = mesh_node_load(s, n);
11    mesh_nl x = {mesh_mailbox_get64(mb, r + MESH_NL_OBS_NS), mesh_mailbox_get64(mb, r + MESH_NL_LOST_NS), mesh_mailbox_get(mb, r + MESH_NL_GAPS),
12                 mesh_mailbox_get(mb, r + MESH_NL_QUEUE_US)};
13    *max_gap_ns = mesh_mailbox_get(mb, r + MESH_NL_MAX_GAP_NS);
14    *win = mesh_mailbox_get(mb, r + MESH_NL_WIN);
15    return x;
16}
17
18void mesh_prog_sampler_reset(mesh_prog_sampler *p) {
19    for (size_t i = 0; i < p->nwin; i++) free(p->win[i].lat);
20    memset(p, 0, sizeof *p);
21}
22
23static void pf(uint8_t *b, size_t o, float v) { memcpy(b + o, &v, 4); }
24static void pu(uint8_t *b, size_t o, uint32_t v) { for (int i = 0; i < 4; i++) b[o + (size_t)i] = (uint8_t)(v >> (8 * i)); }
25static void pu64(uint8_t *b, size_t o, uint64_t v) { for (int i = 0; i < 8; i++) b[o + (size_t)i] = (uint8_t)(v >> (8 * i)); }
26
27static int cmp_u32(const void *a, const void *b) {
28    uint32_t x = *(const uint32_t *)a, y = *(const uint32_t *)b;
29    return x < y ? -1 : x > y;
30}
31
32void mesh_prog_sample(mesh_prog_sampler *p, mesh_mailbox mb, uint32_t s, double dt, uint32_t win, uint8_t out[MESH_PLOAD_BYTES], mesh_nl *dsum,
33                      uint32_t *max_gap_out) {
34    memset(out, 0, MESH_PLOAD_BYTES);
35    size_t e = mesh_prog_entry(s);
36    uint64_t hash = mesh_mailbox_get64(mb, e + MESH_PT_PTX_HASH);
37    uint32_t loads = mesh_mailbox_get(mb, e + MESH_PT_LOADS);
38    uint32_t nodes = mesh_mailbox_get(mb, e + MESH_PT_NODES);
39    if (nodes > MESH_PROG_MAX_NODES) nodes = MESH_PROG_MAX_NODES;
40    bool fresh = !p->started || hash != p->hash || loads != p->loads || nodes != p->nodes;
41    /* The CPU executors' cost (wasm lanes, native workers): cumulative, baselined at the image's load. */
42    uint32_t exec = mesh_mailbox_get(mb, mesh_prog_ext(s) + MESH_PX_EXEC);
43    bool host = exec == MESH_BACKEND_WASM || exec == MESH_BACKEND_NATIVE;
44    uint64_t cpu, rss;
45    uint32_t units;
46    mesh_slot_use_get(&mesh_cpuuse_slots[s], &cpu, &rss, &units);
47    if (fresh) {
48        mesh_prog_sampler_reset(p);
49        p->started = true;
50        p->hash = hash;
51        p->loads = loads;
52        p->nodes = nodes;
53        p->cpu_base = p->cpu_last = cpu;
54    }
55    uint64_t cpu_d = cpu > p->cpu_last ? cpu - p->cpu_last : 0;
56    if (cpu > p->cpu_last) p->cpu_last = cpu;
57    /* GPU share: the node load records; the longest gap of the window that just closed. */
58    mesh_nl d = {0, 0, 0, 0};
59    uint32_t max_gap = 0;
60    uint64_t busy_ns = 0, busy_tasks = 0;
61    for (uint32_t n = 0; n < nodes; n++) {
62        uint32_t mg, w;
63        mesh_nl now = mesh_nl_read(mb, s, n, &mg, &w);
64        if (w == win && mg > max_gap) max_gap = mg;
65        size_t r = mesh_node_rec(s, n);
66        uint64_t bn = mesh_mailbox_get64(mb, r + MESH_N_BUSY_NS);
67        uint32_t bt = mesh_mailbox_get(mb, r + MESH_N_TASKS);
68        if (!fresh) {
69            mesh_nl dd = mesh_nl_delta(&now, &p->nl[n]);
70            mesh_nl_add(&d, &dd);
71            if (bn >= p->busy_ns[n]) { busy_ns += bn - p->busy_ns[n]; busy_tasks += (uint64_t)(uint32_t)(bt - p->busy_tasks[n]); }
72        }
73        p->nl[n] = now;
74        p->busy_ns[n] = bn;
75        p->busy_tasks[n] = bt;
76    }
77    /* The trace records written since the last sample (at most a ring's worth). */
78    uint32_t tn = mesh_mailbox_get(mb, mesh_prog_ctl(s) + MESH_P_TRACE_N);
79    uint32_t *lat = NULL;
80    size_t nlat = 0;
81    if (!fresh) {
82        uint32_t nw = tn - p->trace_n;
83        if (nw > MESH_TRACE_PER_PROG) nw = MESH_TRACE_PER_PROG;
84        lat = mesh_xmalloc((nw ? nw : 1) * sizeof *lat);
85        for (uint32_t i = tn - nw; i != tn; i++) {
86            size_t r = mesh_trace_rec(s, i);
87            uint32_t w0 = mesh_mailbox_get(mb, r);
88            if ((w0 & 0xffu) != s || (w0 >> 16) == MESH_OP_INIT) continue;
89            lat[nlat++] = mesh_mailbox_get(mb, r + 12) - mesh_mailbox_get(mb, r + 8);
90        }
91    }
92    p->trace_n = tn;
93    uint32_t done = mesh_mailbox_get(mb, mesh_prog_ctl(s) + MESH_P_DONE);
94    uint64_t tasks = fresh ? 0 : (uint64_t)(uint32_t)(done - p->done);
95    p->done = done;
96    if (fresh) {
97        pu(out, MESH_PL_FLAGS, MESH_PLF_LOADED | (host ? MESH_PLF_HOST : 0));
98        if (host) { pu64(out, MESH_PL_RSS, rss); pu(out, MESH_PL_UNITS, units); }
99        free(lat);
100        *dsum = d;
101        *max_gap_out = 0;
102        return;
103    }
104    if (p->nwin == MESH_LOAD_WINDOW) {
105        free(p->win[0].lat);
106        memmove(p->win, p->win + 1, (MESH_LOAD_WINDOW - 1) * sizeof *p->win);
107        p->nwin--;
108    }
109    mesh_prog_window *nwn = &p->win[p->nwin++];
110    nwn->d = d;
111    nwn->lat = lat;
112    nwn->nlat = nlat;
113    nwn->busy_ns = busy_ns;
114    nwn->busy_tasks = busy_tasks;
115    nwn->tasks = tasks;
116    nwn->cpu_d = cpu_d;
117    mesh_nl w = {0, 0, 0, 0};
118    uint64_t wb0 = 0, wb1 = 0, wdone = 0, wcpu = 0;
119    size_t wt = 0, nall = 0;
120    for (size_t i = 0; i < p->nwin; i++) nall += p->win[i].nlat;
121    uint32_t *all = mesh_xmalloc((nall ? nall : 1) * sizeof *all);
122    size_t k = 0;
123    for (size_t i = 0; i < p->nwin; i++) {
124        mesh_prog_window *x = &p->win[i];
125        mesh_nl_add(&w, &x->d);
126        memcpy(all + k, x->lat, x->nlat * sizeof *all);
127        k += x->nlat;
128        wb0 += x->busy_ns;
129        wb1 += x->busy_tasks;
130        wt++;
131        wdone += x->tasks;
132        wcpu += x->cpu_d;
133    }
134    qsort(all, nall, sizeof *all, cmp_u32);
135    if (wb1 > 0) { p->has_task_us = true; p->task_us = (float)((double)wb0 / (double)wb1 / 1e3); }
136    mesh_share one = mesh_share_of(&d), ten = mesh_share_of(&w);
137    uint32_t flags = MESH_PLF_LOADED;
138    if ((one.has_share || ten.has_share) && ten.has_share) {
139        flags |= MESH_PLF_GPU;
140        pf(out, MESH_PL_SHARE, one.has_share ? one.share : ten.share);
141        pf(out, MESH_PL_SHARE10, ten.share);
142        pf(out, MESH_PL_SLICE_US, ten.slice_us);
143        pf(out, MESH_PL_GAP_US, ten.gap_us);
144        pf(out, MESH_PL_QUEUE_US, ten.queue_us);
145    }
146    pf(out, MESH_PL_MAX_GAP_US, (float)max_gap / 1e3f);
147    pf(out, MESH_PL_OBS, (float)((double)d.obs / 1e9 / dt));
148    if (nall) {
149        flags |= MESH_PLF_LAT;
150        pu(out, MESH_PL_LAT_N, (uint32_t)nall);
151        const size_t offs[3] = {MESH_PL_P50_US, MESH_PL_P90_US, MESH_PL_P99_US};
152        const double qs[3] = {0.5, 0.9, 0.99};
153        for (int i = 0; i < 3; i++) pf(out, offs[i], (float)mesh_quantile(all, nall, qs[i]) / 1e3f);
154        pf(out, MESH_PL_MAX_US, (float)all[nall - 1] / 1e3f);
155    }
156    free(all);
157    if (p->has_task_us) { flags |= MESH_PLF_TASK; pf(out, MESH_PL_TASK_US, p->task_us); }
158    pf(out, MESH_PL_TASKS_S, (float)((double)wdone / ((double)wt * dt)));
159    if (nodes > 0) {
160        double b = ((double)busy_ns / 1e9) / ((double)nodes * dt);
161        pf(out, MESH_PL_BUSY, (float)(b < 1.0 ? b : 1.0));
162    }
163    if (host) {
164        flags |= MESH_PLF_HOST;
165        pu64(out, MESH_PL_CPU_NS, cpu > p->cpu_base ? cpu - p->cpu_base : 0);
166        pu64(out, MESH_PL_RSS, rss);
167        pf(out, MESH_PL_CPU, (float)((double)cpu_d / 1e9 / dt));
168        pf(out, MESH_PL_CPU10, (float)((double)wcpu / 1e9 / ((double)wt * dt)));
169        if (wdone > 0) pf(out, MESH_PL_CPU_TASK_US, (float)((double)wcpu / 1e3 / (double)wdone));
170        pu(out, MESH_PL_UNITS, units);
171    }
172    pu(out, MESH_PL_FLAGS, flags);
173    *dsum = d;
174    *max_gap_out = max_gap;
175}