summaryrefslogtreecommitdiff
path: root/proxy_await.c
diff options
context:
space:
mode:
Diffstat (limited to 'proxy_await.c')
-rw-r--r--proxy_await.c56
1 files changed, 54 insertions, 2 deletions
diff --git a/proxy_await.c b/proxy_await.c
index 1486068..5e379db 100644
--- a/proxy_await.c
+++ b/proxy_await.c
@@ -50,6 +50,7 @@ int mcplib_await(lua_State *L) {
case AWAIT_OK:
case AWAIT_FIRST:
case AWAIT_FASTGOOD:
+ case AWAIT_BACKGROUND:
break;
default:
proxy_lua_error(L, "invalid type argument tp mcp.await");
@@ -176,6 +177,40 @@ static void mcp_queue_await_io(conn *c, lua_State *Lc, mcp_request_t *rq, int aw
return;
}
+static void mcp_queue_await_dummy_io(conn *c, lua_State *Lc, int await_ref) {
+ io_queue_t *q = conn_io_queue_get(c, IO_QUEUE_PROXY);
+
+ io_pending_proxy_t *p = do_cache_alloc(c->thread->io_cache);
+ if (p == NULL) {
+ WSTAT_INCR(c, proxy_conn_oom, 1);
+ proxy_lua_error(Lc, "out of memory allocating from IO cache");
+ return;
+ }
+
+ // this is a re-cast structure, so assert that we never outsize it.
+ assert(sizeof(io_pending_t) >= sizeof(io_pending_proxy_t));
+ memset(p, 0, sizeof(io_pending_proxy_t));
+ // set up back references.
+ p->io_queue_type = IO_QUEUE_PROXY;
+ p->thread = c->thread;
+ p->c = c;
+ p->resp = NULL;
+
+ // await specific
+ p->is_await = true;
+ p->await_ref = await_ref;
+ p->await_background = true;
+
+ // Dummy IO has no backend, and no request attached.
+
+ // All we need to do is link into the batch chain.
+ p->next = q->stack_ctx;
+ q->stack_ctx = p;
+ P_DEBUG("%s: queued\n", __func__);
+
+ return;
+}
+
// TODO (v2): need to get this code running under pcall().
// It looks like a bulk of this code can move into mcplib_await(),
// and then here post-yield we can add the conn and coro_ref to the right
@@ -224,6 +259,12 @@ int mcplib_await_run(conn *c, mc_resp *resp, lua_State *L, int coro_ref) {
}
P_DEBUG("%s: argtable len: %d\n", __func__, n);
+ if (aw->type == AWAIT_BACKGROUND) {
+ mcp_queue_await_dummy_io(c, L, await_ref);
+ aw->pending++;
+ aw->wait_for = 0;
+ }
+
lua_pop(L, 1); // remove table key.
aw->resp = resp; // cuddle the current mc_resp to fill later
@@ -262,7 +303,13 @@ int mcplib_await_return(io_pending_proxy_t *p) {
// TODO (v2): for GOOD or OK cases, it might be better to return the
// last object as valid if there are otherwise zero valids?
// Think we just have to count valids...
- if (!aw->completed) {
+ if (aw->type == AWAIT_BACKGROUND) {
+ // in the background case, we never want to collect responses.
+ if (p->await_background) {
+ // found the dummy IO, complete and return conn to worker.
+ completing = true;
+ }
+ } else if (!aw->completed) {
valid = true; // always collect results unless we are completed.
if (aw->wait_for > 0) {
bool is_good = false;
@@ -298,6 +345,9 @@ int mcplib_await_return(io_pending_proxy_t *p) {
}
}
break;
+ case AWAIT_BACKGROUND:
+ // In background mode we don't wait for any response.
+ break;
}
if (is_good) {
@@ -335,7 +385,9 @@ int mcplib_await_return(io_pending_proxy_t *p) {
}
// lose our internal mcpres reference regardless.
- luaL_unref(L, LUA_REGISTRYINDEX, p->mcpres_ref);
+ if (p->mcpres_ref) {
+ luaL_unref(L, LUA_REGISTRYINDEX, p->mcpres_ref);
+ }
// our await_ref is shared, so we don't need to release it.
if (completing) {