X-Git-Url: http://git.liburcu.org/?a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=b70026242abadc5c2ba54913be6f8de6e743b921;hb=2fc6b1ab74fcff341b8c55260b4b0bea1efd1d7a;hp=ea46ec72276117359b32fa499b584357b5aeb401;hpb=0072e5e28fb5f18daffb930be7efd55d2474e6ad;p=lttng-tools.git

diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c
index ea46ec722..b70026242 100644
--- a/src/bin/lttng-relayd/main.c
+++ b/src/bin/lttng-relayd/main.c
@@ -70,6 +70,7 @@
 #include "stream.h"
 #include "connection.h"
 #include "tracefile-array.h"
+#include "tcp_keep_alive.h"
 
 /* command line options */
 char *opt_output_path;
@@ -161,10 +162,11 @@ static struct option long_options[] = {
 	{ "output", 1, 0, 'o', },
 	{ "verbose", 0, 0, 'v', },
 	{ "config", 1, 0, 'f' },
+	{ "version", 0, 0, 'V' },
 	{ NULL, 0, 0, 0, },
 };
 
-static const char *config_ignore_options[] = { "help", "config" };
+static const char *config_ignore_options[] = { "help", "config", "version" };
 
 /*
  * Take an option from the getopt output and set it in the right variable to be
@@ -255,6 +257,9 @@ static int set_option(int opt, const char *arg, const char *optname)
 			perror("exec");
 		}
 		exit(EXIT_FAILURE);
+	case 'V':
+		fprintf(stdout, "%s\n", VERSION);
+		exit(EXIT_SUCCESS);
 	case 'o':
 		if (lttng_is_setuid_setgid()) {
 			WARN("Getting '%s' argument from setuid/setgid binary refused for security reasons.",
@@ -887,6 +892,15 @@ restart:
 					lttcomm_destroy_sock(newsock);
 					goto error;
 				}
+
+				ret = socket_apply_keep_alive_config(newsock->fd);
+				if (ret < 0) {
+					ERR("Failed to apply TCP keep-alive configuration on socket (%i)",
+							newsock->fd);
+					lttcomm_destroy_sock(newsock);
+					goto error;
+				}
+
 				new_conn = connection_create(newsock, type);
 				if (!new_conn) {
 					lttcomm_destroy_sock(newsock);
@@ -965,12 +979,16 @@ static void *relay_thread_dispatcher(void *data)
 
 	health_code_update();
 
-	while (!CMM_LOAD_SHARED(dispatch_thread_exit)) {
+	for (;;) {
 		health_code_update();
 
 		/* Atomically prepare the queue futex */
 		futex_nto1_prepare(&relay_conn_queue.futex);
 
+		if (CMM_LOAD_SHARED(dispatch_thread_exit)) {
+			break;
+		}
+
 		do {
 			health_code_update();
 
@@ -1942,6 +1960,7 @@ static int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr,
 	struct lttcomm_relayd_generic_reply reply;
 	struct relay_stream *stream;
 	uint64_t net_seq_num;
+	size_t msg_len;
 
 	assert(conn);
 
@@ -1953,9 +1972,12 @@ static int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr,
 		goto end_no_session;
 	}
 
+	msg_len = lttcomm_relayd_index_len(
+			lttng_to_index_major(conn->major, conn->minor),
+			lttng_to_index_minor(conn->major, conn->minor));
 	ret = conn->sock->ops->recvmsg(conn->sock, &index_info,
-			sizeof(index_info), 0);
-	if (ret < sizeof(index_info)) {
+			msg_len, 0);
+	if (ret < msg_len) {
 		if (ret == 0) {
 			/* Orderly shutdown. Not necessary to print an error. */
 			DBG("Socket %d did an orderly shutdown", conn->sock->fd);
@@ -2179,41 +2201,36 @@ static int handle_index_data(struct relay_stream *stream, uint64_t net_seq_num,
 		goto end;
 	}
 
-	if (rotate_index || !stream->index_fd) {
-		int fd;
+	if (rotate_index || !stream->index_file) {
+		uint32_t major, minor;
 
-		/* Put ref on previous index_fd. */
-		if (stream->index_fd) {
-			stream_fd_put(stream->index_fd);
-			stream->index_fd = NULL;
+		/* Put ref on previous index_file. */
+		if (stream->index_file) {
+			lttng_index_file_put(stream->index_file);
+			stream->index_file = NULL;
 		}
-
-		fd = index_create_file(stream->path_name, stream->channel_name,
+		major = stream->trace->session->major;
+		minor = stream->trace->session->minor;
+		stream->index_file = lttng_index_file_create(stream->path_name,
+				stream->channel_name,
 			        -1, -1, stream->tracefile_size,
-				tracefile_array_get_file_index_head(stream->tfa));
-		if (fd < 0) {
+				tracefile_array_get_file_index_head(stream->tfa),
+				lttng_to_index_major(major, minor),
+				lttng_to_index_minor(major, minor));
+		if (!stream->index_file) {
 			ret = -1;
 			/* Put self-ref for this index due to error. */
 			relay_index_put(index);
-			goto end;
-		}
-		stream->index_fd = stream_fd_create(fd);
-		if (!stream->index_fd) {
-			ret = -1;
-			if (close(fd)) {
-				PERROR("Error closing FD %d", fd);
-			}
-			/* Put self-ref for this index due to error. */
-			relay_index_put(index);
-			/* Will put the local ref. */
+			index = NULL;
 			goto end;
 		}
 	}
 
-	if (relay_index_set_fd(index, stream->index_fd, data_offset)) {
+	if (relay_index_set_file(index, stream->index_file, data_offset)) {
 		ret = -1;
 		/* Put self-ref for this index due to error. */
 		relay_index_put(index);
+		index = NULL;
 		goto end;
 	}
 
@@ -2227,6 +2244,7 @@ static int handle_index_data(struct relay_stream *stream, uint64_t net_seq_num,
 	} else {
 		/* Put self-ref for this index due to error. */
 		relay_index_put(index);
+		index = NULL;
 		ret = -1;
 	}
 end:
@@ -2676,6 +2694,11 @@ error:
 			destroy_conn,
 			sock_n.node) {
 		health_code_update();
+
+		if (session_abort(destroy_conn->session)) {
+			assert(0);
+		}
+
 		/*
 		 * No need to grab another ref, because we own
 		 * destroy_conn.
@@ -2928,6 +2951,12 @@ exit_init_data:
 	health_app_destroy(health_relayd);
 exit_health_app_create:
 exit_options:
+	/*
+	 * Wait for all pending call_rcu work to complete before tearing
+	 * down data structures. call_rcu worker may be trying to
+	 * perform lookups in those structures.
+	 */
+	rcu_barrier();
 	relayd_cleanup();
 
 	/* Ensure all prior call_rcu are done. */