X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;ds=sidebyside;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=009621a6f7b1c4701a0857848346f29c66b8511a;hb=beaad64cecee395058e37c8b33dc50af99d771a4;hp=41afccabdfa33b7e045b38a57a825d4d6246ab4b;hpb=0d9c5d7750202aa5f46cd7c6f642fb31fe1556c4;p=lttng-tools.git

diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c
index 41afccabd..009621a6f 100644
--- a/src/bin/lttng-relayd/main.c
+++ b/src/bin/lttng-relayd/main.c
@@ -1273,14 +1273,16 @@ int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr,
 
 	if (data_buffer_size < data_size) {
 		/* In case the realloc fails, we can free the memory */
-		char *tmp_data_ptr = data_buffer;
-		data_buffer = realloc(data_buffer, data_size);
-		if (!data_buffer) {
+		char *tmp_data_ptr;
+
+		tmp_data_ptr = realloc(data_buffer, data_size);
+		if (!tmp_data_ptr) {
 			ERR("Allocating data buffer");
-			free(tmp_data_ptr);
+			free(data_buffer);
 			ret = -1;
 			goto end;
 		}
+		data_buffer = tmp_data_ptr;
 		data_buffer_size = data_size;
 	}
 	memset(data_buffer, 0, data_size);
@@ -1354,7 +1356,7 @@ int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr,
 	 * structure considering that the other side will adapt.
 	 */
 
-	ret = sscanf(VERSION, "%u.%u", &reply.major, &reply.minor);
+	ret = sscanf(VERSION, "%10u.%10u", &reply.major, &reply.minor);
 	if (ret < 2) {
 		ERR("Error in scanning version");
 		ret = -1;
@@ -1693,14 +1695,16 @@ int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht)
 
 	data_size = be32toh(data_hdr.data_size);
 	if (data_buffer_size < data_size) {
-		char *tmp_data_ptr = data_buffer;
-		data_buffer = realloc(data_buffer, data_size);
-		if (!data_buffer) {
+		char *tmp_data_ptr;
+
+		tmp_data_ptr = realloc(data_buffer, data_size);
+		if (!tmp_data_ptr) {
 			ERR("Allocating data buffer");
-			free(tmp_data_ptr);
+			free(data_buffer);
 			ret = -1;
 			goto end_unlock;
 		}
+		data_buffer = tmp_data_ptr;
 		data_buffer_size = data_size;
 	}
 	memset(data_buffer, 0, data_size);
@@ -1837,8 +1841,8 @@ void relay_del_connection(struct lttng_ht *relay_connections_ht,
 static
 void *relay_thread_worker(void *data)
 {
-	int i, ret, pollfd, err = -1;
-	uint32_t revents, nb_fd;
+	int ret, err = -1, last_seen_data_fd = -1;
+	uint32_t nb_fd;
 	struct relay_command *relay_connection;
 	struct lttng_poll_event events;
 	struct lttng_ht *relay_connections_ht;
@@ -1873,9 +1877,11 @@ void *relay_thread_worker(void *data)
 		goto error;
 	}
 
+restart:
 	while (1) {
+		int idx = -1, i, seen_control = 0, last_notdel_data_fd = -1;
+
 		/* Infinite blocking call, waiting for transmission */
-	restart:
 		DBG3("Relayd worker thread polling...");
 		ret = lttng_poll_wait(&events, -1);
 		if (ret < 0) {
@@ -1890,10 +1896,15 @@ void *relay_thread_worker(void *data)
 
 		nb_fd = ret;
 
+		/*
+		 * Process control. The control connection is prioritised so we don't
+		 * starve it with high throughout put tracing data on the data
+		 * connection.
+		 */
 		for (i = 0; i < nb_fd; i++) {
 			/* Fetch once the poll data */
-			revents = LTTNG_POLL_GETEV(&events, i);
-			pollfd = LTTNG_POLL_GETFD(&events, i);
+			uint32_t revents = LTTNG_POLL_GETEV(&events, i);
+			int pollfd = LTTNG_POLL_GETFD(&events, i);
 
 			/* Thread quit pipe has been closed. Killing thread. */
 			ret = check_thread_quit_pipe(pollfd, revents);
@@ -1915,7 +1926,7 @@ void *relay_thread_worker(void *data)
 						goto error;
 					}
 				}
-			} else if (revents > 0) {
+			} else if (revents) {
 				rcu_read_lock();
 				lttng_ht_lookup(relay_connections_ht,
 						(void *)((unsigned long) pollfd),
@@ -1935,12 +1946,18 @@ void *relay_thread_worker(void *data)
 					relay_del_connection(relay_connections_ht,
 							streams_ht, &iter,
 							relay_connection);
+					if (last_seen_data_fd == pollfd) {
+						last_seen_data_fd = last_notdel_data_fd;
+					}
 				} else if (revents & (LPOLLHUP | LPOLLRDHUP)) {
 					DBG("Socket %d hung up", pollfd);
 					relay_cleanup_poll_connection(&events, pollfd);
 					relay_del_connection(relay_connections_ht,
 							streams_ht, &iter,
 							relay_connection);
+					if (last_seen_data_fd == pollfd) {
+						last_seen_data_fd = last_notdel_data_fd;
+					}
 				} else if (revents & LPOLLIN) {
 					/* control socket */
 					if (relay_connection->type == RELAY_CONTROL) {
@@ -1962,34 +1979,101 @@ void *relay_thread_worker(void *data)
 							ret = relay_process_control(&recv_hdr,
 									relay_connection,
 									streams_ht);
-							/*
-							 * there was an error in processing a control
-							 * command: clear the session
-							 * */
 							if (ret < 0) {
+								/* Clear the session on error. */
 								relay_cleanup_poll_connection(&events, pollfd);
 								relay_del_connection(relay_connections_ht,
 										streams_ht, &iter,
 										relay_connection);
 								DBG("Connection closed with %d", pollfd);
 							}
+							seen_control = 1;
 						}
-						/* data socket */
-					} else if (relay_connection->type == RELAY_DATA) {
-						ret = relay_process_data(relay_connection, streams_ht);
-						/* connection closed */
-						if (ret < 0) {
-							relay_cleanup_poll_connection(&events, pollfd);
-							relay_del_connection(relay_connections_ht,
-									streams_ht, &iter,
-									relay_connection);
-							DBG("Data connection closed with %d", pollfd);
-						}
+					} else {
+						/*
+						 * Flag the last seen data fd not deleted. It will be
+						 * used as the last seen fd if any fd gets deleted in
+						 * this first loop.
+						 */
+						last_notdel_data_fd = pollfd;
+					}
+				}
+				rcu_read_unlock();
+			}
+		}
+
+		/*
+		 * The last loop handled a control request, go back to poll to make
+		 * sure we prioritise the control socket.
+		 */
+		if (seen_control) {
+			continue;
+		}
+
+		if (last_seen_data_fd >= 0) {
+			for (i = 0; i < nb_fd; i++) {
+				int pollfd = LTTNG_POLL_GETFD(&events, i);
+				if (last_seen_data_fd == pollfd) {
+					idx = i;
+					break;
+				}
+			}
+		}
+
+		/* Process data connection. */
+		for (i = idx + 1; i < nb_fd; i++) {
+			/* Fetch the poll data. */
+			uint32_t revents = LTTNG_POLL_GETEV(&events, i);
+			int pollfd = LTTNG_POLL_GETFD(&events, i);
+
+			/* Skip the command pipe. It's handled in the first loop. */
+			if (pollfd == relay_cmd_pipe[0]) {
+				continue;
+			}
+
+			if (revents) {
+				rcu_read_lock();
+				lttng_ht_lookup(relay_connections_ht,
+						(void *)((unsigned long) pollfd),
+						&iter);
+				node = lttng_ht_iter_get_node_ulong(&iter);
+				if (node == NULL) {
+					/* Skip it. Might be removed before. */
+					rcu_read_unlock();
+					continue;
+				}
+				relay_connection = caa_container_of(node,
+						struct relay_command, sock_n);
+
+				if (revents & LPOLLIN) {
+					if (relay_connection->type != RELAY_DATA) {
+						continue;
+					}
+
+					ret = relay_process_data(relay_connection, streams_ht);
+					/* connection closed */
+					if (ret < 0) {
+						relay_cleanup_poll_connection(&events, pollfd);
+						relay_del_connection(relay_connections_ht,
+								streams_ht, &iter,
+								relay_connection);
+						DBG("Data connection closed with %d", pollfd);
+						/*
+						 * Every goto restart call sets the last seen fd where
+						 * here we don't really care since we gracefully
+						 * continue the loop after the connection is deleted.
+						 */
+					} else {
+						/* Keep last seen port. */
+						last_seen_data_fd = pollfd;
+						rcu_read_unlock();
+						goto restart;
 					}
 				}
 				rcu_read_unlock();
 			}
 		}
+		last_seen_data_fd = -1;
 	}
 
 exit:
@@ -2054,7 +2138,7 @@ int main(int argc, char **argv)
 
 	/* Parse arguments */
 	progname = argv[0];
-	if ((ret = parse_args(argc, argv) < 0)) {
+	if ((ret = parse_args(argc, argv)) < 0) {
 		goto exit;
 	}