1/* pipesink-w: programs/pipesink on the wasm executor (alpha-rust-mesh programs/pipesink-w/wasm;
2 * mesh-abi views::pipedemo, K_*). Drains up to DRAIN digests a task; checks each against mesh-ref's
3 * SHA-256 of the msg rebuilt from its seq; histograms the latency since the source's stamp (the
4 * host clock: the same for every executor). */
5#include "hist_quantile.h"
6#include "mesh/abi/view_pipedemo.h"
7#include "mesh/ref/sha256d.h"
8#include "mesh/rt_wasm.h"
9
10#define DRAIN 512u
11/* An idle task waits this long (ns) for a record. */
12#define WAIT_NS 1000000ull
13
14static void check(const mesh_ctx *ctx, const uint32_t w[12], uint64_t now) {
15    uint64_t stamp = (uint64_t)w[MESH_PIPEDEMO_R_STAMP_LO] | (uint64_t)w[MESH_PIPEDEMO_R_STAMP_HI] << 32;
16    uint32_t m[16];
17    mesh_pipedemo_msg(w[MESH_PIPEDEMO_R_SEQ], stamp, w[MESH_PIPEDEMO_R_ORIGIN] & 0xffu, m);
18    uint8_t bytes[64];
19    for (int i = 0; i < 16; i++)
20        for (int k = 0; k < 4; k++) bytes[4 * i + k] = (uint8_t)(m[i] >> (8 * k));
21    uint8_t want[32], got[32];
22    mesh_sha256d(bytes, sizeof bytes, want);
23    for (int i = 0; i < 8; i++)
24        for (int k = 0; k < 4; k++) got[4 * i + k] = (uint8_t)(w[MESH_PIPEDEMO_R_PAYLOAD + i] >> (24 - 8 * k));
25    mesh_view v = mesh_ctx_view(ctx);
26    mesh_view_add32(v, MESH_PIPEDEMO_K_RECORDS, 1);
27    bool same = true;
28    for (int i = 0; i < 32; i++) same = same && got[i] == want[i];
29    if (!same) {
30        mesh_view_add32(v, MESH_PIPEDEMO_K_BAD, 1);
31        mesh_ctx_count(ctx, 1, 1);
32        mesh_view_write_u32s(v, MESH_PIPEDEMO_K_LAST_BAD, w, 12);
33    }
34    if (w[MESH_PIPEDEMO_R_ORIGIN] & MESH_PIPEDEMO_PATH_GPU) {
35        mesh_view_add32(v, MESH_PIPEDEMO_K_GPU, 1);
36        mesh_ctx_count(ctx, 2, 1);
37    } else {
38        mesh_view_add32(v, MESH_PIPEDEMO_K_CPU, 1);
39        mesh_ctx_count(ctx, 3, 1);
40    }
41    uint64_t d = now > stamp ? (now - stamp) / 1000 : 0;
42    uint32_t us = d > UINT32_MAX ? UINT32_MAX : (uint32_t)d;
43    mesh_view_add32(v, MESH_PIPEDEMO_K_HIST + 4 * (us < MESH_PIPEDEMO_K_BINS - 1 ? us : MESH_PIPEDEMO_K_BINS - 1), 1);
44    if (us > mesh_view_u32(v, MESH_PIPEDEMO_K_MAX_US)) mesh_view_set_u32(v, MESH_PIPEDEMO_K_MAX_US, us);
45}
46
47static uint32_t hist[MESH_PIPEDEMO_K_BINS];
48
49static void stats(const mesh_ctx *ctx) {
50    mesh_view v = mesh_ctx_view(ctx);
51    mesh_view_read_u32s(v, MESH_PIPEDEMO_K_HIST, hist, MESH_PIPEDEMO_K_BINS);
52    mesh_view_set_u32(v, MESH_PIPEDEMO_K_P50_US, hist_quantile(hist, MESH_PIPEDEMO_K_BINS, 500));
53    mesh_view_set_u32(v, MESH_PIPEDEMO_K_P90_US, hist_quantile(hist, MESH_PIPEDEMO_K_BINS, 900));
54    mesh_view_set_u32(v, MESH_PIPEDEMO_K_P99_US, hist_quantile(hist, MESH_PIPEDEMO_K_BINS, 990));
55}
56
57static uint32_t drain(const mesh_ctx *ctx) {
58    uint32_t w[12];
59    uint32_t n = 0, len = 0;
60    /* An empty inbox: wait for a record (bounded, past the driver's tick) before the task ends. */
61    uint64_t t0 = mesh_ctx_host_ns(ctx);
62    while (!mesh_ctx_pipe_len(ctx, mesh_ctx_slot(ctx), &len) || len == 0) {
63        if (mesh_ctx_host_ns(ctx) - t0 > WAIT_NS) {
64            mesh_view_add32(mesh_ctx_view(ctx), MESH_PIPEDEMO_K_EMPTY, 1);
65            return 0;
66        }
67    }
68    while (n < DRAIN && mesh_ctx_pop_u32s(ctx, w, 12)) {
69        check(ctx, w, mesh_ctx_host_ns(ctx));
70        n++;
71    }
72    mesh_ctx_count(ctx, 0, n);
73    if (mesh_ctx_ticket(ctx) % MESH_PIPEDEMO_K_STATS_EVERY == 0) stats(ctx);
74    return n;
75}
76
77static void init(const mesh_ctx *ctx) { (void)ctx; }
78
79static uint32_t task(const mesh_ctx *ctx, uint32_t op, uint32_t arg) {
80    (void)arg;
81    switch (op) {
82    case MESH_OP_TASK: return drain(ctx);
83    case MESH_PIPEDEMO_OP_STATS: stats(ctx); return 0;
84    case MESH_PIPEDEMO_OP_SINK_RESET: {
85        mesh_view v = mesh_ctx_view(ctx);
86        for (uint32_t o = 0; o < MESH_PIPEDEMO_K_HIST; o += 4) mesh_view_set_u32(v, o, 0);
87        static const uint32_t zeros[MESH_PIPEDEMO_K_BINS];
88        mesh_view_write_u32s(v, MESH_PIPEDEMO_K_HIST, zeros, MESH_PIPEDEMO_K_BINS);
89        return 0;
90    }
91    default: return MESH_R_BAD_OP;
92    }
93}
94
95MESH_PROGRAM(init, task)