Fix: code refactoring of viewer streams in relayd

[lttng-tools.git] / src / bin / lttng-relayd / main.c

diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c
index 1a93ab2e688143452be836106ec4134b4220b1ce..166e473e998ecd4d135afcb91a3b6592b73ca8bc 100644 (file)
--- a/src/bin/lttng-relayd/main.c
+++ b/src/bin/lttng-relayd/main.c
@@ -60,6 +60,7 @@
 #include "live.h"
 #include "health-relayd.h"
 #include "testpoint.h"
+#include "viewer-stream.h"
 
 /* command line options */
 char *opt_output_path;
@@ -898,7 +899,7 @@ static void destroy_stream(struct relay_stream *stream)
                }
        }
 
-       vstream = live_find_viewer_stream_by_id(stream->stream_handle);
+       vstream = viewer_stream_find_by_id(stream->stream_handle);
        if (vstream) {
                /*
                 * Set the last good value into the viewer stream. This is done
@@ -1024,6 +1025,7 @@ int relay_create_session(struct lttcomm_relayd_hdr *recv_hdr,
        session->sock = cmd->sock;
        session->minor = cmd->minor;
        session->major = cmd->major;
+       pthread_mutex_init(&session->viewer_ready_lock, NULL);
        cmd->session = session;
 
        reply.session_id = htobe64(session->id);
@@ -1071,6 +1073,8 @@ void set_viewer_ready_flag(struct relay_command *cmd)
 {
        struct relay_stream_recv_handle *node, *tmp_node;
 
+       pthread_mutex_lock(&cmd->session->viewer_ready_lock);
+
        cds_list_for_each_entry_safe(node, tmp_node, &cmd->recv_head, node) {
                struct relay_stream *stream;
 
@@ -1093,6 +1097,7 @@ void set_viewer_ready_flag(struct relay_command *cmd)
                free(node);
        }
 
+       pthread_mutex_unlock(&cmd->session->viewer_ready_lock);
        return;
 }
 
@@ -1992,6 +1997,11 @@ int relay_streams_sent(struct lttcomm_relayd_hdr *recv_hdr,
         */
        set_viewer_ready_flag(cmd);
 
+       /*
+        * Inform the viewer that there are new streams in the session.
+        */
+       uatomic_set(&cmd->session->new_streams, 1);
+
        reply.ret_code = htobe32(LTTNG_OK);
        send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply, sizeof(reply), 0);
        if (send_ret < 0) {
@@ -2234,7 +2244,7 @@ int relay_process_data(struct relay_command *cmd)
                                (stream->oldest_tracefile_id + 1) %
                                stream->tracefile_count;
                }
-               vstream = live_find_viewer_stream_by_id(stream->stream_handle);
+               vstream = viewer_stream_find_by_id(stream->stream_handle);
                if (vstream) {
                        /*
                         * The viewer is reading a file about to be