Charge lent answers to the payload budget

A read answered with extents of the store mapping was charged nothing,
so clients that stopped reading could keep any number of old file
versions mapped. The runtime core now takes resources a host sends in
pieces: body.data is NULL, body.size is charged like any other resource,
the outcome has no body to read and cannot be forwarded. pure-serve hands
the mapping reference over as such a resource, so a lent answer holds its
bytes of budget until it is in the socket and its ticket releases the
mapping; past the budget a read gets 503 payload_budget, as a copied one
would. The async core grows 52 bytes (5,776 of its 6,144).

4 files changed, 67 insertions, 32 deletions

diff --git a/prototypes/connections/runtime.c b/prototypes/connections/runtime.c
index 0bfb7bf..4a857c7 100644
--- a/prototypes/connections/runtime.c
+++ b/prototypes/connections/runtime.c
@@ -231,7 +231,7 @@ bool pure_runtime_step(PureRuntime *r) {
             r->budget - r->retained, &w->resource, &outcome.revision);
         if (outcome.status == PURE_OK) {
             if (w->resource.body.size > r->budget - r->retained) outcome.status = PURE_BUDGET;
-            else if (w->resource.body.size && (!w->resource.body.data || !w->resource.release))
+            else if (w->resource.body.size && !w->resource.release)
                 outcome.status = PURE_STORE_ERROR;
             else {
                 w->charged = w->resource.body.size;
@@ -300,7 +300,7 @@ bool pure_runtime_complete(PureTicket ticket, PureResult status,
         if (outcome.status == PURE_OK) {
             if (acquired.body.size > r->budget - r->retained)
                 outcome.status = PURE_BUDGET;
-            else if (acquired.body.size && (!acquired.body.data || !acquired.release))
+            else if (acquired.body.size && !acquired.release)
                 outcome.status = PURE_STORE_ERROR;
             else {
                 w->resource = acquired;
@@ -343,7 +343,8 @@ PureResult pure_runtime_forward(PureTicket source, PureRuntime *r,
                                 const char *key, int64_t expected, PureTicket *out) {
     PureWork *parent = lookup(source, true);
     if (!out || !parent || source.runtime->running || parent->state != DONE ||
-        parent->outcome.status != PURE_OK) return PURE_INVALID;
+        parent->outcome.status != PURE_OK || (parent->outcome.body.size && !parent->outcome.body.data))
+        return PURE_INVALID;
     PureResult status = admission(r, PURE_INPUT_WRITE, key, expected, parent->outcome.body.size);
     if (status != PURE_OK) return status;
     if (parent->pins == SIZE_MAX) return PURE_LIMIT;
diff --git a/prototypes/connections/runtime.h b/prototypes/connections/runtime.h
index 0add777..70f5d08 100644
--- a/prototypes/connections/runtime.h
+++ b/prototypes/connections/runtime.h
@@ -20,7 +20,10 @@ typedef enum { PURE_INPUT_READ, PURE_INPUT_WRITE } PureInput;
 
 /* The host owns this resource until it successfully returns it to the core.
  * release must end all access and free/unmap allocation exactly once. It runs
- * while that domain is busy; calls back into the same domain are rejected. */
+ * while that domain is busy; calls back into the same domain are rejected.
+ * A resource the host sends in pieces (such as extents of a mapping) has no
+ * contiguous body: body.data is NULL and body.size is still what it holds and
+ * is charged. Its outcome has no body to read and cannot be forwarded. */
 typedef struct {
     PureSpan body;
     void *allocation;
diff --git a/prototypes/connections/serve.c b/prototypes/connections/serve.c
index a3e991f..65ca1fb 100644
--- a/prototypes/connections/serve.c
+++ b/prototypes/connections/serve.c
@@ -61,13 +61,13 @@ typedef struct Connection {
     PureTicket ticket;
     struct iovec *out, small[4]; /* the answer: its head, then its body's extents */
     size_t out_count, out_at;
-    PureMapping *mapping;        /* keeps store extents the answer borrows mapped */
     char head[320];
     char in[HEADER_LIMIT]; /* the header being read, then pipelined bytes */
 } Connection;
 
 /* One request applied in the current burst. A read answered from the store
- * lends its extents: extents point into mapping, which the answer keeps. */
+ * lends its extents: they point into a store mapping, which the resource holds
+ * and the runtime charges and releases with the ticket. */
 typedef struct {
     Connection *connection;
     PureTicket ticket;
@@ -77,8 +77,7 @@ typedef struct {
     PureResource resource;
     PureSpan reply;
     struct iovec *extents;
-    size_t extent_count, extent_size;
-    PureMapping *mapping;
+    size_t extent_count;
     char key[65];
 } Applied;
 
@@ -130,12 +129,10 @@ static void release_ticket(void *owner) {
     c->body = NULL;
 }
 
-/* Everything an answer borrowed goes back: the request's payload and ticket,
- * the store extents. */
+/* Everything an answer borrowed goes back with its ticket: the request's
+ * payload, a copied reply or the lent store extents. */
 static void finish_output(Connection *c) {
     release_ticket(c);
-    pure_mapping_release(c->mapping);
-    c->mapping = NULL;
     if (c->out != c->small) free(c->out);
     c->out = NULL;
     c->out_count = c->out_at = 0;
@@ -154,17 +151,16 @@ static void close_connection(Connection *c) {
     --host->active;
 }
 
-/* Answer with a body of `count` extents (`size` bytes in all). The answer
- * takes the mapping reference; one writev sends as much as the socket takes. */
+/* Answer with a body of `count` extents (`size` bytes in all); the ticket
+ * keeps them alive. One writev sends as much as the socket takes. */
 static void respond_extents(Connection *c, int status, const char *reason, const char *error, int64_t revision,
-                            const struct iovec *extents, size_t count, size_t size, PureMapping *mapping) {
+                            const struct iovec *extents, size_t count, size_t size) {
     char tag[48] = "";
     if (revision > 0) snprintf(tag, sizeof(tag), "ETag: \"%" PRId64 "\"\r\n", revision);
     int length = snprintf(c->head, sizeof(c->head),
         "HTTP/1.1 %d %s\r\nContent-Length: %zu\r\n%sContent-Type: application/octet-stream\r\n"
         "Cache-Control: no-store\r\nX-Pure-Error: %s\r\n%s\r\n",
         status, reason, size, c->keep_alive ? "" : "Connection: close\r\n", error, tag);
-    c->mapping = mapping;
     c->out = count < sizeof(c->small) / sizeof(*c->small) ? c->small : malloc((count + 1) * sizeof(*c->out));
     c->phase = WRITING;
     if (!c->out) { c->keep_alive = false; c->out = c->small; count = 0; } /* cannot send it: the close tells */
@@ -177,7 +173,7 @@ static void respond_extents(Connection *c, int status, const char *reason, const
 static void respond(Connection *c, int status, const char *reason, const char *error, int64_t revision,
                     PureSpan body) {
     struct iovec extent = {(void *)body.data, body.size};
-    respond_extents(c, status, reason, error, revision, &extent, body.size ? 1 : 0, body.size, NULL);
+    respond_extents(c, status, reason, error, revision, &extent, body.size ? 1 : 0, body.size);
 }
 
 /* A request the host cannot read further: answer and close. */
@@ -390,6 +386,11 @@ static void release_copy(void *allocation, size_t size) {
     free(allocation);
 }
 
+static void release_loan(void *allocation, size_t size) {
+    (void)size;
+    pure_mapping_release(allocation);
+}
+
 static bool allocate_ingress(void *context, size_t size, PureResource *out) {
     (void)context;
     void *bytes = malloc(size);
@@ -453,8 +454,8 @@ static void apply_effect(Host *host, const PureEffect *effect, Applied *out) {
         if (gather.extents && pure_app_reply(app, gather_extents, &gather) == PURE_OK && gather.held) {
             out->extents = gather.extents;
             out->extent_count = gather.count;
-            out->extent_size = reply.size;
-            out->mapping = mapping;
+            out->resource = (PureResource){.body = {NULL, reply.size}, .allocation = mapping,
+                                           .allocation_size = reply.size, .release = release_loan};
             return;
         }
         free(gather.extents);
@@ -478,20 +479,16 @@ static void finish_burst(Host *host, size_t count) {
         if (!pure_runtime_complete(a->ticket, status, a->input == PURE_INPUT_READ ? &a->resource : NULL,
                                    status == PURE_OK ? a->reply : (PureSpan){0}, a->revision)) abort();
         Connection *c = a->connection;
-        bool waiting = c && c->phase == WAITING && c->ticket.generation == a->ticket.generation;
-        if (waiting && status == PURE_OK && a->mapping) {
-            respond_extents(c, 200, "OK", "none", a->revision, a->extents, a->extent_count, a->extent_size, a->mapping);
-            a->mapping = NULL;
-        }
-        pure_mapping_release(a->mapping);
-        free(a->extents);
-        if (!waiting) continue;
-        if (c->phase == WAITING) {
+        if (c && c->phase == WAITING && c->ticket.generation == a->ticket.generation) {
             PureOutcome result;
             if (!pure_runtime_result(a->ticket, &result)) abort();
-            answer(c, result);
+            /* A lent answer's body is its extents (the runtime charged them). */
+            if (result.status == PURE_OK && a->extents)
+                respond_extents(c, 200, "OK", "none", result.revision, a->extents, a->extent_count, result.body.size);
+            else answer(c, result);
+            drive(c); /* most answers fit the socket buffer at once */
         }
-        drive(c); /* most answers fit the socket buffer at once */
+        free(a->extents);
     }
 }
 
@@ -581,7 +578,6 @@ static void accept_all(Host *host) {
         c->ticket = (PureTicket){0};
         c->out = NULL;
         c->out_count = c->out_at = 0;
-        c->mapping = NULL;
         c->body = NULL;
         c->in_used = 0;
         struct epoll_event event = {.events = EPOLLIN | EPOLLOUT | EPOLLRDHUP | EPOLLET, .data.ptr = c};
diff --git a/prototypes/connections/test-async.c b/prototypes/connections/test-async.c
index 6efaf0b..ad5f15a 100644
--- a/prototypes/connections/test-async.c
+++ b/prototypes/connections/test-async.c
@@ -237,6 +237,40 @@ static void rejection_and_budget(const char *directory) {
     CHECK(runtime.active == 0 && runtime.retained == 0);
 }
 
+/* A resource sent in pieces is charged like any other and released with its
+ * ticket (or at once when it does not fit); it cannot be forwarded. */
+static int loans_released;
+static void release_loan(void *allocation, size_t size) {
+    (void)allocation;
+    (void)size;
+    ++loans_released;
+}
+
+static void vectored_loan(const char *directory) {
+    static int mapping;
+    PureRuntime runtime, other;
+    PureWork slots[2], other_slots[1];
+    CHECK(pure_runtime_init(&runtime, slots, 2, &identity, async_host(directory), 6));
+    CHECK(pure_runtime_init(&other, other_slots, 1, &identity, async_host(directory), 64));
+    CHECK(pure_runtime_enable_async(&runtime) && pure_runtime_enable_async(&other));
+    PureTicket first, second;
+    CHECK(pure_runtime_accept(&runtime, PURE_INPUT_READ, "lent", 0, 0, &first) == PURE_OK);
+    CHECK(pure_runtime_accept(&runtime, PURE_INPUT_READ, "lent", 0, 0, &second) == PURE_OK);
+    PureEffect one, two;
+    CHECK(pure_runtime_take(&runtime, &one) == PURE_TAKE_EFFECT && pure_runtime_take(&runtime, &two) == PURE_TAKE_EFFECT);
+    PureResource a = {.body = {NULL, 4}, .allocation = &mapping, .allocation_size = 4, .release = release_loan}, b = a;
+    CHECK(pure_runtime_complete(one.ticket, PURE_OK, &a, (PureSpan){0}, 1) && !a.release);
+    PureOutcome result;
+    CHECK(pure_runtime_result(first, &result) && result.status == PURE_OK && !result.body.data && result.body.size == 4);
+    CHECK(runtime.retained == 4 && loans_released == 0);
+    CHECK(pure_runtime_complete(two.ticket, PURE_OK, &b, (PureSpan){0}, 1));
+    CHECK(pure_runtime_result(second, &result) && result.status == PURE_BUDGET && loans_released == 1);
+    PureTicket forwarded;
+    CHECK(pure_runtime_forward(first, &other, "copy", 0, &forwarded) == PURE_INVALID);
+    CHECK(pure_runtime_release(first) && loans_released == 2 && runtime.retained == 0);
+    CHECK(pure_runtime_release(second) && runtime.active == 0 && other.active == 0);
+}
+
 static void authored_document(const char *directory) {
     PureProgram program;
     char error[160];
@@ -317,6 +351,7 @@ int main(void) {
     CHECK(directory && pure_store_init(directory) == PURE_OK);
     worker_and_forward(directory);
     rejection_and_budget(directory);
+    vectored_loan(directory);
     authored_document(directory);
     DIR *opened = opendir(directory);
     CHECK(opened);
@@ -326,5 +361,5 @@ int main(void) {
         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\"}");
+    puts("{\"async_effects\":\"authored-document-worker-overlap-release-forward-budget-recovery-vectored-loan\"}");
 }