diff options
| -rw-r--r-- | Makefile | 5 | ||||
| -rw-r--r-- | core/module.c | 19 | ||||
| -rw-r--r-- | core/module.h | 4 | ||||
| -rw-r--r-- | link/link.c | 110 | ||||
| -rw-r--r-- | link/link.h | 48 | ||||
| -rw-r--r-- | research/experiments/019-idle-links/transport.c | 21 | ||||
| -rw-r--r-- | serve/serve.c | 35 |
7 files changed, 137 insertions, 105 deletions
@@ -38,6 +38,11 @@ $(BUILD)/obj/%.o: %.c $(HEADERS) @mkdir -p $(dir $@) $(CC) $(CPPFLAGS) $(CFLAGS) $(INCLUDES) -c $< -o $@ +# The links know nothing of the core: they build without its headers. +$(BUILD)/obj/link/%.o: link/%.c $(wildcard link/*.h) + @mkdir -p $(dir $@) + $(CC) $(CPPFLAGS) $(CFLAGS) -c $< -o $@ + $(BUILD)/libpure.a: $(OBJECTS) rm -f $@ $(AR) rcs $@ $^ diff --git a/core/module.c b/core/module.c index cb4fe1a..8ac3b48 100644 --- a/core/module.c +++ b/core/module.c @@ -1094,6 +1094,25 @@ static int find(const char (*names)[PURE_MODULE_NAME], unsigned count, const cha int pure_module_state(const PureModule *m, const char *name) { return find(m->state_names, m->states, name); } int pure_module_input(const PureModule *m, const char *name) { return find(m->input_names, m->inputs, name); } +static uint32_t fnv(uint32_t h, const void *bytes, size_t size) { + const unsigned char *p = bytes; + for (size_t i = 0; i < size; ++i) h = (h ^ p[i]) * 16777619u; + return h; +} + +uint32_t pure_module_shape(const PureModule *m, uint32_t input) { + const PureInputShape *s = &m->input[input]; + uint32_t h = fnv(2166136261u, &s->prefix, sizeof(s->prefix)); + for (uint32_t f = 0; f < s->field_count; ++f) { + const PureField *field = &m->field[s->field_start + f]; + unsigned char fixed[7] = {field->type, (unsigned char)field->rows, (unsigned char)(field->rows >> 8), + (unsigned char)field->offset, (unsigned char)(field->offset >> 8), + (unsigned char)(field->offset >> 16), (unsigned char)(field->offset >> 24)}; + h = fnv(fnv(h, field->name, strlen(field->name) + 1), fixed, sizeof(fixed)); + } + return h; +} + int pure_module_view(const PureModule *m, const char *name) { for (uint32_t i = 0; i < m->views; ++i) if (!m->view_table[i] && !strcmp(m->view_names[i], name)) return (int)i; diff --git a/core/module.h b/core/module.h index 3431c87..0b45ba2 100644 --- a/core/module.h +++ b/core/module.h @@ -211,6 +211,10 @@ int pure_module_state(const PureModule *, const char *name); int pure_module_input(const PureModule *, const char *name); int pure_module_view(const PureModule *, const char *name); int pure_module_column(const PureModule *, uint32_t table, const char *name); +/* What a message of an input looks like, as 32 bits: its fields' names, + * types and places (FNV-1a). A sender's output and the input it sends to + * across nodes must agree on it (link/link.h). */ +uint32_t pure_module_shape(const PureModule *, uint32_t input); /* The index state a number view shows as its position (a view that is just * `at.position`), or -1: what a host moves with pure_app_point. */ int pure_module_position(const PureModule *, uint32_t view); diff --git a/link/link.c b/link/link.c index 9c6794b..103dcf5 100644 --- a/link/link.c +++ b/link/link.c @@ -89,19 +89,6 @@ static uint32_t fnv(uint32_t h, const void *bytes, size_t size) { static uint32_t key_id(const unsigned char key[PURE_SEAL_KEY]) { return fnv(2166136261u, key, PURE_SEAL_KEY); } -/* What a message of input looks like: its fields' names, types and places. */ -static uint32_t shape_of(const PureModule *m, uint32_t input) { - const PureInputShape *s = &m->input[input]; - uint32_t h = fnv(2166136261u, &s->prefix, sizeof(s->prefix)); - for (uint32_t f = 0; f < s->field_count; ++f) { - const PureField *field = &m->field[s->field_start + f]; - unsigned char fixed[7] = {field->type, (unsigned char)field->rows, (unsigned char)(field->rows >> 8)}; - put32(fixed + 3, field->offset); - h = fnv(fnv(h, field->name, strlen(field->name) + 1), fixed, sizeof(fixed)); - } - return h; -} - static uint64_t random64(void) { uint64_t value = 0; while (getrandom(&value, sizeof(value), 0) != sizeof(value) || !value) {} @@ -164,10 +151,11 @@ static size_t put_names(unsigned char *p, const char *source, const char *input) } /* A source and an input's names from p (size bytes); how many it takes, or 0. */ -static size_t get_names(const unsigned char *p, size_t size, char source[PURE_SOURCE + 1], char input[PURE_MODULE_NAME]) { - if (size < 1 || !p[0] || p[0] > PURE_SOURCE || size < 2u + p[0]) return 0; +static size_t get_names(const unsigned char *p, size_t size, char source[PURE_LINK_NAME + 1], + char input[PURE_LINK_NAME + 1]) { + if (size < 1 || !p[0] || p[0] > PURE_LINK_NAME || size < 2u + p[0]) return 0; size_t a = p[0], b = p[1 + a]; - if (!b || b >= PURE_MODULE_NAME || size < 2 + a + b) return 0; + if (!b || b > PURE_LINK_NAME || size < 2 + a + b) return 0; memcpy(source, p + 1, a); source[a] = 0; memcpy(input, p + 2 + a, b); @@ -249,9 +237,8 @@ typedef struct { } Flight; struct PureLink { - const PureModule *module; - uint32_t output, shape, own_id, peer_id; - char host[256], port[8], input[PURE_MODULE_NAME], source[PURE_SOURCE + 1]; + uint32_t shape, own_id, peer_id; + char host[256], port[8], input[PURE_LINK_NAME + 1], source[PURE_LINK_NAME + 1]; const PureSealKey *own; unsigned char own_public[PURE_SEAL_KEY], peer[PURE_SEAL_KEY], secret[32]; bool peered; @@ -293,23 +280,17 @@ struct PureLink { PureLinkCounts counts; }; -PureLink *pure_link_open(const PureModule *m, const char *output, const char *url, const char *source, - const PureSealKey *own, const unsigned char own_public[PURE_SEAL_KEY], char *error, - size_t size) { - int index = pure_module_input(m, output); - if (index < 0 || m->input[index].kind != PURE_KIND_OUTPUT) { - snprintf(error, size, "no output '%s'", output); - return NULL; - } +PureLink *pure_link_open(const char *url, const char *source, uint32_t shape, const PureSealKey *own, + const unsigned char own_public[PURE_SEAL_KEY], char *error, size_t size) { if (!own) { snprintf(error, size, "a pure:// link needs the node's --key"); return NULL; } size_t length = strlen(source); const char *host = url + 7, *slash = strchr(host, '/'), *colon = slash ? memchr(host, ':', (size_t)(slash - host)) : NULL; - if (!length || length > PURE_SOURCE || strncmp(url, "pure://", 7) || !slash || !colon || colon == host || + if (!length || length > PURE_LINK_NAME || strncmp(url, "pure://", 7) || !slash || !colon || colon == host || (size_t)(colon - host) >= 256 || slash - colon - 1 < 1 || slash - colon - 1 > 5 || !slash[1] || - strlen(slash + 1) >= PURE_MODULE_NAME) { + strlen(slash + 1) > PURE_LINK_NAME) { snprintf(error, size, "a link is pure://host:port/INPUT"); return NULL; } @@ -319,9 +300,7 @@ PureLink *pure_link_open(const PureModule *m, const char *output, const char *ur snprintf(error, size, "out of memory"); return NULL; } - l->module = m; - l->output = (uint32_t)index; - l->shape = shape_of(m, (uint32_t)index); + l->shape = shape; memcpy(l->host, host, (size_t)(colon - host)); memcpy(l->port, colon + 1, (size_t)(slash - colon - 1)); strcpy(l->input, slash + 1); @@ -339,8 +318,6 @@ void pure_link_peer(PureLink *l, const unsigned char peer[PURE_SEAL_KEY]) { l->peered = true; } -uint32_t pure_link_output(const PureLink *l) { return l->output; } - static Record *record_at(PureLink *l, size_t k) { return &l->records[(l->head + k) % l->capacity]; } /* The queued record numbered seq, or NULL (stored already, or not queued). */ @@ -397,17 +374,17 @@ static bool ring_grow(PureLink *l, size_t size) { return true; } -bool pure_link_queue(PureLink *l, uint64_t seq, PureSpan payload) { - Record r = {.seq = seq, .size = (uint32_t)payload.size}; - r.pieces = payload.size ? (uint32_t)((payload.size + PIECE - 1) / PIECE) : 1; - if (payload.size > RECORD_MOST || (r.pieces > 1 && !(r.more = calloc(r.pieces, sizeof(Piece))))) return false; +bool pure_link_queue(PureLink *l, uint64_t seq, const unsigned char *bytes, size_t size) { + Record r = {.seq = seq, .size = (uint32_t)size}; + r.pieces = size ? (uint32_t)((size + PIECE - 1) / PIECE) : 1; + if (size > RECORD_MOST || (r.pieces > 1 && !(r.more = calloc(r.pieces, sizeof(Piece))))) return false; pthread_mutex_lock(&l->lock); bool ok = seq > l->queued && (!l->count || seq == l->queued + 1); - size_t at = ok ? ring_place(l, payload.size) : SIZE_MAX; - if (ok && at == SIZE_MAX) ok = ring_grow(l, payload.size) && (at = ring_place(l, payload.size)) != SIZE_MAX; + size_t at = ok ? ring_place(l, size) : SIZE_MAX; + if (ok && at == SIZE_MAX) ok = ring_grow(l, size) && (at = ring_place(l, size)) != SIZE_MAX; if (ok) { r.at = at; - if (payload.size) memcpy(l->ring + at, payload.data, payload.size); + if (size) memcpy(l->ring + at, bytes, size); } if (ok && l->count == l->capacity) { size_t capacity = l->capacity ? 2 * l->capacity : 64; @@ -568,7 +545,7 @@ static void on_ack(PureLink *l, const unsigned char *p, size_t size, uint64_t no } static void on_stored(PureLink *l, const unsigned char *p, size_t size, uint64_t now) { - char source[PURE_SOURCE + 1], input[PURE_MODULE_NAME]; + char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1]; size_t names = get_names(p, size, source, input); if (!names || size < names + 13 || strcmp(source, l->source) || strcmp(input, l->input)) return; uint64_t heard = get64(p + names); @@ -624,7 +601,7 @@ static void take_answer(PureLink *l, unsigned char *packet, ssize_t n, uint64_t on_ack(l, plain + at, size - at, now); at += 13 + 16u * plain[at + 12]; } else if (type == F_STORED) { - char source[PURE_SOURCE + 1], input[PURE_MODULE_NAME]; + char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1]; size_t names = get_names(plain + at, size - at, source, input); if (!names || at + names + 13 > size) break; on_stored(l, plain + at, size - at, now); @@ -988,8 +965,8 @@ typedef struct { } Held; typedef struct Stream { Active *active; - char source[PURE_SOURCE + 1]; - uint32_t input; + char source[PURE_LINK_NAME + 1], input_name[PURE_LINK_NAME + 1]; + uint32_t input, shape; /* the host's number for the input; the messages' shape */ uint64_t heard; /* the last number heard (stored), once known */ bool known, out, stored_due, gap, ready; /* The line: once the last heard is known, whole records from the one @@ -1030,7 +1007,7 @@ static const uint64_t IDLE_FIRST = 30000 * MS, SWEEP_EVERY = 250 * MS, TRIM_WAIT enum { TRIM_AFTER = 1024 }; struct PureLinkListener { - const PureModule *module; + PureLinkHost host; const PureSealKey *own; unsigned char own_public[PURE_SEAL_KEY]; uint32_t own_id; @@ -1062,31 +1039,25 @@ struct PureLinkListener { size_t ready_count, ready_capacity; Run **due; /* runs to answer */ size_t due_count, due_capacity; - bool (*deliver)(void *, const PureLinkDelivery *); - void *context; unsigned char *batch; size_t batch_capacity; PureLinkCounts counts; }; -PureLinkListener *pure_link_listen(const PureModule *m, unsigned port, const PureSealKey *own, - const unsigned char own_public[PURE_SEAL_KEY], - bool (*deliver)(void *, const PureLinkDelivery *), void *context, char *error, - size_t size) { +PureLinkListener *pure_link_listen(unsigned port, const PureSealKey *own, const unsigned char own_public[PURE_SEAL_KEY], + const PureLinkHost *host, char *error, size_t size) { PureLinkListener *s = own ? calloc(1, sizeof(*s)) : NULL; if (!s || pthread_mutex_init(&s->lock, NULL)) { free(s); snprintf(error, size, own ? "out of memory" : "--link-port needs the node's --key"); return NULL; } - s->module = m; + s->host = *host; s->own = own; memcpy(s->own_public, own_public, PURE_SEAL_KEY); s->own_id = key_id(own_public); s->port = port; s->idle = IDLE_FIRST; - s->deliver = deliver; - s->context = context; s->fd = s->wake = -1; return s; } @@ -1128,8 +1099,8 @@ bool pure_link_allow(PureLinkListener *s, const char *source, const unsigned cha size_t size) { size_t length = strlen(source), p = SIZE_MAX; uint32_t id = key_id(key); - if (!length || length > PURE_SOURCE) { - snprintf(error, size, "a source's name is 1 to %d bytes", PURE_SOURCE); + if (!length || length > PURE_LINK_NAME) { + snprintf(error, size, "a source's name is 1 to %d bytes", PURE_LINK_NAME); return false; } for (size_t slot = s->index_size ? id & (s->index_size - 1) : 0; s->index_size && s->index[slot]; @@ -1431,18 +1402,15 @@ static bool allowed(const PureLinkListener *s, const Peer *peer, const char *sou } /* The stream of a peer's source to input, made the first time: NULL when the - * source is not allowed from the peer, or input takes no such messages. */ + * source is not allowed from the peer, or the host has no input that takes + * such messages. */ static Stream *stream_of(PureLinkListener *s, Active *a, const char *source, const char *input, uint32_t shape) { - int index = pure_module_input(s->module, input); - if (index < 0) return NULL; for (unsigned k = 0; k < a->stream_count; ++k) { Stream *st = a->streams[k]; - if (st->input == (uint32_t)index && !strcmp(st->source, source)) return st; + if (!strcmp(st->input_name, input) && !strcmp(st->source, source)) return st->shape == shape ? st : NULL; } - const PureInputShape *in = &s->module->input[index]; - if (!allowed(s, &s->peers[a->peer], source) || in->kind != PURE_KIND_MESSAGE || in->keyed || - shape_of(s->module, (uint32_t)index) != shape) - return NULL; + int index = allowed(s, &s->peers[a->peer], source) ? s->host.input(s->host.context, input, shape) : -1; + if (index < 0) return NULL; size_t capacity = a->stream_capacity; Stream **streams = room(a->streams, &capacity, a->stream_count, sizeof(*streams), 2), *st; if (!streams) return NULL; @@ -1451,7 +1419,9 @@ static Stream *stream_of(PureLinkListener *s, Active *a, const char *source, con if (!(st = calloc(1, sizeof(*st)))) return NULL; st->active = a; st->input = (uint32_t)index; + st->shape = shape; snprintf(st->source, sizeof(st->source), "%s", source); + snprintf(st->input_name, sizeof(st->input_name), "%s", input); a->streams[a->stream_count++] = st; return st; } @@ -1524,7 +1494,7 @@ static bool hand_over(PureLinkListener *s, Stream *st) { size_t bytes = 0; for (; n < st->line_count && n < BATCH; ++n) bytes += 4 + (size_t)get32(st->line + st->line_head + bytes); PureLinkDelivery d = {st->input, st->source, st->line_first, n, st->line + st->line_head, bytes, st}; - if (!s->deliver(s->context, &d)) return false; + if (!s->host.deliver(s->host.context, &d)) return false; st->out = true; ++s->counts.deliveries; return true; @@ -1548,7 +1518,7 @@ static bool hand_over(PureLinkListener *s, Stream *st) { at += 4 + held_at(st, k)->size; } PureLinkDelivery d = {st->input, st->source, held_at(st, 0)->seq, (unsigned)n, s->batch, bytes, st}; - if (!s->deliver(s->context, &d)) return false; + if (!s->host.deliver(s->host.context, &d)) return false; st->out = true; ++s->counts.deliveries; return true; @@ -1594,7 +1564,7 @@ static bool take_from(PureLinkListener *s, Peer *peer, unsigned char *packet, si for (size_t at = 0; at < size;) { unsigned type = plain[at++]; if (type == F_STREAM) { - char source[PURE_SOURCE + 1], input[PURE_MODULE_NAME]; + char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1]; size_t names = get_names(plain + at, size - at, source, input); if (!names || at + names + 4 > size) break; st = stream_of(s, a, source, input, get32(plain + at + names)); @@ -1687,9 +1657,9 @@ static void answer(PureLinkListener *s, uint64_t now) { for (unsigned j = 0; j < a->stream_count; ++j) { Stream *st = a->streams[j]; if (!(st->stored_due || r->ack_due) || st->run != r || st->run_id != r->run || !st->known) continue; - if (at + 2 + PURE_SOURCE + PURE_MODULE_NAME + 14 > BODY) break; + if (at + strlen(st->source) + strlen(st->input_name) + 16 > BODY) break; /* type, names, 13 */ body[at++] = F_STORED; - at += put_names(body + at, st->source, s->module->input_names[st->input]); + at += put_names(body + at, st->source, st->input_name); put64(body + at, st->heard); body[at + 8] = st->gap ? STORED_GAP : 0; size_t room = st->held_bytes + (st->line_size - st->line_head) < HELD_MOST ? CREDIT : 1; diff --git a/link/link.h b/link/link.h index 83cc645..1ee9473 100644 --- a/link/link.h +++ b/link/link.h @@ -1,6 +1,5 @@ #ifndef PURE_LINK_H #define PURE_LINK_H -#include "module.h" #include "seal.h" /* Links without connections (docs/design/011-pure-transport.md): one node's @@ -47,19 +46,21 @@ * Losses are detected as RFC 9002 says; congestion control is its NewReno * (RFC 9002 ยง7). */ +/* A source's or an input's name is 1 to PURE_LINK_NAME bytes. */ +enum { PURE_LINK_NAME = 64 }; + /* The sending side: one output's saved sends to url, pure://host:port/INPUT, - * as source (a Pure receiver's name for it), from own (this node's key, - * whose public half is own_public) to the receiver's public key, given by - * pure_link_peer before pure_link_start. The owner queues and trims as it - * does a wire's (serve/deliver.h): a send counts as delivered once the - * receiver says it saved it. */ + * as source (a Pure receiver's name for it), their messages of shape (the + * receiver's input must have the same: core/module.h's pure_module_shape), + * from own (this node's key, whose public half is own_public) to the + * receiver's public key, given by pure_link_peer before pure_link_start. + * The owner queues and trims as it does a wire's (serve/deliver.h): a send + * counts as delivered once the receiver says it saved it. */ typedef struct PureLink PureLink; -PureLink *pure_link_open(const PureModule *m, const char *output, const char *url, const char *source, - const PureSealKey *own, const unsigned char own_public[PURE_SEAL_KEY], char *error, - size_t size); +PureLink *pure_link_open(const char *url, const char *source, uint32_t shape, const PureSealKey *own, + const unsigned char own_public[PURE_SEAL_KEY], char *error, size_t size); void pure_link_peer(PureLink *, const unsigned char peer[PURE_SEAL_KEY]); -uint32_t pure_link_output(const PureLink *); -bool pure_link_queue(PureLink *, uint64_t seq, PureSpan payload); +bool pure_link_queue(PureLink *, uint64_t seq, const unsigned char *bytes, size_t size); uint64_t pure_link_queued(PureLink *); /* The last number the receiver saved; any thread. */ uint64_t pure_link_delivered(PureLink *); @@ -69,12 +70,14 @@ void pure_link_close(PureLink *); /* The receiving side: packets on a UDP port, from the senders allowed (a * source name and the key it seals with, a few bytes each however many - * there are: see pure_link_idle). Whole records of a stream, in - * order, go to deliver (on the listener's thread) as one delivery; the host - * applies them from the one after the source's last heard number, stops at - * a gap, saves, and then calls pure_link_done (any thread) with the last - * number heard. Records wait here until then, and only the saved ones are - * acknowledged as stored. */ + * there are: see pure_link_idle). A stream (a source to an input) goes to + * the host's input its name and shape find (its number, or -1: the stream + * is refused). Whole records of a stream, in order, go to deliver (both on + * the listener's thread) as one delivery; the host applies them from the + * one after the source's last heard number, stops at a gap, saves, and then + * calls pure_link_done (any thread) with the last number heard. Records + * wait here until then, and only the saved ones are acknowledged as + * stored. */ typedef struct { uint32_t input; const char *source; @@ -84,11 +87,14 @@ typedef struct { size_t size; void *token; /* for pure_link_done */ } PureLinkDelivery; +typedef struct { + int (*input)(void *context, const char *name, uint32_t shape); + bool (*deliver)(void *context, const PureLinkDelivery *); + void *context; +} PureLinkHost; typedef struct PureLinkListener PureLinkListener; -PureLinkListener *pure_link_listen(const PureModule *m, unsigned port, const PureSealKey *own, - const unsigned char own_public[PURE_SEAL_KEY], - bool (*deliver)(void *, const PureLinkDelivery *), void *context, char *error, - size_t size); +PureLinkListener *pure_link_listen(unsigned port, const PureSealKey *own, const unsigned char own_public[PURE_SEAL_KEY], + const PureLinkHost *host, char *error, size_t size); bool pure_link_allow(PureLinkListener *, const char *source, const unsigned char key[PURE_SEAL_KEY], char *error, size_t size); /* A sender quiet for ms (30 s at first) with nothing of it waiting lets its diff --git a/research/experiments/019-idle-links/transport.c b/research/experiments/019-idle-links/transport.c index 9972cd3..c4c9de1 100644 --- a/research/experiments/019-idle-links/transport.c +++ b/research/experiments/019-idle-links/transport.c @@ -50,6 +50,13 @@ static bool deliver(void *context, const PureLinkDelivery *d) { static void notify(void *context) { (void)context; } +/* The receiver's input a stream goes to: "order", of the sender's shape. */ +static int input_of(void *context, const char *name, uint32_t shape) { + (void)context; + int index = pure_module_input(&receiver, name); + return index >= 0 && pure_module_shape(&receiver, (uint32_t)index) == shape ? index : -1; +} + static uint64_t now_ns(void) { struct timespec t; clock_gettime(CLOCK_MONOTONIC, &t); @@ -131,23 +138,25 @@ int main(int argc, char **argv) { return 1; unsigned port = ntohs(any.sin_port); close(probe); - if (!(listener = pure_link_listen(&receiver, port, b, public_b, deliver, NULL, error, sizeof(error))) || + if (!(listener = pure_link_listen(port, b, public_b, &(PureLinkHost){input_of, deliver, NULL}, error, sizeof(error))) || !pure_link_allow(listener, "bench.ordered", public_a, error, sizeof(error)) || !pure_link_serve(listener, error, sizeof(error))) return fprintf(stderr, "transport: %s\n", error), 1; char url[128]; snprintf(url, sizeof(url), "pure://127.0.0.1:%u/order", port); - PureLink *link = pure_link_open(&sender, "ordered", url, "bench.ordered", a, public_a, error, sizeof(error)); - if (!link) return fprintf(stderr, "transport: %s\n", error), 1; - pure_link_peer(link, public_b); int output = pure_module_input(&sender, "ordered"); + PureLink *link = output < 0 ? NULL + : pure_link_open(url, "bench.ordered", pure_module_shape(&sender, (uint32_t)output), a, + public_a, error, sizeof(error)); + if (!link) return fprintf(stderr, "transport: %s\n", output < 0 ? "no output 'ordered'" : error), 1; + pure_link_peer(link, public_b); payload = malloc(size + 256); payload_size = order(&sender, (uint32_t)output, size, payload); for (size_t i = 0; i < payload_size; ++i) payload[i] ^= (unsigned char)(i * 7); /* bytes that differ by place */ uint64_t first = 1; if (!window) /* a backlog: all queued before the link starts */ for (; first <= count; ++first) - if (!pure_link_queue(link, first, (PureSpan){payload, payload_size})) return fputs("transport: cannot queue\n", stderr), 1; + if (!pure_link_queue(link, first, payload, payload_size)) return fputs("transport: cannot queue\n", stderr), 1; uint64_t began = now_ns(), cpu_link = 0, cpu_listen = thread_cpu("pure-listen"), cpu_all = thread_cpu(NULL); if (!pure_link_start(link, notify, NULL, error, sizeof(error))) return fprintf(stderr, "transport: %s\n", error), 1; for (uint64_t seq = first; seq <= count; ++seq) { @@ -155,7 +164,7 @@ int main(int argc, char **argv) { struct timespec pause = {0, 50000}; nanosleep(&pause, NULL); } - if (!pure_link_queue(link, seq, (PureSpan){payload, payload_size})) return fputs("transport: cannot queue\n", stderr), 1; + if (!pure_link_queue(link, seq, payload, payload_size)) return fputs("transport: cannot queue\n", stderr), 1; } while (pure_link_delivered(link) < count) { struct timespec pause = {0, 50000}; diff --git a/serve/serve.c b/serve/serve.c index d89fb13..5688d79 100644 --- a/serve/serve.c +++ b/serve/serve.c @@ -197,9 +197,11 @@ typedef struct { unsigned wire_count; uint64_t trimmed[PURE_MODULE_INPUTS], answers; bool held[PURE_MODULE_INPUTS]; /* an answer failed to apply: the wire's sends stay until a restart */ - /* The owner's: outputs linked to Pure receivers (pure://), and how far each was trimmed. */ + /* The owner's: outputs linked to Pure receivers (pure://), which output + * each is, and how far each was trimmed. */ PureLink *links[PURE_MODULE_INPUTS]; unsigned link_count; + uint32_t link_output[PURE_MODULE_INPUTS]; uint64_t link_trimmed[PURE_MODULE_INPUTS]; struct Writer *writer; /* the thread writing the app's checkpoints, once one was due */ uint64_t checkpoints; /* images it wrote */ @@ -1501,7 +1503,7 @@ typedef struct { Executor *executor; Fresh *fresh; size_t count, capacity; uint6 /* An output's outlet: its wires first, then its links. */ static uint32_t outlet_output(Executor *ex, unsigned i) { - return i < ex->wire_count ? pure_wire_output(ex->wires[i]) : pure_link_output(ex->links[i - ex->wire_count]); + return i < ex->wire_count ? pure_wire_output(ex->wires[i]) : ex->link_output[i - ex->wire_count]; } static uint64_t outlet_queued(Executor *ex, unsigned i) { return i < ex->wire_count ? pure_wire_queued(ex->wires[i]) : pure_link_queued(ex->links[i - ex->wire_count]); @@ -1541,7 +1543,8 @@ static void after_save(Executor *ex) { const Fresh *f = &q.fresh[k]; if (blocked[f->wire]) continue; blocked[f->wire] = f->wire < ex->wire_count ? !pure_wire_queue(ex->wires[f->wire], f->seq, f->payload) - : !pure_link_queue(ex->links[f->wire - ex->wire_count], f->seq, f->payload); + : !pure_link_queue(ex->links[f->wire - ex->wire_count], f->seq, + f->payload.data, f->payload.size); } free(q.fresh); } @@ -1574,7 +1577,7 @@ static bool take_deliveries(Executor *ex) { for (unsigned i = 0; i < ex->link_count; ++i) { uint64_t stored = pure_link_delivered(ex->links[i]); if (stored <= ex->link_trimmed[i]) continue; - pure_app_trim(&ex->app, pure_link_output(ex->links[i]), stored); + pure_app_trim(&ex->app, ex->link_output[i], stored); ex->link_trimmed[i] = stored; } return answered; @@ -2331,6 +2334,17 @@ static void stop_shard(Host *host) { free(host->views); } +/* The input a link's stream goes to: one taking messages (not keyed) of the + * sender's shape. */ +static int link_input(void *context, const char *name, uint32_t shape) { + const PureModule *m = ((Owner *)context)->executor.app.module; + int index = pure_module_input(m, name); + if (index < 0 || m->input[index].kind != PURE_KIND_MESSAGE || m->input[index].keyed || + pure_module_shape(m, (uint32_t)index) != shape) + return -1; + return index; +} + /* Records a link delivered, queued at the owner as a request is: applied * with the next burst and done (pure_link_done) once its save is over. */ static PureLinkListener *link_listener; @@ -2393,13 +2407,17 @@ static bool open_wires(Executor *ex, const PureModule *m, const char *const *sen output[length] = 0; int written = snprintf(source, sizeof(source), "%s.%s", name, output); if (!strncmp(equals + 1, "pure://", 7)) { /* a link: no connection, sealed with the node's key */ - PureLink *link = written > 0 && written <= PURE_SOURCE - ? pure_link_open(m, output, equals + 1, source, seal_key, own_public, error, sizeof(error)) + int index = pure_module_input(m, output); + if (index < 0 || m->input[index].kind != PURE_KIND_OUTPUT) snprintf(error, sizeof(error), "no output '%s'", output); + PureLink *link = index >= 0 && m->input[index].kind == PURE_KIND_OUTPUT && written > 0 && written <= PURE_SOURCE + ? pure_link_open(equals + 1, source, pure_module_shape(m, (uint32_t)index), seal_key, + own_public, error, sizeof(error)) : NULL; if (!link) { fprintf(stderr, "--send %s: %s\n", sends[i], written > PURE_SOURCE ? "the name and output are longer than 64 bytes" : error); return false; } + ex->link_output[ex->link_count] = (uint32_t)index; ex->links[ex->link_count++] = link; continue; } @@ -2440,7 +2458,7 @@ static bool wire_seals(Executor *ex, const PureModule *m, const char *const *sea for (unsigned k = 0; k < ex->wire_count; ++k) if (!strcmp(m->input_names[pure_wire_output(ex->wires[k])], output)) wire = ex->wires[k]; for (unsigned k = 0; k < ex->link_count; ++k) - if (!strcmp(m->input_names[pure_link_output(ex->links[k])], output)) link = ex->links[k]; + if (!strcmp(m->input_names[ex->link_output[k]], output)) link = ex->links[k]; } if (link && pure_seal_peer(equals + 1, peer)) { /* the link's receiver */ pure_link_peer(link, peer); @@ -2657,7 +2675,8 @@ int main(int argc, char **argv) { fputs("--link-port needs an app with states and the node's --key\n", stderr); return 2; } - link_listener = pure_link_listen(&module, link_port, seal_key, own_public, link_deliver, &owner, error, sizeof(error)); + link_listener = pure_link_listen(link_port, seal_key, own_public, &(PureLinkHost){link_input, link_deliver, &owner}, + error, sizeof(error)); bool allowed = link_listener != NULL && allow_senders(link_listener, senders_path, error, sizeof(error)); for (unsigned i = 0; allowed && i < sender_count; ++i) allowed = pure_link_allow(link_listener, senders[i].name, senders[i].key, error, sizeof(error)); |