1/* jobs.rs's Out and durable. */
2#include "mesh/jobs/out.h"
3
4#include <stdlib.h>
5#include <string.h>
6
7void mj_out_free(mj_out *o) {
8    mj_frame_free(&o->frame);
9    mj_log_free(&o->log);
10    if (o->kind == MJ_OUT_RUN) mj_lease_msg_free(&o->run);
11    free(o->text);
12    memset(o, 0, sizeof *o);
13}
14
15void mj_outs_push(mj_outs *v, mj_out o) {
16    if (v->n == v->cap) {
17        v->cap = v->cap ? 2 * v->cap : 16;
18        v->v = mxrealloc(v->v, v->cap * sizeof *v->v);
19    }
20    v->v[v->n++] = o;
21}
22
23void mj_outs_append(mj_outs *dst, mj_outs *src) {
24    for (size_t i = 0; i < src->n; i++) mj_outs_push(dst, src->v[i]);
25    free(src->v);
26    *src = (mj_outs){0};
27}
28
29void mj_outs_free(mj_outs *v) {
30    for (size_t i = 0; i < v->n; i++) mj_out_free(&v->v[i]);
31    free(v->v);
32    *v = (mj_outs){0};
33}
34
35mj_out mj_out_send(uint32_t to, mj_frame f) { return (mj_out){.kind = MJ_OUT_SEND, .to = to, .frame = f}; }
36mj_out mj_out_push_blob(uint32_t to, mj_job_id job, mj_hash hash) { return (mj_out){.kind = MJ_OUT_PUSH_BLOB, .to = to, .job = job, .hash = hash}; }
37mj_out mj_out_fetch_blob(mj_job_id job, mj_hash hash, uint32_t holders) {
38    return (mj_out){.kind = MJ_OUT_FETCH_BLOB, .job = job, .hash = hash, .holders = holders};
39}
40mj_out mj_out_blob_done(mj_job_id job, mj_hash hash) { return (mj_out){.kind = MJ_OUT_BLOB_DONE, .job = job, .hash = hash}; }
41mj_out mj_out_persist(mj_log l) { return (mj_out){.kind = MJ_OUT_PERSIST, .log = l}; }
42mj_out mj_out_run(mj_lease_msg l) { return (mj_out){.kind = MJ_OUT_RUN, .run = l}; }
43mj_out mj_out_cut(uint64_t lease, uint64_t cut) { return (mj_out){.kind = MJ_OUT_CUT, .lease = lease, .cut = cut}; }
44mj_out mj_out_answer(uint64_t token, bool ok, char *text) { return (mj_out){.kind = MJ_OUT_ANSWER, .token = token, .ok = ok, .text = text}; }
45mj_out mj_out_log(char *text) { return (mj_out){.kind = MJ_OUT_LOG, .text = text}; }
46
47mj_outs mj_durable(mj_outs *outs, const mj_disk *disk, bool *ok, char **err) {
48    mj_outs v = {0};
49    bool dirty = false;
50    *ok = true;
51    size_t i = 0;
52    for (; i < outs->n; i++) {
53        mj_out *o = &outs->v[i];
54        if (o->kind == MJ_OUT_PERSIST) {
55            if (!disk->append(disk->ctx, &o->log, err)) {
56                *ok = false;
57                break;
58            }
59            dirty = true;
60            mj_out_free(o);
61            continue;
62        }
63        if ((o->kind == MJ_OUT_SEND || o->kind == MJ_OUT_ANSWER) && dirty) {
64            if (!disk->sync(disk->ctx, err)) {
65                *ok = false;
66                break;
67            }
68            dirty = false;
69        }
70        mj_outs_push(&v, *o);
71        memset(o, 0, sizeof *o);
72    }
73    if (*ok && dirty && !disk->sync(disk->ctx, err)) *ok = false;
74    /* What was not acted on is dropped. */
75    for (; i < outs->n; i++) mj_out_free(&outs->v[i]);
76    free(outs->v);
77    *outs = (mj_outs){0};
78    return v;
79}