1/* pipesink: the pipe demo's sink as a native worker (view pipedemo, K_*). Each
2 * continuous task drains up to DRAIN digests from its inbox; each is checked
3 * against mesh-ref's SHA-256d of the msg rebuilt from its seq (an
4 * implementation independent of the hashing stage's), and its latency -- the
5 * host clock now minus the source's stamp: one clock for every executor --
6 * goes into a 1 us histogram in the view.
7 * Ported from alpha-rust-mesh programs/pipesink/src/main.rs. */
8#include "pipesink.h"
9
10#include "hist_quantile.h"
11#include "mesh/abi/ops.h"
12#include "mesh/abi/view_pipedemo.h"
13#include "mesh/ref/sha256d.h"
14
15#include <string.h>
16
17#define DRAIN 512u
18/* An idle task waits this long (ns) for a record. */
19#define WAIT_NS UINT64_C(1000000)
20
21static void check(const mesh_worker_ctx *ctx, const uint8_t rec[48], uint64_t now) {
22    uint32_t w[12];
23    for (int i = 0; i < 12; i++) w[i] = (uint32_t)rec[4 * i] | (uint32_t)rec[4 * i + 1] << 8 | (uint32_t)rec[4 * i + 2] << 16 | (uint32_t)rec[4 * i + 3] << 24;
24    uint64_t stamp = w[MESH_PIPEDEMO_R_STAMP_LO] | (uint64_t)w[MESH_PIPEDEMO_R_STAMP_HI] << 32;
25    uint32_t m[16];
26    mesh_pipedemo_msg(w[MESH_PIPEDEMO_R_SEQ], stamp, w[MESH_PIPEDEMO_R_ORIGIN] & 0xff, m);
27    uint8_t bytes[64], want[32], got[32];
28    for (int i = 0; i < 16; i++)
29        for (int k = 0; k < 4; k++) bytes[4 * i + k] = (uint8_t)(m[i] >> (8 * k));
30    mesh_sha256d(bytes, sizeof bytes, want);
31    for (int i = 0; i < 8; i++)
32        for (int k = 0; k < 4; k++) got[4 * i + k] = (uint8_t)(w[MESH_PIPEDEMO_R_PAYLOAD + i] >> (24 - 8 * k));
33    mesh_view v = mesh_ctx_view(ctx);
34    mesh_view_add_u32(v, MESH_PIPEDEMO_K_RECORDS, 1);
35    if (memcmp(got, want, 32) != 0) {
36        mesh_view_add_u32(v, MESH_PIPEDEMO_K_BAD, 1);
37        mesh_ctx_count(ctx, 1, 1);
38        mesh_view_write(v, MESH_PIPEDEMO_K_LAST_BAD, rec, 48);
39    }
40    if (w[MESH_PIPEDEMO_R_ORIGIN] & MESH_PIPEDEMO_PATH_GPU) {
41        mesh_view_add_u32(v, MESH_PIPEDEMO_K_GPU, 1);
42        mesh_ctx_count(ctx, 2, 1);
43    } else {
44        mesh_view_add_u32(v, MESH_PIPEDEMO_K_CPU, 1);
45        mesh_ctx_count(ctx, 3, 1);
46    }
47    uint64_t d = now > stamp ? (now - stamp) / 1000 : 0;
48    uint32_t us = d > UINT32_MAX ? UINT32_MAX : (uint32_t)d;
49    mesh_view_add_u32(v, MESH_PIPEDEMO_K_HIST + 4 * (us < MESH_PIPEDEMO_K_BINS - 1 ? us : MESH_PIPEDEMO_K_BINS - 1), 1);
50    if (us > mesh_view_u32(v, MESH_PIPEDEMO_K_MAX_US)) mesh_view_set_u32(v, MESH_PIPEDEMO_K_MAX_US, us);
51}
52
53static void stats(const mesh_worker_ctx *ctx) {
54    mesh_view v = mesh_ctx_view(ctx);
55    static _Thread_local uint32_t hist[MESH_PIPEDEMO_K_BINS];
56    for (uint32_t i = 0; i < MESH_PIPEDEMO_K_BINS; i++) hist[i] = mesh_view_u32(v, MESH_PIPEDEMO_K_HIST + 4 * i);
57    const uint32_t q[3][2] = {{MESH_PIPEDEMO_K_P50_US, 500}, {MESH_PIPEDEMO_K_P90_US, 900}, {MESH_PIPEDEMO_K_P99_US, 990}};
58    for (int i = 0; i < 3; i++) mesh_view_set_u32(v, q[i][0], hist_quantile(hist, MESH_PIPEDEMO_K_BINS, q[i][1]));
59}
60
61static uint32_t drain(const mesh_worker_ctx *ctx) {
62    uint8_t rec[48];
63    uint32_t n = 0, len = 0;
64    /* An empty inbox: wait for a record (bounded, past the driver's tick) before the task ends. */
65    uint64_t t0 = mesh_ctx_host_ns(ctx);
66    while (!mesh_ctx_pipe_len(ctx, mesh_ctx_slot(ctx), &len) || len == 0) {
67        if (mesh_ctx_host_ns(ctx) - t0 > WAIT_NS) {
68            mesh_view_add_u32(mesh_ctx_view(ctx), MESH_PIPEDEMO_K_EMPTY, 1);
69            return 0;
70        }
71    }
72    while (n < DRAIN && mesh_ctx_pop(ctx, rec, sizeof rec)) {
73        check(ctx, rec, mesh_ctx_host_ns(ctx));
74        n++;
75    }
76    mesh_ctx_count(ctx, 0, n);
77    if (mesh_ctx_ticket(ctx) % MESH_PIPEDEMO_K_STATS_EVERY == 0) stats(ctx);
78    return n;
79}
80
81uint32_t pipesink_task(const mesh_worker_ctx *ctx, uint32_t op, uint32_t arg, void *user) {
82    (void)arg, (void)user;
83    switch (op) {
84    case MESH_OP_TASK: return drain(ctx);
85    case MESH_PIPEDEMO_OP_STATS: stats(ctx); return 0;
86    case MESH_PIPEDEMO_OP_SINK_RESET: {
87        mesh_view v = mesh_ctx_view(ctx);
88        for (uint32_t o = 0; o < MESH_PIPEDEMO_K_HIST; o += 4) mesh_view_set_u32(v, o, 0);
89        for (uint32_t i = 0; i < MESH_PIPEDEMO_K_BINS; i++) mesh_view_set_u32(v, MESH_PIPEDEMO_K_HIST + 4 * i, 0);
90        return 0;
91    }
92    default: return MESH_R_BAD_OP;
93    }
94}
95
96void pipesink_init(const mesh_worker_ctx *ctx, void *user) { (void)ctx, (void)user; }