X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Flive.c;h=06c1c15134cf63af055729b6b28f75d0489962ce;hp=b2e8e7c8bb1305b07c121172ecb1d2e760339348;hb=b5a6470f372799b28d3d20603c1c0c8e5871dd63;hpb=c3b7390bf0fe3bcb1d5364acddcbe0fd107303d3

diff --git a/src/bin/lttng-relayd/live.c b/src/bin/lttng-relayd/live.c
index b2e8e7c8b..06c1c1513 100644
--- a/src/bin/lttng-relayd/live.c
+++ b/src/bin/lttng-relayd/live.c
@@ -153,32 +153,33 @@ ssize_t send_response(struct lttcomm_sock *sock, void *buf, size_t size)
 }
 
 /*
- * Atomically check if new streams got added in the session since the last
- * check and reset the flag to 0.
+ * Atomically check if new streams got added in one of the sessions attached
+ * and reset the flag to 0.
  *
  * Returns 1 if new streams got added, 0 if nothing changed, a negative value
  * on error.
  */
 static
-int check_new_streams(uint64_t session_id, struct lttng_ht *sessions_ht)
+int check_new_streams(struct relay_connection *conn)
 {
-	int ret;
-	unsigned long current_val;
 	struct relay_session *session;
+	unsigned long current_val;
+	int ret = 0;
 
-	assert(sessions_ht);
-
-	session = session_find_by_id(sessions_ht, session_id);
-	if (!session) {
-		DBG("Relay session %" PRIu64 " not found", session_id);
-		ret = -1;
-		goto error;
+	if (!conn->viewer_session) {
+		goto end;
+	}
+	cds_list_for_each_entry(session,
+			&conn->viewer_session->sessions_head,
+			viewer_session_list) {
+		current_val = uatomic_cmpxchg(&session->new_streams, 1, 0);
+		ret = current_val;
+		if (ret == 1) {
+			goto end;
+		}
 	}
 
-	current_val = uatomic_cmpxchg(&session->new_streams, 1, 0);
-	ret = current_val;
-
-error:
+end:
 	return ret;
 }
 
@@ -349,7 +350,7 @@ void stop_threads(void)
 
 	/* Stopping all threads */
 	DBG("Terminating all live threads");
-	ret = notify_thread_pipe(live_conn_pipe[1]);
+	ret = notify_thread_pipe(thread_quit_pipe[1]);
 	if (ret < 0) {
 		ERR("write error on thread quit pipe");
 	}
@@ -378,7 +379,7 @@ int create_thread_poll_set(struct lttng_poll_event *events, int size)
 	}
 
 	/* Add quit pipe */
-	ret = lttng_poll_add(events, live_conn_pipe[0], LPOLLIN | LPOLLERR);
+	ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN | LPOLLERR);
 	if (ret < 0) {
 		goto error;
 	}
@@ -395,9 +396,9 @@ error:
  * Return 1 if it was triggered else 0;
  */
 static
-int check_live_conn_pipe(int fd, uint32_t events)
+int check_thread_quit_pipe(int fd, uint32_t events)
 {
-	if (fd == live_conn_pipe[0] && (events & LPOLLIN)) {
+	if (fd == thread_quit_pipe[0] && (events & LPOLLIN)) {
 		return 1;
 	}
 
@@ -514,7 +515,7 @@ restart:
 			pollfd = LTTNG_POLL_GETFD(&events, i);
 
 			/* Thread quit pipe has been closed. Killing thread. */
-			ret = check_live_conn_pipe(pollfd, revents);
+			ret = check_thread_quit_pipe(pollfd, revents);
 			if (ret) {
 				err = 0;
 				goto exit;
@@ -693,6 +694,7 @@ int viewer_connect(struct relay_connection *conn)
 
 	health_code_update();
 
+	memset(&reply, 0, sizeof(reply));
 	reply.major = RELAYD_VERSION_COMM_MAJOR;
 	reply.minor = RELAYD_VERSION_COMM_MINOR;
 
@@ -809,6 +811,98 @@ end:
 	return ret;
 }
 
+/*
+ * Check if a connection is attached to a session.
+ * Return 1 if attached, 0 if not attached, a negative value on error.
+ */
+static
+int session_attached(struct relay_connection *conn, uint64_t session_id)
+{
+	struct relay_session *session;
+	int found = 0;
+
+	if (!conn->viewer_session) {
+		goto end;
+	}
+	cds_list_for_each_entry(session,
+			&conn->viewer_session->sessions_head,
+			viewer_session_list) {
+		if (session->id == session_id) {
+			found = 1;
+			goto end;
+		}
+	}
+
+end:
+	return found;
+}
+
+/*
+ * Delete all streams for a specific session ID.
+ */
+static void destroy_viewer_streams_by_session(struct relay_session *session)
+{
+	struct relay_viewer_stream *stream;
+	struct lttng_ht_iter iter;
+
+	assert(session);
+
+	rcu_read_lock();
+	cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, stream,
+			stream_n.node) {
+		struct ctf_trace *ctf_trace;
+
+		health_code_update();
+		if (stream->session_id != session->id) {
+			continue;
+		}
+
+		ctf_trace = ctf_trace_find_by_path(session->ctf_traces_ht,
+				stream->path_name);
+		assert(ctf_trace);
+
+		viewer_stream_delete(stream);
+
+		if (stream->metadata_flag) {
+			ctf_trace->metadata_sent = 0;
+			ctf_trace->viewer_metadata_stream = NULL;
+		}
+
+		viewer_stream_destroy(ctf_trace, stream);
+	}
+	rcu_read_unlock();
+}
+
+static void try_destroy_streams(struct relay_session *session)
+{
+	struct ctf_trace *ctf_trace;
+	struct lttng_ht_iter iter;
+
+	assert(session);
+
+	cds_lfht_for_each_entry(session->ctf_traces_ht->ht, &iter.iter, ctf_trace,
+			node.node) {
+		/* Attempt to destroy the ctf trace of that session. */
+		ctf_trace_try_destroy(session, ctf_trace);
+	}
+}
+
+/*
+ * Cleanup a session.
+ */
+static void cleanup_session(struct relay_connection *conn,
+		struct relay_session *session)
+{
+	/*
+	 * Very important that this is done before destroying the session so we
+	 * can put back every viewer stream reference from the ctf_trace.
+	 */
+	destroy_viewer_streams_by_session(session);
+	try_destroy_streams(session);
+	cds_list_del(&session->viewer_session_list);
+	session_viewer_try_destroy(conn->sessions_ht, session);
+}
+
 /*
  * Send the viewer the list of current sessions.
  */
@@ -820,6 +914,7 @@ int viewer_get_new_streams(struct relay_connection *conn)
 	struct lttng_viewer_new_streams_request request;
 	struct lttng_viewer_new_streams_response response;
 	struct relay_session *session;
+	uint64_t session_id;
 
 	assert(conn);
 
@@ -832,32 +927,26 @@ int viewer_get_new_streams(struct relay_connection *conn)
 	if (ret < 0) {
 		goto error;
 	}
+	session_id = be64toh(request.session_id);
 
 	health_code_update();
 
 	rcu_read_lock();
-	session = session_find_by_id(conn->sessions_ht,
-			be64toh(request.session_id));
+	session = session_find_by_id(conn->sessions_ht, session_id);
 	if (!session) {
-		DBG("Relay session %" PRIu64 " not found",
-				be64toh(request.session_id));
+		DBG("Relay session %" PRIu64 " not found", session_id);
 		response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_ERR);
 		goto send_reply;
 	}
 
-	if (conn->session_id == session->id) {
-		/* We confirmed the viewer is asking for the same session. */
-		send_streams = 1;
-		response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_OK);
-	} else {
+	if (!session_attached(conn, session_id)) {
 		send_streams = 0;
 		response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_ERR);
 		goto send_reply;
 	}
 
-	if (!send_streams) {
-		goto send_reply;
-	}
+	send_streams = 1;
+	response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_OK);
 
 	ret = make_viewer_streams(session, LTTNG_VIEWER_SEEK_LAST, NULL, &nb_unsent,
 			&nb_created);
@@ -868,6 +957,23 @@ int viewer_get_new_streams(struct relay_connection *conn)
 	nb_streams = nb_created + nb_unsent;
 	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 (nb_streams == 0 && session->close_flag) {
+		send_streams = 0;
+		response.status = htobe32(LTTNG_VIEWER_NEW_STREAMS_HUP);
+		/*
+		 * Remove the session from the attached list of the connection
+		 * and try to destroy it.
+		 */
+		cds_list_del(&session->viewer_session_list);
+		cleanup_session(conn, session);
+		goto send_reply;
+	}
+
 send_reply:
 	health_code_update();
 	ret = send_response(conn->sock, &response, sizeof(response));
@@ -956,8 +1062,8 @@ int viewer_attach_session(struct relay_connection *conn)
 	} else {
 		send_streams = 1;
 		response.status = htobe32(LTTNG_VIEWER_ATTACH_OK);
-		conn->session_id = session->id;
-		conn->session = session;
+		cds_list_add(&session->viewer_session_list,
+				&conn->viewer_session->sessions_head);
 	}
 
 	switch (be32toh(request.seek)) {
@@ -972,10 +1078,6 @@ int viewer_attach_session(struct relay_connection *conn)
 		goto send_reply;
 	}
 
-	if (!send_streams) {
-		goto send_reply;
-	}
-
 	ret = make_viewer_streams(session, seek_type, &nb_streams, NULL, NULL);
 	if (ret < 0) {
 		goto end_unlock;
@@ -1041,14 +1143,14 @@ int viewer_get_next_index(struct relay_connection *conn)
 	health_code_update();
 
 	rcu_read_lock();
-	session = session_find_by_id(conn->sessions_ht, conn->session_id);
-	if (!session) {
+	vstream = viewer_stream_find_by_id(be64toh(request_index.stream_id));
+	if (!vstream) {
 		ret = -1;
 		goto end_unlock;
 	}
 
-	vstream = viewer_stream_find_by_id(be64toh(request_index.stream_id));
-	if (!vstream) {
+	session = session_find_by_id(conn->sessions_ht, vstream->session_id);
+	if (!session) {
 		ret = -1;
 		goto end_unlock;
 	}
@@ -1087,23 +1189,25 @@ int viewer_get_next_index(struct relay_connection *conn)
 	rstream = stream_find_by_id(relay_streams_ht, vstream->stream_handle);
 	assert(rstream);
 
+	pthread_mutex_lock(&rstream->viewer_stream_rotation_lock);
 	if (!rstream->close_flag) {
 		if (vstream->abort_flag) {
 			/* Rotate on abort (overwrite). */
 			DBG("Viewer rotate because of overwrite");
 			ret = viewer_stream_rotate(vstream, rstream);
 			if (ret < 0) {
+				pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
 				goto end_unlock;
 			} else if (ret == 1) {
 				viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_HUP);
 				viewer_stream_delete(vstream);
 				viewer_stream_destroy(ctf_trace, vstream);
+				pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
 				goto send_reply;
 			}
 			/* ret == 0 means successful so we continue. */
 		}
 
-		pthread_mutex_lock(&rstream->viewer_stream_rotation_lock);
 		if (rstream->tracefile_count_current == vstream->tracefile_count_current) {
 			if (rstream->beacon_ts_end != -1ULL &&
 				vstream->last_sent_index == rstream->total_index_received) {
@@ -1124,24 +1228,25 @@ int viewer_get_next_index(struct relay_connection *conn)
 				goto send_reply;
 			}
 		}
-		pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
 	} else if (rstream->close_flag && vstream->close_write_flag &&
 			vstream->total_index_received == vstream->last_sent_index) {
 		/* Last index sent and current tracefile closed in write */
 		viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_HUP);
 		viewer_stream_delete(vstream);
 		viewer_stream_destroy(ctf_trace, vstream);
+		pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
 		goto send_reply;
 	} else {
 		vstream->close_write_flag = 1;
 	}
+	pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
 
 	if (!ctf_trace->metadata_received ||
 			ctf_trace->metadata_received > ctf_trace->metadata_sent) {
 		viewer_index.flags |= LTTNG_VIEWER_FLAG_NEW_METADATA;
 	}
 
-	ret = check_new_streams(vstream->session_id, conn->sessions_ht);
+	ret = check_new_streams(conn);
 	if (ret < 0) {
 		goto end_unlock;
 	} else if (ret == 1) {
@@ -1171,10 +1276,15 @@ int viewer_get_next_index(struct relay_connection *conn)
 			sizeof(packet_index));
 	pthread_mutex_unlock(&vstream->overwrite_lock);
 	if (ret < sizeof(packet_index)) {
+		unsigned int close_write_flag;
+
+		pthread_mutex_lock(&rstream->viewer_stream_rotation_lock);
+		close_write_flag = vstream->close_write_flag;
+		pthread_mutex_unlock(&rstream->viewer_stream_rotation_lock);
 		/*
 		 * The tracefile is closed in write, so we read up to EOF.
 		 */
-		if (vstream->close_write_flag == 1) {
+		if (close_write_flag == 1) {
 			viewer_index.status = htobe32(LTTNG_VIEWER_INDEX_RETRY);
 			/* Rotate on normal EOF */
 			ret = viewer_stream_rotate(vstream, rstream);
@@ -1242,6 +1352,7 @@ 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 *stream;
+	struct relay_session *session;
 	struct ctf_trace *ctf_trace;
 
 	assert(conn);
@@ -1265,7 +1376,13 @@ int viewer_get_packet(struct relay_connection *conn)
 		goto error;
 	}
 
-	ctf_trace = ctf_trace_find_by_path(conn->session->ctf_traces_ht,
+	session = session_find_by_id(conn->sessions_ht, stream->session_id);
+	if (!session) {
+		ret = -1;
+		goto error;
+	}
+
+	ctf_trace = ctf_trace_find_by_path(session->ctf_traces_ht,
 			stream->path_name);
 	assert(ctf_trace);
 
@@ -1303,7 +1420,7 @@ int viewer_get_packet(struct relay_connection *conn)
 		goto send_reply;
 	}
 
-	ret = check_new_streams(stream->session_id, conn->sessions_ht);
+	ret = check_new_streams(conn);
 	if (ret < 0) {
 		goto end_unlock;
 	} else if (ret == 1) {
@@ -1403,6 +1520,7 @@ int viewer_get_metadata(struct relay_connection *conn)
 	struct lttng_viewer_metadata_packet reply;
 	struct relay_viewer_stream *stream;
 	struct ctf_trace *ctf_trace;
+	struct relay_session *session;
 
 	assert(conn);
 
@@ -1423,7 +1541,13 @@ int viewer_get_metadata(struct relay_connection *conn)
 		goto error;
 	}
 
-	ctf_trace = ctf_trace_find_by_path(conn->session->ctf_traces_ht,
+	session = session_find_by_id(conn->sessions_ht, stream->session_id);
+	if (!session) {
+		ret = -1;
+		goto error;
+	}
+
+	ctf_trace = ctf_trace_find_by_path(session->ctf_traces_ht,
 			stream->path_name);
 	assert(ctf_trace);
 	assert(ctf_trace->metadata_sent <= ctf_trace->metadata_received);
@@ -1511,7 +1635,7 @@ int viewer_create_session(struct relay_connection *conn)
 	DBG("Viewer create session received");
 
 	resp.status = htobe32(LTTNG_VIEWER_CREATE_SESSION_OK);
-	conn->viewer_session = zmalloc(sizeof(conn->viewer_session));
+	conn->viewer_session = zmalloc(sizeof(*conn->viewer_session));
 	if (!conn->viewer_session) {
 		ERR("Allocation viewer session");
 		resp.status = htobe32(LTTNG_VIEWER_CREATE_SESSION_ERR);
@@ -1621,56 +1745,6 @@ void cleanup_connection_pollfd(struct lttng_poll_event *events, int pollfd)
 	}
 }
 
-/*
- * Delete all streams for a specific session ID.
- */
-static void destroy_viewer_streams_by_session(struct relay_session *session)
-{
-	struct relay_viewer_stream *stream;
-	struct lttng_ht_iter iter;
-
-	assert(session);
-
-	rcu_read_lock();
-	cds_lfht_for_each_entry(viewer_streams_ht->ht, &iter.iter, stream,
-			stream_n.node) {
-		struct ctf_trace *ctf_trace;
-
-		health_code_update();
-		if (stream->session_id != session->id) {
-			continue;
-		}
-
-		ctf_trace = ctf_trace_find_by_path(session->ctf_traces_ht,
-				stream->path_name);
-		assert(ctf_trace);
-
-		viewer_stream_delete(stream);
-
-		if (stream->metadata_flag) {
-			ctf_trace->metadata_sent = 0;
-			ctf_trace->viewer_metadata_stream = NULL;
-		}
-
-		viewer_stream_destroy(ctf_trace, stream);
-	}
-	rcu_read_unlock();
-}
-
-static void try_destroy_streams(struct relay_session *session)
-{
-	struct ctf_trace *ctf_trace;
-	struct lttng_ht_iter iter;
-
-	assert(session);
-
-	cds_lfht_for_each_entry(session->ctf_traces_ht->ht, &iter.iter, ctf_trace,
-			node.node) {
-		/* Attempt to destroy the ctf trace of that session. */
-		ctf_trace_try_destroy(session, ctf_trace);
-	}
-}
-
 /*
  * Delete and destroy a connection.
  *
@@ -1679,28 +1753,27 @@ static void try_destroy_streams(struct relay_session *session)
 static void destroy_connection(struct lttng_ht *relay_connections_ht,
 		struct relay_connection *conn)
 {
-	struct relay_session *session;
+	struct relay_session *session, *tmp_session;
 
 	assert(relay_connections_ht);
 	assert(conn);
 
-	DBG("Cleaning connection of session ID %" PRIu64, conn->session_id);
-
 	connection_delete(relay_connections_ht, conn);
 
+	if (!conn->viewer_session) {
+		goto end;
+	}
+
 	rcu_read_lock();
-	session = session_find_by_id(conn->sessions_ht, conn->session_id);
-	if (session) {
-		/*
-		 * Very important that this is done before destroying the session so we
-		 * can put back every viewer stream reference from the ctf_trace.
-		 */
-		destroy_viewer_streams_by_session(session);
-		try_destroy_streams(session);
-		session_viewer_try_destroy(conn->sessions_ht, session);
+	cds_list_for_each_entry_safe(session, tmp_session,
+			&conn->viewer_session->sessions_head,
+			viewer_session_list) {
+		DBG("Cleaning connection of session ID %" PRIu64, session->id);
+		cleanup_session(conn, session);
 	}
 	rcu_read_unlock();
 
+end:
 	connection_destroy(conn);
 }
 
@@ -1782,7 +1855,7 @@ restart:
 			health_code_update();
 
 			/* Thread quit pipe has been closed. Killing thread. */
-			ret = check_live_conn_pipe(pollfd, revents);
+			ret = check_thread_quit_pipe(pollfd, revents);
 			if (ret) {
 				err = 0;
 				goto exit;