LithosAnanake/v4/system/storage.c

/* storage.c -- what is on a storage port.  See storage.h.
 * docs/v4.0.0/MESH.md section 8.
 *
 * Nothing here uses the C library: the bare-metal kernel links this file.
 */
#include "v4/storage.h"

void v4_store_init(v4_store *s, const v4_store_ops *ops, void *common, void *private_chain, v4_cell number)
{
    unsigned i;

    s->ops = ops;
    s->common = common;
    s->private_chain = private_chain;
    s->number = number;
    s->in.count = 0;
    for (i = 0; i < V4_STORE_PENDING; i++) s->writing[i].used = 0;
    s->head = 0;
    s->count = 0;
    s->out.count = 0;
    s->out_at = 0;
    s->out_ready = 0;
    s->taken = 0;
}

/* The chain a number means and the block in it; or 0, with why in *status. */
static void *chain_for(const v4_store *s, uint64_t n, uint32_t *block, int *status)
{
    *status = V4_STORE_RANGE;
    if (n == 0 || n > V4_STORE_TOP) return 0;
    if (s->private_chain) {
        uint64_t k = V4_STORE_TOP - n, have = s->ops->blocks(s->private_chain);
        if (have > V4_STORE_FIRST && k < have - V4_STORE_FIRST) { *block = (uint32_t)(V4_STORE_FIRST + k); return s->private_chain; }
    }
    *block = (uint32_t)n;
    if (s->common) return s->common;
    if (s->private_chain && n < V4_STORE_FIRST) return s->private_chain;
    *status = V4_STORE_NONE;
    return 0;
}

static void queue(v4_store *s, v4_cell to, int read, uint64_t block, int status)
{
    unsigned at = (s->head + s->count) % V4_STORE_PENDING;

    if (s->count == V4_STORE_PENDING) return;       /* take does not let it come to this */
    s->answer[at].to = to;
    s->answer[at].read = read;
    s->answer[at].block = block;
    s->answer[at].status = status;
    s->count++;
}

/* A block of data has come from `from`: it is written where that sender's
 * BLOCK_WRITE said, if there was one. */
static int write_block(v4_store *s, v4_cell from)
{
    unsigned i, k;
    uint32_t block;
    int status;
    void *chain;

    for (i = 0; i < V4_STORE_PENDING && !(s->writing[i].used && s->writing[i].from == from); i++) { }
    if (i == V4_STORE_PENDING) return V4_STORE_REFUSED;
    s->writing[i].used = 0;
    if (v4_message_length(&s->in) != V4_STORE_BYTES) return V4_STORE_REFUSED;
    chain = chain_for(s, s->writing[i].block, &block, &status);
    if (!chain) return status;
    for (k = 0; k < V4_STORE_BYTES; k++) s->bytes[k] = (unsigned char)v4_message_char(&s->in, k);
    return s->ops->write(chain, block, s->bytes);
}

/* The message in s->in is whole. */
static void served(v4_store *s)
{
    v4_cell from = v4_message_from(&s->in), type = v4_message_type(&s->in);
    uint64_t block = (uint64_t)(v4_ucell)s->in.word[V4_MSG_HEADER];
    unsigned i, free_at = V4_STORE_PENDING;

    s->taken++;
    if (v4_message_to(&s->in) != s->number) return;
    if (type == V4_MSG_BLOCK_DATA) { queue(s, from, 0, 0, write_block(s, from)); return; }
    if (type != V4_MSG_BLOCK_READ && type != V4_MSG_BLOCK_WRITE) return;
    if (v4_message_length(&s->in) != 4) { queue(s, from, 0, 0, V4_STORE_REFUSED); return; }
    if (type == V4_MSG_BLOCK_READ) { queue(s, from, 1, block, 0); return; }

    /* a write: its number is kept until its block comes.  A sender's second
     * write before then takes the place of its first. */
    for (i = 0; i < V4_STORE_PENDING; i++) {
        if (s->writing[i].used && s->writing[i].from == from) { free_at = i; break; }
        if (!s->writing[i].used && free_at == V4_STORE_PENDING) free_at = i;
    }
    if (free_at == V4_STORE_PENDING) { queue(s, from, 0, 0, V4_STORE_REFUSED); return; }
    s->writing[free_at].used = 1;
    s->writing[free_at].from = from;
    s->writing[free_at].block = block;
}

int v4_store_take(void *self, v4_cell value)
{
    v4_store *s = (v4_store *)self;

    if (s->in.count == 0 && s->count == V4_STORE_PENDING) return 0;
    if (v4_message_word(&s->in, value)) {
        served(s);
        s->in.count = 0;
    }
    return 1;
}

/* The oldest answer waiting becomes the message in s->out. */
static void answer(v4_store *s)
{
    v4_cell to = s->answer[s->head].to;
    int status = s->answer[s->head].status;
    unsigned i;

    if (s->answer[s->head].read) {
        uint32_t block;
        void *chain = chain_for(s, s->answer[s->head].block, &block, &status);
        if (chain) status = s->ops->read(chain, block, s->bytes);
    } else if (status == V4_STORE_OK) {
        status = -1;                                 /* a write that worked: said below */
    }
    s->head = (s->head + 1) % V4_STORE_PENDING;
    s->count--;

    s->out.word[0] = to;
    s->out.word[1] = s->number;
    s->out.word[3] = 0;
    s->out.word[4] = 0;
    s->out.word[5] = 0;
    if (status == V4_STORE_OK) {                     /* a read that worked: the block */
        s->out.word[2] = V4_MSG_BLOCK_DATA;
        s->out.word[6] = (v4_cell)V4_STORE_BYTES;
        for (i = 0; i < V4_STORE_BYTES / 4u; i++)
            s->out.word[V4_MSG_HEADER + i] = (v4_cell)((v4_ucell)s->bytes[4u * i] | (v4_ucell)s->bytes[4u * i + 1u] << 8
                                                       | (v4_ucell)s->bytes[4u * i + 2u] << 16 | (v4_ucell)s->bytes[4u * i + 3u] << 24);
        s->out.count = V4_MSG_HEADER + V4_STORE_BYTES / 4u;
    } else {
        s->out.word[2] = V4_MSG_BLOCK_DONE;
        s->out.word[6] = 4;
        s->out.word[V4_MSG_HEADER] = status < 0 ? V4_STORE_OK : status;
        s->out.count = V4_MSG_HEADER + 1u;
    }
    s->out_at = 0;
    s->out_ready = 1;
}

int v4_store_give(void *self, v4_cell *value)
{
    v4_store *s = (v4_store *)self;

    if (!s->out_ready) {
        if (s->count == 0) return 0;
        answer(s);
    }
    *value = s->out.word[s->out_at++];
    if (s->out_at == s->out.count) s->out_ready = 0;
    return 1;
}