1/* pipesrc-n: the pipe demo's source as a native worker (programs/pipesrc's
2 * twin, view pipedemo). Each continuous task pushes up to S_BATCH records:
3 * msgs into pipehash's inbox (mode 0), or digests it hashed itself into a sink
4 * (mode 1). A full pipe ends the task (backpressure); no consumer counts as gone.
5 * Ported from alpha-rust-mesh programs/pipesrc-n/src/main.rs. */
6#include "pipesrc.h"
7
8#include "mesh/abi/ops.h"
9#include "mesh/abi/pipe.h"
10#include "mesh/abi/view_pipedemo.h"
11
12#include <time.h>
13
14void pipesrc_init(const mesh_worker_ctx *ctx, void *user) {
15    (void)user;
16    mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_MODE, MESH_PIPEDEMO_MODE_GPU);
17    mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_BATCH, MESH_PIPEDEMO_SRC_BATCH);
18}
19
20static bool dst(const mesh_worker_ctx *ctx, uint32_t mode, uint32_t *d) {
21    if (mode == MESH_PIPEDEMO_MODE_GPU) return mesh_ctx_find(ctx, "pipehash", d);
22    if (mode == MESH_PIPEDEMO_MODE_DIRECT) return mesh_ctx_find(ctx, "pipesink", d) || mesh_ctx_find(ctx, "pipesink-w", d);
23    return false;
24}
25
26static void le_bytes(const uint32_t *w, size_t n, uint8_t *out) {
27    for (size_t i = 0; i < n; i++)
28        for (int k = 0; k < 4; k++) out[4 * i + k] = (uint8_t)(w[i] >> (8 * k));
29}
30
31static void gone(const mesh_worker_ctx *ctx) {
32    mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_GONE, mesh_ctx_view_u32(ctx, MESH_PIPEDEMO_S_GONE) + 1);
33    mesh_ctx_count(ctx, 2, 1);
34}
35
36static uint32_t send(const mesh_worker_ctx *ctx) {
37    uint32_t mode = mesh_ctx_view_u32(ctx, MESH_PIPEDEMO_S_MODE);
38    if (mode == MESH_PIPEDEMO_MODE_OFF) {
39        struct timespec ts = {0, 200000};
40        nanosleep(&ts, NULL);
41        return 0;
42    }
43    uint64_t iv = (uint64_t)mesh_ctx_view_u32(ctx, MESH_PIPEDEMO_S_INTERVAL_US) * 1000;
44    uint64_t now = mesh_ctx_host_ns(ctx);
45    /* Paced: wait out the interval in the task (bounded), so the pace is not the driver's tick. */
46    uint64_t last = mesh_ctx_view_u64(ctx, MESH_PIPEDEMO_S_LAST_NS);
47    if (iv > 0 && now - last < iv) {
48        if (iv - (now - last) > 2000000) return 0;
49        while (now - last < iv) now = mesh_ctx_host_ns(ctx);
50    }
51    mesh_ctx_set_u64(ctx, MESH_PIPEDEMO_S_LAST_NS, now);
52    uint32_t d;
53    if (!dst(ctx, mode, &d)) {
54        gone(ctx);
55        return 0;
56    }
57    mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_DST, d + 1);
58    uint32_t seq = mesh_ctx_view_u32(ctx, MESH_PIPEDEMO_S_NEXT_SEQ), sent = 0, full = 0, batch = mesh_ctx_view_u32(ctx, MESH_PIPEDEMO_S_BATCH);
59    for (uint32_t i = 0; i < batch; i++) {
60        uint32_t m[16], dg[12];
61        uint8_t rec[64];
62        size_t n;
63        mesh_pipedemo_msg(seq, mesh_ctx_host_ns(ctx), MESH_PIPEDEMO_ORIGIN_NATIVE, m);
64        if (mode == MESH_PIPEDEMO_MODE_GPU) {
65            le_bytes(m, 16, rec);
66            n = 64;
67        } else {
68            mesh_pipedemo_digest(m, 0, dg);
69            le_bytes(dg, 12, rec);
70            n = 48;
71        }
72        int32_t code = mesh_ctx_push_code(ctx, d, rec, n);
73        if (code == MESH_PIPE_OK) {
74            seq++;
75            sent++;
76        } else if (code == MESH_PIPE_NONE) {
77            full = 1;
78            break;
79        } else {
80            gone(ctx);
81            break;
82        }
83    }
84    mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_NEXT_SEQ, seq);
85    mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_SENT, mesh_ctx_view_u32(ctx, MESH_PIPEDEMO_S_SENT) + sent);
86    mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_FULL, mesh_ctx_view_u32(ctx, MESH_PIPEDEMO_S_FULL) + full);
87    mesh_ctx_count(ctx, 0, sent);
88    mesh_ctx_count(ctx, 1, full);
89    return sent;
90}
91
92uint32_t pipesrc_task(const mesh_worker_ctx *ctx, uint32_t op, uint32_t arg, void *user) {
93    (void)user;
94    switch (op) {
95    case MESH_OP_TASK: return send(ctx);
96    case MESH_PIPEDEMO_OP_MODE:
97        if (arg > MESH_PIPEDEMO_MODE_OFF) return MESH_R_BAD_OP;
98        mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_MODE, arg);
99        return 0;
100    case MESH_PIPEDEMO_OP_BATCH:
101        if (arg < 1 || arg > 4096) return MESH_R_BAD_OP;
102        mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_BATCH, arg);
103        return 0;
104    case MESH_PIPEDEMO_OP_INTERVAL: mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_INTERVAL_US, arg); return 0;
105    case MESH_PIPEDEMO_OP_RESET:
106        mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_SENT, 0);
107        mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_FULL, 0);
108        mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_S_GONE, 0);
109        return 0;
110    default: return MESH_R_BAD_OP;
111    }
112}