Fix: don't expose empty streams

[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 5d3449d4f451f49aaf64a6f3506a4b93c1affc3c..aa0586814e6ba55adeb6c2b35d5c31f1789bcc51 100644 (file)
--- a/src/bin/lttng-relayd/live.c
+++ b/src/bin/lttng-relayd/live.c
@@ -313,6 +313,12 @@ int make_viewer_streams(struct relay_session *session,
                        if (!stream->published) {
                                goto next;
                        }
+                       /*
+                        * Stream has no data, don't consider it yet.
+                        */
+                       if (stream->prev_seq == -1ULL) {
+                               goto next;
+                       }
                        vstream = viewer_stream_get_by_id(stream->stream_handle);
                        if (!vstream) {
                                vstream = viewer_stream_create(stream, seek_t);
@@ -327,17 +333,34 @@ int make_viewer_streams(struct relay_session *session,
                                        /* Update number of created stream counter. */
                                        (*nb_created)++;
                                }
+                               /*
+                                * Ensure a self-reference is preserved even
+                                * after we have put our local reference.
+                                */
+                               viewer_stream_get(vstream);
                        } else {
                                if (!vstream->sent_flag && nb_unsent) {
                                        /* Update number of unsent stream counter. */
                                        (*nb_unsent)++;
                                }
-                               viewer_stream_put(vstream);
                        }
                        /* Update number of total stream counter. */
                        if (nb_total) {
-                               (*nb_total)++;
+                               if (stream->is_metadata) {
+                                       if (!stream->closed ||
+                                                       stream->metadata_received > vstream->metadata_sent) {
+                                               (*nb_total)++;
+                                       }
+                               } else {
+                                       if (!stream->closed ||
+                                               !(((int64_t) (stream->prev_seq - stream->last_net_seq_num)) >= 0)) {
+
+                                               (*nb_total)++;
+                                       }
+                               }
                        }
+                       /* Put local reference. */
+                       viewer_stream_put(vstream);
                next:
                        stream_put(stream);
                }
@@ -911,7 +934,8 @@ int viewer_get_new_streams(struct relay_connection *conn)
        response.streams_count = htobe32(nb_streams);
 
        /*
-        * If the session is closed, HUP when there are no more streams.
+        * If the session is closed, HUP when there are no more streams
+        * with data.
         */
        if (closed && nb_total == 0) {
                send_streams = 0;
@@ -1273,8 +1297,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. */
@@ -1397,7 +1421,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);
@@ -1458,8 +1484,6 @@ int viewer_get_packet(struct relay_connection *conn)
        struct lttng_viewer_get_packet get_packet_info;
        struct lttng_viewer_trace_packet reply;
        struct relay_viewer_stream *vstream = NULL;
-       struct ctf_trace *ctf_trace;
-       struct relay_viewer_stream *metadata_viewer_stream = NULL;
 
        DBG2("Relay get data packet");
 
@@ -1477,60 +1501,11 @@ int viewer_get_packet(struct relay_connection *conn)
 
        vstream = viewer_stream_get_by_id(be64toh(get_packet_info.stream_id));
        if (!vstream) {
-               goto error;
-       }
-
-       ctf_trace = vstream->stream->trace;
-
-       /* metadata_viewer_stream may be NULL. */
-       metadata_viewer_stream =
-                       ctf_trace_get_viewer_metadata_stream(ctf_trace);
-
-       if (metadata_viewer_stream) {
-               bool get_packet_err = false;
-
-               pthread_mutex_lock(&metadata_viewer_stream->stream->lock);
-               DBG("get packet metadata check: recv %" PRIu64 " sent %" PRIu64,
-                       metadata_viewer_stream->stream->metadata_received,
-                       metadata_viewer_stream->metadata_sent);
-               if (!metadata_viewer_stream->stream->metadata_received ||
-                               metadata_viewer_stream->stream->metadata_received >
-                                       metadata_viewer_stream->metadata_sent) {
-                       /*
-                        * We prevent the client from reading a data stream as
-                        * long as there is metadata left to consume. This
-                        * ensures that the client won't receive data of which
-                        * it can't make sense.
-                        */
-                       get_packet_err = true;
-               }
-               pthread_mutex_unlock(&metadata_viewer_stream->stream->lock);
-               viewer_stream_put(metadata_viewer_stream);
-               if (get_packet_err) {
-                       reply.status = htobe32(LTTNG_VIEWER_GET_PACKET_ERR);
-                       reply.flags |= LTTNG_VIEWER_FLAG_NEW_METADATA;
-                       goto send_reply_nolock;
-               }
+               reply.status = htobe32(LTTNG_VIEWER_GET_PACKET_ERR);
+               goto send_reply_nolock;
        }
 
        pthread_mutex_lock(&vstream->stream->lock);
-       /*
-        * The vstream->stream_fd used here has been opened by
-        * get_next_index. It is opened there because this is what
-        * allows us to grab a reference to the file with stream lock
-        * held, thus protecting us against overwrite caused by
-        * tracefile rotation. Since tracefile rotation unlinks the old
-        * data file, we are ensured that we won't have our data
-        * overwritten under us.
-        */
-       ret = check_new_streams(conn);
-       if (ret < 0) {
-               goto end_free;
-       } else if (ret == 1) {
-               reply.status = htobe32(LTTNG_VIEWER_GET_PACKET_ERR);
-               reply.flags |= LTTNG_VIEWER_FLAG_NEW_STREAM;
-               goto send_reply;
-       }
 
        len = be32toh(get_packet_info.len);
        data = zmalloc(len);