Fix: relayd reply error to client if cannot find viewer stream

[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 283aacc3c124d32645673666c6c741e1eeeca7f5..7b1602d36a757c9638e2bec0dc73c8b953b58f09 100644 (file)
--- a/src/bin/lttng-relayd/live.c
+++ b/src/bin/lttng-relayd/live.c
@@ -266,7 +266,7 @@ end_unlock:
 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)
+               uint32_t *nb_created, bool *closed)
 {
        int ret;
        struct lttng_ht_iter iter;
@@ -280,6 +280,10 @@ int make_viewer_streams(struct relay_session *session,
         */
        pthread_mutex_lock(&session->lock);
 
+       if (session->connection_closed) {
+               *closed = true;
+       }
+
        /*
         * Create viewer streams for relay streams that are ready to be
         * used for a the given session id only.
@@ -861,6 +865,7 @@ int viewer_get_new_streams(struct relay_connection *conn)
        struct lttng_viewer_new_streams_response response;
        struct relay_session *session;
        uint64_t session_id;
+       bool closed = false;
 
        assert(conn);
 
@@ -896,7 +901,7 @@ int viewer_get_new_streams(struct relay_connection *conn)
        response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_OK);
 
        ret = make_viewer_streams(session, LTTNG_VIEWER_SEEK_LAST, &nb_total, &nb_unsent,
-                       &nb_created);
+                       &nb_created, &closed);
        if (ret < 0) {
                goto end_put_session;
        }
@@ -905,12 +910,11 @@ int viewer_get_new_streams(struct relay_connection *conn)
        response.streams_count = htobe32(nb_streams);
 
        /*
-        * If the session is closed and we have no new streams to send,
-        * it means that the viewer has already received the whole trace
-        * for this session and should now close it.
+        * If the session is closed, HUP when there are no more streams.
         */
-       if (nb_total == 0 && session->connection_closed) {
+       if (closed && nb_total == 0) {
                send_streams = 0;
+               response.streams_count = 0;
                response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_HUP);
                goto send_reply;
        }
@@ -963,6 +967,7 @@ int viewer_attach_session(struct relay_connection *conn)
        struct lttng_viewer_attach_session_request request;
        struct lttng_viewer_attach_session_response response;
        struct relay_session *session = NULL;
+       bool closed = false;
 
        assert(conn);
 
@@ -1021,12 +1026,26 @@ int viewer_attach_session(struct relay_connection *conn)
                goto send_reply;
        }
 
-       ret = make_viewer_streams(session, seek_type, &nb_streams, NULL, NULL);
+       ret = make_viewer_streams(session, seek_type, &nb_streams, NULL,
+                       NULL, &closed);
        if (ret < 0) {
                goto end_put_session;
        }
        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,
+        * so we don't allow the viewer to attach, even if there are
+        * streams available.
+        */
+       if (closed) {
+               send_streams = 0;
+               response.streams_count = 0;
+               response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_HUP);
+               goto send_reply;
+       }
+
 send_reply:
        health_code_update();
        ret = send_response(conn->sock, &response, sizeof(response));
@@ -1253,8 +1272,8 @@ int viewer_get_next_index(struct relay_connection *conn)
 
        vstream = viewer_stream_get_by_id(be64toh(request_index.stream_id));
        if (!vstream) {
-               ret = -1;
-               goto end;
+               viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_ERR);
+               goto send_reply;
        }
 
        /* Use back. ref. Protected by refcounts. */
@@ -1377,7 +1396,9 @@ int viewer_get_next_index(struct relay_connection *conn)
        viewer_index.stream_id = packet_index.stream_id;
 
 send_reply:
-       pthread_mutex_unlock(&rstream->lock);
+       if (rstream) {
+               pthread_mutex_unlock(&rstream->lock);
+       }
 
        if (metadata_viewer_stream) {
                pthread_mutex_lock(&metadata_viewer_stream->stream->lock);
@@ -1609,7 +1630,15 @@ int viewer_get_metadata(struct relay_connection *conn)
 
        vstream = viewer_stream_get_by_id(be64toh(request.stream_id));
        if (!vstream) {
-               reply.status = htobe32(LTTNG_VIEWER_NO_NEW_METADATA);
+               /*
+                * The metadata stream can be closed by a CLOSE command
+                * just before we attach. It can also be closed by
+                * per-pid tracing during tracing. Therefore, it is
+                * possible that we cannot find this viewer stream.
+                * Reply back to the client with an error if we cannot
+                * find it.
+                */
+               reply.status = htobe32(LTTNG_VIEWER_METADATA_ERR);
                goto send_reply;
        }
        pthread_mutex_lock(&vstream->stream->lock);