X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;ds=sidebyside;f=src%2Fbin%2Flttng-relayd%2Flive.c;h=aa0586814e6ba55adeb6c2b35d5c31f1789bcc51;hb=c0bae11d346fa301d993430a2cf33b3c426e3140;hp=5d3449d4f451f49aaf64a6f3506a4b93c1affc3c;hpb=3b463131be76ea826a7e74f9b88422573e3a2076;p=lttng-tools.git

diff --git a/src/bin/lttng-relayd/live.c b/src/bin/lttng-relayd/live.c
index 5d3449d4f..aa0586814 100644
--- 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);