1/* The perf board (shared.rs PerfBoard): every member's last F_PERF, and the link RTTs measured from the
2 * echoes of ours. */
3#include "relay.h"
4
5#include <stdlib.h>
6#include <string.h>
7
8void board_init(board_t *b) { pthread_mutex_init(&b->mu, NULL); }
9
10void board_free(board_t *b) {
11    for (uint32_t k = 0; k < MAXS; k++)
12        if (b->has[k]) msr_free(&b->rep[k]);
13    pthread_mutex_destroy(&b->mu);
14}
15
16/* A deep copy (the text form is lossy: it rounds the floats). */
17static msr_report report_clone(const msr_report *r) {
18    msr_report c = {.t_us = r->t_us, .cpus = r->cpus, .big = r->big, .little = r->little, .ram_mb = r->ram_mb, .disk_free_mb = r->disk_free_mb,
19                    .heap_kb = r->heap_kb, .slots_free = r->slots_free};
20    for (size_t i = 0; i < r->nprogs; i++) msr_add_prog(&c, r->progs[i].name, &r->progs[i].perf);
21    for (size_t i = 0; i < r->nfails; i++) msr_add_fail(&c, r->fails[i].name, r->fails[i].hash, r->fails[i].unsupported, r->fails[i].why);
22    for (size_t i = 0; i < r->nrtt; i++) msr_add_rtt(&c, r->rtt[i].k, r->rtt[i].us);
23    for (size_t i = 0; i < r->necho; i++) msr_add_echo(&c, r->echo[i].k, r->echo[i].t, r->echo[i].held);
24    return c;
25}
26
27void board_report(mpr_relay *r, uint32_t from, const char *text) {
28    msr_report rep;
29    char *err = NULL;
30    if (!msr_decode(text, &rep, &err)) {
31        relay_log(r, true, mt_fmt("peer: system %u: %s", from, err));
32        free(err);
33        return;
34    }
35    if (from >= MAXS) {
36        msr_free(&rep);
37        return;
38    }
39    uint64_t now = mpr_mono_us();
40    board_t *b = &r->board;
41    pthread_mutex_lock(&b->mu);
42    /* Its echo of our last send: the link's round trip less the time it held it. */
43    for (size_t i = 0; i < rep.necho; i++)
44        if (rep.echo[i].k == r->me) {
45            uint64_t t = rep.echo[i].t, held = rep.echo[i].held;
46            uint64_t rtt = now > t ? now - t : 0;
47            rtt = rtt > held ? rtt - held : 0;
48            if (t > 0 && rtt < 10000000) b->rtt_us[from] = mof32_some((float)rtt);
49            break;
50        }
51    if (b->has[from]) msr_free(&b->rep[from]);
52    b->rep[from] = rep;
53    b->at_us[from] = now;
54    b->has[from] = true;
55    pthread_mutex_unlock(&b->mu);
56}
57
58void mpr_board_own(mpr_relay *r, const msr_report *rep) {
59    board_t *b = &r->board;
60    pthread_mutex_lock(&b->mu);
61    if (b->has[r->me]) msr_free(&b->rep[r->me]);
62    b->rep[r->me] = report_clone(rep);
63    b->at_us[r->me] = mpr_mono_us();
64    b->has[r->me] = true;
65    pthread_mutex_unlock(&b->mu);
66}
67
68size_t mpr_board_echoes(mpr_relay *r, uint32_t *k, uint64_t *t, uint64_t *held) {
69    board_t *b = &r->board;
70    size_t n = 0;
71    uint64_t now = mpr_mono_us();
72    pthread_mutex_lock(&b->mu);
73    for (uint32_t j = 0; j < r->n; j++)
74        if (j != r->me && b->has[j]) k[n] = j, t[n] = b->rep[j].t_us, held[n++] = now - b->at_us[j];
75    pthread_mutex_unlock(&b->mu);
76    return n;
77}