Foundational work to allow async/await on multiple file descriptors

A slightly different version of this patch has been sitting on a local
branch for almost two years, so I decided to take a look this weekend
and ended up rewriting it this weekend so it's compatible with the
current version of Lwan's event loop.

While this won't allow a request handler to await on multiple file
descriptors just yet, it lays down the foundations to do so by making
all the necessary plumbing changes to allow a coro_yield() from a
request handler to receive a pointer to the struct lwan_connection that
caused the request handler to be awoken.

The actual changes are minimal (with little to no overhead), but some
comments had to be added throghout to explain certain things as the
ownership of certain values isn't clear from the struct definition
alone anymore.

No APIs have been implemented to allow waiting for multiple file
descriptors at once yet, but they will probably follow in the next few
weeks.  I got to make sure it also works with the HTTP connection
coroutine as well as any other file descriptor a handler might decide
to use (e.g.  a pubsub subscription); this would allow, for instance, a
chat application to wait on both the WebSocket connection and the
pubsub subscription and react as fast as it's able to.  Let's see
what I end up coming up with.

Tested to compile only; have not tested if this feature works yet,
but normal requests seems to work fine.

Author: lpereira

src/lib/lwan-request.c

@@ -2023,27 +2023,28 @@ make_async_yield_value(int fd, enum lwan_connection_coro_yield event)
     return (int64_t)(((uint64_t)fd << 32 | event));
 }
 
-static inline void async_await_fd(struct coro *coro,
-                                  int fd,
-                                  enum lwan_connection_coro_yield events)
+static inline struct lwan_connection *async_await_fd(
+    struct coro *coro, int fd, enum lwan_connection_coro_yield events)
 {
     assert(events >= CONN_CORO_ASYNC_AWAIT_READ &&
            events <= CONN_CORO_ASYNC_AWAIT_READ_WRITE);
 
-    return (void)coro_yield(coro, make_async_yield_value(fd, events));
+    int64_t from_coro = coro_yield(coro, make_async_yield_value(fd, events));
+    return (struct lwan_connection *)(intptr_t)from_coro;
 }
 
-void lwan_request_await_read(struct lwan_request *r, int fd)
+struct lwan_connection *lwan_request_await_read(struct lwan_request *r, int fd)
 {
     return async_await_fd(r->conn->coro, fd, CONN_CORO_ASYNC_AWAIT_READ);
 }
 
-void lwan_request_await_write(struct lwan_request *r, int fd)
+struct lwan_connection *lwan_request_await_write(struct lwan_request *r, int fd)
 {
     return async_await_fd(r->conn->coro, fd, CONN_CORO_ASYNC_AWAIT_WRITE);
 }
 
-void lwan_request_await_read_write(struct lwan_request *r, int fd)
+struct lwan_connection *lwan_request_await_read_write(struct lwan_request *r,
+                                                      int fd)
 {
     return async_await_fd(r->conn->coro, fd, CONN_CORO_ASYNC_AWAIT_READ_WRITE);
 }

src/lib/lwan-thread.c

@@ -522,11 +522,23 @@ static void update_epoll_flags(const struct timeout_queue *tq,
         lwan_status_perror("epoll_ctl");
 }
 
-static void clear_async_await_flag(void *data)
+static void unasync_await_conn(void *data1, void *data2)
 {
-    struct lwan_connection *async_fd_conn = data;
+    struct lwan_connection *async_fd_conn = data1;
 
-    async_fd_conn->flags &= ~CONN_ASYNC_AWAIT;
+    async_fd_conn->flags &= ~(CONN_ASYNC_AWAIT | CONN_HUNG_UP);
+    async_fd_conn->thread = data2;
+
+    /* If this file descriptor number is used again in the future as an HTTP
+     * connection, we need the coro pointer to be NULL so a new coroutine is
+     * created!  */
+    async_fd_conn->coro = NULL;
+
+    /* While not strictly necessary, make sure that prev/next point to
+     * something valid rather than whatever junk was left from when their
+     * storage was used for the parent pointer.  */
+    async_fd_conn->prev = -1;
+    async_fd_conn->next = -1;
 }
 
 static enum lwan_connection_coro_yield
@@ -558,13 +570,31 @@ resume_async(const struct timeout_queue *tq,
 
         op = EPOLL_CTL_MOD;
     } else {
+        await_fd_conn->parent = conn;
+
+        /* We assert() in the timeout queue that we're not freeing a
+         * coroutine when CONN_ASYNC_AWAIT is set in the connection, and are
+         * careful to not ever do that.  This makes us get away with struct
+         * coro not being refcounted, even though this kinda feels like
+         * running with scissors.  */
+        assert(!await_fd_conn->coro);
+        await_fd_conn->coro = conn->coro;
+
+        /* Since scheduling is performed during startup, we gotta take note
+         * of which thread was originally supposed to handle this particular
+         * file descriptor once we're done borrowing this lwan_connection
+         * for the awaited file descriptor.  */
+        struct lwan_thread *old_thread = await_fd_conn->thread;
+        await_fd_conn->thread = conn->thread;
+
         op = EPOLL_CTL_ADD;
         flags |= CONN_ASYNC_AWAIT;
-        coro_defer(conn->coro, clear_async_await_flag, await_fd_conn);
+
+        coro_defer2(conn->coro, unasync_await_conn, await_fd_conn, old_thread);
     }
 
     struct epoll_event event = {.events = conn_flags_to_epoll_events(flags),
-                                .data.ptr = conn};
+                                .data.ptr = await_fd_conn};
     if (LIKELY(!epoll_ctl(epoll_fd, op, await_fd, &event))) {
         await_fd_conn->flags &= ~CONN_EVENTS_MASK;
         await_fd_conn->flags |= flags;
@@ -575,24 +605,27 @@ resume_async(const struct timeout_queue *tq,
 }
 
 static ALWAYS_INLINE void resume_coro(struct timeout_queue *tq,
-                                      struct lwan_connection *conn,
+                                      struct lwan_connection *conn_to_resume,
+                                      struct lwan_connection *conn_to_yield,
                                       int epoll_fd)
 {
-    assert(conn->coro);
+    assert(conn_to_resume->coro);
+    assert(conn_to_yield->coro);
 
-    int64_t from_coro = coro_resume(conn->coro);
+    int64_t from_coro = coro_resume_value(conn_to_resume->coro,
+                                          (int64_t)(intptr_t)conn_to_yield);
     enum lwan_connection_coro_yield yield_result = from_coro & 0xffffffff;
 
     if (UNLIKELY(yield_result >= CONN_CORO_ASYNC)) {
         yield_result =
-            resume_async(tq, yield_result, from_coro, conn, epoll_fd);
+            resume_async(tq, yield_result, from_coro, conn_to_resume, epoll_fd);
     }
 
     if (UNLIKELY(yield_result == CONN_CORO_ABORT)) {
-        timeout_queue_expire(tq, conn);
+        timeout_queue_expire(tq, conn_to_resume);
     } else {
-        update_epoll_flags(tq, conn, epoll_fd, yield_result);
-        timeout_queue_move_to_last(tq, conn);
+        update_epoll_flags(tq, conn_to_resume, epoll_fd, yield_result);
+        timeout_queue_move_to_last(tq, conn_to_resume);
     }
 }
 
@@ -696,8 +729,7 @@ static ALWAYS_INLINE bool spawn_coro(struct lwan_connection *conn,
 #endif
 
     assert(!conn->coro);
-    assert(!(conn->flags & CONN_ASYNC_AWAIT));
-    assert(!(conn->flags & CONN_AWAITED_FD));
+    assert(!(conn->flags & (CONN_ASYNC_AWAIT | CONN_HUNG_UP)));
     assert(!(conn->flags & CONN_LISTENER));
     assert(t);
     assert((uintptr_t)t >= (uintptr_t)tq->lwan->thread.threads);
@@ -978,7 +1010,36 @@ static void *thread_io_loop(void *data)
         for (struct epoll_event *event = events; n_fds--; event++) {
             struct lwan_connection *conn = event->data.ptr;
 
-            assert(!(conn->flags & CONN_ASYNC_AWAIT));
+            if (conn->flags & CONN_ASYNC_AWAIT) {
+                /* Assert that the connection is part of the conns array,
+                 * since the storage for conn->parent is shared with
+                 * prev/next. */
+                assert(conn->parent >= lwan->conns);
+                assert(conn->parent <= &lwan->conns[lwan->thread.max_fd]);
+
+                /* Also validate that conn->parent is in fact a HTTP client
+                 * connection and not an awaited fd! */
+                assert(!(conn->parent->flags & CONN_ASYNC_AWAIT));
+
+                /* CONN_ASYNC_AWAIT conns *must* have a coro and thread as
+                 * it's the same as the HTTP client coro for API
+                 * consistency, as struct lwan_connection isn't opaque.  (If
+                 * it were opaque, or at least a private API, though, we
+                 * might be able to get away with reusing the space for
+                 * these two pointers for something else in some cases.
+                 * This has not been necessary yet, but might become useful
+                 * in the future.) */
+                assert(conn->coro);
+                assert(conn->coro == conn->parent->coro);
+                assert(conn->thread == conn->parent->thread);
+
+                if (UNLIKELY(events->events & (EPOLLRDHUP | EPOLLHUP)))
+                    conn->flags |= CONN_HUNG_UP;
+
+                resume_coro(&tq, conn->parent, conn, epoll_fd);
+
+                continue;
+            }
 
             if (conn->flags & CONN_LISTENER) {
                 if (LIKELY(accept_waiting_clients(t, conn)))
@@ -989,10 +1050,8 @@ static void *thread_io_loop(void *data)
             }
 
             if (UNLIKELY(event->events & (EPOLLRDHUP | EPOLLHUP))) {
-                if ((conn->flags & CONN_AWAITED_FD) != CONN_SUSPENDED_MASK) {
-                    timeout_queue_expire(&tq, conn);
-                    continue;
-                }
+                timeout_queue_expire(&tq, conn);
+                continue;
             }
 
             if (!conn->coro) {
@@ -1004,7 +1063,7 @@ static void *thread_io_loop(void *data)
                 created_coros = true;
             }
 
-            resume_coro(&tq, conn, epoll_fd);
+            resume_coro(&tq, conn, conn, epoll_fd);
         }
 
         if (created_coros)

src/lib/lwan-tq.c

@@ -38,6 +38,8 @@ timeout_queue_idx_to_node(struct timeout_queue *tq, int idx)
 inline void timeout_queue_insert(struct timeout_queue *tq,
                                  struct lwan_connection *new_node)
 {
+    assert(!(new_node->flags & (CONN_HUNG_UP | CONN_ASYNC_AWAIT)));
+
     new_node->next = -1;
     new_node->prev = tq->head.prev;
     struct lwan_connection *prev = timeout_queue_idx_to_node(tq, tq->head.prev);
@@ -87,6 +89,8 @@ void timeout_queue_init(struct timeout_queue *tq, const struct lwan *lwan)
 void timeout_queue_expire(struct timeout_queue *tq,
                           struct lwan_connection *conn)
 {
+    assert(!(conn->flags & (CONN_HUNG_UP | CONN_ASYNC_AWAIT)));
+
     timeout_queue_remove(tq, conn);
 
     if (LIKELY(conn->coro)) {

src/lib/lwan.h

@@ -275,7 +275,6 @@ enum lwan_connection_flags {
     CONN_SUSPENDED_MASK = 1 << 5,
     CONN_SUSPENDED = (EPOLLRDHUP << CONN_EPOLL_EVENT_SHIFT) | CONN_SUSPENDED_MASK,
     CONN_HAS_REMOVE_SLEEP_DEFER = 1 << 6,
-    CONN_AWAITED_FD = CONN_SUSPENDED_MASK | CONN_HAS_REMOVE_SLEEP_DEFER,
 
     /* Used when HTTP pipelining has been detected.  This enables usage of the
      * MSG_MORE flags when sending responses to batch as many short responses
@@ -298,7 +297,12 @@ enum lwan_connection_flags {
 
     CONN_USE_DYNAMIC_BUFFER = 1 << 12,
 
-    CONN_FLAG_LAST = CONN_USE_DYNAMIC_BUFFER,
+    /* Only valid when CONN_ASYNC_AWAIT is set. Set on file descriptors that
+     * got (EPOLLHUP|EPOLLRDHUP) events from epoll so that request handlers
+     * can deal with this fact.  */
+    CONN_HUNG_UP = 1 << 13,
+
+    CONN_FLAG_LAST = CONN_HUNG_UP,
 };
 
 static_assert(CONN_FLAG_LAST < ((1 << 15) - 1),
@@ -373,10 +377,35 @@ struct lwan_connection {
     /* This structure is exactly 32-bytes on x86-64. If it is changed,
      * make sure the scheduler (lwan-thread.c) is updated as well. */
     enum lwan_connection_flags flags;
+
     unsigned int time_to_expire;
+
     struct coro *coro;
     struct lwan_thread *thread;
-    int prev, next; /* for timeout queue */
+
+    /* This union is here to support async/await when a handler is waiting
+     * on multiple file descriptors.  By storing a pointer to the parent
+     * connection here, we're able to register the awaited file descriptor
+     * in epoll using a pointer to the awaited file descriptor struct,
+     * allowing us to yield to the handler this information and signal which
+     * file descriptor caused the handler to be awoken.  (We can yield just
+     * the file descriptor plus another integer with values to signal things
+     * like timeouts and whatnot.  Future problems!)
+     *
+     * Also, when CONN_ASYNC_AWAIT is set, `coro` points to parent->coro,
+     * so that conn->coro is consistently usable.  Gotta be careful though,
+     * because struct coros are not refcounted and this could explode with
+     * a double free. */
+    union {
+        /* For HTTP client connections handling inside the timeout queue */
+        struct {
+            int prev;
+            int next;
+        };
+
+        /* For awaited file descriptor, only valid if flags&CONN_ASYNC_AWAIT */
+        struct lwan_connection *parent;
+    };
 };
 
 struct lwan_proxy {
@@ -639,9 +668,11 @@ void lwan_response_websocket_write_binary(struct lwan_request *request);
 int lwan_response_websocket_read(struct lwan_request *request);
 int lwan_response_websocket_read_hint(struct lwan_request *request, size_t size_hint);
 
-void lwan_request_await_read(struct lwan_request *r, int fd);
-void lwan_request_await_write(struct lwan_request *r, int fd);
-void lwan_request_await_read_write(struct lwan_request *r, int fd);
+struct lwan_connection *lwan_request_await_read(struct lwan_request *r, int fd);
+struct lwan_connection *lwan_request_await_write(struct lwan_request *r,
+                                                 int fd);
+struct lwan_connection *lwan_request_await_read_write(struct lwan_request *r,
+                                                      int fd);
 ssize_t lwan_request_async_read(struct lwan_request *r, int fd, void *buf, size_t len);
 ssize_t lwan_request_async_read_flags(struct lwan_request *request, int fd, void *buf, size_t len, int flags);
 ssize_t lwan_request_async_write(struct lwan_request *r, int fd, const void *buf, size_t len);