==> src/samplers.c <==1/* The load and NIC samplers. */
2#include "mesh/samplers.h"
3#include "mesh/cpuuse.h"
4#include "mesh/load_prog.h"
5#include "mesh/log.h"
6#include "mesh/nic.h"
7
8#include <stdlib.h>
9#include <string.h>
10
11void (*mesh_load_perf)(mesh_shared *sh, const mesh_host_sample *h, const uint8_t *cls, size_t ncls, const uint8_t (*recs)[MESH_PLOAD_BYTES], size_t n);
12
13#define SAMPLE_NS 1000000000ll
14
15static void sleep_until(int64_t t0) {
16 int64_t left = SAMPLE_NS - (mesh_mono_ns() - t0);
17 if (left > 0) mesh_sleep_us(left / 1000);
18}
19
20/* Says an error once per distinct text. */
21static void once(char **said, const char *e, const char *what) {
22 if (*said && strcmp(*said, e) == 0) return;
23 mesh_error("%s: %s", what, e);
24 free(*said);
25 *said = mesh_xstrdup(e);
26}
27
28void mesh_load_sampler(mesh_shared *sh) {
29 static uint8_t cls[MESH_MAX_CPUS];
30 const char *src = mesh_core_classes(cls, MESH_MAX_CPUS);
31 bool psi = mesh_has_psi();
32 mesh_host_sample *prev = mesh_xcalloc(1, sizeof *prev), *h = mesh_xcalloc(1, sizeof *h);
33 bool has_prev = false;
34 int64_t prev_at = 0;
35 mesh_prog_sampler *progs = mesh_xcalloc(MESH_PROG_SLOTS, sizeof *progs);
36 mesh_nl gwin[MESH_LOAD_WINDOW];
37 size_t ngwin = 0;
38 uint32_t samples = 0;
39 char *said = NULL;
40 bool logged = false;
41 mesh_host_derived *der = mesh_xcalloc(1, sizeof *der);
42 uint8_t (*recs)[MESH_PLOAD_BYTES] = mesh_xcalloc(MESH_PROG_SLOTS, MESH_PLOAD_BYTES);
43 for (;;) {
44 int64_t t0 = mesh_mono_ns();
45 char *err = NULL;
46 if (mesh_host_read(h, sh->programs, psi, &err) == 0) {
47 if (said) { mesh_info("load: host counters readable again"); free(said); said = NULL; }
48 if (has_prev) {
49 double dt = (double)(t0 - prev_at) / 1e9;
50 mesh_mailbox mb = sh->mb;
51 /* Close the load window: this sample takes the longest gaps of window `win`. */
52 uint32_t win = mesh_mailbox_get(mb, MESH_H_LOAD_WIN);
53 mesh_nl d = {0, 0, 0, 0};
54 uint32_t max_gap = 0, execs[MESH_PROG_SLOTS];
55 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) {
56 execs[s] = mesh_mailbox_get(mb, mesh_prog_ext(s) + MESH_PX_EXEC);
57 if (mesh_mailbox_get(mb, mesh_prog_entry(s) + MESH_PT_STATE) == MESH_ST_EMPTY) {
58 mesh_prog_sampler_reset(&progs[s]);
59 memset(recs[s], 0, MESH_PLOAD_BYTES);
60 continue;
61 }
62 mesh_nl pd;
63 uint32_t mg;
64 mesh_prog_sample(&progs[s], mb, s, dt, win, recs[s], &pd, &mg);
65 mesh_nl_add(&d, &pd);
66 if (mg > max_gap) max_gap = mg;
67 }
68 mesh_mailbox_set(mb, MESH_H_LOAD_WIN, win + 1);
69 if (ngwin == MESH_LOAD_WINDOW) { memmove(gwin, gwin + 1, (MESH_LOAD_WINDOW - 1) * sizeof *gwin); ngwin--; }
70 gwin[ngwin++] = d;
71 mesh_nl w = {0, 0, 0, 0};
72 for (size_t i = 0; i < ngwin; i++) mesh_nl_add(&w, &gwin[i]);
73 mesh_host_derive(h, prev, cls, MESH_MAX_CPUS, dt, der);
74 samples++;
75 mesh_share s1 = mesh_share_of(&d), s10 = mesh_share_of(&w);
76 uint64_t up = mesh_host_block_uptime_ms(&sh->host);
77 uint8_t m[MESH_MACHINE_BYTES];
78 mesh_machine_bytes(m, h, der, cls, MESH_MAX_CPUS, &s1, &s10, max_gap, (float)((double)d.gaps / dt), (float)((double)d.obs / 1e9 / dt),
79 up ? up : 1, samples, (uint32_t)((mesh_mono_ns() - t0) / 1000));
80 mesh_exec_totals(m, (const uint8_t (*)[MESH_PLOAD_BYTES])recs, execs, MESH_PROG_SLOTS, atomic_load(&mesh_cpuuse_workers),
81 atomic_load(&mesh_cpuuse_worker_spawns), atomic_load(&mesh_cpuuse_worker_crashes), mesh_log_dropped());
82 uint8_t rep[MESH_LOADREP_BYTES];
83 mesh_loadrep(rep, m);
84 uint32_t me = sh->me < MESH_MAX_SYSTEMS ? sh->me : 0;
85 uint64_t at = up ? up : 1;
86 uint8_t atb[8];
87 for (int i = 0; i < 8; i++) atb[i] = (uint8_t)(at >> (8 * i));
88 mesh_host_field f[3 + MESH_PROG_SLOTS];
89 f[0] = (mesh_host_field){MESH_S_LOAD, m, sizeof m};
90 f[1] = (mesh_host_field){mesh_mload(me) + MESH_ML_SHARE, rep, sizeof rep};
91 f[2] = (mesh_host_field){mesh_mload(me) + MESH_ML_AT_MS, atb, 8};
92 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) f[3 + s] = (mesh_host_field){mesh_pload(s), recs[s], MESH_PLOAD_BYTES};
93 mesh_host_block_set_many(&sh->host, f, 3 + MESH_PROG_SLOTS);
94 /* The load page: the same bytes for GPU programs. */
95 uint8_t *hb = mesh_xmalloc(MESH_SNAP_HOST);
96 mesh_host_block_bytes(&sh->host, hb);
97 mesh_mailbox_write(mb, MESH_LOAD_PAGE, hb + MESH_S_LOAD, MESH_S_LOAD_END - MESH_S_LOAD, MESH_S_LOAD_END - MESH_S_LOAD);
98 free(hb);
99 if (sh->peer) {
100 uint8_t fr[1 + MESH_LOADREP_BYTES];
101 fr[0] = MESH_F_LOADREP;
102 memcpy(fr + 1, rep, sizeof rep);
103 mesh_shared_broadcast(sh, fr, sizeof fr);
104 if (mesh_load_perf) mesh_load_perf(sh, h, cls, MESH_MAX_CPUS, (const uint8_t (*)[MESH_PLOAD_BYTES])recs, MESH_PROG_SLOTS);
105 }
106 if (!logged) {
107 logged = true;
108 size_t nb = 0, nl = 0, on = 0;
109 for (size_t i = 0; i < MESH_MAX_CPUS; i++) { nb += cls[i] == MESH_CORE_BIG; nl += cls[i] == MESH_CORE_LITTLE; }
110 for (size_t i = 0; i < h->ncpus; i++) on += h->have[i];
111 mesh_info("load: sampling every 1s: %zu CPUs (%zu big, %zu little, by %s), PSI %s, sample cost %lld us", on, nb, nl, src, psi ? "yes" : "no",
112 (long long)((mesh_mono_ns() - t0) / 1000));
113 }
114 }
115 mesh_host_sample *t = prev;
116 prev = h;
117 h = t;
118 has_prev = true;
119 prev_at = t0;
120 } else {
121 /* Loud once per distinct error; S_LOAD_AT_MS stops advancing, so viewers show the sample's age. */
122 once(&said, err, "load: cannot read the host counters");
123 free(err);
124 }
125 if (atomic_load(sh->quit)) break;
126 sleep_until(t0);
127 }
128 for (uint32_t s = 0; s < MESH_PROG_SLOTS; s++) mesh_prog_sampler_reset(&progs[s]);
129 free(progs);
130 free(prev);
131 free(h);
132 free(der);
133 free(recs);
134 free(said);
135}
136
137void mesh_net_sampler(mesh_shared *sh) {
138 mesh_nics nics = {0};
139 char *said = NULL;
140 uint32_t samples = 0;
141 for (;;) {
142 int64_t t0 = mesh_mono_ns();
143 mesh_nic_raw *raw;
144 size_t nraw;
145 char *err = NULL;
146 if (mesh_nic_read(&raw, &nraw, &err) == 0) {
147 if (said) { mesh_info("net: interface counters readable again"); free(said); said = NULL; }
148 mesh_nic_now *now;
149 size_t nnow;
150 mesh_nics_sample(&nics, raw, nraw, &now, &nnow);
151 free(raw);
152 uint8_t *h = mesh_xmalloc(MESH_SNAP_HOST);
153 mesh_host_block_bytes(&sh->host, h);
154 /* The interfaces that carry (or carried) a link, and which links (sorted by name, as the Rust's BTreeMap). */
155 char names[MESH_MAX_SYSTEMS][17];
156 uint32_t masks[MESH_MAX_SYSTEMS];
157 size_t non = 0;
158 for (uint32_t k = 0; k < MESH_MAX_SYSTEMS; k++) {
159 const uint8_t *b = h + mesh_link(k) + MESH_L_IFACE;
160 char nm[17];
161 size_t len = 0;
162 while (len < 16 && b[len]) len++;
163 memcpy(nm, b, len);
164 nm[len] = 0;
165 if (!len) continue;
166 size_t i = 0;
167 while (i < non && strcmp(names[i], nm) != 0) i++;
168 if (i == non) { strcpy(names[non], nm); masks[non] = 0; non++; }
169 masks[i] |= 1u << k;
170 }
171 free(h);
172 for (size_t i = 1; i < non; i++)
173 for (size_t j = i; j > 0 && strcmp(names[j - 1], names[j]) > 0; j--) {
174 char t[17];
175 strcpy(t, names[j]);
176 strcpy(names[j], names[j - 1]);
177 strcpy(names[j - 1], t);
178 uint32_t mm = masks[j];
179 masks[j] = masks[j - 1];
180 masks[j - 1] = mm;
181 }
182 uint8_t recs[MESH_MAX_NICS][MESH_NIC_BYTES];
183 size_t nrec = 0;
184 static const uint64_t zero[8];
185 for (size_t i = 0; i < non && nrec < MESH_MAX_NICS; i++) {
186 const uint64_t *v = mesh_nic_lookup(now, nnow, names[i]);
187 if (v) mesh_nic_record(recs[nrec++], names[i], v, masks[i], 1);
188 else mesh_nic_record(recs[nrec++], names[i], zero, masks[i], 0); /* gone: NI_COUNT 0 says so */
189 }
190 free(now);
191 samples++;
192 uint8_t total[MESH_NIC_BYTES];
193 mesh_nic_record(total, "total", nics.total, 0, nics.counted);
194 uint64_t up = mesh_host_block_uptime_ms(&sh->host);
195 uint64_t at = up ? up : 1;
196 uint8_t atb[8], cnt[4], smp[4];
197 for (int i = 0; i < 8; i++) atb[i] = (uint8_t)(at >> (8 * i));
198 for (int i = 0; i < 4; i++) { cnt[i] = (uint8_t)((uint32_t)nrec >> (8 * i)); smp[i] = (uint8_t)(samples >> (8 * i)); }
199 mesh_host_field f[4 + MESH_MAX_NICS] = {{MESH_S_NET_TOTAL, total, MESH_NIC_BYTES}, {MESH_S_NET_NICS, cnt, 4}, {MESH_S_NET_SAMPLES, smp, 4},
200 {MESH_S_NET_AT_MS, atb, 8}};
201 for (size_t i = 0; i < nrec; i++) f[4 + i] = (mesh_host_field){mesh_nic(i), recs[i], MESH_NIC_BYTES};
202 mesh_host_block_set_many(&sh->host, f, 4 + nrec);
203 } else {
204 once(&said, err, "net: cannot read the interface counters");
205 free(err);
206 }
207 if (atomic_load(sh->quit)) break;
208 mesh_sleep_us(SAMPLE_NS / 1000);
209 (void)t0;
210 }
211 mesh_nics_free(&nics);
212 free(said);
213}