1/* Ticket rings and the publishing policies. */
2#include "mesh/publish.h"
3#include "mesh/buf.h"
4
5#include <stdlib.h>
6#include <string.h>
7
8void mesh_pring_init(mesh_pring *r, uint32_t slot) {
9    memset(r, 0, sizeof *r);
10    r->ctl = mesh_prog_ctl(slot);
11}
12
13void mesh_pring_free(mesh_pring *r) {
14    for (size_t i = 0; i < r->n; i++) mesh_reply_drop(r->inflight[i].kind.reply);
15    free(r->inflight);
16    r->inflight = NULL;
17    r->n = r->cap = 0;
18}
19
20static size_t entry(const mesh_pring *r, uint32_t t) { return r->ctl + MESH_P_RING + MESH_ENTRY_BYTES * (size_t)(t % MESH_PROG_RING); }
21
22uint32_t mesh_pring_room(const mesh_pring *r) {
23    if (!r->n) return MESH_PROG_RING;
24    uint32_t used = r->tail - r->inflight[0].ticket;
25    return used >= MESH_PROG_RING ? 0 : MESH_PROG_RING - used;
26}
27
28bool mesh_pring_can_publish(const mesh_pring *r) { return mesh_pring_room(r) > 0; }
29
30bool mesh_pring_publish(mesh_pring *r, mesh_mailbox mb, uint32_t op, uint32_t arg, mesh_kind kind, uint32_t *ticket) {
31    if (!mesh_pring_can_publish(r)) return false;
32    uint32_t t = r->tail;
33    size_t e = entry(r, t);
34    mesh_mailbox_set(mb, e + MESH_E_OP, op);
35    mesh_mailbox_set(mb, e + MESH_E_ARG, arg);
36    mesh_mailbox_set(mb, e + MESH_E_SLOT, t % MESH_PROG_RESULTS);
37    mesh_mailbox_publish(mb, e + MESH_E_SEQ, t + 1);
38    r->tail = t + 1;
39    mesh_mailbox_set64(mb, r->ctl + MESH_P_RING_TAIL, r->tail);
40    if (r->n == r->cap) {
41        r->cap = r->cap ? 2 * r->cap : 16;
42        r->inflight = mesh_xrealloc(r->inflight, r->cap * sizeof *r->inflight);
43    }
44    r->inflight[r->n].ticket = t;
45    r->inflight[r->n].kind = kind;
46    r->n++;
47    if (ticket) *ticket = t;
48    return true;
49}
50
51void mesh_pring_collect(mesh_pring *r, mesh_mailbox mb, mesh_done **done, size_t *n) {
52    *done = NULL;
53    *n = 0;
54    if (!r->n) return;
55    size_t keep = 0, k = 0;
56    mesh_done *d = NULL;
57    for (size_t i = 0; i < r->n; i++) {
58        mesh_out o = r->inflight[i];
59        size_t c = o.ticket % MESH_PROG_RESULTS;
60        if (mesh_mailbox_get(mb, r->ctl + MESH_P_MARKS + 4 * c) == o.ticket + 1) {
61            if (!d) d = mesh_xmalloc(r->n * sizeof *d);
62            d[k].out = o;
63            d[k].result = mesh_mailbox_get(mb, r->ctl + MESH_P_RESULTS + 4 * c);
64            k++;
65        } else r->inflight[keep++] = o;
66    }
67    r->n = keep;
68    *done = d;
69    *n = k;
70}
71
72bool mesh_pring_idle(const mesh_pring *r) { return r->n == 0; }
73
74static uint32_t count(const mesh_pring *r, mesh_tk tk) {
75    uint32_t c = 0;
76    for (size_t i = 0; i < r->n; i++) c += r->inflight[i].kind.tk == tk;
77    return c;
78}
79
80uint32_t mesh_pring_auto_in_flight(const mesh_pring *r) { return count(r, MESH_TK_AUTO); }
81uint32_t mesh_pring_follow_in_flight(const mesh_pring *r) { return count(r, MESH_TK_FOLLOW); }
82size_t mesh_pring_in_flight(const mesh_pring *r) { return r->n; }
83bool mesh_pring_has_req(const mesh_pring *r) { return count(r, MESH_TK_REQ) > 0; }
84bool mesh_pring_has_init(const mesh_pring *r) { return count(r, MESH_TK_INIT) > 0; }
85
86void mesh_pring_rewind(mesh_pring *r, mesh_mailbox mb, mesh_out **outs, size_t *n) {
87    mesh_mailbox_set(mb, r->ctl + MESH_P_RING_HEAD, r->tail);
88    mesh_mailbox_set(mb, r->ctl + MESH_P_ONLINE, 0);
89    mesh_mailbox_set(mb, r->ctl + MESH_P_HALTED, 0);
90    *outs = r->inflight;
91    *n = r->n;
92    r->inflight = NULL;
93    r->n = r->cap = 0;
94}
95
96uint32_t mesh_due(const mesh_manifest *m, uint32_t n, const mesh_pring *r, int64_t *last, int64_t now) {
97    uint32_t a = mesh_pring_auto_in_flight(r);
98    if (m->batch && a > 0) return 0;
99    switch (m->policy) {
100    case MESH_POLICY_CONTINUOUS: return n - (a < n ? a : n);
101    case MESH_POLICY_PERIODIC: {
102        int64_t period = (int64_t)(1000000000ull / m->hz);
103        int64_t since = now > *last ? now - *last : 0;
104        if (since < period) return 0;
105        /* Keep the cadence; after a stall (or a stop) restart it from now. */
106        *last = since < 2 * period ? *last + period : now;
107        return n;
108    }
109    default: return 0;
110    }
111}
112
113uint32_t mesh_publish_auto(mesh_mailbox mb, const mesh_manifest *m, uint32_t view_off, mesh_pring *r, int64_t *last, int64_t now) {
114    uint32_t k = m->fanout;
115    if (m->batch && mesh_pring_room(r) < k) return 0;
116    uint32_t n = mesh_due(m, k, r, last, now);
117    uint32_t base = m->arg_mode == MESH_ARG_VIEW_U32 ? mesh_mailbox_get(mb, (size_t)view_off + m->arg_off) : 0;
118    uint32_t sent = 0;
119    for (uint32_t i = 0; i < n; i++) {
120        if (!mesh_pring_publish(r, mb, MESH_OP_TASK, base + i, mesh_kind_of(MESH_TK_AUTO), NULL)) break;
121        sent++;
122    }
123    return sent;
124}
125
126uint32_t mesh_publish_parts(mesh_mailbox mb, const mesh_manifest *m, uint32_t view_off, mesh_pring *r, int64_t *last,
127                            int64_t now, uint32_t parts) {
128    uint32_t k = 0;
129    for (int b = 0; b < 32; b++) k += (parts >> b) & 1u;
130    if (k == 0 || mesh_pring_room(r) < k) return 0;
131    if (mesh_due(m, k, r, last, now) < k) return 0;
132    uint32_t base = m->arg_mode == MESH_ARG_VIEW_U32 ? mesh_mailbox_get(mb, (size_t)view_off + m->arg_off) : 0;
133    uint32_t sent = 0;
134    for (uint32_t b = 0; b < 32; b++) {
135        if (!((parts >> b) & 1u)) continue;
136        if (!mesh_pring_publish(r, mb, MESH_OP_TASK, base + b, mesh_kind_of(MESH_TK_AUTO), NULL)) break;
137        sent++;
138    }
139    return sent;
140}