1/* pipesrc: the pipe demo's source on the wasm executor (alpha-rust-mesh programs/pipesrc/wasm;
2 * mesh-abi views::pipedemo). Each continuous task pushes up to S_BATCH records: msgs into
3 * pipehash's inbox (mode 0, the GPU hashes them), or digests it hashed itself straight into a sink
4 * (mode 1). A full pipe ends the task (backpressure: nothing is dropped, the record is pushed by a
5 * later task); no consumer loaded counts as gone. Every record carries the host clock at its push. */
6#include "mesh/abi/view_pipedemo.h"
7#include "mesh/rt_wasm.h"
8
9static void init(const mesh_ctx *ctx) {
10    mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_MODE, MESH_PIPEDEMO_MODE_GPU);
11    mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_BATCH, MESH_PIPEDEMO_SRC_BATCH);
12}
13
14static bool dst(const mesh_ctx *ctx, uint32_t mode, uint32_t *slot) {
15    switch (mode) {
16    case MESH_PIPEDEMO_MODE_GPU: return mesh_ctx_find(ctx, "pipehash", 8, slot);
17    case MESH_PIPEDEMO_MODE_DIRECT: return mesh_ctx_find(ctx, "pipesink", 8, slot) || mesh_ctx_find(ctx, "pipesink-w", 10, slot);
18    default: return false;
19    }
20}
21
22static void gone(const mesh_ctx *ctx) {
23    mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_GONE, mesh_ctx_view_u32(ctx, MESH_PIPEDEMO_S_GONE) + 1u);
24    mesh_ctx_count(ctx, 2, 1);
25}
26
27static uint32_t send(const mesh_ctx *ctx) {
28    uint32_t mode = mesh_ctx_view_u32(ctx, MESH_PIPEDEMO_S_MODE);
29    if (mode == MESH_PIPEDEMO_MODE_OFF) return 0;
30    uint64_t iv = (uint64_t)mesh_ctx_view_u32(ctx, MESH_PIPEDEMO_S_INTERVAL_US) * 1000u;
31    /* Paced: wait out the interval in the task (bounded), so the pace is not the driver's tick. */
32    uint64_t now = mesh_ctx_host_ns(ctx);
33    uint64_t last = mesh_ctx_view_u64(ctx, MESH_PIPEDEMO_S_LAST_NS);
34    if (iv > 0 && now - last < iv) {
35        if (iv - (now - last) > 2000000u) return 0;
36        while (now - last < iv) now = mesh_ctx_host_ns(ctx);
37    }
38    mesh_ctx_set_u64(ctx, MESH_PIPEDEMO_S_LAST_NS, now);
39    uint32_t d;
40    if (!dst(ctx, mode, &d)) {
41        gone(ctx);
42        return 0;
43    }
44    mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_DST, d + 1u);
45    uint32_t seq = mesh_ctx_view_u32(ctx, MESH_PIPEDEMO_S_NEXT_SEQ), sent = 0, full = 0;
46    uint32_t batch = mesh_ctx_view_u32(ctx, MESH_PIPEDEMO_S_BATCH);
47    for (uint32_t i = 0; i < batch; i++) {
48        uint32_t m[16];
49        mesh_pipedemo_msg(seq, mesh_ctx_host_ns(ctx), MESH_PIPEDEMO_ORIGIN_WASM, m);
50        int32_t code;
51        if (mode == MESH_PIPEDEMO_MODE_GPU) {
52            code = mesh_ctx_push_code(ctx, d, m, sizeof m);
53        } else {
54            uint32_t g[12];
55            mesh_pipedemo_digest(m, 0, g);
56            code = mesh_ctx_push_code(ctx, d, g, sizeof g);
57        }
58        if (code == MESH_PIPE_OK) {
59            seq++;
60            sent++;
61        } else if (code == MESH_PIPE_NONE) {
62            full = 1;
63            break;
64        } else {
65            gone(ctx);
66            break;
67        }
68    }
69    mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_NEXT_SEQ, seq);
70    mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_SENT, mesh_ctx_view_u32(ctx, MESH_PIPEDEMO_S_SENT) + sent);
71    mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_FULL, mesh_ctx_view_u32(ctx, MESH_PIPEDEMO_S_FULL) + full);
72    mesh_ctx_count(ctx, 0, sent);
73    mesh_ctx_count(ctx, 1, full);
74    return sent;
75}
76
77static uint32_t task(const mesh_ctx *ctx, uint32_t op, uint32_t arg) {
78    switch (op) {
79    case MESH_OP_TASK: return send(ctx);
80    case MESH_PIPEDEMO_OP_MODE:
81        if (arg > MESH_PIPEDEMO_MODE_OFF) return MESH_R_BAD_OP;
82        mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_MODE, arg);
83        return 0;
84    case MESH_PIPEDEMO_OP_BATCH:
85        if (arg < 1 || arg > 4096) return MESH_R_BAD_OP;
86        mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_BATCH, arg);
87        return 0;
88    case MESH_PIPEDEMO_OP_INTERVAL: mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_INTERVAL_US, arg); return 0;
89    case MESH_PIPEDEMO_OP_RESET:
90        mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_SENT, 0);
91        mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_FULL, 0);
92        mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_GONE, 0);
93        return 0;
94    default: return MESH_R_BAD_OP;
95    }
96}
97
98MESH_PROGRAM(init, task)