==> src/lease.c <==1/* The node loop, completion, faults, the lockstep round. */
2#include "mesh/lease.h"
3
4#include "mesh/abi/base.h"
5#include "mesh/abi/control_block.h"
6#include "mesh/abi/exchange.h"
7#include "mesh/abi/header.h"
8#include "mesh/abi/node.h"
9#include "mesh/abi/pipe.h"
10#include "mesh/abi/ring.h"
11#include "mesh/abi/trace.h"
12
13#include <sched.h>
14#include <stdlib.h>
15#include <time.h>
16
17uint64_t mesh_host_ns(void) {
18 struct timespec ts;
19 clock_gettime(CLOCK_MONOTONIC, &ts);
20 return (uint64_t)ts.tv_sec * 1000000000ull + (uint64_t)ts.tv_nsec;
21}
22
23static void sleep_ns(long ns) {
24 struct timespec ts = {0, ns};
25 nanosleep(&ts, NULL);
26}
27
28void mesh_lease_init(mesh_lease *l, mesh_mailbox mb, uint32_t slot, uint32_t first, uint32_t count, bool blink_gate,
29 _Atomic bool *done, mesh_lane_stats *stats) {
30 l->mb = mb;
31 l->slot = slot;
32 l->first = first;
33 l->count = count;
34 l->blink_gate = blink_gate;
35 l->done = done;
36 l->stats = stats;
37 l->beat_at = mesh_host_ns();
38}
39
40/* Up to max published tickets in order, each claimed by a CAS on P_RING_HEAD. */
41static size_t take(const mesh_lease *l, size_t max, mesh_taken *out) {
42 mesh_mailbox mb = l->mb;
43 size_t n = 0;
44 if (l->blink_gate && mesh_mailbox_get(mb, MESH_H_BLINK) != 0) return 0;
45 size_t head = mesh_lease_ctl(l, MESH_P_RING_HEAD);
46 while (n < max) {
47 uint32_t t = mesh_mailbox_get(mb, head);
48 size_t e = mesh_lease_ctl(l, MESH_P_RING + MESH_ENTRY_BYTES * (size_t)(t % MESH_PROG_RING));
49 if (mesh_mailbox_get(mb, e + MESH_E_SEQ) != t + 1u) break;
50 /* Read the entry before the claim: once head moves past t the publisher may reuse it. */
51 mesh_taken k = {t, mesh_mailbox_get(mb, e + MESH_E_OP), mesh_mailbox_get(mb, e + MESH_E_ARG),
52 mesh_mailbox_get(mb, e + MESH_E_SLOT) % MESH_PROG_RESULTS};
53 if (mesh_mailbox_cas(mb, head, t, t + 1u) == t) out[n++] = k;
54 }
55 return n;
56}
57
58/* Nothing published: heartbeat like idle nodes, poll again soon (sooner right after work: the
59 * publisher's next batch usually follows at once). True: this turn was a heartbeat. */
60static bool idle(mesh_lease *l, uint64_t since) {
61 uint64_t now = mesh_host_ns();
62 bool beat = now - l->beat_at >= MESH_LEASE_HEARTBEAT_NS;
63 if (beat) {
64 l->beat_at = now;
65 for (uint32_t n = l->first; n < l->first + l->count; n++) {
66 size_t r = mesh_node_rec(l->slot, n) + MESH_N_HEARTBEAT;
67 mesh_mailbox_set(l->mb, r, mesh_mailbox_get(l->mb, r) + 1u);
68 }
69 }
70 uint64_t idle_ns = now - since;
71 if (idle_ns < 200000) sched_yield();
72 else if (idle_ns < 50000000) sleep_ns(30000);
73 else sleep_ns(250000);
74 return beat;
75}
76
77void mesh_lease_run(mesh_lease *l, const mesh_runner *runner, void *self) {
78 mesh_mailbox mb = l->mb;
79 for (uint32_t n = l->first; n < l->first + l->count; n++) mesh_mailbox_set(mb, mesh_node_rec(l->slot, n) + MESH_N_SM, n);
80 atomic_thread_fence(memory_order_seq_cst);
81 mesh_mailbox_fetch_add(mb, mesh_lease_ctl(l, MESH_P_ONLINE), l->count);
82 size_t max = runner->max_take(self);
83 if (max > MESH_PROG_RING) max = MESH_PROG_RING;
84 if (max == 0) max = 1;
85 mesh_taken taken[MESH_PROG_RING];
86 mesh_words w = mesh_mailbox_words(mb);
87 uint64_t idle_since = mesh_host_ns();
88 for (;;) {
89 bool shutdown = mesh_mailbox_get(mb, mesh_lease_ctl(l, MESH_P_SHUTDOWN)) != 0;
90 size_t n = take(l, max, taken);
91 if (n == 0) {
92 /* A ticket published before the shutdown still runs (above); now none is. */
93 if (shutdown) break;
94 if (idle(l, idle_since) && runner->heartbeat) runner->heartbeat(self);
95 continue;
96 }
97 uint32_t parity = 0;
98 if (l->blink_gate) parity = mesh_pipe_flight_begin(&w, l->slot);
99 /* false: the fault is recorded (P_FAULT); the driver fails the slot and shuts it down. */
100 bool ok = runner->run(self, l, taken, n);
101 if (l->blink_gate) mesh_pipe_flight_end(&w, l->slot, parity);
102 if (ok) idle_since = mesh_host_ns();
103 }
104 atomic_thread_fence(memory_order_seq_cst);
105 mesh_mailbox_fetch_add(mb, mesh_lease_ctl(l, MESH_P_HALTED), l->count);
106 atomic_store(l->done, true);
107}
108
109void mesh_lease_complete(mesh_lease *l, uint32_t node, const mesh_taken *k, uint32_t result, uint64_t start, uint64_t end) {
110 mesh_mailbox mb = l->mb;
111 mesh_mailbox_set(mb, mesh_lease_ctl(l, MESH_P_RESULTS + 4 * (size_t)k->cslot), result);
112 mesh_mailbox_publish(mb, mesh_lease_ctl(l, MESH_P_MARKS + 4 * (size_t)k->cslot), k->t + 1u);
113 mesh_mailbox_fetch_add(mb, mesh_lease_ctl(l, MESH_P_DONE), 1);
114 size_t r = mesh_node_rec(l->slot, node);
115 mesh_mailbox_set(mb, r + MESH_N_TASKS, mesh_mailbox_get(mb, r + MESH_N_TASKS) + 1u);
116 mesh_mailbox_set(mb, r + MESH_N_LAST_OP, k->op);
117 mesh_mailbox_set64(mb, r + MESH_N_BUSY_NS, mesh_mailbox_get64(mb, r + MESH_N_BUSY_NS) + (end > start ? end - start : 0));
118 mesh_mailbox_set(mb, r + MESH_N_HEARTBEAT, mesh_mailbox_get(mb, r + MESH_N_HEARTBEAT) + 1u);
119 uint32_t tn = mesh_mailbox_fetch_add(mb, mesh_lease_ctl(l, MESH_P_TRACE_N), 1);
120 size_t tr = mesh_trace_rec(l->slot, tn);
121 mesh_mailbox_set(mb, tr, l->slot | node << 8 | k->op << 16);
122 mesh_mailbox_set(mb, tr + 4, k->t);
123 mesh_mailbox_set(mb, tr + 8, (uint32_t)start);
124 mesh_mailbox_set(mb, tr + 12, (uint32_t)end);
125 if (l->stats) atomic_fetch_add(&l->stats->tasks, 1);
126}
127
128void mesh_lease_fault(const mesh_lease *l, uint32_t code) {
129 size_t f = mesh_lease_ctl(l, MESH_P_FAULT);
130 if (mesh_mailbox_get(l->mb, f) == 0) mesh_mailbox_cas(l->mb, f, 0, code);
131}
132
133uint32_t mesh_lease_exchange(mesh_lease *l, uint64_t n) {
134 uint64_t waited = 0;
135 uint32_t st = mesh_round(l->mb, n, &waited);
136 if (l->stats) {
137 atomic_fetch_add(&l->stats->rounds, 1);
138 atomic_fetch_add(&l->stats->round_wait_ns, waited);
139 }
140 return st;
141}
142
143uint32_t mesh_round(mesh_mailbox mb, uint64_t n, uint64_t *waited_ns) {
144 uint32_t me = mesh_mailbox_get(mb, MESH_H_GROUP_RANK), sys = mesh_mailbox_get(mb, MESH_H_GROUP_N);
145 if (waited_ns) *waited_ns = 0;
146 if (!(sys >= 2 && sys <= MESH_MAX_SYSTEMS && me < sys)) return MESH_ROUND_TIMEOUT;
147 atomic_thread_fence(memory_order_seq_cst);
148 if (mesh_mailbox_get64(mb, mesh_ex_counter(me)) < n) mesh_mailbox_publish64(mb, mesh_ex_counter(me), n);
149 uint64_t t0 = mesh_host_ns();
150 uint64_t budget = (uint64_t)MESH_PEER_POLLS * MESH_PEER_POLL_NS;
151 bool ok;
152 for (;;) {
153 ok = true;
154 for (uint32_t j = 0; j < sys && ok; j++)
155 if (j != me && mesh_mailbox_get64(mb, mesh_ex_counter(j)) < n) ok = false;
156 if (ok) break;
157 if (mesh_host_ns() - t0 >= budget) break;
158 }
159 atomic_thread_fence(memory_order_seq_cst);
160 if (waited_ns) *waited_ns = mesh_host_ns() - t0;
161 return ok ? MESH_ROUND_OK : MESH_ROUND_TIMEOUT;
162}