X-Git-Url: https://git.liburcu.org/?a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Flive.c;fp=src%2Fbin%2Flttng-relayd%2Flive.c;h=9c1c64d81c6aa4161697348858aa42d0a828512d;hb=5d2369e3971b6b9d7f26591727241e2c414b5ece;hp=0355cf9d91f06cf95c62a84efd61ccb07fa29f60;hpb=883b91c5a6b247de16c816e7772a945fd43cc216;p=lttng-tools.git

diff --git a/src/bin/lttng-relayd/live.c b/src/bin/lttng-relayd/live.c
index 0355cf9d9..9c1c64d81 100644
--- a/src/bin/lttng-relayd/live.c
+++ b/src/bin/lttng-relayd/live.c
@@ -88,6 +88,27 @@ static uint64_t last_relay_viewer_session_id;
 static pthread_mutex_t last_relay_viewer_session_id_lock =
 		PTHREAD_MUTEX_INITIALIZER;
 
+static const char *
+lttng_viewer_next_index_return_code_str(enum lttng_viewer_next_index_return_code code)
+{
+	switch (code) {
+	case LTTNG_VIEWER_INDEX_OK:
+		return "INDEX_OK";
+	case LTTNG_VIEWER_INDEX_RETRY:
+		return "INDEX_RETRY";
+	case LTTNG_VIEWER_INDEX_HUP:
+		return "INDEX_HUP";
+	case LTTNG_VIEWER_INDEX_ERR:
+		return "INDEX_ERR";
+	case LTTNG_VIEWER_INDEX_INACTIVE:
+		return "INDEX_INACTIVE";
+	case LTTNG_VIEWER_INDEX_EOF:
+		return "INDEX_EOF";
+	default:
+		abort();
+	}
+}
+
 /*
  * Cleanup the daemon
  */
@@ -156,28 +177,30 @@ static
 int check_new_streams(struct relay_connection *conn)
 {
 	struct relay_session *session;
-	unsigned long current_val;
 	int ret = 0;
 
+	rcu_read_lock();
 	if (!conn->viewer_session) {
 		goto end;
 	}
-	rcu_read_lock();
-	cds_list_for_each_entry_rcu(session,
-			&conn->viewer_session->session_list,
-			viewer_session_node) {
+
+	cds_list_for_each_entry_rcu(
+		session, &conn->viewer_session->session_list, viewer_session_node)
+	{
 		if (!session_get(session)) {
 			continue;
 		}
-		current_val = uatomic_cmpxchg(&session->new_streams, 1, 0);
-		ret = current_val;
+
+		ret = uatomic_read(&session->new_streams);
 		session_put(session);
 		if (ret == 1) {
 			goto end;
 		}
 	}
+
 end:
 	rcu_read_unlock();
+	DBG("Viewer connection has%s new streams: socket_fd = %d", ret == 0 ? " no" : "", conn->sock->fd);
 	return ret;
 }
 
@@ -1187,6 +1210,8 @@ int viewer_get_new_streams(struct relay_connection *conn)
 		response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_ERR);
 		goto send_reply_unlock;
 	}
+
+	uatomic_set(&session->new_streams, 0);
 	send_streams = 1;
 	response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_OK);
 
@@ -1601,6 +1626,7 @@ int viewer_get_next_index(struct relay_connection *conn)
 	struct relay_stream *rstream = NULL;
 	struct ctf_trace *ctf_trace = NULL;
 	struct relay_viewer_stream *metadata_viewer_stream = NULL;
+	bool attached_sessions_have_new_streams = false;
 
 	assert(conn);
 
@@ -1648,6 +1674,17 @@ int viewer_get_next_index(struct relay_connection *conn)
 		goto send_reply;
 	}
 
+	ret = check_new_streams(conn);
+	if (ret < 0) {
+		viewer_index.status = LTTNG_VIEWER_INDEX_ERR;
+		ERR("Error checking for new streams in the attached sessions, returning status=%s",
+		    lttng_viewer_next_index_return_code_str(
+			    (enum lttng_viewer_next_index_return_code) viewer_index.status));
+		goto send_reply;
+	} else if (ret == 1) {
+		attached_sessions_have_new_streams = true;
+	}
+
 	if (rstream->ongoing_rotation.is_set) {
 		/* Rotation is ongoing, try again later. */
 		viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_RETRY);
@@ -1710,6 +1747,7 @@ int viewer_get_next_index(struct relay_connection *conn)
 		 */
 		goto send_reply;
 	}
+
 	/* At this point, ret is 0 thus we will be able to read the index. */
 	assert(!ret);
 
@@ -1769,14 +1807,6 @@ int viewer_get_next_index(struct relay_connection *conn)
 		vstream->stream_file.handle = fs_handle;
 	}
 
-	ret = check_new_streams(conn);
-	if (ret < 0) {
-		viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_ERR);
-		goto send_reply;
-	} else if (ret == 1) {
-		viewer_index.flags |= LTTNG_VIEWER_FLAG_NEW_STREAM;
-	}
-
 	ret = lttng_index_file_read(vstream->index_file, &packet_index);
 	if (ret) {
 		ERR("Relay error reading index file");
@@ -1821,6 +1851,10 @@ send_reply:
 		pthread_mutex_unlock(&metadata_viewer_stream->stream->lock);
 	}
 
+	if (attached_sessions_have_new_streams) {
+		viewer_index.flags |= LTTNG_VIEWER_FLAG_NEW_STREAM;
+	}
+
 	viewer_index.flags = htobe32(viewer_index.flags);
 	health_code_update();