X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=fcc7fc00ae55c844b935ef065028e3526c384257;hp=f23d0391cc6fac44de193ae9513fe53b86af99d0;hb=2f21a469eee9bc3352b439b1e64fe6bbd5088fea;hpb=25c6c5dcd5c2d503a8521dda5853d6e3074fdc33

diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c
index f23d0391c..fcc7fc00a 100644
--- a/src/bin/lttng-relayd/main.c
+++ b/src/bin/lttng-relayd/main.c
@@ -1106,16 +1106,19 @@ static int relay_create_session(const struct lttcomm_relayd_hdr *recv_hdr,
 
 	memset(&reply, 0, sizeof(reply));
 
-	switch (conn->minor) {
-	case 1:
-	case 2:
-	case 3:
-		break;
-	case 4: /* LTTng sessiond 2.4 */
-	default:
+	if (conn->minor < 4) {
+		/* From 2.1 to 2.3 */
+		ret = 0;
+	} else if (conn->minor >= 4 && conn->minor < 11) {
+		/* From 2.4 to 2.10 */
 		ret = cmd_create_session_2_4(payload, session_name,
 			hostname, &live_timer, &snapshot);
+	} else {
+		/* From 2.11 to ... */
+		ret = cmd_create_session_2_11(payload, session_name,
+			hostname, &live_timer, &snapshot);
 	}
+
 	if (ret < 0) {
 		goto send_reply;
 	}
@@ -1202,17 +1205,20 @@ static int relay_add_stream(const struct lttcomm_relayd_hdr *recv_hdr,
 		goto end_no_session;
 	}
 
-	switch (session->minor) {
-	case 1: /* LTTng sessiond 2.1. Allocates path_name and channel_name. */
+	if (session->minor == 1) {
+		/* For 2.1 */
 		ret = cmd_recv_stream_2_1(payload, &path_name,
 			&channel_name);
-		break;
-	case 2: /* LTTng sessiond 2.2. Allocates path_name and channel_name. */
-	default:
+	} else if (session->minor > 1 && session->minor < 11) {
+		/* From 2.2 to 2.10 */
 		ret = cmd_recv_stream_2_2(payload, &path_name,
 			&channel_name, &tracefile_size, &tracefile_count);
-		break;
+	} else {
+		/* From 2.11 to ... */
+		ret = cmd_recv_stream_2_11(payload, &path_name,
+			&channel_name, &tracefile_size, &tracefile_count);
 	}
+
 	if (ret < 0) {
 		goto send_reply;
 	}
@@ -3344,10 +3350,14 @@ static enum relay_connection_status relay_process_data_receive_payload(
 	uint64_t left_to_receive = state->left_to_receive;
 	struct relay_session *session;
 
+	DBG3("Receiving data for stream id %" PRIu64 " seqnum %" PRIu64 ", %" PRIu64" bytes received, %" PRIu64 " bytes left to receive",
+			state->header.stream_id, state->header.net_seq_num,
+			state->received, left_to_receive);
+
 	stream = stream_get_by_id(state->header.stream_id);
 	if (!stream) {
 		/* Protocol error. */
-		DBG("relay_process_data_receive_payload: Cannot find stream %" PRIu64,
+		ERR("relay_process_data_receive_payload: cannot find stream %" PRIu64,
 				state->header.stream_id);
 		status = RELAY_CONNECTION_STATUS_ERROR;
 		goto end;
@@ -3355,10 +3365,13 @@ static enum relay_connection_status relay_process_data_receive_payload(
 
 	pthread_mutex_lock(&stream->lock);
 	session = stream->trace->session;
-
-	DBG3("Receiving data for stream id %" PRIu64 " seqnum %" PRIu64 ", %" PRIu64" bytes received, %" PRIu64 " bytes left to receive",
-			state->header.stream_id, state->header.net_seq_num,
-			state->received, left_to_receive);
+	if (!conn->session) {
+		ret = connection_set_session(conn, session);
+		if (ret) {
+			status = RELAY_CONNECTION_STATUS_ERROR;
+			goto end_stream_unlock;
+		}
+	}
 
 	/*
 	 * The size of the "chunk" received on any iteration is bounded by:
@@ -3686,6 +3699,26 @@ restart:
 
 					status = relay_process_control(ctrl_conn);
 					if (status != RELAY_CONNECTION_STATUS_OK) {
+						/*
+						 * On socket error flag the session as aborted to force
+						 * the cleanup of its stream otherwise it can leak
+						 * during the lifetime of the relayd.
+						 *
+						 * This prevents situations in which streams can be
+						 * left opened because an index was received, the
+						 * control connection is closed, and the data
+						 * connection is closed (uncleanly) before the packet's
+						 * data provided.
+						 *
+						 * Since the control connection encountered an error,
+						 * it is okay to be conservative and close the
+						 * session right now as we can't rely on the protocol
+						 * being respected anymore.
+						 */
+						if (status == RELAY_CONNECTION_STATUS_ERROR) {
+							session_abort(ctrl_conn->session);
+						}
+
 						/* Clear the connection on error or close. */
 						relay_thread_close_connection(&events,
 								pollfd,
@@ -3765,6 +3798,25 @@ restart:
 				status = relay_process_data(data_conn);
 				/* Connection closed or error. */
 				if (status != RELAY_CONNECTION_STATUS_OK) {
+					/*
+					 * On socket error flag the session as aborted to force
+					 * the cleanup of its stream otherwise it can leak
+					 * during the lifetime of the relayd.
+					 *
+					 * This prevents situations in which streams can be
+					 * left opened because an index was received, the
+					 * control connection is closed, and the data
+					 * connection is closed (uncleanly) before the packet's
+					 * data provided.
+					 *
+					 * Since the data connection encountered an error,
+					 * it is okay to be conservative and close the
+					 * session right now as we can't rely on the protocol
+					 * being respected anymore.
+					 */
+					if (status == RELAY_CONNECTION_STATUS_ERROR) {
+						session_abort(data_conn->session);
+					}
 					relay_thread_close_connection(&events, pollfd,
 							data_conn);
 					/*
@@ -3796,16 +3848,14 @@ restart:
 
 exit:
 error:
-	/* Cleanup reamaining connection object. */
+	/* Cleanup remaining connection object. */
 	rcu_read_lock();
 	cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter,
 			destroy_conn,
 			sock_n.node) {
 		health_code_update();
 
-		if (session_abort(destroy_conn->session)) {
-			assert(0);
-		}
+		session_abort(destroy_conn->session);
 
 		/*
 		 * No need to grab another ref, because we own