MEDIUM: mux_h2: Implement the takeover() method.

Implement a takeover() method in the mux_h2, so that other threads may
take an idle connection over if they need it.

diff --git a/src/mux_h2.c b/src/mux_h2.c
index 0d42324d78f26f6303cdf8c777e42def52416682..d10c525c90fa90b835638dc08281b586859b5281 100644 (file)
--- a/src/mux_h2.c
+++ b/src/mux_h2.c
@@ -3517,17 +3517,58 @@ schedule:
 /* this is the tasklet referenced in h2c->wait_event.tasklet */
 static struct task *h2_io_cb(struct task *t, void *ctx, unsigned short status)
 {
-       struct h2c *h2c = ctx;
+       struct connection *conn;
+       struct tasklet *tl = (struct tasklet *)t;
+       int conn_in_list;
+       struct h2c *h2c;
        int ret = 0;
 
-       TRACE_ENTER(H2_EV_H2C_WAKE, h2c->conn);
+
+       HA_SPIN_LOCK(OTHER_LOCK, &toremove_lock[tid]);
+       if (t->context == NULL) {
+               /* The connection has been taken over by another thread,
+                * we're no longer responsible for it, so just free the
+                * tasklet, and do nothing.
+                */
+               HA_SPIN_UNLOCK(OTHER_LOCK, &toremove_lock[tid]);
+               tasklet_free(tl);
+               return NULL;
+       }
+       h2c = ctx;
+       conn = h2c->conn;
+
+       TRACE_ENTER(H2_EV_H2C_WAKE, conn);
+
+       conn_in_list = conn->flags & CO_FL_LIST_MASK;
+
+       /* Remove the connection from the list, to be sure nobody attempts
+        * to use it while we handle the I/O events
+        */
+       if (conn_in_list)
+               MT_LIST_DEL(&conn->list);
+
+       HA_SPIN_UNLOCK(OTHER_LOCK, &toremove_lock[tid]);
 
        if (!(h2c->wait_event.events & SUB_RETRY_SEND))
                ret = h2_send(h2c);
        if (!(h2c->wait_event.events & SUB_RETRY_RECV))
                ret |= h2_recv(h2c);
        if (ret || b_data(&h2c->dbuf))
-               h2_process(h2c);
+               ret = h2_process(h2c);
+
+       /* If we were in an idle list, we want to add it back into it,
+        * unless h2_process() returned -1, which mean it has destroyed
+        * the connection (testing !ret is enough, if h2_process() wasn't
+        * called then ret will be 0 anyway.
+        */
+       if (!ret && conn_in_list) {
+               struct server *srv = objt_server(conn->target);
+
+               if (conn_in_list == CO_FL_SAFE_LIST)
+                       MT_LIST_ADDQ(&srv->safe_conns[tid], &conn->list);
+               else
+                       MT_LIST_ADDQ(&srv->idle_conns[tid], &conn->list);
+       }
 
        TRACE_LEAVE(H2_EV_H2C_WAKE);
        return NULL;
@@ -3676,6 +3717,22 @@ static struct task *h2_timeout_task(struct task *t, void *context, unsigned shor
                return t;
        }
 
+       /* We're about to destroy the connection, so make sure nobody attempts
+        * to steal it from us.
+        */
+       HA_SPIN_LOCK(OTHER_LOCK, &toremove_lock[tid]);
+
+       if (h2c && h2c->conn->flags & CO_FL_LIST_MASK)
+               MT_LIST_DEL(&h2c->conn->list);
+
+       /* Somebody already stole the connection from us, so we should not
+        * free it, we just have to free the task.
+        */
+       if (!t->context)
+               h2c = NULL;
+
+       HA_SPIN_UNLOCK(OTHER_LOCK, &toremove_lock[tid]);
+
        task_destroy(t);
 
        if (!h2c) {
@@ -5979,6 +6036,50 @@ static void h2_show_fd(struct buffer *msg, struct connection *conn)
        }
 }
 
+/* Migrate the the connection to the current thread.
+ * Return 0 if successful, non-zero otherwise.
+ * Expected to be called with the old thread lock held.
+ */
+static int h2_takeover(struct connection *conn)
+{
+       struct h2c *h2c = conn->ctx;
+
+       if (fd_takeover(conn->handle.fd, conn) != 0)
+               return -1;
+       if (h2c->wait_event.events)
+               h2c->conn->xprt->unsubscribe(h2c->conn, h2c->conn->xprt_ctx,
+                   h2c->wait_event.events, &h2c->wait_event);
+       /* To let the tasklet know it should free itself, and do nothing else,
+        * set its context to NULL.
+        */
+       h2c->wait_event.tasklet->context = NULL;
+       tasklet_wakeup(h2c->wait_event.tasklet);
+       if (h2c->task) {
+               h2c->task->context = NULL;
+               /* Wake the task, to let it free itself */
+               task_wakeup(h2c->task, TASK_WOKEN_OTHER);
+
+               h2c->task = task_new(tid_bit);
+               if (!h2c->task) {
+                       h2_release(h2c);
+                       return -1;
+               }
+               h2c->task->process = h2_timeout_task;
+               h2c->task->context = h2c;
+       }
+       h2c->wait_event.tasklet = tasklet_new();
+       if (!h2c->wait_event.tasklet) {
+               h2_release(h2c);
+               return -1;
+       }
+       h2c->wait_event.tasklet->process = h2_io_cb;
+       h2c->wait_event.tasklet->context = h2c;
+       h2c->conn->xprt->subscribe(h2c->conn, h2c->conn->xprt_ctx,
+                                  SUB_RETRY_RECV, &h2c->wait_event);
+
+       return 0;
+}
+
 /*******************************************************/
 /* functions below are dedicated to the config parsers */
 /*******************************************************/
@@ -6070,6 +6171,7 @@ static const struct mux_ops h2_ops = {
        .shutw = h2_shutw,
        .ctl = h2_ctl,
        .show_fd = h2_show_fd,
+       .takeover = h2_takeover,
        .flags = MX_FL_CLEAN_ABRT|MX_FL_HTX,
        .name = "H2",
 };