Establish native micro runtime checkpoint

1 files changed, 330 insertions, 0 deletions

diff --git a/prototypes/connections/test-async.c b/prototypes/connections/test-async.c
new file mode 100644
index 0000000..6efaf0b
--- /dev/null
+++ b/prototypes/connections/test-async.c
@@ -0,0 +1,330 @@
+#define _POSIX_C_SOURCE 200809L
+#include "runtime.h"
+#include "store-port.h"
+#include "store.h"
+#include <dirent.h>
+#include <poll.h>
+#include <pthread.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+#define CHECK(test) do { if (!(test)) { fprintf(stderr, "line %d: %s\n", __LINE__, #test); exit(1); } } while (0)
+
+static const PureProgram identity = {
+    .read = {.to_end = true}, .write = {.to_end = true},
+    .reply = {.to_end = true}, .reply_input = true
+};
+
+typedef struct {
+    PureEffect effect;
+    const PureProgram *definition;
+    const char *directory;
+    int start, done;
+    PureResult status;
+    PureSpan reply;
+    int64_t revision;
+} Worker;
+
+static void *write_after_signal(void *context) {
+    Worker *worker = context;
+    char byte = 0;
+    if (read(worker->start, &byte, 1) != 1) worker->status = PURE_STORE_ERROR;
+    else {
+        PureHost store = pure_store_host(worker->directory);
+        worker->status = store.write(store.context, worker->definition,
+                                     worker->effect.key, worker->effect.expected,
+                                     worker->effect.value, &worker->reply,
+                                     &worker->revision);
+    }
+    (void)write(worker->done, &byte, 1);
+    return NULL;
+}
+
+static PureHost async_host(const char *directory) {
+    PureHost host = pure_store_host(directory);
+    host.read = NULL;
+    host.write = NULL;
+    return host;
+}
+
+static PureTicket submit(PureRuntime *runtime, const char *key,
+                         const void *body, size_t size) {
+    PureTicket ticket;
+    CHECK(pure_runtime_accept(runtime, PURE_INPUT_WRITE, key, 0, size, &ticket) == PURE_OK);
+    if (size) memcpy(pure_runtime_ingress(ticket, NULL), body, size);
+    CHECK(pure_runtime_ready(ticket));
+    return ticket;
+}
+
+static void expect(const char *directory, const char *key, const void *body, size_t size) {
+    PureStoredView view;
+    CHECK(pure_store_get(directory, key, 64, &view) == PURE_OK);
+    CHECK(view.body.size == size && !memcmp(view.body.data, body, size));
+    pure_store_release(&view);
+}
+
+static void worker_and_forward(const char *directory) {
+    PureRuntime source, destination;
+    PureWork source_slots[2], destination_slot[1];
+    CHECK(pure_runtime_init(&source, source_slots, 2, &identity,
+                            async_host(directory), 64));
+    PureTicket ignored = {0};
+    CHECK(pure_runtime_accept(&source, PURE_INPUT_WRITE, "denied", 0, 1,
+                              &ignored) == PURE_DENIED);
+    CHECK(pure_runtime_enable_async(&source));
+    PureTicket slow = submit(&source, "slow", "alpha", 5);
+    PureTicket fast = submit(&source, "fast", "beta", 4);
+    CHECK(!pure_runtime_step(&source)); /* No synchronous host callback. */
+    PureEffect deferred;
+    CHECK(pure_runtime_take(&source, &deferred) == PURE_TAKE_EFFECT);
+    CHECK(deferred.ticket.generation == slow.generation &&
+          deferred.value.size == 5 && !memcmp(deferred.value.data, "alpha", 5));
+    CHECK(!pure_runtime_cancel(slow));
+    int start_pipe[2], done_pipe[2];
+    CHECK(!pipe(start_pipe) && !pipe(done_pipe));
+    Worker worker = {.effect = deferred, .definition = &identity, .directory = directory,
+                     .start = start_pipe[0], .done = done_pipe[1]};
+    pthread_t thread;
+    CHECK(!pthread_create(&thread, NULL, write_after_signal, &worker));
+    CHECK(pure_runtime_release(slow));
+    CHECK(source.active == 2 && source.retained == 9);
+    PureEffect immediate;
+    CHECK(pure_runtime_take(&source, &immediate) == PURE_TAKE_EFFECT);
+    int64_t revision = 0;
+    PureHost store = pure_store_host(directory);
+    PureSpan reply = {0};
+    PureResult written = store.write(store.context, &identity, immediate.key,
+                                     immediate.expected, immediate.value, &reply, &revision);
+    CHECK(pure_runtime_complete(immediate.ticket, written, NULL, reply, revision));
+    PureOutcome result;
+    CHECK(pure_runtime_result(fast, &result) && result.status == PURE_OK &&
+          result.body.size == 4 && !memcmp(result.body.data, "beta", 4));
+    CHECK(pure_runtime_release(fast));
+    expect(directory, "fast", "beta", 4);
+    PureStoredView absent;
+    CHECK(pure_store_get(directory, "slow", 64, &absent) == PURE_NOT_FOUND);
+    pure_runtime_stop(&source);
+    CHECK(!pure_runtime_restart(&source, &identity, async_host(directory), 64));
+    CHECK(pure_runtime_accept(&source, PURE_INPUT_READ, "fast", 0, 0,
+                              &ignored) == PURE_CLOSED);
+    CHECK(write(start_pipe[1], "x", 1) == 1);
+    struct pollfd done = {.fd = done_pipe[0], .events = POLLIN};
+    CHECK(poll(&done, 1, 5000) == 1 && (done.revents & POLLIN));
+    CHECK(!pthread_join(thread, NULL));
+    CHECK(pure_runtime_complete(deferred.ticket, worker.status, NULL,
+                                worker.reply, worker.revision));
+    CHECK(!pure_runtime_complete(deferred.ticket, PURE_OK, NULL, (PureSpan){0}, 1));
+    CHECK(!pure_runtime_result(slow, &result)); /* Caller released before I/O ended. */
+    CHECK(source.active == 0 && source.retained == 0 && source.commits == 2);
+    expect(directory, "slow", "alpha", 5);
+    CHECK(!close(start_pipe[0]) && !close(start_pipe[1]) &&
+          !close(done_pipe[0]) && !close(done_pipe[1]));
+    CHECK(pure_runtime_restart(&source, &identity, async_host(directory), 64));
+
+    CHECK(pure_runtime_init(&destination, destination_slot, 1, &identity,
+                            async_host(directory), 64));
+    CHECK(pure_runtime_enable_async(&destination));
+    PureTicket read_ticket;
+    CHECK(pure_runtime_accept(&source, PURE_INPUT_READ, "fast", 0, 0,
+                              &read_ticket) == PURE_OK);
+    PureEffect read_effect;
+    CHECK(pure_runtime_take(&source, &read_effect) == PURE_TAKE_EFFECT &&
+          read_effect.input == PURE_INPUT_READ);
+    PureResource mapping = {0};
+    revision = 0;
+    PureResult read_status = store.read(store.context, &identity, read_effect.key,
+                                        read_effect.available, &mapping, &revision);
+    CHECK(pure_runtime_complete(read_effect.ticket, read_status, &mapping,
+                                (PureSpan){0}, revision));
+    CHECK(!mapping.allocation);
+    CHECK(pure_runtime_result(read_ticket, &result) && result.status == PURE_OK &&
+          result.body.size == 4);
+    PureTicket forwarded;
+    CHECK(pure_runtime_forward(read_ticket, &destination, "copy", 0,
+                               &forwarded) == PURE_OK);
+    const unsigned char *borrowed = result.body.data;
+    CHECK(pure_runtime_release(read_ticket));
+    CHECK(source.active == 1 && source.retained == 4);
+    PureEffect output;
+    CHECK(pure_runtime_take(&destination, &output) == PURE_TAKE_EFFECT);
+    CHECK(output.value.data == borrowed && output.value.size == 4);
+    revision = 0;
+    reply = (PureSpan){0};
+    written = store.write(store.context, &identity, output.key, output.expected,
+                          output.value, &reply, &revision);
+    CHECK(pure_runtime_complete(output.ticket, written, NULL, reply, revision));
+    CHECK(pure_runtime_result(forwarded, &result) && result.status == PURE_OK &&
+          result.body.data == borrowed);
+    CHECK(pure_runtime_release(forwarded));
+    CHECK(source.active == 0 && source.retained == 0 &&
+          destination.active == 0 && destination.retained == 0);
+    expect(directory, "copy", "beta", 4);
+}
+
+static void rejection_and_budget(const char *directory) {
+    PureProgram invalid = identity;
+    invalid.input_required = 4;
+    PureRuntime runtime;
+    PureWork slots[2];
+    CHECK(pure_runtime_init(&runtime, slots, 2, &invalid,
+                            async_host(directory), 6));
+    CHECK(pure_runtime_enable_async(&runtime));
+    PureTicket rejected = submit(&runtime, "bad", "x", 1);
+    PureEffect effect;
+    CHECK(pure_runtime_take(&runtime, &effect) == PURE_TAKE_EFFECT);
+    PureHost store = pure_store_host(directory);
+    PureSpan reply = {0};
+    int64_t revision = 0;
+    PureResult status = store.write(store.context, &invalid, effect.key,
+                                    effect.expected, effect.value, &reply, &revision);
+    CHECK(status == PURE_LIMIT);
+    CHECK(pure_runtime_complete(effect.ticket, status, NULL, reply, revision));
+    PureOutcome result;
+    CHECK(pure_runtime_result(rejected, &result) && result.status == PURE_LIMIT);
+    CHECK(!pure_runtime_complete(rejected, PURE_OK, NULL, (PureSpan){0}, 1));
+    CHECK(pure_runtime_release(rejected));
+    CHECK(runtime.active == 0 && runtime.retained == 0);
+
+    CHECK(pure_runtime_restart(&runtime, &identity, async_host(directory), 6));
+    PureTicket cancelled = submit(&runtime, "unused", "x", 1);
+    CHECK(pure_runtime_cancel(cancelled));
+    CHECK(pure_runtime_result(cancelled, &result) && result.status == PURE_CANCELLED);
+    CHECK(pure_runtime_release(cancelled));
+    PureTicket invalid_read;
+    CHECK(pure_runtime_accept(&runtime, PURE_INPUT_READ, "fast", 0, 0,
+                              &invalid_read) == PURE_OK);
+    PureEffect invalid_effect;
+    CHECK(pure_runtime_take(&runtime, &invalid_effect) == PURE_TAKE_EFFECT);
+    PureResource foreign = {.body = {(const unsigned char *)"x", 1}};
+    CHECK(!pure_runtime_complete(invalid_effect.ticket, PURE_OK, &foreign,
+                                 (PureSpan){0}, 1));
+    CHECK(foreign.body.size == 1 && runtime.active == 1);
+    CHECK(pure_runtime_complete(invalid_effect.ticket, PURE_STORE_ERROR, NULL,
+                                (PureSpan){0}, 0));
+    CHECK(pure_runtime_release(invalid_read));
+    PureTicket bad_reply = submit(&runtime, "bad_reply", "x", 1);
+    PureEffect bad_effect;
+    CHECK(pure_runtime_take(&runtime, &bad_effect) == PURE_TAKE_EFFECT);
+    CHECK(!pure_runtime_complete(bad_effect.ticket, PURE_OK, NULL,
+                                 (PureSpan){(const unsigned char *)"outside", 7}, 1));
+    CHECK(runtime.active == 1 && runtime.retained == 1);
+    CHECK(pure_runtime_complete(bad_effect.ticket, PURE_STORE_ERROR, NULL,
+                                (PureSpan){0}, 0));
+    CHECK(pure_runtime_release(bad_reply) && runtime.active == 0 && runtime.retained == 0);
+    PureTicket first, second;
+    CHECK(pure_runtime_accept(&runtime, PURE_INPUT_READ, "fast", 0, 0, &first) == PURE_OK);
+    CHECK(pure_runtime_accept(&runtime, PURE_INPUT_READ, "fast", 0, 0, &second) == PURE_OK);
+    PureEffect one, two;
+    CHECK(pure_runtime_take(&runtime, &one) == PURE_TAKE_EFFECT);
+    CHECK(pure_runtime_take(&runtime, &two) == PURE_TAKE_EFFECT);
+    CHECK(pure_runtime_take(&runtime, &effect) == PURE_TAKE_EMPTY);
+    PureResource a = {0}, b = {0};
+    int64_t revision_a = 0, revision_b = 0;
+    CHECK(store.read(store.context, &identity, one.key, one.available,
+                     &a, &revision_a) == PURE_OK);
+    CHECK(store.read(store.context, &identity, two.key, two.available,
+                     &b, &revision_b) == PURE_OK);
+    CHECK(pure_runtime_complete(two.ticket, PURE_OK, &b, (PureSpan){0}, revision_b));
+    CHECK(!b.allocation);
+    CHECK(pure_runtime_complete(one.ticket, PURE_OK, &a, (PureSpan){0}, revision_a));
+    CHECK(!a.allocation);
+    CHECK(pure_runtime_result(first, &result) && result.status == PURE_BUDGET);
+    CHECK(pure_runtime_result(second, &result) && result.status == PURE_OK);
+    CHECK(runtime.budget_rejections == 1 && runtime.retained == 4);
+    CHECK(pure_runtime_release(first) && pure_runtime_release(second));
+    CHECK(runtime.active == 0 && runtime.retained == 0);
+}
+
+static void authored_document(const char *directory) {
+    PureProgram program;
+    char error[160];
+    CHECK(pure_program_load("examples/editor.pure", &program, error, sizeof(error)));
+    PureRuntime runtime;
+    PureWork slot[1];
+    CHECK(pure_runtime_init(&runtime, slot, 1, &program,
+                            async_host(directory), 64));
+    CHECK(pure_runtime_enable_async(&runtime));
+    const unsigned char create[] = {0,0,0,0, 0,0,0,0, 'H','i'};
+    PureTicket ticket = submit(&runtime, "document", create, sizeof(create));
+    PureEffect effect;
+    CHECK(pure_runtime_take(&runtime, &effect) == PURE_TAKE_EFFECT &&
+          effect.value.size == sizeof(create));
+    PureSpan value, reply = {0};
+    size_t offset, remove;
+    CHECK(pure_program_splice(&program, effect.value, &offset, &remove) &&
+          pure_program_input(&program, effect.value, &value, &reply) &&
+          offset == 0 && remove == 0 && value.size == 2 &&
+          !memcmp(value.data, "Hi", 2));
+    PureHost store = pure_store_host(directory);
+    int64_t revision = 0;
+    PureResult status = store.write(store.context, &program, effect.key,
+                                    effect.expected, effect.value, &reply, &revision);
+    CHECK(pure_runtime_complete(effect.ticket, status, NULL, reply, revision));
+    PureOutcome result;
+    CHECK(pure_runtime_result(ticket, &result) && result.status == PURE_OK &&
+          result.body.size == 2 && !memcmp(result.body.data, "Hi", 2));
+    CHECK(pure_runtime_release(ticket));
+
+    const unsigned char append[] = {2,0,0,0, 0,0,0,0, '!'};
+    CHECK(pure_runtime_accept(&runtime, PURE_INPUT_WRITE, "document", 1,
+                              sizeof(append), &ticket) == PURE_OK);
+    memcpy(pure_runtime_ingress(ticket, NULL), append, sizeof(append));
+    CHECK(pure_runtime_ready(ticket));
+    CHECK(pure_runtime_take(&runtime, &effect) == PURE_TAKE_EFFECT &&
+          effect.value.size == sizeof(append));
+    CHECK(pure_program_splice(&program, effect.value, &offset, &remove) &&
+          offset == 2 && remove == 0);
+    reply = (PureSpan){0};
+    status = store.write(store.context, &program, effect.key, effect.expected,
+                         effect.value, &reply, &revision);
+    CHECK(pure_runtime_complete(effect.ticket, status, NULL, reply, revision));
+    CHECK(pure_runtime_result(ticket, &result) && result.status == PURE_OK &&
+          result.body.size == 1 && result.body.data[0] == '!');
+    CHECK(pure_runtime_release(ticket));
+    expect(directory, "document", "Hi!", 3);
+
+    ticket = submit(&runtime, "document", append, sizeof(append));
+    CHECK(pure_runtime_take(&runtime, &effect) == PURE_TAKE_EFFECT);
+    reply = (PureSpan){0};
+    status = store.write(store.context, &program, effect.key, effect.expected,
+                         effect.value, &reply, &revision);
+    CHECK(pure_runtime_complete(effect.ticket, status, NULL, reply, revision));
+    CHECK(pure_runtime_result(ticket, &result) && result.status == PURE_CONFLICT &&
+          !result.body.size);
+    CHECK(pure_runtime_release(ticket));
+    expect(directory, "document", "Hi!", 3);
+
+    CHECK(pure_runtime_accept(&runtime, PURE_INPUT_READ, "document", 0, 0,
+                              &ticket) == PURE_OK);
+    CHECK(pure_runtime_take(&runtime, &effect) == PURE_TAKE_EFFECT);
+    PureResource mapping = {0};
+    status = store.read(store.context, &program, effect.key,
+                        effect.available, &mapping, &revision);
+    CHECK(pure_runtime_complete(effect.ticket, status, &mapping,
+                                (PureSpan){0}, revision));
+    CHECK(pure_runtime_result(ticket, &result) && result.status == PURE_OK &&
+          result.body.size == 3 && !memcmp(result.body.data, "Hi!", 3));
+    CHECK(pure_runtime_release(ticket));
+    CHECK(runtime.active == 0 && runtime.retained == 0 &&
+          runtime.commits == 2 && runtime.conflicts == 1 && runtime.gets == 1);
+}
+
+int main(void) {
+    char template[] = "/tmp/pure-async-effects-XXXXXX";
+    char *directory = mkdtemp(template);
+    CHECK(directory && pure_store_init(directory) == PURE_OK);
+    worker_and_forward(directory);
+    rejection_and_budget(directory);
+    authored_document(directory);
+    DIR *opened = opendir(directory);
+    CHECK(opened);
+    struct dirent *entry;
+    while ((entry = readdir(opened))) {
+        if (!strcmp(entry->d_name, ".") || !strcmp(entry->d_name, "..")) continue;
+        CHECK(unlinkat(dirfd(opened), entry->d_name, 0) == 0);
+    }
+    CHECK(!closedir(opened) && !rmdir(directory));
+    puts("{\"async_effects\":\"authored-document-worker-overlap-release-forward-budget-recovery\"}");
+}