1/* place.rs Placer: one slot's placement state machine (see the header). */
2#include "mesh/placer.h"
3#include "mesh/abi/place_ops.h"
4#include "mesh/abi/placement.h"
5#include "mesh/rust_fmt.h"
6
7#include <stdio.h>
8#include <stdlib.h>
9#include <string.h>
10
11typedef enum { PAUSING, SETTING } phase_t;
12
13/* Who asked for a change: a local request, or nobody (a re-plan after the up members changed). */
14typedef struct {
15    bool local;
16    uint64_t handle;
17} asker_t;
18
19typedef struct {
20    uint32_t k;
21    mpw_bytes state;
22    uint32_t part_ns;
23} member_state;
24
25/* A change the leader runs among the up members. */
26typedef struct {
27    uint32_t id;
28    mpm_mode from;
29    mpm_target to;
30    mpm_mode mode; /* resolved at Setting */
31    phase_t phase;
32    member_state states[MESH_PLACE_PARTS_MAX];
33    size_t nstates;
34    uint32_t waiting, members;
35    uint64_t since;
36    asker_t asker;
37} change_t;
38
39struct mpl_placer {
40    char name[MESH_NAME_BYTES + 1];
41    mps_spec spec;
42    uint32_t me, n;
43    mpm_mode mode;
44    bool paused;
45    uint32_t part_ns;
46    bool has_coord;
47    uint32_t coord;
48    uint32_t up, members;
49    bool has_request;
50    mt_strs request;
51    mpc_composer composer;
52    bool has_follow_inflight;
53    uint32_t follow_inflight;
54    bool has_follow_next;
55    mpw_frame follow_next;
56    bool has_change;
57    change_t change;
58    bool has_pause_for;
59    uint32_t pause_id, pause_leader;
60    bool has_join_at;
61    uint64_t join_at;
62    uint32_t next_id;
63    mpl_stats stats;
64    void (*log)(void *ctx, const char *line);
65    void *log_ctx;
66};
67
68static uint64_t since_us(uint64_t now, uint64_t t) { return now > t ? now - t : 0; }
69
70static void push(mpl_acts *a, mpl_act x) {
71    if (a->n == a->cap) {
72        a->cap = a->cap ? a->cap * 2 : 8;
73        a->v = mt_realloc(a->v, a->cap * sizeof *a->v);
74    }
75    a->v[a->n++] = x;
76}
77
78void mpl_acts_free(mpl_acts *a) {
79    for (size_t i = 0; i < a->n; i++) {
80        mpw_bytes_free(&a->v[i].bytes);
81        free(a->v[i].text);
82    }
83    free(a->v);
84    *a = (mpl_acts){0};
85}
86
87static void act_log(mpl_acts *a, char *text) { push(a, (mpl_act){.kind = MPL_LOG, .text = text}); }
88
89static void change_free(change_t *c) {
90    for (size_t i = 0; i < c->nstates; i++) mpw_bytes_free(&c->states[i].state);
91    c->nstates = 0;
92}
93
94mpl_placer *mpl_new(const char *name, mps_spec spec, uint32_t me, uint32_t n) {
95    mpl_placer *p = mt_alloc(sizeof *p);
96    size_t k = strlen(name);
97    if (k > MESH_NAME_BYTES) k = MESH_NAME_BYTES;
98    memcpy(p->name, name, k);
99    p->spec = spec;
100    p->me = me;
101    p->n = n;
102    p->mode = mpm_local();
103    p->up = p->members = 1u << me;
104    p->next_id = 1;
105    return p;
106}
107
108void mpl_free(mpl_placer *p) {
109    if (!p) return;
110    mps_free(&p->spec);
111    mt_strs_free(&p->request);
112    mpc_clear(&p->composer);
113    if (p->has_follow_next) mpw_frame_free(&p->follow_next);
114    if (p->has_change) change_free(&p->change);
115    free(p);
116}
117
118void mpl_set_log(mpl_placer *p, void (*log)(void *ctx, const char *line), void *ctx) {
119    p->log = log;
120    p->log_ctx = ctx;
121}
122
123const mpm_mode *mpl_mode(const mpl_placer *p) { return &p->mode; }
124mpm_role mpl_role(const mpl_placer *p) { return mpm_role_of(&p->mode, p->me); }
125bool mpl_paused(const mpl_placer *p) { return p->paused; }
126bool mpl_busy(const mpl_placer *p) { return p->has_change || p->has_pause_for; }
127uint32_t mpl_members(const mpl_placer *p) { return p->members; }
128uint32_t mpl_my_parts(const mpl_placer *p) { return mpm_parts_of(&p->mode, p->me, p->spec.fanout); }
129const mpl_stats *mpl_stats_of(const mpl_placer *p) { return &p->stats; }
130void mpl_set_part_ns(mpl_placer *p, uint32_t ns) { p->part_ns = ns; }
131const mps_spec *mpl_spec(const mpl_placer *p) { return &p->spec; }
132
133/* The other up members. */
134static uint32_t others(const mpl_placer *p) { return p->up & mpm_bits(p->n) & ~(1u << p->me); }
135static bool leading(const mpl_placer *p) { return p->has_coord && p->coord == p->me; }
136/* The members a mode needs (pin k: k; split: those with parts). */
137static uint32_t involved(const mpm_mode *m) { return m->kind == MPM_LOCAL ? 0 : m->kind == MPM_PIN ? 1u << m->pin : mpm_members(&m->a); }
138
139bool mpl_parse(const mpl_placer *p, char *const *words, size_t n, mpm_target *out, char **err) { return mpm_parse(words, n, p->n, p->spec.fanout, out, err); }
140
141bool mpl_plan(const mpl_placer *p, const mt_strs *words, mpm_target *out, char **err) {
142    uint32_t up[MESH_PLACE_PARTS_MAX];
143    size_t nup = 0;
144    for (uint32_t k = 0; k < p->n; k++)
145        if (p->up >> k & 1) up[nup++] = k;
146    mpm_target t;
147    if (!mpl_parse(p, words->v, words->n, &t, err)) return false;
148    if (t.is_auto) t.auto_mask &= p->up;
149    else if (t.mode.kind == MPM_PIN && !(p->up >> t.mode.pin & 1)) t.mode = mpm_pin(p->me);
150    else if (t.mode.kind == MPM_SPLIT && !((mpm_members(&t.mode.a) & ~p->up) == 0 && words->n != 1)) {
151        /* `split` alone: every up member; a spec naming a member that is down: its up members. */
152        uint32_t named[MESH_PLACE_PARTS_MAX];
153        size_t nn = 0;
154        if (words->n == 1) memcpy(named, up, nup * sizeof *up), nn = nup;
155        else {
156            for (uint32_t b = 0; b < t.mode.a.fanout; b++) {
157                uint32_t k = t.mode.a.of[b];
158                bool have = false;
159                for (size_t i = 0; i < nn; i++) have |= named[i] == k;
160                if ((p->up >> k & 1) && !have) named[nn++] = k;
161            }
162            if (!nn) memcpy(named, up, nup * sizeof *up), nn = nup;
163        }
164        t.mode = mpm_split(mpm_even(named, nn, p->spec.fanout));
165    }
166    *out = t;
167    return true;
168}
169
170static void msg(const mpl_placer *p, mpl_acts *a, mpl_dest to, uint32_t op, uint32_t id, mpm_mode mode, uint32_t aux, const uint8_t *bytes, size_t n) {
171    mpw_place m = {.op = op, .id = id, .mode = mode, .aux = aux, .bytes = (uint8_t *)bytes, .nbytes = n};
172    memcpy(m.name, p->name, sizeof m.name);
173    push(a, (mpl_act){.kind = MPL_SEND, .dest = to, .bytes = mpw_place_encode(&m)});
174}
175
176static mpl_dest to_k(uint32_t k) { return (mpl_dest){false, k}; }
177static mpl_dest to_mask(uint32_t m) { return (mpl_dest){true, m}; }
178
179/* The mode changes here (resets the per-mode frame state). */
180static void set_mode(mpl_placer *p, mpm_mode m, mpl_acts *a) {
181    p->mode = m;
182    mpc_clear(&p->composer);
183    p->has_follow_inflight = false;
184    if (p->has_follow_next) mpw_frame_free(&p->follow_next);
185    p->has_follow_next = false;
186    push(a, (mpl_act){.kind = MPL_MODE, .mode = m});
187}
188
189/* Answers a request: its asker, or a log line for a re-plan. Takes text. */
190static void answer(const mpl_placer *p, asker_t who, bool ok, char *text, mpl_acts *a) {
191    if (who.local) push(a, (mpl_act){.kind = MPL_ANSWER, .handle = who.handle, .ok = ok, .text = text});
192    else {
193        act_log(a, mt_fmt("place %s: %s", p->name, text));
194        free(text);
195    }
196}
197
198/* The leader starts a change: pause every other up member, collect their state. */
199static void begin(mpl_placer *p, mpm_target to, asker_t asker, uint64_t now, bool force, mpl_acts *a) {
200    if (p->has_change) {
201        answer(p, asker, false, mt_fmt("%s: a placement change is in progress", p->name), a);
202        return;
203    }
204    mpm_target cur = {.mode = p->mode};
205    if (mpm_target_eq(&to, &cur) && p->members == p->up && !force) {
206        char *l = mpm_label(&p->mode);
207        answer(p, asker, true, mt_fmt("%s already", l), a);
208        free(l);
209        return;
210    }
211    uint32_t id = p->next_id++;
212    p->paused = true;
213    char *label = to.is_auto ? mt_strdup("split auto") : mpm_label(&to.mode);
214    uint32_t o = others(p);
215    p->has_change = true;
216    p->change = (change_t){.id = id, .from = p->mode, .to = to, .mode = p->mode, .phase = PAUSING, .waiting = o, .members = p->up, .since = now, .asker = asker};
217    char *from = mpm_label(&p->mode), hx[24];
218    act_log(a, mt_fmt("place %s: %s -> %s (change %u): pausing the up members %s", p->name, from, label, id, rf_hex_alt(hx, p->up)));
219    free(from), free(label);
220    if (o) msg(p, a, to_mask(o), MESH_PL_PAUSE, id, p->mode, 0, NULL, 0);
221}
222
223static member_state *state_of(change_t *c, uint32_t k) {
224    for (size_t i = 0; i < c->nstates; i++)
225        if (c->states[i].k == k) return &c->states[i];
226    return NULL;
227}
228
229/* Removes k's state from the change into *out (true), or false. */
230static bool take_state(change_t *c, uint32_t k, mpw_bytes *out) {
231    for (size_t i = 0; i < c->nstates; i++)
232        if (c->states[i].k == k) {
233            *out = c->states[i].state;
234            c->states[i] = c->states[--c->nstates];
235            return true;
236        }
237    return false;
238}
239
240/* The mode a change lands on (split auto: every member's time per part, ours included). */
241static mpm_mode resolve(const mpl_placer *p, change_t *c) {
242    if (!c->to.is_auto) return c->to.mode;
243    uint32_t ks[MESH_PLACE_PARTS_MAX], ns[MESH_PLACE_PARTS_MAX];
244    size_t k = 0;
245    for (uint32_t s = 0; s < p->n; s++)
246        if (c->to.auto_mask >> s & 1) {
247            member_state *st = state_of(c, s);
248            ks[k] = s;
249            ns[k++] = s == p->me ? p->part_ns : st ? st->part_ns : 0;
250        }
251    mpm_mode m = mpm_split(mpm_auto(ks, ns, k, p->spec.fanout));
252    if (p->log) {
253        mt_buf b = {0};
254        mt_printf(&b, "place %s: split auto: time per part ", p->name);
255        for (size_t i = 0; i < k; i++) mt_printf(&b, i ? ", system %u %.1f us" : "system %u %.1f us", ks[i], (double)ns[i] / 1e3);
256        char *l = mpm_label(&m);
257        mt_printf(&b, " -> %s", l);
258        free(l);
259        p->log(p->log_ctx, mt_str(&b));
260        mt_buf_free(&b);
261    }
262    return m;
263}
264
265/* The leader re-runs the committed request over the up members. */
266static void replan(mpl_placer *p, uint64_t now, mpl_acts *a) {
267    p->has_join_at = false;
268    mt_strs words = p->has_request ? mt_strs_clone(&p->request) : (mt_strs){0};
269    if (!p->has_request) mt_push(&words, "local");
270    mpm_target t;
271    char *err = NULL;
272    if (mpl_plan(p, &words, &t, &err)) begin(p, t, (asker_t){false, 0}, now, true, a);
273    else {
274        char *j = mt_join(&words, " ");
275        char *q = rf_debug_str(j);
276        act_log(a, mt_fmt("place %s: re-plan of %s: %s", p->name, q, err));
277        free(j), free(q), free(err);
278    }
279    mt_strs_free(&words);
280}
281
282mpl_acts mpl_request(mpl_placer *p, const mt_strs *words, uint64_t handle, uint64_t now) {
283    mpl_acts a = {0};
284    if (!leading(p)) {
285        char *who = p->has_coord ? mt_fmt("the leader, system %u", p->coord) : mt_strdup("a leader (there is none: no majority)");
286        push(&a, (mpl_act){.kind = MPL_ANSWER, .handle = handle, .text = mt_fmt("peer: placement changes are coordinated by %s", who)});
287        free(who);
288        return a;
289    }
290    mpm_target to;
291    char *err = NULL;
292    bool ok = mpl_plan(p, words, &to, &err);
293    if (ok) {
294        mpm_mode m = to.is_auto ? mpm_split(mpm_even((uint32_t[]){0}, 1, p->spec.fanout)) : to.mode;
295        ok = mpm_allowed(&m, p->spec.cap, &err) && (to.is_auto || mpm_fits(&m, p->n, p->spec.fanout, &err));
296    }
297    if (!ok) {
298        push(&a, (mpl_act){.kind = MPL_ANSWER, .handle = handle, .text = mt_fmt("%s: %s", p->name, err)});
299        free(err);
300        return a;
301    }
302    mt_strs_free(&p->request);
303    p->has_request = !(words->n == 1 && strcmp(words->v[0], "local") == 0);
304    if (p->has_request) p->request = mt_strs_clone(words);
305    begin(p, to, (asker_t){true, handle}, now, false, &a);
306    return a;
307}
308
309mpl_acts mpl_view(mpl_placer *p, uint32_t up, bool has_coord, uint32_t coord, const mt_strs *request, uint64_t now) {
310    mpl_acts a = {0};
311    p->up = (up | 1u << p->me) & mpm_bits(p->n);
312    bool was_has = p->has_coord;
313    uint32_t was = p->coord;
314    p->has_coord = has_coord;
315    p->coord = has_coord ? coord : 0;
316    mt_strs_free(&p->request);
317    p->has_request = request != NULL;
318    if (request) p->request = mt_strs_clone(request);
319    /* Members gone during a change: nothing more to wait for from them. */
320    if (p->has_change) p->change.waiting &= p->up, p->change.members &= p->up;
321    if (p->has_change && !leading(p)) {
322        p->has_change = false;
323        change_free(&p->change);
324        p->paused = false;
325        answer(p, p->change.asker, false, mt_strdup("peer: the leader changed during the change"), &a);
326    }
327    if (p->has_pause_for && (!(p->up >> p->pause_leader & 1) || !has_coord || coord != p->pause_leader)) p->has_pause_for = false, p->paused = false;
328    uint32_t missing = involved(&p->mode) & ~p->up;
329    bool coord_changed = was_has != has_coord || (has_coord && was != coord);
330    char h1[24], h2[24];
331    if (leading(p)) {
332        if (missing && !p->has_change) {
333            char *l = mpm_label(&p->mode);
334            act_log(&a, mt_fmt("place %s: system(s) %s of %s are gone: re-planning over the up members %s", p->name, rf_hex_alt(h1, missing), l, rf_hex_alt(h2, p->up)));
335            free(l);
336            replan(p, now, &a);
337        } else if ((p->up & ~p->members) || (coord_changed && p->members != p->up)) {
338            if (!p->has_join_at) p->has_join_at = true, p->join_at = now + MPL_JOIN_DWELL_US;
339        } else p->has_join_at = false;
340    } else {
341        p->has_join_at = false;
342        if (!has_coord && missing && p->mode.kind != MPM_LOCAL) {
343            char *l = mpm_label(&p->mode);
344            act_log(&a, mt_fmt("place %s: system(s) %s are gone and there is no leader to re-plan: %s -> local on this machine", p->name, rf_hex_alt(h1, missing), l));
345            free(l);
346            p->paused = false;
347            set_mode(p, mpm_local(), &a);
348            p->members = 1u << p->me;
349        }
350    }
351    return a;
352}
353
354/* Setting with nothing more to wait for: the change is done. */
355static void finish(mpl_placer *p, uint64_t now, mpl_acts *a) {
356    if (!p->has_change || p->change.phase != SETTING || p->change.waiting) return;
357    change_t c = p->change;
358    p->has_change = false;
359    p->paused = false;
360    double ms = (double)since_us(now, c.since) / 1e3;
361    char *l = mpm_label(&c.mode), hx[24];
362    act_log(a, mt_fmt("place %s: %s on the up members %s in %.1f ms", p->name, l, rf_hex_alt(hx, c.members), ms));
363    answer(p, c.asker, true, mt_fmt("%s (state from system %u) in %.1f ms", l, mpm_authority(&c.from, &c.mode), ms), a);
364    free(l);
365    change_free(&c);
366}
367
368static void show(mpl_placer *p, uint32_t frame, uint32_t sys, uint8_t *buf, size_t n, uint32_t lat_us, mpl_acts *a) {
369    p->stats.shown++;
370    if (p->mode.kind == MPM_SPLIT) {
371        p->stats.lat_n++;
372        p->stats.lat_sum_us += lat_us;
373        if (lat_us > p->stats.lat_max_us) p->stats.lat_max_us = lat_us;
374    }
375    push(a, (mpl_act){.kind = MPL_SHOW, .frame = frame, .sys = sys, .bytes = {buf, n, n}, .lat_us = lat_us});
376}
377
378/* Parts into the composer; a whole frame is shown. */
379static bool compose(mpl_placer *p, uint32_t frame, uint32_t parts, uint32_t sys, const uint8_t *data, size_t n, uint64_t now, mpl_acts *a) {
380    mpc_whole w;
381    char *err = NULL;
382    int r = mpc_add(&p->composer, &p->spec, frame, parts, sys, data, n, now, &w, &err);
383    if (r == 1) {
384        show(p, w.frame, w.sys, w.buf, w.nbuf, (uint32_t)since_us(now, w.t0), a);
385        return true;
386    }
387    if (r < 0) {
388        act_log(a, mt_fmt("place %s: %s", p->name, err));
389        free(err);
390    }
391    return false;
392}
393
394static void try_follow(mpl_placer *p, uint64_t now, mpl_acts *a) {
395    if (p->has_follow_inflight || !p->has_follow_next) return;
396    mpw_frame m = p->follow_next;
397    p->has_follow_next = false;
398    mpw_bytes st = {0};
399    mpw_put(&st, m.state, m.nstate);
400    push(a, (mpl_act){.kind = MPL_WRITE_STATE, .bytes = st});
401    mpc_whole w;
402    char *err = NULL;
403    int r = mpc_add(&p->composer, &p->spec, m.frame, m.parts, m.sys, m.data, m.ndata, now, &w, &err);
404    if (r < 0) {
405        act_log(a, mt_fmt("place %s: %s", p->name, err));
406        free(err);
407        mpw_frame_free(&m);
408        return;
409    }
410    if (r == 1) free(w.buf); /* a follower's own parts complete it later */
411    p->has_follow_inflight = true;
412    p->follow_inflight = m.frame;
413    push(a, (mpl_act){.kind = MPL_FOLLOW, .frame = m.frame});
414    mpw_frame_free(&m);
415}
416
417mpl_acts mpl_tick(mpl_placer *p, uint64_t now, bool drained, mpl_state_fn state, void *ctx) {
418    mpl_acts a = {0};
419    if (p->has_pause_for && drained) {
420        p->has_pause_for = false;
421        mpw_bytes st = state(ctx);
422        msg(p, &a, to_k(p->pause_leader), MESH_PL_PAUSED, p->pause_id, p->mode, p->part_ns, st.b, st.n);
423        mpw_bytes_free(&st);
424    }
425    if (leading(p) && !p->has_change && p->has_join_at && now >= p->join_at) {
426        char hx[24];
427        act_log(&a, mt_fmt("place %s: members %s up again: re-planning", p->name, rf_hex_alt(hx, p->up & ~p->members)));
428        replan(p, now, &a);
429    }
430    if (p->has_change) {
431        change_t *c = &p->change;
432        if (since_us(now, c->since) > MPL_CHANGE_TIMEOUT_US) {
433            mt_buf miss = {0};
434            for (uint32_t k = 0; k < p->n; k++)
435                if (c->waiting >> k & 1) mt_printf(&miss, miss.len ? ",%u" : "%u", k);
436            char *from = mpm_label(&c->from);
437            char *why = mt_fmt("%s: the change did not finish in 5s (system(s) %s did not %s); back to %s", p->name, mt_str(&miss), c->phase == PAUSING ? "drain" : "apply it", from);
438            free(from);
439            mt_buf_free(&miss);
440            change_t cc = *c;
441            p->has_change = false;
442            act_log(&a, mt_strdup(why));
443            if (!mpm_mode_eq(&p->mode, &cc.from)) set_mode(p, cc.from, &a);
444            uint32_t o = others(p);
445            if (o) msg(p, &a, to_mask(o), MESH_PL_FAIL, cc.id, cc.from, 0, (const uint8_t *)why, strlen(why));
446            p->paused = false;
447            answer(p, cc.asker, false, why, &a);
448            change_free(&cc);
449        } else if (c->phase == PAUSING && drained && c->waiting == 0) {
450            mpm_mode to = resolve(p, c);
451            /* The state that continues: the old mode's authority's; if it is gone, the new authority's
452             * own copy (a mirror or follower holds the last frame's state). */
453            uint32_t auth = mpm_authority(&c->from, &to), src;
454            mpw_bytes st;
455            if (auth == p->me) src = p->me, st = state(ctx);
456            else if (take_state(c, auth, &st)) src = auth;
457            else {
458                mpm_mode local = mpm_local();
459                uint32_t na = mpm_authority(&local, &to);
460                bool had = take_state(c, na, &st);
461                if (had && na != p->me) src = na;
462                else {
463                    if (had) mpw_bytes_free(&st);
464                    src = p->me, st = state(ctx);
465                }
466            }
467            if (src != auth) act_log(&a, mt_fmt("place %s: system %u's state is gone: the game continues from system %u's copy", p->name, auth, src));
468            c->phase = SETTING;
469            c->mode = to;
470            c->waiting = others(p) & c->members;
471            if (src != p->me) {
472                mpw_bytes copy = {0};
473                mpw_put(&copy, st.b, st.n);
474                push(&a, (mpl_act){.kind = MPL_WRITE_STATE, .bytes = copy});
475            }
476            set_mode(p, to, &a);
477            p->members = c->members;
478            /* PL_SET's aux: the members (bits 0..8) and the system whose state continues (bits 8..16). */
479            if (c->waiting) msg(p, &a, to_mask(c->waiting), MESH_PL_SET, c->id, to, c->members | src << 8, st.b, st.n);
480            mpw_bytes_free(&st);
481            finish(p, now, &a);
482        }
483    }
484    if (mpl_role(p) == MPM_ROLE_FOLLOWER && !p->paused) try_follow(p, now, &a);
485    return a;
486}
487
488mpl_acts mpl_peer_place(mpl_placer *p, mpw_place m, uint32_t from, uint64_t now) {
489    mpl_acts a = {0};
490    if (m.op == MESH_PL_PAUSED && leading(p)) {
491        if (p->has_change && p->change.id == m.id && p->change.phase == PAUSING) {
492            change_t *c = &p->change;
493            member_state *s = state_of(c, from);
494            if (!s) s = &c->states[c->nstates++], s->k = from;
495            else mpw_bytes_free(&s->state);
496            s->state = (mpw_bytes){m.bytes, m.nbytes, m.nbytes};
497            s->part_ns = m.aux;
498            m.bytes = NULL;
499            c->waiting &= ~(1u << from);
500        } else act_log(&a, mt_fmt("place %s: stale PL_PAUSED %u from system %u", p->name, m.id, from));
501    } else if (m.op == MESH_PL_DONE && leading(p)) {
502        if (p->has_change && p->change.id == m.id && p->change.phase == SETTING) {
503            p->change.waiting &= ~(1u << from);
504            finish(p, now, &a);
505        } else act_log(&a, mt_fmt("place %s: stale PL_DONE %u from system %u", p->name, m.id, from));
506    } else if (m.op == MESH_PL_PAUSE) {
507        p->paused = true;
508        p->has_pause_for = true;
509        p->pause_id = m.id;
510        p->pause_leader = from;
511    } else if (m.op == MESH_PL_SET) {
512        char *err = NULL;
513        if (!mpm_fits(&m.mode, p->n, p->spec.fanout, &err)) {
514            char *l = mpm_label(&m.mode);
515            act_log(&a, mt_fmt("place %s: system %u set %s which does not fit here (%s); local", p->name, from, l, err));
516            free(l), free(err);
517            set_mode(p, mpm_local(), &a);
518        } else {
519            /* The state continues from the system the leader took it from (aux bits 8..16). */
520            if ((m.aux >> 8 & 0xff) != p->me) {
521                push(&a, (mpl_act){.kind = MPL_WRITE_STATE, .bytes = {m.bytes, m.nbytes, m.nbytes}});
522                m.bytes = NULL;
523            }
524            set_mode(p, m.mode, &a);
525            p->members = m.aux & 0xff & mpm_bits(p->n);
526        }
527        p->paused = false;
528        p->has_pause_for = false;
529        msg(p, &a, to_k(from), MESH_PL_DONE, m.id, p->mode, 0, NULL, 0);
530    } else if (m.op == MESH_PL_FAIL) {
531        char *r = mt_strndup((const char *)m.bytes, m.nbytes);
532        act_log(&a, mt_fmt("place %s: system %u rolled the change back: %s", p->name, from, r));
533        free(r);
534        if (!mpm_mode_eq(&p->mode, &m.mode)) set_mode(p, m.mode, &a);
535        p->paused = false;
536        p->has_pause_for = false;
537    } else {
538        char lead[24];
539        if (p->has_coord) snprintf(lead, sizeof lead, "Some(%u)", p->coord);
540        else strcpy(lead, "None");
541        act_log(&a, mt_fmt("place %s: F_PLACE op %u from system %u is not for system %u (the leader is %s); ignored", p->name, m.op, from, p->me, lead));
542    }
543    mpw_place_free(&m);
544    return a;
545}
546
547static mpw_bytes frame_bytes(const mpl_placer *p, uint32_t frame, uint32_t parts, const uint8_t *st, size_t nst, const uint8_t *data, size_t nd) {
548    mpw_frame f = {.frame = frame, .parts = parts, .sys = p->me, .state = (uint8_t *)st, .nstate = nst, .data = (uint8_t *)data, .ndata = nd};
549    memcpy(f.name, p->name, sizeof f.name);
550    return mpw_frame_encode(&f);
551}
552
553mpl_acts mpl_own_frame(mpl_placer *p, uint32_t frame, mpw_bytes state, mpw_bytes data, uint64_t now) {
554    mpl_acts a = {0};
555    uint32_t parts = mpl_my_parts(p);
556    mpl_dest to = to_mask(p->members & mpm_bits(p->n) & ~(1u << p->me));
557    mpm_role r = mpl_role(p);
558    if (r == MPM_ROLE_RUNNER) {
559        size_t nb = mps_out_bytes(&p->spec);
560        uint8_t *buf = mt_alloc(nb ? nb : 1);
561        char *err = NULL;
562        if (!mpc_scatter(&p->spec, buf, nb, parts, data.b, data.n, &err)) {
563            act_log(&a, mt_fmt("place %s: %s", p->name, err));
564            free(err), free(buf);
565            goto done;
566        }
567        show(p, frame, mpc_sys_nibbles(parts, p->me), buf, nb, 0, &a);
568        push(&a, (mpl_act){.kind = MPL_SEND, .dest = to, .bytes = frame_bytes(p, frame, parts, state.b, state.n, data.b, data.n)});
569    } else if (r == MPM_ROLE_OWNER) {
570        mpc_whole w;
571        char *err = NULL;
572        int x = mpc_add(&p->composer, &p->spec, frame, parts, p->me, data.b, data.n, now, &w, &err);
573        if (x == 1) show(p, w.frame, w.sys, w.buf, w.nbuf, 0, &a);
574        else if (x < 0) {
575            act_log(&a, mt_fmt("place %s: %s", p->name, err));
576            free(err);
577        }
578        push(&a, (mpl_act){.kind = MPL_SEND, .dest = to, .bytes = frame_bytes(p, frame, parts, state.b, state.n, data.b, data.n)});
579    }
580done:
581    mpw_bytes_free(&state);
582    mpw_bytes_free(&data);
583    return a;
584}
585
586mpl_acts mpl_follow_done(mpl_placer *p, mpw_bytes data, uint64_t now) {
587    mpl_acts a = {0};
588    if (!p->has_follow_inflight) goto done;
589    p->has_follow_inflight = false;
590    uint32_t frame = p->follow_inflight;
591    if (mpl_role(p) != MPM_ROLE_FOLLOWER) goto done;
592    uint32_t parts = mpl_my_parts(p);
593    compose(p, frame, parts, p->me, data.b, data.n, now, &a);
594    push(&a, (mpl_act){.kind = MPL_SEND, .dest = to_mask(p->members & mpm_bits(p->n) & ~(1u << p->me)), .bytes = frame_bytes(p, frame, parts, NULL, 0, data.b, data.n)});
595    if (!p->paused) try_follow(p, now, &a);
596done:
597    mpw_bytes_free(&data);
598    return a;
599}
600
601void mpl_follow_lost(mpl_placer *p) { p->has_follow_inflight = false; }
602
603mpl_acts mpl_peer_frame(mpl_placer *p, mpw_frame m, uint64_t now) {
604    mpl_acts a = {0};
605    /* The sender must render exactly these parts in our mode; the state comes from the authority only. */
606    uint32_t theirs = mpm_parts_of(&p->mode, m.sys, p->spec.fanout);
607    uint32_t auth = mpm_input_to(&p->mode, p->me);
608    bool valid = m.sys != p->me && theirs && m.parts == theirs && (m.nstate == 0 || (m.sys == auth && m.nstate == mps_state_bytes(&p->spec)));
609    if (!valid) {
610        p->stats.ignored++;
611        mpw_frame_free(&m);
612        return a;
613    }
614    mpm_role r = mpl_role(p);
615    if (r == MPM_ROLE_MIRROR) {
616        if (m.nstate) {
617            mpw_bytes st = {0};
618            mpw_put(&st, m.state, m.nstate);
619            push(&a, (mpl_act){.kind = MPL_WRITE_STATE, .bytes = st});
620        }
621        compose(p, m.frame, m.parts, m.sys, m.data, m.ndata, now, &a);
622    } else if (r == MPM_ROLE_FOLLOWER && m.sys == auth) {
623        if (!m.nstate) {
624            p->stats.ignored++;
625            mpw_frame_free(&m);
626            return a;
627        }
628        /* The newest wins: a frame still waiting is replaced (the owner is ahead). */
629        if (p->has_follow_next) {
630            p->stats.ignored++;
631            mpw_frame_free(&p->follow_next);
632        }
633        p->follow_next = m;
634        p->has_follow_next = true;
635        m = (mpw_frame){0};
636        if (!p->paused) try_follow(p, now, &a);
637    } else if (r == MPM_ROLE_FOLLOWER || r == MPM_ROLE_OWNER) {
638        /* Another follower's parts (to the owner, or to a fellow follower). */
639        if (!compose(p, m.frame, m.parts, m.sys, m.data, m.ndata, now, &a) && r == MPM_ROLE_OWNER && p->composer.n == 0) p->stats.ignored++;
640    } else p->stats.ignored++;
641    mpw_frame_free(&m);
642    return a;
643}
644
645mpl_acts mpl_drop_all(mpl_placer *p, const char *why) {
646    mpl_acts a = {0};
647    if (p->has_change) {
648        p->has_change = false;
649        answer(p, p->change.asker, false, mt_strdup(why), &a);
650        change_free(&p->change);
651    }
652    return a;
653}