Fix: relayd: live: some listed sessions are not attacheable

[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 84fa1435196057c6a506b019eedf4bcf477da49b..8c4f501aa3446afd5b5cb5d7e72e42bf82dd2d11 100644 (file)
--- a/src/bin/lttng-relayd/live.c
+++ b/src/bin/lttng-relayd/live.c
@@ -197,7 +197,7 @@ end:
  */
 static
 ssize_t send_viewer_streams(struct lttcomm_sock *sock,
  */
 static
 ssize_t send_viewer_streams(struct lttcomm_sock *sock,

-               struct relay_session *session, unsigned int ignore_sent_flag)
+               uint64_t session_id, unsigned int ignore_sent_flag)

 {
        ssize_t ret;
        struct lttng_viewer_stream send_stream;
 {
        ssize_t ret;
        struct lttng_viewer_stream send_stream;


@@ -218,7 +218,7 @@ ssize_t send_viewer_streams(struct lttcomm_sock *sock,
 
                pthread_mutex_lock(&vstream->stream->lock);
                /* Ignore if not the same session. */
 
                pthread_mutex_lock(&vstream->stream->lock);
                /* Ignore if not the same session. */

-               if (vstream->stream->trace->session->id != session->id ||
+               if (vstream->stream->trace->session->id != session_id ||

                                (!ignore_sent_flag && vstream->sent_flag)) {
                        pthread_mutex_unlock(&vstream->stream->lock);
                        viewer_stream_put(vstream);
                                (!ignore_sent_flag && vstream->sent_flag)) {
                        pthread_mutex_unlock(&vstream->stream->lock);
                        viewer_stream_put(vstream);


@@ -271,24 +271,25 @@ end_unlock:
  * viewer stream of the session, the number of unsent stream and the number of
  * stream created. Those counters can be NULL and thus will be ignored.
  *
  * viewer stream of the session, the number of unsent stream and the number of
  * stream created. Those counters can be NULL and thus will be ignored.
  *

+ * session must be locked to ensure that we see either none or all initial
+ * streams for a session, but no intermediate state..
+ *

  * Return 0 on success or else a negative value.
  */
  * Return 0 on success or else a negative value.
  */

-static
-int make_viewer_streams(struct relay_session *session,
-               enum lttng_viewer_seek seek_t, uint32_t *nb_total, uint32_t *nb_unsent,
-               uint32_t *nb_created, bool *closed)
+static int make_viewer_streams(struct relay_session *session,
+               struct lttng_trace_chunk *viewer_trace_chunk,
+               enum lttng_viewer_seek seek_t,
+               uint32_t *nb_total,
+               uint32_t *nb_unsent,
+               uint32_t *nb_created,
+               bool *closed)

 {
        int ret;
        struct lttng_ht_iter iter;
        struct ctf_trace *ctf_trace;
 
        assert(session);
 {
        int ret;
        struct lttng_ht_iter iter;
        struct ctf_trace *ctf_trace;
 
        assert(session);

-
-       /*
-        * Hold the session lock to ensure that we see either none or
-        * all initial streams for a session, but no intermediate state.
-        */
-       pthread_mutex_lock(&session->lock);
+       ASSERT_LOCKED(session->lock);

 
        if (session->connection_closed) {
                *closed = true;
 
        if (session->connection_closed) {
                *closed = true;


@@ -323,7 +324,8 @@ int make_viewer_streams(struct relay_session *session,
                        }
                        vstream = viewer_stream_get_by_id(stream->stream_handle);
                        if (!vstream) {
                        }
                        vstream = viewer_stream_get_by_id(stream->stream_handle);
                        if (!vstream) {

-                               vstream = viewer_stream_create(stream, seek_t);
+                               vstream = viewer_stream_create(stream,
+                                               viewer_trace_chunk, seek_t);

                                if (!vstream) {
                                        ret = -1;
                                        ctf_trace_put(ctf_trace);
                                if (!vstream) {
                                        ret = -1;
                                        ctf_trace_put(ctf_trace);


@@ -376,7 +378,6 @@ int make_viewer_streams(struct relay_session *session,
 
 error_unlock:
        rcu_read_unlock();
 
 error_unlock:
        rcu_read_unlock();

-       pthread_mutex_unlock(&session->lock);

        return ret;
 }
 
        return ret;
 }
 


@@ -830,9 +831,18 @@ int viewer_list_sessions(struct relay_connection *conn)
 
                health_code_update();
 
 
                health_code_update();
 

+               pthread_mutex_lock(&session->lock);

                if (session->connection_closed) {
                        /* Skip closed session */
                if (session->connection_closed) {
                        /* Skip closed session */

-                       continue;
+                       goto next_session;
+               }
+               if (!session->current_trace_chunk) {
+                       /*
+                        * Skip un-attachable session. It is either
+                        * being destroyed or has not had a trace
+                        * chunk created against it yet.
+                        */
+                       goto next_session;

                }
 
                if (count >= buf_count) {
                }
 
                if (count >= buf_count) {


@@ -843,7 +853,7 @@ int viewer_list_sessions(struct relay_connection *conn)
                                new_buf_count * sizeof(*send_session_buf));
                        if (!newbuf) {
                                ret = -1;
                                new_buf_count * sizeof(*send_session_buf));
                        if (!newbuf) {
                                ret = -1;

-                               break;
+                               goto break_loop;

                        }
                        send_session_buf = newbuf;
                        buf_count = new_buf_count;
                        }
                        send_session_buf = newbuf;
                        buf_count = new_buf_count;


@@ -853,12 +863,12 @@ int viewer_list_sessions(struct relay_connection *conn)
                                session->session_name,
                                sizeof(send_session->session_name))) {
                        ret = -1;
                                session->session_name,
                                sizeof(send_session->session_name))) {
                        ret = -1;

-                       break;
+                       goto break_loop;

                }
                if (lttng_strncpy(send_session->hostname, session->hostname,
                                sizeof(send_session->hostname))) {
                        ret = -1;
                }
                if (lttng_strncpy(send_session->hostname, session->hostname,
                                sizeof(send_session->hostname))) {
                        ret = -1;

-                       break;
+                       goto break_loop;

                }
                send_session->id = htobe64(session->id);
                send_session->live_timer = htobe32(session->live_timer);
                }
                send_session->id = htobe64(session->id);
                send_session->live_timer = htobe32(session->live_timer);


@@ -869,6 +879,12 @@ int viewer_list_sessions(struct relay_connection *conn)
                }
                send_session->streams = htobe32(session->stream_count);
                count++;
                }
                send_session->streams = htobe32(session->stream_count);
                count++;

+       next_session:
+               pthread_mutex_unlock(&session->lock);
+               continue;
+       break_loop:
+               pthread_mutex_unlock(&session->lock);
+               break;

        }
        rcu_read_unlock();
        if (ret < 0) {
        }
        rcu_read_unlock();
        if (ret < 0) {


@@ -909,7 +925,7 @@ int viewer_get_new_streams(struct relay_connection *conn)
        uint32_t nb_created = 0, nb_unsent = 0, nb_streams = 0, nb_total = 0;
        struct lttng_viewer_new_streams_request request;
        struct lttng_viewer_new_streams_response response;
        uint32_t nb_created = 0, nb_unsent = 0, nb_streams = 0, nb_total = 0;
        struct lttng_viewer_new_streams_request request;
        struct lttng_viewer_new_streams_response response;

-       struct relay_session *session;
+       struct relay_session *session = NULL;

        uint64_t session_id;
        bool closed = false;
 
        uint64_t session_id;
        bool closed = false;
 


@@ -946,11 +962,24 @@ int viewer_get_new_streams(struct relay_connection *conn)
        send_streams = 1;
        response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_OK);
 
        send_streams = 1;
        response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_OK);
 

-       ret = make_viewer_streams(session, LTTNG_VIEWER_SEEK_LAST, &nb_total, &nb_unsent,
+       pthread_mutex_lock(&session->lock);
+       if (!conn->viewer_session->current_trace_chunk &&
+                       session->current_trace_chunk) {
+               ret = viewer_session_set_trace_chunk(conn->viewer_session,
+                               session->current_trace_chunk);
+               if (ret) {
+                       goto error_unlock_session;
+               }
+       }
+       ret = make_viewer_streams(session,
+                       conn->viewer_session->current_trace_chunk,
+                       LTTNG_VIEWER_SEEK_LAST, &nb_total, &nb_unsent,

                        &nb_created, &closed);
        if (ret < 0) {
                        &nb_created, &closed);
        if (ret < 0) {

-               goto end_put_session;
+               goto error_unlock_session;

        }
        }

+       pthread_mutex_unlock(&session->lock);
+

        /* Only send back the newly created streams with the unsent ones. */
        nb_streams = nb_created + nb_unsent;
        response.streams_count = htobe32(nb_streams);
        /* Only send back the newly created streams with the unsent ones. */
        nb_streams = nb_created + nb_unsent;
        response.streams_count = htobe32(nb_streams);


@@ -988,7 +1017,7 @@ send_reply:
         * streams that were not sent from that point will be sent to
         * the viewer.
         */
         * streams that were not sent from that point will be sent to
         * the viewer.
         */

-       ret = send_viewer_streams(conn->sock, session, 0);
+       ret = send_viewer_streams(conn->sock, session_id, 0);

        if (ret < 0) {
                goto end_put_session;
        }
        if (ret < 0) {
                goto end_put_session;
        }


@@ -999,6 +1028,10 @@ end_put_session:
        }
 error:
        return ret;
        }
 error:
        return ret;

+error_unlock_session:
+       pthread_mutex_unlock(&session->lock);
+       session_put(session);
+       return ret;

 }
 
 /*
 }
 
 /*


@@ -1015,6 +1048,7 @@ int viewer_attach_session(struct relay_connection *conn)
        struct lttng_viewer_attach_session_response response;
        struct relay_session *session = NULL;
        bool closed = false;
        struct lttng_viewer_attach_session_response response;
        struct relay_session *session = NULL;
        bool closed = false;

+       uint64_t session_id;

 
        assert(conn);
 
 
        assert(conn);
 


@@ -1026,6 +1060,7 @@ int viewer_attach_session(struct relay_connection *conn)
                goto error;
        }
 
                goto error;
        }
 

+       session_id = be64toh(request.session_id);

        health_code_update();
 
        memset(&response, 0, sizeof(response));
        health_code_update();
 
        memset(&response, 0, sizeof(response));


@@ -1036,16 +1071,15 @@ int viewer_attach_session(struct relay_connection *conn)
                goto send_reply;
        }
 
                goto send_reply;
        }
 

-       session = session_get_by_id(be64toh(request.session_id));
+       session = session_get_by_id(session_id);

        if (!session) {
        if (!session) {

-               DBG("Relay session %" PRIu64 " not found",
-                               (uint64_t) be64toh(request.session_id));
+               DBG("Relay session %" PRIu64 " not found", session_id);

                response.status = htobe32(LTTNG_VIEWER_ATTACH_UNK);
                goto send_reply;
        }
                response.status = htobe32(LTTNG_VIEWER_ATTACH_UNK);
                goto send_reply;
        }

-       DBG("Attach session ID %" PRIu64 " received",
-               (uint64_t) be64toh(request.session_id));
+       DBG("Attach session ID %" PRIu64 " received", session_id);

 
 

+       pthread_mutex_lock(&session->lock);

        if (session->live_timer == 0) {
                DBG("Not live session");
                response.status = htobe32(LTTNG_VIEWER_ATTACH_NOT_LIVE);
        if (session->live_timer == 0) {
                DBG("Not live session");
                response.status = htobe32(LTTNG_VIEWER_ATTACH_NOT_LIVE);


@@ -1073,13 +1107,25 @@ int viewer_attach_session(struct relay_connection *conn)
                goto send_reply;
        }
 
                goto send_reply;
        }
 

-       ret = make_viewer_streams(session, seek_type, &nb_streams, NULL,
-                       NULL, &closed);
+       if (!conn->viewer_session->current_trace_chunk &&
+                       session->current_trace_chunk) {
+               ret = viewer_session_set_trace_chunk(conn->viewer_session,
+                               session->current_trace_chunk);
+               if (ret) {
+                       goto end_put_session;
+               }
+       }
+       ret = make_viewer_streams(session,
+                       conn->viewer_session->current_trace_chunk, seek_type,
+                       &nb_streams, NULL, NULL, &closed);

        if (ret < 0) {
                goto end_put_session;
        }
        if (ret < 0) {
                goto end_put_session;
        }

-       response.streams_count = htobe32(nb_streams);
+       pthread_mutex_unlock(&session->lock);
+       session_put(session);
+       session = NULL;

 
 

+       response.streams_count = htobe32(nb_streams);

        /*
         * If the session is closed when the viewer is attaching, it
         * means some of the streams may have been concurrently removed,
        /*
         * If the session is closed when the viewer is attaching, it
         * means some of the streams may have been concurrently removed,


@@ -1111,13 +1157,14 @@ send_reply:
        }
 
        /* Send stream and ignore the sent flag. */
        }
 
        /* Send stream and ignore the sent flag. */

-       ret = send_viewer_streams(conn->sock, session, 1);
+       ret = send_viewer_streams(conn->sock, session_id, 1);

        if (ret < 0) {
                goto end_put_session;
        }
 
 end_put_session:
        if (session) {
        if (ret < 0) {
                goto end_put_session;
        }
 
 end_put_session:
        if (session) {

+               pthread_mutex_unlock(&session->lock);

                session_put(session);
        }
 error:
                session_put(session);
        }
 error:


@@ -1153,7 +1200,7 @@ static int try_open_index(struct relay_viewer_stream *vstream,
                goto end;
        }
        vstream->index_file = lttng_index_file_create_from_trace_chunk_read_only(
                goto end;
        }
        vstream->index_file = lttng_index_file_create_from_trace_chunk_read_only(

-                       rstream->trace_chunk, rstream->path_name,
+                       vstream->stream_file.trace_chunk, rstream->path_name,

                        rstream->channel_name, rstream->tracefile_size,
                        vstream->current_tracefile_id,
                        lttng_to_index_major(connection_major, connection_minor),
                        rstream->channel_name, rstream->tracefile_size,
                        vstream->current_tracefile_id,
                        lttng_to_index_major(connection_major, connection_minor),


@@ -2023,8 +2070,13 @@ restart:
                                        if (ret < 0) {
                                                goto error;
                                        }
                                        if (ret < 0) {
                                                goto error;
                                        }

-                                       lttng_poll_add(&events, conn->sock->fd,
+                                       ret = lttng_poll_add(&events,
+                                                       conn->sock->fd,

                                                        LPOLLIN | LPOLLRDHUP);
                                                        LPOLLIN | LPOLLRDHUP);

+                                       if (ret) {
+                                               ERR("Failed to add new live connection file descriptor to poll set");
+                                               goto error;
+                                       }

                                        connection_ht_add(viewer_connections_ht, conn);
                                        DBG("Connection socket %d added to poll", conn->sock->fd);
                                } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
                                        connection_ht_add(viewer_connections_ht, conn);
                                        DBG("Connection socket %d added to poll", conn->sock->fd);
                                } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {