1/* jobd.rs's jobs_text, job_line, status_text and results_text. */
2#include "mesh/jobd/text.h"
3#include "mesh/jobs/lid.h"
4
5#include <stdio.h>
6#include <stdlib.h>
7#include <string.h>
8
9const char *mjd_role(const mjd_text_cx *cx) {
10    if (mj_node_active(cx->node)) return "leader";
11    if (mj_node_leading(cx->node)) return "leader-merging";
12    return cx->peered ? "follower" : "single";
13}
14
15static void leader_word(const mj_view *v, char out[16]) {
16    if (v->has_leader) snprintf(out, 16, "%u", v->leader);
17    else strcpy(out, "-");
18}
19
20/* Items of job id in the leader's leases (false: not leading). */
21static bool leased_of(const mj_node *n, mj_job_id id, uint64_t *sum) {
22    const momap *ls = mj_node_leases(n);
23    if (!ls) return false;
24    *sum = 0;
25    momap_iter it;
26    momap_iter_init(&it, ls);
27    for (const mj_lease *x = momap_iter_next(&it, NULL); x; x = momap_iter_next(&it, NULL))
28        if (x->job == id) *sum += mj_ranges_len(&x->out);
29    return true;
30}
31
32void mjd_job_line(const mjd_text_cx *cx, const mj_job *j, msb *out) {
33    mj_job_id id = j->spec.id;
34    uint64_t leased, queued, end = j->finished ? j->finish_at : cx->unix_ms;
35    uint64_t elapsed = end > j->spec.submitted_ms ? end - j->spec.submitted_ms : 0;
36    char w[MJ_JOB_WORD_MAX];
37    mj_job_word(id, w);
38    msb_printf(out, "job %s program %s state %s priority %d items ", w, j->spec.program, mj_job_state(j), j->spec.priority);
39    if (mj_spec_endless(&j->spec)) msb_puts(out, "endless");
40    else msb_printf(out, "%llu", (unsigned long long)mj_job_total(j));
41    uint64_t done = mj_ranges_len(&j->done);
42    msb_printf(out, " done %llu failed %zu leased ", (unsigned long long)done, momap_len(j->failed));
43    if (leased_of(cx->node, id, &leased)) msb_printf(out, "%llu", (unsigned long long)leased);
44    else msb_putc(out, '-');
45    msb_puts(out, " queued ");
46    if (mj_node_queued(cx->node, id, &queued)) msb_printf(out, "%llu", (unsigned long long)queued);
47    else msb_putc(out, '-');
48    msb_printf(out, " dupes %llu mode %s elapsed_ms %llu rate %.1f\n", (unsigned long long)j->dupes, mj_mode_word(j->spec.mode),
49               (unsigned long long)elapsed, (double)done * 1000.0 / (double)(elapsed ? elapsed : 1));
50}
51
52char *mjd_jobs_text(const mjd_text_cx *cx) {
53    const mj_node *n = cx->node;
54    const mj_table *t = mj_node_committed(n);
55    size_t jobs = 0, active = 0;
56    uint64_t id;
57    for (const mj_job *j = mj_table_first(t, &id); j; j = mj_table_next(t, &id)) {
58        jobs++;
59        active += mj_job_active(j);
60    }
61    char lw[16];
62    leader_word(&cx->view, lw);
63    msb b = {0};
64    msb_printf(&b, "jobs system %u leader %s role %s lid %llx commit %llu jobs %zu active %zu held %zu unconfirmed %zu staged %zu parked %zu rate %.1f\n",
65               n->cfg.me, lw, mjd_role(cx), (unsigned long long)mj_node_lid(n), (unsigned long long)mj_node_commit_n(n), jobs, active,
66               momap_len(n->held), mj_node_unconfirmed(n), mj_node_staged(n), mj_node_parked(n), mj_node_my_rate(n));
67    for (const mj_job *j = mj_table_first(t, &id); j; j = mj_table_next(t, &id)) mjd_job_line(cx, j, &b);
68    const mj_mpmap *rates = mj_node_rates(n);
69    if (rates)
70        for (size_t i = 0; i < rates->n; i++)
71            msb_printf(&b, "rate system %u program %s items_per_s %.1f\n", rates->v[i].member, rates->v[i].program, rates->v[i].rate.ips);
72    return msb_take(&b);
73}
74
75static bool no_job(mj_job_id id, char **text) {
76    char w[MJ_JOB_WORD_MAX];
77    mj_job_word(id, w);
78    *text = mfmt("no job %s (committed here)", w);
79    return false;
80}
81
82bool mjd_status_text(const mjd_text_cx *cx, mj_job_id id, char **text) {
83    const mj_job *j = mj_table_get(mj_node_committed(cx->node), id);
84    if (!j) return no_job(id, text);
85    msb b = {0};
86    mjd_job_line(cx, j, &b);
87    msb_puts(&b, "spec ");
88    mj_spec_encode(&j->spec, &b);
89    msb_putc(&b, '\n');
90    /* Per member: who ran how many (committed). */
91    uint64_t k;
92    for (const uint64_t *c = momap_first(j->by_member, &k); c; c = momap_next(j->by_member, &k, &k))
93        msb_printf(&b, "member %llu done %llu\n", (unsigned long long)k, (unsigned long long)*c);
94    const momap *ls = mj_node_leases(cx->node);
95    if (ls) {
96        uint64_t lk;
97        for (const mj_lease *x = momap_first(ls, &lk); x; x = momap_next(ls, &lk, &lk)) {
98            if (x->job != id) continue;
99            msb_printf(&b, "lease %llx system %u items %llu left %llu (", (unsigned long long)x->id, x->member, (unsigned long long)x->size,
100                       (unsigned long long)mj_ranges_len(&x->out));
101            mj_ranges_word(&x->out, &b);
102            msb_puts(&b, ")\n");
103        }
104    }
105    /* The lease choice (F4's filter and score), as the leader made it last. */
106    const mj_picks *p = mj_node_picks(cx->node, id);
107    if (p)
108        for (size_t i = 0; i < p->n; i++) {
109            char *l = mj_pick_line(&p->v[i]);
110            msb_printf(&b, "%s\n", l);
111            free(l);
112        }
113    else if (mj_job_active(j)) {
114        char lw[16];
115        leader_word(&cx->view, lw);
116        msb_printf(&b, "pick on the leader (system %s): its `job status` explains the lease choice\n", lw);
117    }
118    if (j->cancel) msb_printf(&b, "cancel %s\n", j->cancel);
119    size_t shown = 0;
120    for (const mj_failed *f = momap_first(j->failed, &k); f && shown < 50; f = momap_next(j->failed, &k, &k), shown++)
121        msb_printf(&b, "failed %llu tries %u %s\n", (unsigned long long)k, f->tries, f->why);
122    size_t retried = momap_len(j->fails);
123    if (retried > 0) msb_printf(&b, "retried %zu item(s)\n", retried);
124    *text = msb_take(&b);
125    return true;
126}
127
128bool mjd_results_text(const mjd_text_cx *cx, mj_job_id id, uint64_t from, uint64_t n, char **text) {
129    const mj_job *j = mj_table_get(mj_node_committed(cx->node), id);
130    if (!j) return no_job(id, text);
131    msb s = {0};
132    uint64_t c = 0, i = 0, item;
133    momap_iter it;
134    momap_iter_init(&it, j->results);
135    for (const mj_res *r = momap_iter_next1(&it, &item); r && c < n; r = momap_iter_next1(&it, &item), i++) {
136        if (i < from) continue;
137        msb_printf(&s, "%llu %u lease %llx member %u", (unsigned long long)item, r->result, (unsigned long long)r->lease, r->member);
138        if (r->out_len) {
139            msb_puts(&s, " out ");
140            for (size_t q = 0; q < r->out_len; q++) msb_printf(&s, "%02x", r->out[q]);
141        }
142        if (r->has_blob) {
143            char x[65];
144            mj_hex(&r->blob.hash, x);
145            msb_printf(&s, " blob %s len %u holders 0x%x", x, r->blob.len, r->blob.holders);
146        }
147        msb_putc(&s, '\n');
148        c++;
149    }
150    char w[MJ_JOB_WORD_MAX];
151    mj_job_word(id, w);
152    msb b = {0};
153    msb_printf(&b, "results %s total %zu from %llu count %llu done %llu failed %zu dupes %llu state %s\n%s", w, momap_len(j->results),
154               (unsigned long long)from, (unsigned long long)c, (unsigned long long)mj_ranges_len(&j->done), momap_len(j->failed),
155               (unsigned long long)j->dupes, mj_job_state(j), msb_str(&s));
156    msb_free(&s);
157    *text = msb_take(&b);
158    return true;
159}