aboutsummaryrefslogtreecommitdiffstats
path: root/prototypes
diff options
context:
space:
mode:
Diffstat (limited to 'prototypes')
-rw-r--r--prototypes/connections/runtime.c7
-rw-r--r--prototypes/connections/runtime.h5
-rw-r--r--prototypes/connections/serve.c50
-rw-r--r--prototypes/connections/test-async.c37
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\"}");
}
generated by cgit v1.2.3 (git 2.46.0) at 2026-10-20 13:33:38 +0000