lib-auth-client: auth-master - Make sending of requests fully asynchronous

diff --git a/src/lib-auth-client/auth-master-private.h b/src/lib-auth-client/auth-master-private.h
index e9fed3c2dfeca08f8ca0bfe076b5713ea43a4e72..9098b20eb4a689317b0c07149e7c6ce00d55656b 100644 (file)
--- a/src/lib-auth-client/auth-master-private.h
+++ b/src/lib-auth-client/auth-master-private.h
@@ -13,7 +13,9 @@
 #define MAX_OUTBUF_SIZE 1024
 
 enum auth_master_request_state {
-       AUTH_MASTER_REQUEST_STATE_SENT = 0,
+       AUTH_MASTER_REQUEST_STATE_SUBMITTED = 0,
+       AUTH_MASTER_REQUEST_STATE_SENT,
+       AUTH_MASTER_REQUEST_STATE_REPLIED_MORE,
        AUTH_MASTER_REQUEST_STATE_REPLIED,
        AUTH_MASTER_REQUEST_STATE_FINISHED,
        AUTH_MASTER_REQUEST_STATE_ABORTED,
@@ -67,6 +69,7 @@ struct auth_master_connection {
        unsigned int id_counter;
        HASH_TABLE(void *, struct auth_master_request *) requests;
        struct auth_master_request *requests_head, *requests_tail;
+       struct auth_master_request *requests_unsent;
        unsigned int requests_count;
 
        unsigned int timeout_msecs;
@@ -86,8 +89,10 @@ struct auth_master_connection {
 unsigned int
 auth_master_request_get_timeout_msecs(struct auth_master_request *req);
 
+void auth_master_request_send(struct auth_master_request *req);
 int auth_master_request_got_reply(struct auth_master_request **_req,
                                  const char *reply, const char *const *args);
+
 void auth_master_request_fail(struct auth_master_request **_req,
                              const char *reason);
 
@@ -101,4 +106,6 @@ void auth_master_stop_idle(struct auth_master_connection *conn);
 void auth_master_connection_update_timeout(struct auth_master_connection *conn);
 void auth_master_connection_start_timeout(struct auth_master_connection *conn);
 
+void auth_master_handle_requests(struct auth_master_connection *conn);
+
 #endif

diff --git a/src/lib-auth-client/auth-master-request.c b/src/lib-auth-client/auth-master-request.c
index 4941f7418335f0736b2faec7b8702ee4056a64e0..5e4e5a169ce6d21faa6bbfc8ced9481987bcdb0d 100644 (file)
--- a/src/lib-auth-client/auth-master-request.c
+++ b/src/lib-auth-client/auth-master-request.c
@@ -43,6 +43,8 @@ static void auth_master_request_remove(struct auth_master_request *req)
 
        if (req->sent)
                hash_table_remove(conn->requests, POINTER_CAST(req->id));
+       if (req == conn->requests_unsent)
+               conn->requests_unsent = conn->requests_unsent->next;
        DLLIST2_REMOVE(&conn->requests_head, &conn->requests_tail, req);
        conn->requests_count--;
 
@@ -168,6 +170,7 @@ int auth_master_request_got_reply(struct auth_master_request **_req,
 
        ret = auth_master_request_callback(req, &mreply);
        if (ret == 0) {
+               req->state = AUTH_MASTER_REQUEST_STATE_REPLIED_MORE;
                if (conn->waiting) {
                        i_assert(conn->ioloop != NULL);
                        io_loop_stop(conn->ioloop);
@@ -223,11 +226,13 @@ void auth_master_request_fail(struct auth_master_request **_req,
        auth_master_request_abort(_req);
 }
 
-static void auth_master_request_send(struct auth_master_request *req)
+void auth_master_request_send(struct auth_master_request *req)
 {
        struct auth_master_connection *conn = req->conn;
        const char *id_str = dec2str(req->id);
 
+       i_assert(req->state == AUTH_MASTER_REQUEST_STATE_SUBMITTED);
+
        const struct const_iovec iov[] = {
                { req->cmd, strlen(req->cmd), },
                { "\t", 1 },
@@ -239,6 +244,13 @@ static void auth_master_request_send(struct auth_master_request *req)
        unsigned int iovc = N_ELEMENTS(iov);
 
        o_stream_nsendv(conn->conn.output, iov, iovc);
+
+       req->state = AUTH_MASTER_REQUEST_STATE_SENT;
+
+       hash_table_insert(conn->requests, POINTER_CAST(req->id), req);
+       req->sent = TRUE;
+
+       e_debug(req->event, "Sent");
 }
 
 #undef auth_master_request
@@ -272,6 +284,11 @@ auth_master_request(struct auth_master_connection *conn, const char *cmd,
 
        DLLIST2_APPEND(&conn->requests_head, &conn->requests_tail, req);
        conn->requests_count++;
+       if (conn->requests_unsent == NULL)
+               conn->requests_unsent = conn->requests_tail;
+
+       auth_master_connection_start_timeout(conn);
+       auth_master_stop_idle(conn);
 
        req->cmd = p_strdup(req->pool, cmd);
        if (args_size > 0)
@@ -280,63 +297,10 @@ auth_master_request(struct auth_master_connection *conn, const char *cmd,
 
        e_debug(req->event, "Created");
 
+       auth_master_handle_requests(conn);
        return req;
 }
 
-int auth_master_request_submit(struct auth_master_request **_req)
-{
-       struct auth_master_request *req = *_req;
-       struct auth_master_connection *conn = req->conn;
-
-       if (req == NULL)
-               return -1;
-
-       if (!conn->connected) {
-               if (auth_master_connect(conn) < 0) {
-                       // FIXME: handle
-                       /* we couldn't connect to auth now,
-                          so we probably can't in future either. */
-                       auth_master_request_unref(_req);
-                       return -1;
-               }
-               // FIXME: allow asynchronous connection
-               i_assert(conn->connected);
-               connection_input_resume(&conn->conn);
-       }
-
-       o_stream_cork(conn->conn.output);
-       if (!conn->sent_handshake) {
-               const struct connection_settings *set = &conn->conn.list->set;
-
-               o_stream_nsend_str(conn->conn.output,
-                       t_strdup_printf("VERSION\t%u\t%u\n",
-                                       set->major_version,
-                                       set->minor_version));
-               conn->sent_handshake = TRUE;
-       }
-
-       auth_master_request_send(req);
-       o_stream_uncork(conn->conn.output);
-
-       if (o_stream_flush(conn->conn.output) < 0) {
-               e_error(conn->conn.event, "write(auth socket) failed: %s",
-                       o_stream_get_error(conn->conn.output));
-               auth_master_disconnect(conn);
-               auth_master_request_unref(_req);
-               return -1;
-       }
-
-       hash_table_insert(conn->requests, POINTER_CAST(req->id), req);
-       req->sent = TRUE;
-
-       auth_master_connection_start_timeout(conn);
-       auth_master_stop_idle(conn);
-
-       e_debug(req->event, "Submitted");
-
-       return 0;
-}
-
 static void auth_master_request_stop(struct auth_master_request *req)
 {
        struct auth_master_connection *conn = req->conn;
@@ -386,14 +350,18 @@ bool auth_master_request_wait(struct auth_master_request *req)
                 io_loop_have_immediate_timeouts(ioloop));
 
        auth_master_request_ref(req);
-       req->state = AUTH_MASTER_REQUEST_STATE_SENT;
 
        /* add stop handler */
        to = timeout_add_short(100, auth_master_request_stop, req);
 
        conn->waiting = TRUE;
-       while (req->state < AUTH_MASTER_REQUEST_STATE_REPLIED)
-               io_loop_run(conn->ioloop);
+       if (req->state < AUTH_MASTER_REQUEST_STATE_REPLIED) {
+               if (req->state == AUTH_MASTER_REQUEST_STATE_REPLIED_MORE)
+                       req->state = AUTH_MASTER_REQUEST_STATE_SENT;
+               do
+                       io_loop_run(ioloop);
+               while (req->state < AUTH_MASTER_REQUEST_STATE_REPLIED_MORE);
+       }
        conn->waiting = waiting;
 
        e_debug(req->event, "Finished waiting for request");

diff --git a/src/lib-auth-client/auth-master.c b/src/lib-auth-client/auth-master.c
index 8883a062cb058968e3ddc5257b0f2527809e1529..d095214583750945327f7775b45d052a52cfa24e 100644 (file)
--- a/src/lib-auth-client/auth-master.c
+++ b/src/lib-auth-client/auth-master.c
@@ -371,6 +371,64 @@ auth_master_input_args(struct connection *_conn, const char *const *args)
        return (ret > 0 ? 0 : 1);
 }
 
+static void
+auth_master_handle_output_error(struct auth_master_connection *conn)
+{
+       struct ostream *output = conn->conn.output;
+
+       if (output->stream_errno != EPIPE &&
+           output->stream_errno != ECONNRESET) {
+               e_error(conn->conn.event, "write(%s) failed: %s",
+                       o_stream_get_name(output), o_stream_get_error(output));
+       } else {
+               e_error(conn->conn.event, "Remote disconnected");
+       }
+       auth_master_disconnect(conn);
+}
+
+static int
+auth_master_connection_output(struct auth_master_connection *conn)
+{
+       int ret;
+
+       if ((ret = o_stream_flush(conn->conn.output)) <= 0) {
+               if (ret < 0)
+                       auth_master_handle_output_error(conn);
+               return ret;
+       }
+
+       if (o_stream_get_buffer_used_size(conn->conn.output) >= MAX_OUTBUF_SIZE)
+               return 1;
+
+       o_stream_cork(conn->conn.output);
+       if (!conn->sent_handshake) {
+               const struct connection_settings *set = &conn->conn.list->set;
+
+               o_stream_nsend_str(conn->conn.output,
+                       t_strdup_printf("VERSION\t%u\t%u\n",
+                                       set->major_version,
+                                       set->minor_version));
+               conn->sent_handshake = TRUE;
+       }
+
+       e_debug(conn->conn.event, "Sending requests");
+
+       while (conn->requests_unsent != NULL) {
+               auth_master_request_send(conn->requests_unsent);
+               conn->requests_unsent = conn->requests_unsent->next;
+               if (o_stream_get_buffer_used_size(conn->conn.output) >=
+                   MAX_OUTBUF_SIZE)
+                       break;
+       }
+
+       if (conn->conn.output != NULL &&
+           o_stream_uncork_flush(conn->conn.output) < 0) {
+               auth_master_handle_output_error(conn);
+               return -1;
+       }
+       return 1;
+}
+
 static int auth_master_input_line(struct connection *_conn, const char *line)
 {
        struct auth_master_connection *conn =
@@ -398,6 +456,10 @@ static void auth_master_connected(struct connection *_conn, bool success)
        i_assert(success);
 
        conn->connected = TRUE;
+
+       o_stream_set_flush_callback(_conn->output,
+                                   auth_master_connection_output, conn);
+       auth_master_handle_requests(conn);
 }
 
 static void auth_master_connect_timeout(struct auth_master_connection *conn)
@@ -445,6 +507,23 @@ int auth_master_connect(struct auth_master_connection *conn)
                                          auth_master_connect_timeout, conn);
        return 0;
 }
+
+void auth_master_handle_requests(struct auth_master_connection *conn)
+{
+       if (conn->requests_unsent == NULL)
+               return;
+
+       if (!conn->connected) {
+               e_debug(conn->conn.event, "Need to connect");
+
+               (void)auth_master_connect(conn);
+               return;
+       }
+
+       i_assert(conn->conn.output != NULL);
+       o_stream_set_flush_pending(conn->conn.output, TRUE);
+}
+
 void auth_master_switch_ioloop_to(struct auth_master_connection *conn,
                                  struct ioloop *ioloop)
 {
@@ -822,6 +901,10 @@ int auth_master_pass_lookup(struct auth_master_connection *conn,
                *fields_r = NULL;
                return 0;
        }
+       if (auth_master_connect(conn) < 0) {
+               *fields_r = empty_str_array;
+               return -1;
+       }
 
        i_zero(&lookup);
        lookup.conn = conn;
@@ -845,11 +928,6 @@ int auth_master_pass_lookup(struct auth_master_connection *conn,
 
        req = auth_master_request(conn, "PASS", str_data(args), str_len(args),
                                  auth_lookup_reply_callback, &lookup);
-       if (auth_master_request_submit(&req) < 0) {
-               *fields_r = empty_str_array;
-               event_unref(&lookup.event);
-               return -1;
-       }
 
        auth_master_request_set_event(req, lookup.event);
        (void)auth_master_request_wait(req);
@@ -902,6 +980,10 @@ int auth_master_user_lookup(struct auth_master_connection *conn,
                *fields_r = NULL;
                return 0;
        }
+       if (auth_master_connect(conn) < 0) {
+               *fields_r = empty_str_array;
+               return -1;
+       }
 
        i_zero(&lookup);
        lookup.conn = conn;
@@ -925,11 +1007,6 @@ int auth_master_user_lookup(struct auth_master_connection *conn,
 
        req = auth_master_request(conn, "USER", str_data(args), str_len(args),
                                  auth_lookup_reply_callback, &lookup);
-       if (auth_master_request_submit(&req) < 0) {
-               *fields_r = empty_str_array;
-               event_unref(&lookup.event);
-               return -1;
-       }
 
        auth_master_request_set_event(req, lookup.event);
        (void)auth_master_request_wait(req);
@@ -1089,10 +1166,7 @@ auth_master_user_list_init(struct auth_master_connection *conn,
        ctx->req = auth_master_request(conn, "LIST",
                                       str_data(args), str_len(args),
                                       auth_user_list_reply_callback, ctx);
-       if (auth_master_request_submit(&ctx->req) < 0)
-               ctx->failed = TRUE;
-       else
-               auth_master_request_set_event(ctx->req, ctx->event);
+       auth_master_request_set_event(ctx->req, ctx->event);
 
        connection_input_halt(&conn->conn);
 
@@ -1221,6 +1295,9 @@ int auth_master_cache_flush(struct auth_master_connection *conn,
        struct auth_master_request *req;
        string_t *args;
 
+       if (auth_master_connect(conn) < 0)
+               return -1;
+
        i_zero(&ctx);
        ctx.conn = conn;
 
@@ -1242,12 +1319,8 @@ int auth_master_cache_flush(struct auth_master_connection *conn,
        req = auth_master_request(conn, "CACHE-FLUSH",
                                  str_data(args), str_len(args),
                                  auth_cache_flush_reply_callback, &ctx);
-       if (auth_master_request_submit(&req) < 0)
-               ctx.failed = TRUE;
-       else {
-               auth_master_request_set_event(req, ctx.event);
-               (void)auth_master_request_wait(req);
-       }
+       auth_master_request_set_event(req, ctx.event);
+       (void)auth_master_request_wait(req);
 
        if (ctx.failed)
                e_debug(ctx.event, "Cache flush failed");

diff --git a/src/lib-auth-client/auth-master.h b/src/lib-auth-client/auth-master.h
index e7765f95bd928dbae7dd80d64250a5f58ff21c81..449a6855e3d7ea895486a4ec1279b2ef04ac9f7c 100644 (file)
--- a/src/lib-auth-client/auth-master.h
+++ b/src/lib-auth-client/auth-master.h
@@ -47,8 +47,6 @@ auth_master_request(struct auth_master_connection *conn, const char *cmd,
                        typeof(context))), \
                (auth_master_request_callback_t *)callback, context)
 
-int auth_master_request_submit(struct auth_master_request **_req);
-
 void auth_master_request_set_event(struct auth_master_request *req,
                                   struct event *event);