lib: connection_input_resume() - Set added IO as having pending input

This makes sense, because when the connection is halted it might already
have input pending in the istream. So doing this always automatically
can prevent unexpected hangs. It shouldn't be harmful to do this even if
there's no input pending.

diff --git a/src/lib/connection.c b/src/lib/connection.c
index 91b1e34ca033a836240856a8496131badb10d046..214255e9959ab166d3060b17a6d1929ea0a60380 100644 (file)
--- a/src/lib/connection.c
+++ b/src/lib/connection.c
@@ -202,7 +202,8 @@ void connection_input_halt(struct connection *conn)
        timeout_remove(&conn->to);
 }
 
-void connection_input_resume(struct connection *conn)
+static void
+connection_input_resume_full(struct connection *conn, bool set_io_pending)
 {
        i_assert(!conn->disconnected);
 
@@ -211,9 +212,13 @@ void connection_input_resume(struct connection *conn)
        } else if (conn->input != NULL) {
                conn->io = io_add_istream_to(conn->ioloop, conn->input,
                                             *conn->v.input, conn);
+               if (set_io_pending)
+                       io_set_pending(conn->io);
        } else if (conn->fd_in != -1) {
                conn->io = io_add_to(conn->ioloop, conn->fd_in, IO_READ,
                                     *conn->v.input, conn);
+               if (set_io_pending)
+                       io_set_pending(conn->io);
        }
        if (conn->input_idle_timeout_secs != 0 && conn->to == NULL) {
                conn->to = timeout_add_to(conn->ioloop,
@@ -222,6 +227,11 @@ void connection_input_resume(struct connection *conn)
        }
 }
 
+void connection_input_resume(struct connection *conn)
+{
+       connection_input_resume_full(conn, TRUE);
+}
+
 static void
 connection_update_property_label(struct connection *conn)
 {
@@ -452,7 +462,7 @@ static void connection_init_streams(struct connection *conn)
 
        conn->disconnected = FALSE;
        i_assert(conn->to == NULL);
-       connection_input_resume(conn);
+       connection_input_resume_full(conn, FALSE);
        i_assert(conn->to != NULL || conn->input_idle_timeout_secs == 0);
        if (set->major_version != 0 && !set->dont_send_version) {
                e_debug(conn->event, "Sending version handshake");
@@ -669,7 +679,7 @@ void connection_init_from_streams(struct connection_list *list,
        connection_update_stream_names(conn);
        
        conn->disconnected = FALSE;
-       connection_input_resume(conn);
+       connection_input_resume_full(conn, FALSE);
 
        if (conn->v.client_connected != NULL)
                conn->v.client_connected(conn, TRUE);
@@ -878,7 +888,7 @@ void connection_set_handlers(struct connection *conn,
         if (conn->v.connect_timeout == NULL)
                 conn->v.connect_timeout = connection_connect_timeout;
        if (!conn->disconnected)
-               connection_input_resume(conn);
+               connection_input_resume_full(conn, FALSE);
 }
 
 void connection_set_default_handlers(struct connection *conn)

diff --git a/src/lib/connection.h b/src/lib/connection.h
index 5b1493b43a187c4a9945d9cd8f1a166ac7be4bdd..87aaa5d477b6ca02959fb1ecf5a06928515c16ca 100644 (file)
--- a/src/lib/connection.h
+++ b/src/lib/connection.h
@@ -209,6 +209,8 @@ void connection_disconnect(struct connection *conn);
 void connection_deinit(struct connection *conn);
 
 void connection_input_halt(struct connection *conn);
+/* Resume connection handling. If a new IO was added, it's marked as having
+   pending input. */
 void connection_input_resume(struct connection *conn);
 
 /* Update event fields and log prefix based on connection properties. */

diff --git a/src/lib/test-connection.c b/src/lib/test-connection.c
index f23a62106b5ae9d624c016e308283b779ef39893..8e757d3acecfeda7baccf31fe85f5b5531298c8b 100644 (file)
--- a/src/lib/test-connection.c
+++ b/src/lib/test-connection.c
@@ -57,6 +57,8 @@ static void test_connection_run(const struct connection_settings *set_s,
        for(unsigned int iters = 0; iters < iter_count; iters++) {
                test_assert(socketpair(AF_UNIX, SOCK_STREAM, 0, fds) == 0);
 
+               fd_set_nonblock(fds[0], TRUE);
+               fd_set_nonblock(fds[1], TRUE);
                connection_init_server(servers, conn_s, "client", fds[1], fds[1]);
                connection_init_client_fd(clients, conn_c, "server", fds[0], fds[0]);
 
@@ -398,6 +400,50 @@ static void test_connection_resume(void)
        test_end();
 }
 
+/* BEGIN RESUME PIPELINED TEST */
+static int test_connection_resume_pipelined_input_args(struct connection *conn,
+                                                      const char *const *args)
+{
+       test_assert(args[0] != NULL);
+       if (args[0] == NULL)
+               return -1;
+
+       if (strcmp(args[0], "BEGIN") == 0) {
+               o_stream_nsend_str(conn->output, "HALT\nQUIT\n");
+       } else if (strcmp(args[0], "HALT") == 0) {
+               connection_input_halt(conn);
+               to_resume = timeout_add_short(100, test_connection_resume_continue, conn);
+               return 0;
+       } else if (strcmp(args[0], "QUIT") == 0) {
+               received_quit = TRUE;
+               connection_disconnect(conn);
+       }
+
+       return 1;
+}
+
+static const struct connection_vfuncs resume_pipelined_v =
+{
+       .client_connected = test_connection_resume_client_connected,
+       .input_args = test_connection_resume_pipelined_input_args,
+       .destroy = test_connection_simple_destroy,
+};
+
+static void test_connection_resume_pipelined(void)
+{
+       test_begin("connection resume pipelined");
+
+       was_resumed = received_quit = FALSE;
+       test_connection_run(&server_set, &client_set,
+                           &resume_pipelined_v, &resume_pipelined_v, 1);
+
+       test_assert(was_resumed);
+       test_assert(received_quit);
+       was_resumed = received_quit = FALSE;
+
+       test_end();
+}
+
 /* BEGIN IDLE KILL TEST */
 
 static void
@@ -687,6 +733,7 @@ void test_connection(void)
        test_connection_ping_pong();
        test_connection_input_full();
        test_connection_resume();
+       test_connection_resume_pipelined();
        test_connection_idle_kill();
        test_connection_handshake_failed_version();
        test_connection_handshake_failed_args();