==> src/pipehash.c <==1/* pipehash: the pipe demo's GPU stage (@ajax/mesh-abi-view-pipedemo, H_*). A
2 * task (one warp) finds the sink (pipesink, else pipesink-w), pushes the
3 * digests its node carried from an earlier task, then -- if the sink has room
4 * for 32 more -- pops up to 32 `msg`s from its own inbox (lane i keeps the
5 * i-th), hashes them in parallel (sha256d of the 64 bytes, pipedemo digest)
6 * and pushes the digests on, one collective push each. A digest the sink does
7 * not take within PUSH_WAIT_NS goes into the node's carry in the view, so
8 * nothing is dropped and the next task retries it first. Without room the
9 * warp pops nothing: backpressure up the pipe.
10 * Counters: 0 hashed (pushed on), 1 stalls, 2 carried.
11 * Ported from alpha-rust-mesh programs/pipehash/src/lib.rs. */
12#include "mesh/rt.h"
13
14/* An idle task waits this long for work (ns) before it ends: past the
15 * driver's 500 us tick, so a warp is always waiting on the inbox. */
16#define PIPEHASH_WAIT_NS UINT64_C(1000000)
17/* A push into a full sink retries this long (ns) before the digest is carried. */
18#define PIPEHASH_PUSH_WAIT_NS UINT64_C(200000)
19#define PIPEHASH_DIGEST_WORDS (MESH_PIPEDEMO_DIGEST_BYTES / 4)
20
21static void pipehash_init(const mesh_ctx *ctx) { (void)ctx; }
22
23typedef struct {
24 uint32_t lane, owner;
25 const uint32_t *dg;
26} pipehash_fill;
27static void pipehash_fill_rec(mesh_view v, void *user) {
28 const pipehash_fill *f = user;
29 if (f->lane == f->owner)
30 for (uint32_t k = 0; k < 12; k++) mesh_view_set_u32(v, 4 * k, f->dg[k]);
31}
32
33/* Pushes the digest that lane owner holds (dg) into slot d, retrying a full
34 * sink for PUSH_WAIT_NS: true pushed. */
35static bool pipehash_push(const mesh_ctx *ctx, uint32_t d, uint32_t owner, const uint32_t dg[12]) {
36 pipehash_fill f = {mesh_ctx_lane(ctx), owner, dg};
37 uint64_t t0 = mesh_rt_gtimer();
38 for (;;) {
39 int32_t code = mesh_ctx_push_with(ctx, d, pipehash_fill_rec, &f);
40 if (code == MESH_PIPE_OK) return true;
41 if (code != MESH_PIPE_NONE || mesh_rt_shfl_idx(mesh_rt_gtimer() - t0 > PIPEHASH_PUSH_WAIT_NS ? 1u : 0u, 0) != 0) return false;
42 mesh_rt_nanosleep(2000);
43 }
44}
45
46static uint32_t pipehash_rec(uint32_t j) { return MESH_PIPEDEMO_H_CARRY_REC + j * MESH_PIPEDEMO_DIGEST_BYTES; }
47
48/* The node's carried digests go first; false: some are still carried. */
49static bool pipehash_push_carry(const mesh_ctx *ctx, bool have_d, uint32_t d, mesh_view carry) {
50 uint32_t held = mesh_rt_shfl_idx(mesh_view_u32(carry, 0), 0);
51 if (held == 0) return true;
52 uint32_t lane = mesh_ctx_lane(ctx), i = 0;
53 if (have_d) {
54 while (i < held) {
55 uint32_t dg[12] = {0};
56 if (lane == 0)
57 for (uint32_t k = 0; k < 12; k++) dg[k] = mesh_view_u32(carry, pipehash_rec(i) + 4 * k);
58 if (!pipehash_push(ctx, d, 0, dg)) break;
59 i++;
60 }
61 }
62 if (i == 0) return false;
63 /* Keep the rest at the front (lane k copies word k of each record, front to back). */
64 for (uint32_t j = i; j < held; j++)
65 if (lane < PIPEHASH_DIGEST_WORDS) mesh_view_set_u32(carry, pipehash_rec(j - i) + 4 * lane, mesh_view_u32(carry, pipehash_rec(j) + 4 * lane));
66 if (mesh_ctx_leader(ctx)) {
67 mesh_view_set_u32(carry, 0, held - i);
68 mesh_ctx_count(ctx, 0, i);
69 }
70 return i == held;
71}
72
73typedef struct {
74 uint32_t lane, i;
75 uint32_t *m;
76} pipehash_take;
77static void pipehash_take_rec(mesh_view v, void *user) {
78 pipehash_take *t = user;
79 if (t->lane == t->i)
80 for (uint32_t k = 0; k < 16; k++) t->m[k] = mesh_view_u32(v, 4 * k);
81}
82
83static uint32_t pipehash_task(const mesh_ctx *ctx, uint32_t op, uint32_t arg) {
84 (void)arg;
85 if (op != MESH_OP_TASK) return MESH_R_BAD_OP;
86 uint32_t lane = mesh_ctx_lane(ctx), node = mesh_ctx_node(ctx);
87 if (node >= MESH_PIPEDEMO_H_NODES_MAX) return MESH_R_BAD_OP;
88 mesh_view carry;
89 if (!mesh_view_sub(mesh_ctx_view(ctx), MESH_PIPEDEMO_H_CARRY + node * MESH_PIPEDEMO_H_CARRY_BYTES, MESH_PIPEDEMO_H_CARRY_BYTES, &carry))
90 return MESH_R_BAD_OP;
91 /* Every decision the warp takes together is lane 0's, broadcast: lanes
92 * reading the table or a ring's words at different moments could disagree,
93 * and a collective (push_with, pop_with) needs all 32. */
94 uint32_t d = UINT32_MAX, s;
95 if (mesh_ctx_find(ctx, "pipesink", 8, &s) || mesh_ctx_find(ctx, "pipesink-w", 10, &s)) d = s;
96 d = mesh_rt_shfl_idx(d, 0);
97 bool have_d = d != UINT32_MAX;
98 if (mesh_ctx_leader(ctx)) mesh_ctx_set_u32(ctx, MESH_PIPEDEMO_H_DST, have_d ? d + 1 : 0);
99 if (!pipehash_push_carry(ctx, have_d, d, carry)) {
100 if (mesh_ctx_leader(ctx)) mesh_ctx_count(ctx, 1, 1);
101 mesh_rt_nanosleep(20000);
102 return 0;
103 }
104 if (!have_d) {
105 mesh_rt_nanosleep(20000);
106 return 0;
107 }
108 /* Room for a warp's 32 digests in the sink, and something to hash: else wait (bounded). */
109 uint64_t t0 = mesh_rt_gtimer();
110 for (;;) {
111 uint32_t free_cells = 0, waiting = 0;
112 bool room = mesh_ctx_pipe_free(ctx, d, &free_cells) && free_cells >= 32;
113 bool work = mesh_ctx_pipe_len(ctx, mesh_ctx_slot(ctx), &waiting) && waiting > 0;
114 if (room && work) break;
115 if (mesh_rt_shfl_idx(mesh_rt_gtimer() - t0 > PIPEHASH_WAIT_NS ? 1u : 0u, 0) != 0) {
116 if (!room && mesh_ctx_leader(ctx)) mesh_ctx_count(ctx, 1, 1);
117 return 0;
118 }
119 mesh_rt_nanosleep(1000);
120 }
121 uint32_t m[16] = {0}, n = 0;
122 while (n < 32) {
123 pipehash_take t = {lane, n, m};
124 if (mesh_ctx_pop_with(ctx, pipehash_take_rec, &t) != MESH_PIPE_OK) break;
125 n++;
126 }
127 if (n == 0) return 0;
128 uint32_t dg[12] = {0};
129 if (lane < n) mesh_pipedemo_digest(m, MESH_PIPEDEMO_PATH_GPU, dg);
130 uint32_t i = 0;
131 while (i < n && pipehash_push(ctx, d, i, dg)) i++;
132 /* The sink did not take the rest in time: lane j's digest into carry slot j - i. */
133 if (lane >= i && lane < n)
134 for (uint32_t k = 0; k < 12; k++)
135 mesh_view_set_u32(carry, MESH_PIPEDEMO_H_CARRY_REC + (lane - i) * MESH_PIPEDEMO_DIGEST_BYTES + 4 * k, dg[k]);
136 if (mesh_ctx_leader(ctx)) {
137 mesh_view_set_u32(carry, 0, n - i);
138 mesh_ctx_count(ctx, 0, i);
139 if (i < n) mesh_ctx_count(ctx, 2, n - i);
140 }
141 return n;
142}
143
144MESH_PROGRAM(pipehash_init, pipehash_task);