Fix: check for new streams in all attached sessions

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

diff --git a/src/bin/lttng-relayd/live.c b/src/bin/lttng-relayd/live.c
index 5b9e4248c0e779deadc7b604ace4d928f7c9634f..887066f14c0822bd6c37c676a2fd7b82a79acf09 100644 (file)
--- a/src/bin/lttng-relayd/live.c
+++ b/src/bin/lttng-relayd/live.c
@@ -153,32 +153,33 @@ ssize_t send_response(struct lttcomm_sock *sock, void *buf, size_t size)
 }
 
 /*
- * Atomically check if new streams got added in the session since the last
- * check and reset the flag to 0.
+ * Atomically check if new streams got added in one of the sessions attached
+ * and reset the flag to 0.
  *
  * Returns 1 if new streams got added, 0 if nothing changed, a negative value
  * on error.
  */
 static
-int check_new_streams(uint64_t session_id, struct lttng_ht *sessions_ht)
+int check_new_streams(struct relay_connection *conn)
 {
-       int ret;
-       unsigned long current_val;
        struct relay_session *session;
+       unsigned long current_val;
+       int ret = 0;
 
-       assert(sessions_ht);
-
-       session = session_find_by_id(sessions_ht, session_id);
-       if (!session) {
-               DBG("Relay session %" PRIu64 " not found", session_id);
-               ret = -1;
-               goto error;
+       if (!conn->viewer_session) {
+               goto end;
+       }
+       cds_list_for_each_entry(session,
+                       &conn->viewer_session->sessions_head,
+                       viewer_session_list) {
+               current_val = uatomic_cmpxchg(&session->new_streams, 1, 0);
+               ret = current_val;
+               if (ret == 1) {
+                       goto end;
+               }
        }
 
-       current_val = uatomic_cmpxchg(&session->new_streams, 1, 0);
-       ret = current_val;
-
-error:
+end:
        return ret;
 }
 
@@ -835,6 +836,72 @@ end:
        return found;
 }
 
+/*
+ * Delete all streams for a specific session ID.
+ */
+static void destroy_viewer_streams_by_session(struct relay_session *session)
+{
+       struct relay_viewer_stream *stream;
+       struct lttng_ht_iter iter;
+
+       assert(session);
+
+       rcu_read_lock();
+       cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, stream,
+                       stream_n.node) {
+               struct ctf_trace *ctf_trace;
+
+               health_code_update();
+               if (stream->session_id != session->id) {
+                       continue;
+               }
+
+               ctf_trace = ctf_trace_find_by_path(session->ctf_traces_ht,
+                               stream->path_name);
+               assert(ctf_trace);
+
+               viewer_stream_delete(stream);
+
+               if (stream->metadata_flag) {
+                       ctf_trace->metadata_sent = 0;
+                       ctf_trace->viewer_metadata_stream = NULL;
+               }
+
+               viewer_stream_destroy(ctf_trace, stream);
+       }
+       rcu_read_unlock();
+}
+
+static void try_destroy_streams(struct relay_session *session)
+{
+       struct ctf_trace *ctf_trace;
+       struct lttng_ht_iter iter;
+
+       assert(session);
+
+       cds_lfht_for_each_entry(session->ctf_traces_ht->ht, &iter.iter, ctf_trace,
+                       node.node) {
+               /* Attempt to destroy the ctf trace of that session. */
+               ctf_trace_try_destroy(session, ctf_trace);
+       }
+}
+
+/*
+ * Cleanup a session.
+ */
+static void cleanup_session(struct relay_connection *conn,
+               struct relay_session *session)
+{
+       /*
+        * Very important that this is done before destroying the session so we
+        * can put back every viewer stream reference from the ctf_trace.
+        */
+       destroy_viewer_streams_by_session(session);
+       try_destroy_streams(session);
+       cds_list_del(&session->viewer_session_list);
+       session_viewer_try_destroy(conn->sessions_ht, session);
+}
+
 /*
  * Send the viewer the list of current sessions.
  */
@@ -906,7 +973,7 @@ int viewer_get_new_streams(struct relay_connection *conn)
                 * and try to destroy it.
                 */
                cds_list_del(&session->viewer_session_list);
-               session_viewer_try_destroy(conn->sessions_ht, session);
+               cleanup_session(conn, session);
                goto send_reply;
        }
 
@@ -1183,7 +1250,7 @@ int viewer_get_next_index(struct relay_connection *conn)
                viewer_index.flags |= LTTNG_VIEWER_FLAG_NEW_METADATA;
        }
 
-       ret = check_new_streams(vstream->session_id, conn->sessions_ht);
+       ret = check_new_streams(conn);
        if (ret < 0) {
                goto end_unlock;
        } else if (ret == 1) {
@@ -1352,7 +1419,7 @@ int viewer_get_packet(struct relay_connection *conn)
                goto send_reply;
        }
 
-       ret = check_new_streams(stream->session_id, conn->sessions_ht);
+       ret = check_new_streams(conn);
        if (ret < 0) {
                goto end_unlock;
        } else if (ret == 1) {
@@ -1677,56 +1744,6 @@ void cleanup_connection_pollfd(struct lttng_poll_event *events, int pollfd)
        }
 }
 
-/*
- * Delete all streams for a specific session ID.
- */
-static void destroy_viewer_streams_by_session(struct relay_session *session)
-{
-       struct relay_viewer_stream *stream;
-       struct lttng_ht_iter iter;
-
-       assert(session);
-
-       rcu_read_lock();
-       cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, stream,
-                       stream_n.node) {
-               struct ctf_trace *ctf_trace;
-
-               health_code_update();
-               if (stream->session_id != session->id) {
-                       continue;
-               }
-
-               ctf_trace = ctf_trace_find_by_path(session->ctf_traces_ht,
-                               stream->path_name);
-               assert(ctf_trace);
-
-               viewer_stream_delete(stream);
-
-               if (stream->metadata_flag) {
-                       ctf_trace->metadata_sent = 0;
-                       ctf_trace->viewer_metadata_stream = NULL;
-               }
-
-               viewer_stream_destroy(ctf_trace, stream);
-       }
-       rcu_read_unlock();
-}
-
-static void try_destroy_streams(struct relay_session *session)
-{
-       struct ctf_trace *ctf_trace;
-       struct lttng_ht_iter iter;
-
-       assert(session);
-
-       cds_lfht_for_each_entry(session->ctf_traces_ht->ht, &iter.iter, ctf_trace,
-                       node.node) {
-               /* Attempt to destroy the ctf trace of that session. */
-               ctf_trace_try_destroy(session, ctf_trace);
-       }
-}
-
 /*
  * Delete and destroy a connection.
  *
@@ -1751,14 +1768,7 @@ static void destroy_connection(struct lttng_ht *relay_connections_ht,
                        &conn->viewer_session->sessions_head,
                        viewer_session_list) {
                DBG("Cleaning connection of session ID %" PRIu64, session->id);
-               /*
-                * Very important that this is done before destroying the session so we
-                * can put back every viewer stream reference from the ctf_trace.
-                */
-               destroy_viewer_streams_by_session(session);
-               try_destroy_streams(session);
-               cds_list_del(&session->viewer_session_list);
-               session_viewer_try_destroy(conn->sessions_ht, session);
+               cleanup_session(conn, session);
        }
        rcu_read_unlock();