sessiond: notification: refactor: split transmission and poll update

Split the notification transmission logic and its effect on a
notification_client from the logic tied to the management of the
notification thread.

This is to make it possible to send (or queue) notifications from the
notification thread or another thread. If another thread encounters an
error or a full socket buffer, a future mechanism will allow it to
signal the notification thread to update its private state (e.g. poll
mask).

Signed-off-by: Jérémie Galarneau <jeremie.galarneau@efficios.com>
Change-Id: I35d8943cb11473d82b07a4dbc5a0f093cde25a79

diff --git a/src/bin/lttng-sessiond/notification-thread-events.c b/src/bin/lttng-sessiond/notification-thread-events.c
index 285bcf951aafe3a5bb2801777e7a13eb06dc5ec7..6e507642a224be3affae1540bd762bc2a138849a 100644 (file)
--- a/src/bin/lttng-sessiond/notification-thread-events.c
+++ b/src/bin/lttng-sessiond/notification-thread-events.c
@@ -2720,13 +2720,61 @@ int handle_notification_thread_trigger_unregister_all(
        return error_occurred ? -1 : 0;
 }
 
        return error_occurred ? -1 : 0;
 }
 

+static
+int client_handle_transmission_status(
+               struct notification_client *client,
+               enum client_transmission_status transmission_status,
+               struct notification_thread_state *state)
+{
+       int ret = 0;
+
+       switch (transmission_status) {
+       case CLIENT_TRANSMISSION_STATUS_COMPLETE:
+               ret = lttng_poll_mod(&state->events, client->socket,
+                               CLIENT_POLL_MASK_IN);
+               if (ret) {
+                       goto end;
+               }
+
+               client->communication.outbound.queued_command_reply = false;
+               client->communication.outbound.dropped_notification = false;
+               break;
+       case CLIENT_TRANSMISSION_STATUS_QUEUED:
+               /*
+                * We want to be notified whenever there is buffer space
+                * available to send the rest of the payload.
+                */
+               ret = lttng_poll_mod(&state->events, client->socket,
+                               CLIENT_POLL_MASK_IN_OUT);
+               if (ret) {
+                       goto end;
+               }
+               break;
+       case CLIENT_TRANSMISSION_STATUS_FAIL:
+               ret = notification_thread_client_disconnect(client, state);
+               if (ret) {
+                       goto end;
+               }
+               break;
+       case CLIENT_TRANSMISSION_STATUS_ERROR:
+               ret = -1;
+               goto end;
+       default:
+               abort();
+       }
+end:
+       return ret;
+}
+

 /* Client lock must be acquired by caller. */
 static
 /* Client lock must be acquired by caller. */
 static

-int client_flush_outgoing_queue(struct notification_client *client,
+enum client_transmission_status client_flush_outgoing_queue(
+               struct notification_client *client,

                struct notification_thread_state *state)
 {
        ssize_t ret;
        size_t to_send_count;
                struct notification_thread_state *state)
 {
        ssize_t ret;
        size_t to_send_count;

+       enum client_transmission_status status;

 
        ASSERT_LOCKED(client->lock);
 
 
        ASSERT_LOCKED(client->lock);
 


@@ -2751,41 +2799,29 @@ int client_flush_outgoing_queue(struct notification_client *client,
                                &client->communication.outbound.buffer,
                                to_send_count);
                if (ret) {
                                &client->communication.outbound.buffer,
                                to_send_count);
                if (ret) {

+                       status = CLIENT_TRANSMISSION_STATUS_ERROR;

                        goto error;
                }
                        goto error;
                }

-
-               /*
-                * We want to be notified whenever there is buffer space
-                * available to send the rest of the payload.
-                */
-               ret = lttng_poll_mod(&state->events, client->socket,
-                               CLIENT_POLL_MASK_IN_OUT);
-               if (ret) {
-                       goto error;
-               }
+               status = CLIENT_TRANSMISSION_STATUS_QUEUED;

        } else if (ret < 0) {
                /* Generic error, disconnect the client. */
        } else if (ret < 0) {
                /* Generic error, disconnect the client. */

-               ERR("[notification-thread] Failed to send flush outgoing queue, disconnecting client (socket fd = %i)",
+               ERR("[notification-thread] Failed to flush outgoing queue, disconnecting client (socket fd = %i)",

                                client->socket);
                                client->socket);

-               ret = notification_thread_client_disconnect(client, state);
-               if (ret) {
-                       goto error;
-               }
+               status = CLIENT_TRANSMISSION_STATUS_FAIL;

        } else {
                /* No error and flushed the queue completely. */
                ret = lttng_dynamic_buffer_set_size(
                                &client->communication.outbound.buffer, 0);
                if (ret) {
        } else {
                /* No error and flushed the queue completely. */
                ret = lttng_dynamic_buffer_set_size(
                                &client->communication.outbound.buffer, 0);
                if (ret) {

+                       status = CLIENT_TRANSMISSION_STATUS_ERROR;

                        goto error;
                }
                        goto error;
                }

-               ret = lttng_poll_mod(&state->events, client->socket,
-                               CLIENT_POLL_MASK_IN);
-               if (ret) {
-                       goto error;
-               }
+               status = CLIENT_TRANSMISSION_STATUS_COMPLETE;
+       }

 
 

-               client->communication.outbound.queued_command_reply = false;
-               client->communication.outbound.dropped_notification = false;
+       ret = client_handle_transmission_status(client, status, state);
+       if (ret) {
+               goto error;

        }
 
        return 0;
        }
 
        return 0;

diff --git a/src/bin/lttng-sessiond/notification-thread-internal.h b/src/bin/lttng-sessiond/notification-thread-internal.h
index 5242695f05205171f3e77152841bb6c0df0a1d2e..b278f83d32fbf5fcaaaec646b77075bb6b59df30 100644 (file)
--- a/src/bin/lttng-sessiond/notification-thread-internal.h
+++ b/src/bin/lttng-sessiond/notification-thread-internal.h
@@ -64,4 +64,12 @@ struct channel_info {
        struct rcu_head rcu_node;
 };
 
        struct rcu_head rcu_node;
 };
 

+enum client_transmission_status {
+       CLIENT_TRANSMISSION_STATUS_COMPLETE,
+       CLIENT_TRANSMISSION_STATUS_QUEUED,
+       /* Communication failure. */
+       CLIENT_TRANSMISSION_STATUS_FAIL,
+       /* Fatal error. */
+       CLIENT_TRANSMISSION_STATUS_ERROR,
+};

 #endif /* NOTIFICATION_THREAD_INTERNAL_H */
 #endif /* NOTIFICATION_THREAD_INTERNAL_H */