1/* pipesrc-n on exec's node loop: msgs into pipehash's inbox in order (mode
2 * 0), digests straight into a sink (mode 1), backpressure when full, `gone`
3 * without a consumer, and its ops. */
4#include "harness.h"
5#include "../src/pipesrc.h"
6
7#include "mesh/abi/pipe.h"
8#include "mesh/abi/view_pipedemo.h"
9
10#define SRC 2
11#define HASH 0
12#define SINK 1
13
14int main(void) {
15    mesh_mailbox mb = mesh_mailbox_test();
16    mesh_words w = mesh_mailbox_words(mb);
17    h_install(mb, SRC, "pipesrc-n", 1, 0x100000, 0x100);
18    h_install(mb, HASH, "pipehash", 1, 0x110000, 0x16000);
19    h_inbox(mb, HASH, MESH_PIPEDEMO_H_INBOX, 128, 64, MESH_PIPEDEMO_MSG_TYPE);
20    h_feed(mb, SRC, HASH);
21    h_slot h = {.mb = mb, .slot = SRC, .prog = {pipesrc_init, pipesrc_task, NULL}};
22    h_start(&h);
23    CHECK_U(h_call(&h, MESH_OP_INIT, 0), MESH_R_NONE);
24    size_t v = 0x100000;
25    CHECK_U(mesh_mailbox_get(mb, v + MESH_PIPEDEMO_S_BATCH), MESH_PIPEDEMO_SRC_BATCH);
26    /* Mode 0: one batch of 64 msgs, in seq order on the CPU side. */
27    CHECK_U(h_call(&h, MESH_OP_TASK, 0), 64);
28    CHECK_U(mesh_mailbox_get(mb, v + MESH_PIPEDEMO_S_DST), HASH + 1);
29    uint8_t rec[64];
30    for (uint32_t s = 0; s < 64; s++) {
31        CHECK_I(mesh_pipe_pop(&w, HASH, rec, 64), 1);
32        uint32_t seq = (uint32_t)rec[0] | (uint32_t)rec[1] << 8 | (uint32_t)rec[2] << 16 | (uint32_t)rec[3] << 24;
33        CHECK_U(seq, s);
34        CHECK_U(rec[12], MESH_PIPEDEMO_ORIGIN_NATIVE);
35    }
36    /* Backpressure: the CPU side holds 64; the second batch of 64 fills it, the third is refused. */
37    CHECK_U(h_call(&h, MESH_OP_TASK, 0), 64);
38    CHECK_U(h_call(&h, MESH_OP_TASK, 0), 0);
39    CHECK_U(mesh_mailbox_get(mb, v + MESH_PIPEDEMO_S_FULL), 1);
40    CHECK_U(mesh_mailbox_get(mb, v + MESH_PIPEDEMO_S_SENT), 128);
41    /* Mode 1 with no sink: gone. Then a sink: digests of msgs 128.. */
42    CHECK_U(h_call(&h, MESH_PIPEDEMO_OP_MODE, MESH_PIPEDEMO_MODE_DIRECT), 0);
43    CHECK_U(h_call(&h, MESH_OP_TASK, 0), 0);
44    CHECK_U(mesh_mailbox_get(mb, v + MESH_PIPEDEMO_S_GONE), 1);
45    h_install(mb, SINK, "pipesink", 1, 0x130000, 0x3d000);
46    h_inbox(mb, SINK, MESH_PIPEDEMO_K_INBOX, 256, 48, MESH_PIPEDEMO_DIGEST_TYPE);
47    h_feed(mb, SRC, SINK);
48    CHECK_U(h_call(&h, MESH_PIPEDEMO_OP_BATCH, 5), 0);
49    CHECK_U(h_call(&h, MESH_OP_TASK, 0), 5);
50    uint8_t d[48];
51    CHECK_I(mesh_pipe_pop(&w, SINK, d, 48), 1);
52    uint32_t seq = (uint32_t)d[0] | (uint32_t)d[1] << 8 | (uint32_t)d[2] << 16 | (uint32_t)d[3] << 24;
53    CHECK_U(seq, 128);
54    /* Ops. */
55    CHECK_U(h_call(&h, MESH_PIPEDEMO_OP_MODE, 9), MESH_R_BAD_OP);
56    CHECK_U(h_call(&h, MESH_PIPEDEMO_OP_BATCH, 0), MESH_R_BAD_OP);
57    CHECK_U(h_call(&h, MESH_PIPEDEMO_OP_INTERVAL, 500), 0);
58    CHECK_U(h_call(&h, MESH_PIPEDEMO_OP_RESET, 0), 0);
59    CHECK_U(mesh_mailbox_get(mb, v + MESH_PIPEDEMO_S_SENT), 0);
60    CHECK_U(h_call(&h, MESH_PIPEDEMO_OP_MODE, MESH_PIPEDEMO_MODE_OFF), 0);
61    CHECK_U(h_call(&h, MESH_OP_TASK, 0), 0);
62    h_stop(&h);
63    free(mb.base);
64    return mesh_check_done();
65}