==> src/publish.c <==1/* The jobs thread's publishing (jobd.rs `publish`, `news`, `jobs_recs`): the host block's S_JOB_*
2 * fields, the jobs view's records and blobd's news, every 250 ms. */
3#include "jobs_thread.h"
4
5#include "mesh/abi/host_jobs.h"
6#include "mesh/assemble.h"
7#include "mesh/jobd/text.h"
8#include "mesh/jobs/lid.h"
9#include "mesh/strbuf.h"
10
11#include <stdio.h>
12#include <stdlib.h>
13#include <string.h>
14
15static uint64_t leased_of(const momap *ls, mj_job_id job, bool any_member, uint32_t member) {
16 uint64_t s = 0;
17 momap_iter it;
18 momap_iter_init(&it, ls);
19 for (const mj_lease *l = momap_iter_next(&it, NULL); l; l = momap_iter_next(&it, NULL))
20 if (l->job == job && (any_member || l->member == member)) s += mj_ranges_len(&l->out);
21 return s;
22}
23
24void jd_publish(jd *x) {
25 mesh_host_block *h = &x->sh->host;
26 const mj_node *n = x->node;
27 const mj_table *t = &n->rep.committed;
28 uint32_t active = 0;
29 uint64_t done = 0, failed = 0, dupes = 0, id;
30 for (mj_job *j = mj_table_first(t, &id); j; j = mj_table_next(t, &id)) {
31 if (mj_job_active(j)) active++;
32 uint64_t d = mj_ranges_len(&j->done);
33 done = done + d < done ? UINT64_MAX : done + d;
34 failed += momap_len(j->failed);
35 dupes += j->dupes;
36 }
37 mesh_host_block_set_u32(h, MESH_S_JOBS_ACTIVE, active);
38 mesh_host_block_set_u32(h, MESH_S_JOBS_TOTAL, (uint32_t)momap_len(t->jobs));
39 mesh_host_block_set_u64(h, MESH_S_JOB_DONE, done);
40 mesh_host_block_set_u64(h, MESH_S_JOB_FAILED, failed);
41 mesh_host_block_set_u64(h, MESH_S_JOB_DUPES, dupes);
42 uint64_t leased = 0;
43 const momap *ls = mj_node_leases(n);
44 momap_iter it;
45 if (ls) {
46 momap_iter_init(&it, ls);
47 for (const mj_lease *l = momap_iter_next(&it, NULL); l; l = momap_iter_next(&it, NULL)) leased += mj_ranges_len(&l->out);
48 } else {
49 momap_iter_init(&it, n->held);
50 for (const mj_held *l = momap_iter_next(&it, NULL); l; l = momap_iter_next(&it, NULL)) leased += mj_ranges_len(&l->items);
51 }
52 mesh_host_block_set_u32(h, MESH_S_JOB_LEASED, leased > UINT32_MAX ? UINT32_MAX : (uint32_t)leased);
53 double rate = 0;
54 const mj_mpmap *rates = mj_node_rates(n);
55 if (rates)
56 for (size_t i = 0; i < rates->n; i++) rate += rates->v[i].rate.ips;
57 else rate = mj_node_my_rate(n);
58 mesh_host_block_set_f32(h, MESH_S_JOB_RATE, (float)rate);
59 mesh_host_block_set_u64(h, MESH_S_JOB_COMMIT, mj_node_commit_n(n));
60 mesh_host_block_set_u64(h, MESH_S_JOB_TERM, mj_lid_term(mj_node_lid(n)));
61 mesh_host_block_set_u32(h, MESH_S_JOB_FLAGS, (mj_node_leading(n) ? MESH_JF_LEADER : 0) | (mj_node_active(n) ? MESH_JF_ACTIVE : 0) |
62 (mjd_store_persisted(x->d->store) ? MESH_JF_PERSIST : 0));
63 mesh_host_block_set_u32(h, MESH_S_JOB_HELD, (uint32_t)momap_len(n->held));
64 mesh_recs recs = jd_jobs_recs(x, rate);
65 pthread_mutex_lock(&x->sh->mu);
66 mesh_recs_free(&x->sh->jobs_view_recs);
67 x->sh->jobs_view_recs = recs;
68 pthread_mutex_unlock(&x->sh->mu);
69
70 /* blobd's news: every job's state, the jobs forgotten since, and the committed payload results of the
71 * images it assembles that it does not have yet. */
72 mbd_news *nw = calloc(1, sizeof *nw);
73 if (!nw) abort();
74 size_t cap = momap_len(t->jobs);
75 nw->jobs = mxalloc((cap ? cap : 1) * sizeof *nw->jobs);
76 for (mj_job *j = mj_table_first(t, &id); j; j = mj_table_next(t, &id))
77 nw->jobs[nw->njobs++] = (mbd_job_info){j->spec.id, mstrdup(j->spec.program), j->spec.op, mj_setup_clone(j->spec.setup, j->spec.nsetup),
78 j->spec.nsetup, j->tone, mj_job_active(j), mstrdup(mj_job_state(j))};
79 for (size_t i = 0; i < t->nforgotten; i++) {
80 bool fresh;
81 momap_put1(x->told, t->forgotten[i], &fresh);
82 if (!fresh) continue;
83 nw->forgotten = mxrealloc(nw->forgotten, (nw->nforgotten + 1) * sizeof *nw->forgotten);
84 nw->forgotten[nw->nforgotten++] = t->forgotten[i];
85 }
86 mj_job_id *watch;
87 size_t nwatch = mjd_ext_watch(x->sh, &watch);
88 uint64_t k;
89 for (void *v = momap_first(x->sent, &k); v; v = momap_next(x->sent, &k, &k)) {
90 bool kept = false;
91 for (size_t i = 0; i < nwatch; i++) kept |= watch[i] == k;
92 if (!kept) momap_del1(x->sent, k);
93 }
94 size_t capr = 0;
95 for (size_t w = 0; w < nwatch; w++) {
96 mj_job *j = mj_table_get(t, watch[w]);
97 if (!j) continue;
98 mj_ranges *sent = momap_put1(x->sent, watch[w], NULL);
99 uint64_t item;
100 momap_iter_init(&it, j->results);
101 for (const mj_res *r = momap_iter_next1(&it, &item); r; r = momap_iter_next1(&it, &item)) {
102 uint64_t part;
103 if (mj_ranges_contains(sent, item)) continue;
104 if (!r->has_blob || !masm_part_of(r->out, r->out_len, &part)) continue;
105 mj_ranges_add(sent, item);
106 if (nw->nresults == capr) nw->results = mxrealloc(nw->results, (capr = capr ? 2 * capr : 64) * sizeof *nw->results);
107 nw->results[nw->nresults++] = (mbd_result){watch[w], item, {r->blob.hash, r->blob.len, r->blob.holders, r->member, part}};
108 }
109 }
110 free(watch);
111 mjd_blobs_send(x->sh, &(mjd_bev){.k = MJD_BEV_TABLE, .news = nw});
112}
113
114/* The jobs view's records, from this member's committed table (and, on the leader, its leases, rates
115 * and picks):
116 * - `jobs`: system, leader, role, commit, jobs, active, leases held here, unconfirmed results, items/s, persisted;
117 * - `job` per job: program, state, priority, items, done, failed, leased, queued, dupes, retried items, mode,
118 * progress (0..1; endless: -), items/s over the last ~3 s, ETA, elapsed; text: the cancel reason;
119 * - `share` per (job, member that ran or holds some): done, leased now, the member's measured rate of the program;
120 * - `pick` per (job, member) on the leader: the lease choice (mj_select); text: the score's terms or the reason;
121 * - `result` (the last 3 per job) and `failed` (the first 5 per job): the preview. */
122mesh_recs jd_jobs_recs(jd *x, double rate) {
123 const mj_node *n = x->node;
124 const mj_table *t = &n->rep.committed;
125 mj_view v = jd_view(x);
126 int64_t now = mesh_mono_ns();
127 uint64_t unix_ms = jd_unix_ms();
128 mesh_recs out = MESH_RECS_INIT;
129 mjd_text_cx cx = {n, v, x->peered, unix_ms};
130 uint64_t id;
131 uint32_t active = 0;
132 for (mj_job *j = mj_table_first(t, &id); j; j = mj_table_next(t, &id)) active += mj_job_active(j);
133 mesh_rec *r = mesh_recs_add(&out, "jobs");
134 mesh_rec_kvu(r, "system", x->d->cfg.me);
135 if (v.has_leader) mesh_rec_kvu(r, "leader", v.leader);
136 else mesh_rec_none(r, "leader");
137 mesh_rec_kv(r, "role", mjd_role(&cx));
138 mesh_rec_kvu(r, "commit", mj_node_commit_n(n));
139 mesh_rec_kvu(r, "jobs", momap_len(t->jobs));
140 mesh_rec_kvu(r, "active", active);
141 mesh_rec_kvu(r, "held", momap_len(n->held));
142 mesh_rec_kvu(r, "unconfirmed", mj_node_unconfirmed(n));
143 mesh_rec_f(r, "rate", true, rate, 1);
144 mesh_rec_kvu(r, "persisted", mjd_store_persisted(x->d->store) ? 1 : 0);
145 for (void *hv = momap_first(x->hist, &id); hv; hv = momap_next(x->hist, &id, &id))
146 if (!mj_table_get(t, id)) momap_del1(x->hist, id);
147 const momap *ls = mj_node_leases(n);
148 const mj_mpmap *rates = mj_node_rates(n);
149 for (mj_job *j = mj_table_first(t, &id); j; j = mj_table_next(t, &id)) {
150 char word[MJ_JOB_WORD_MAX];
151 mj_job_word(id, word);
152 uint64_t done = mj_ranges_len(&j->done), nfailed = momap_len(j->failed);
153 bool act = mj_job_active(j);
154 jd_hist *h = momap_put1(x->hist, id, NULL);
155 if (h->n == h->cap) {
156 h->cap = h->cap ? 2 * h->cap : 8;
157 h->at = mxrealloc(h->at, h->cap * sizeof *h->at);
158 h->done = mxrealloc(h->done, h->cap * sizeof *h->done);
159 }
160 h->at[h->n] = now;
161 h->done[h->n++] = done;
162 while (h->n > 2 && now - h->at[1] >= 3000000000LL) {
163 memmove(h->at, h->at + 1, (h->n - 1) * sizeof *h->at);
164 memmove(h->done, h->done + 1, (h->n - 1) * sizeof *h->done);
165 h->n--;
166 }
167 double ips = 0;
168 if (h->n && h->at[h->n - 1] > h->at[0] && act)
169 ips = (double)(h->done[h->n - 1] - h->done[0]) / ((double)(h->at[h->n - 1] - h->at[0]) / 1e9);
170 bool endless = mj_spec_endless(&j->spec);
171 uint64_t total = mj_job_total(j);
172 uint64_t settled = done + nfailed;
173 uint64_t left = total > settled ? total - settled : 0;
174 bool has_eta = !endless && act && ips > 0;
175 r = mesh_recs_add(&out, "job");
176 mesh_rec_kv(r, "job", word);
177 mesh_rec_kv(r, "program", j->spec.program);
178 mesh_rec_kv(r, "state", mj_job_state(j));
179 mesh_rec_kvi(r, "priority", j->spec.priority);
180 if (endless) mesh_rec_kv(r, "items", "endless");
181 else mesh_rec_kvu(r, "items", total);
182 mesh_rec_kvu(r, "done", done);
183 mesh_rec_kvu(r, "failed", nfailed);
184 if (ls) mesh_rec_kvu(r, "leased", leased_of(ls, id, true, 0));
185 else mesh_rec_none(r, "leased");
186 uint64_t q;
187 if (!mj_node_queued(n, id, &q)) mesh_rec_none(r, "queued");
188 else if (endless) mesh_rec_kv(r, "queued", "endless");
189 else mesh_rec_kvu(r, "queued", q);
190 mesh_rec_kvu(r, "dupes", j->dupes);
191 mesh_rec_kvu(r, "retried", momap_len(j->fails));
192 mesh_rec_kv(r, "mode", mj_mode_word(j->spec.mode));
193 mesh_rec_f(r, "progress", !endless, (double)settled / (double)(total ? total : 1), 4);
194 mesh_rec_f(r, "rate", true, ips, 1);
195 mesh_rec_f(r, "eta_s", has_eta, has_eta ? (double)left / ips : 0, 1);
196 uint64_t end = j->finished ? j->finish_at : unix_ms;
197 mesh_rec_kvu(r, "elapsed_ms", end > j->spec.submitted_ms ? end - j->spec.submitted_ms : 0);
198 mesh_rec_kvu(r, "submitted_ms", j->spec.submitted_ms);
199 mesh_rec_text(r, j->cancel ? j->cancel : "");
200 /* The members that ran some of it, and those that hold some now. */
201 uint32_t members = 0;
202 uint64_t m;
203 momap_iter it;
204 momap_iter_init(&it, j->by_member);
205 for (void *c = momap_iter_next1(&it, &m); c; c = momap_iter_next1(&it, &m))
206 if (m < 32) members |= 1u << m;
207 if (ls) {
208 momap_iter_init(&it, ls);
209 for (const mj_lease *l = momap_iter_next(&it, NULL); l; l = momap_iter_next(&it, NULL))
210 if (l->job == id && l->member < 32) members |= 1u << l->member;
211 }
212 for (uint32_t k = 0; k < 32; k++) {
213 if (!(members >> k & 1)) continue;
214 uint64_t held = ls ? leased_of(ls, id, false, k) : 0;
215 /* The rate while it holds items of the job (an idle member's last rate is history). */
216 bool has_rate = rates && held > 0;
217 double kr = 0;
218 if (has_rate)
219 for (const char *p = j->spec.program; p;) {
220 const char *c = strchr(p, ',');
221 char *name = c ? mstrndup(p, (size_t)(c - p)) : mstrdup(p);
222 const mj_mp *mp = mj_mpmap_get(rates, k, name);
223 if (mp) kr += mp->rate.ips;
224 free(name);
225 p = c ? c + 1 : NULL;
226 }
227 r = mesh_recs_add(&out, "share");
228 mesh_rec_kv(r, "job", word);
229 mesh_rec_kvu(r, "member", k);
230 mesh_rec_kvu(r, "done", mj_count_get(j->by_member, k));
231 if (ls) mesh_rec_kvu(r, "leased", held);
232 else mesh_rec_none(r, "leased");
233 mesh_rec_f(r, "rate", has_rate, kr, 1);
234 }
235 const mj_picks *p = mj_node_picks(n, id);
236 for (size_t i = 0; p && i < p->n; i++) {
237 const mj_pick *x1 = &p->v[i];
238 r = mesh_recs_add(&out, "pick");
239 mesh_rec_kv(r, "job", word);
240 mesh_rec_kvu(r, "member", x1->member);
241 mesh_rec_kv(r, "verdict", x1->ok ? "eligible" : "rejected");
242 mesh_rec_kvu(r, "busy", x1->busy ? 1 : 0);
243 mesh_rec_f(r, "score", x1->ok, x1->ok ? x1->score.total : 0, 3);
244 mesh_rec_text(r, x1->ok ? (x1->score.explain ? x1->score.explain : "") : (x1->why ? x1->why : ""));
245 }
246 uint64_t item;
247 int c = 0;
248 for (const mj_res *res = momap_last(j->results, &item); res && c < 3; res = momap_prev(j->results, &item, &item), c++) {
249 char hex[33];
250 size_t nb = res->out_len < 16 ? res->out_len : 16;
251 for (size_t b = 0; b < nb; b++) snprintf(hex + 2 * b, 3, "%02x", res->out[b]);
252 hex[2 * nb] = 0;
253 r = mesh_recs_add(&out, "result");
254 mesh_rec_kv(r, "job", word);
255 mesh_rec_kvu(r, "item", item);
256 mesh_rec_kvu(r, "result", res->result);
257 mesh_rec_kvu(r, "member", res->member);
258 mesh_rec_kv(r, "out", hex);
259 }
260 c = 0;
261 momap_iter_init(&it, j->failed);
262 for (const mj_failed *f = momap_iter_next1(&it, &item); f && c < 5; f = momap_iter_next1(&it, &item), c++) {
263 r = mesh_recs_add(&out, "failed");
264 mesh_rec_kv(r, "job", word);
265 mesh_rec_kvu(r, "item", item);
266 mesh_rec_kvu(r, "tries", f->tries);
267 mesh_rec_text(r, f->why ? f->why : "");
268 }
269 }
270 return out;
271}