1/* @ajax/mesh-abi-pipe: Pipes between programs: push into a connected inbox, pop your own.
2 *
3 * Part of alpha-c-mesh, a C port of alpha-rust-mesh (README.md says from where). */
4#ifndef MESH_ABI_PIPE_H
5#define MESH_ABI_PIPE_H
6
7#include <stdbool.h>
8#include <stddef.h>
9#include <stdint.h>
10#include "mesh/abi/base.h"
11#include "mesh/abi/header.h"
12#include "mesh/abi/program_table.h"
13#include "mesh/abi/pipe_table.h"
14#include "mesh/abi/pipe_ring.h"
15#include "mesh/abi/words.h"
16#include "mesh/abi/pipe_hash.h"
17
18/* The runtimes' return codes (wasm `mesh::pipe_*`, mesh-worker, mesh-rt):
19 * 1 done, 0 full (push) / empty (pop), negative: a PipeErr. */
20#define MESH_PIPE_OK 1
21#define MESH_PIPE_NONE 0
22#define MESH_PIPE_NOT_CONNECTED -1
23#define MESH_PIPE_SIZE -2
24#define MESH_PIPE_GONE -3
25
26/* Why a push or pop could not run (FAULT_PIPE for the first two). Functions
27 * below return 1 for Ok(true), 0 for Ok(false), and -e for Err(e), so that
28 * -NOT_CONNECTED, -SIZE and -GONE equal the runtimes' PIPE_* codes. */
29typedef enum {
30    /* This slot does not feed that one (no connected `feeds` entry), or it has no inbox. */
31    MESH_PIPE_ERR_NOT_CONNECTED = 1,
32    /* The record's size is not the inbox's PP_IN_REC. */
33    MESH_PIPE_ERR_SIZE = 2,
34    /* The consumer is not loaded (or FAILED: a stalled ring is not written to). */
35    MESH_PIPE_ERR_GONE = 3,
36    /* meshd holds the inbox (it is being reset): try later, like a full ring. */
37    MESH_PIPE_ERR_BUSY = 4,
38} mesh_pipe_err;
39
40/* The runtimes' return code of a result: PIPE_OK, PIPE_NONE (full, empty or
41 * busy), or a negative PIPE_* code. */
42static inline int32_t mesh_pipe_code(int r) {
43    if (r > 0) return MESH_PIPE_OK;
44    if (r == 0 || r == -MESH_PIPE_ERR_BUSY) return MESH_PIPE_NONE;
45    if (r == -MESH_PIPE_ERR_NOT_CONNECTED) return MESH_PIPE_NOT_CONNECTED;
46    if (r == -MESH_PIPE_ERR_SIZE) return MESH_PIPE_SIZE;
47    return MESH_PIPE_GONE;
48}
49
50/* Slot s's pipe block and view offset, read under its PT_GEN seqlock into
51 * out: (PT_STATE, PT_VIEW_OFF, PP_IN_OFF, PP_IN_CAP, PP_IN_REC,
52 * PP_FEEDS_MASK, PP_IN_HOLD). False: it kept changing (TRIES reads; the
53 * caller treats it as busy). */
54static inline bool mesh_pipe_read_block(const mesh_words *w, uint32_t s, uint32_t out[7]) {
55    size_t e = mesh_prog_entry(s), p = mesh_pipe_entry(s);
56    for (uint32_t i = 0; i < MESH_PIPE_TRIES; i++) {
57        uint32_t g = mesh_words_load(w, e + MESH_PT_GEN);
58        if (g & 1u) continue;
59        out[0] = mesh_words_load(w, e + MESH_PT_STATE);
60        out[1] = mesh_words_load(w, e + MESH_PT_VIEW_OFF);
61        out[2] = mesh_words_load(w, p + MESH_PP_IN_OFF);
62        out[3] = mesh_words_load(w, p + MESH_PP_IN_CAP);
63        out[4] = mesh_words_load(w, p + MESH_PP_IN_REC);
64        out[5] = mesh_words_load(w, p + MESH_PP_FEEDS_MASK);
65        out[6] = mesh_words_load(w, p + MESH_PP_IN_HOLD);
66        if (mesh_words_load(w, e + MESH_PT_GEN) == g) return true;
67    }
68    return false;
69}
70
71/* Slot c's inbox as slot `me` may push into it (`me == c`: its own, to pop):
72 * 0 with *out, or -e. An empty slot c is Gone before anything else: a
73 * producer that found its consumer just before it was unloaded (its feeds bit
74 * cleared right after) is told so, not faulted. */
75static inline int mesh_pipe_target(const mesh_words *w, uint32_t me, uint32_t c, mesh_pipe_inbox *out) {
76    uint32_t b[7], mine[7];
77    if (me >= MESH_PROG_SLOTS || c >= MESH_PROG_SLOTS) return -MESH_PIPE_ERR_NOT_CONNECTED;
78    if (!mesh_pipe_read_block(w, c, b)) return -MESH_PIPE_ERR_BUSY;
79    uint32_t st = b[0], view = b[1], off = b[2], cap = b[3], rec = b[4], hold = b[6];
80    if (st == MESH_ST_EMPTY) return -MESH_PIPE_ERR_GONE;
81    if (me != c) {
82        if (!mesh_pipe_read_block(w, me, mine)) return -MESH_PIPE_ERR_BUSY;
83        if ((mine[5] & (1u << c)) == 0) return -MESH_PIPE_ERR_NOT_CONNECTED;
84    }
85    if (me != c && st == MESH_ST_FAILED) return -MESH_PIPE_ERR_GONE;
86    if (cap == 0) return -MESH_PIPE_ERR_NOT_CONNECTED;
87    if (me != c && hold != 0) return -MESH_PIPE_ERR_BUSY;
88    out->base = (size_t)view + (size_t)off;
89    out->cap = cap;
90    out->rec = rec;
91    return 0;
92}
93
94/* Producer slot me's pushes in flight: +1 in its word of the current epoch's
95 * parity (returned, for flight_end) before it resolves the consumer's block;
96 * the fence orders that before the block's loads (meshd stores PP_IN_HOLD,
97 * bumps H_PIPE_EPOCH, fences, then reads the old parity's words). */
98static inline uint32_t mesh_pipe_flight_begin(const mesh_words *w, uint32_t me) {
99    uint32_t parity = mesh_words_load(w, MESH_H_PIPE_EPOCH) & 1u;
100    mesh_words_add(w, mesh_flight_word(me, parity), 1);
101    mesh_words_fence(w);
102    return parity;
103}
104/* The push is over (its commit, if any, is ordered before: release). */
105static inline void mesh_pipe_flight_end(const mesh_words *w, uint32_t me, uint32_t parity) {
106    mesh_words_add(w, mesh_flight_word(me, parity), UINT32_MAX);
107}
108
109/* Pushes rec (len bytes) into slot c's inbox, on the CPU side (host code:
110 * wasm lanes, native workers): 1 pushed, 0 full or busy (try again later), -e. */
111static inline int mesh_pipe_push(const mesh_words *w, uint32_t me, uint32_t c, const uint8_t *rec, size_t len) {
112    if (me >= MESH_PROG_SLOTS) return -MESH_PIPE_ERR_NOT_CONNECTED;
113    uint32_t parity = mesh_pipe_flight_begin(w, me);
114    mesh_pipe_inbox in;
115    int r = mesh_pipe_target(w, me, c, &in);
116    if (r == 0) {
117        mesh_pipe_ring ring = mesh_pipe_inbox_side(&in, MESH_PIPE_SIDE_CPU);
118        uint32_t p;
119        if (len != ring.rec) r = -MESH_PIPE_ERR_SIZE;
120        else if (!mesh_pipe_ring_reserve_push(&ring, w, me, &p)) r = 0;
121        else {
122            mesh_words_write(w, mesh_pipe_ring_cell(&ring, p), rec, len);
123            mesh_words_store(w, mesh_pipe_ring_sum(&ring, p), mesh_pipe_rec_sum_at(rec, len, p));
124            mesh_pipe_ring_commit_push(&ring, w, p);
125            r = 1;
126        }
127    }
128    mesh_pipe_flight_end(w, me, parity);
129    return r == -MESH_PIPE_ERR_BUSY ? 0 : r;
130}
131
132/* Pops the next record of slot me's own inbox into out (its PP_IN_REC bytes,
133 * len): 1 popped, 0 empty, -e. */
134static inline int mesh_pipe_pop(const mesh_words *w, uint32_t me, uint8_t *out, size_t len) {
135    mesh_pipe_inbox in;
136    int r = mesh_pipe_target(w, me, me, &in);
137    if (r == -MESH_PIPE_ERR_BUSY) return 0;
138    if (r < 0) return r;
139    if (len != in.rec) return -MESH_PIPE_ERR_SIZE;
140    mesh_pipe_ring ring;
141    uint32_t t;
142    if (!mesh_pipe_inbox_take(&in, w, out, len, &ring, &t)) {
143        mesh_pipe_ring cpu = mesh_pipe_inbox_side(&in, MESH_PIPE_SIDE_CPU);
144        mesh_words_add(w, cpu.base + MESH_PIPE_PI_EMPTY, 1);
145        return 0;
146    }
147    mesh_pipe_ring_commit_pop(&ring, w, t);
148    return 1;
149}
150
151#endif