X-Git-Url: http://git.liburcu.org/?a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=69131a7bb374e9d73666449588fcf6ddc2251553;hb=28e9df91c0ebda73a60579b6187bba4c14c92d43;hp=60b6bf2214e8bee4d0266000a1db592982384940;hpb=53efb85a242809ed5ed21e9ab40effa696ecbc6f;p=lttng-tools.git

diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c
index 60b6bf221..69131a7bb 100644
--- a/src/bin/lttng-relayd/main.c
+++ b/src/bin/lttng-relayd/main.c
@@ -44,6 +44,7 @@
 #include <common/common.h>
 #include <common/compat/poll.h>
 #include <common/compat/socket.h>
+#include <common/compat/endian.h>
 #include <common/defaults.h>
 #include <common/daemonize.h>
 #include <common/futex.h>
@@ -232,6 +233,11 @@ int set_option(int opt, const char *arg, const char *optname)
 		break;
 	case 'g':
 		tracing_group_name = strdup(arg);
+		if (tracing_group_name == NULL) {
+			ret = -errno;
+			PERROR("strdup");
+			goto end;
+		}
 		tracing_group_name_override = 1;
 		break;
 	case 'h':
@@ -250,7 +256,10 @@ int set_option(int opt, const char *arg, const char *optname)
 		if (arg) {
 			lttng_opt_verbose = config_parse_value(arg);
 		} else {
-			lttng_opt_verbose += 1;
+			/* Only 3 level of verbosity (-vvv). */
+			if (lttng_opt_verbose < 3) {
+				lttng_opt_verbose += 1;
+			}
 		}
 		break;
 	default:
@@ -382,8 +391,9 @@ int set_options(int argc, char **argv)
 
 	/* assign default values */
 	if (control_uri == NULL) {
-		ret = asprintf(&default_address, "tcp://0.0.0.0:%d",
-				DEFAULT_NETWORK_CONTROL_PORT);
+		ret = asprintf(&default_address,
+			"tcp://" DEFAULT_NETWORK_CONTROL_BIND_ADDRESS ":%d",
+			DEFAULT_NETWORK_CONTROL_PORT);
 		if (ret < 0) {
 			PERROR("asprintf default data address");
 			goto exit;
@@ -397,8 +407,9 @@ int set_options(int argc, char **argv)
 		}
 	}
 	if (data_uri == NULL) {
-		ret = asprintf(&default_address, "tcp://0.0.0.0:%d",
-				DEFAULT_NETWORK_DATA_PORT);
+		ret = asprintf(&default_address,
+			"tcp://" DEFAULT_NETWORK_DATA_BIND_ADDRESS ":%d",
+			DEFAULT_NETWORK_DATA_PORT);
 		if (ret < 0) {
 			PERROR("asprintf default data address");
 			goto exit;
@@ -412,8 +423,9 @@ int set_options(int argc, char **argv)
 		}
 	}
 	if (live_uri == NULL) {
-		ret = asprintf(&default_address, "tcp://0.0.0.0:%d",
-				DEFAULT_NETWORK_VIEWER_PORT);
+		ret = asprintf(&default_address,
+			"tcp://" DEFAULT_NETWORK_VIEWER_BIND_ADDRESS ":%d",
+			DEFAULT_NETWORK_VIEWER_PORT);
 		if (ret < 0) {
 			PERROR("asprintf default viewer control address");
 			goto exit;
@@ -1198,6 +1210,7 @@ int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr,
 	stream->session_id = session->id;
 	stream->index_fd = -1;
 	stream->read_index_fd = -1;
+	stream->ctf_stream_id = -1ULL;
 	lttng_ht_node_init_u64(&stream->node, stream->stream_handle);
 	pthread_mutex_init(&stream->lock, NULL);
 
@@ -1934,9 +1947,10 @@ int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr,
 		DBG("Received live beacon for stream %" PRIu64, stream->stream_handle);
 
 		/*
-		 * Only flag a stream inactive when it has already received data.
+		 * Only flag a stream inactive when it has already received data
+		 * and no indexes are in flight.
 		 */
-		if (stream->total_index_received > 0) {
+		if (stream->total_index_received > 0 && stream->indexes_in_flight == 0) {
 			stream->beacon_ts_end = be64toh(index_info.timestamp_end);
 		}
 		ret = 0;
@@ -1953,9 +1967,13 @@ int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr,
 			goto end_rcu_unlock;
 		}
 		index_created = 1;
+		stream->indexes_in_flight++;
 	}
 
 	copy_index_control_data(index, &index_info);
+	if (stream->ctf_stream_id == -1ULL) {
+		stream->ctf_stream_id = be64toh(index_info.stream_id);
+	}
 
 	if (index_created) {
 		/*
@@ -1980,6 +1998,8 @@ int relay_recv_index(struct lttcomm_relayd_hdr *recv_hdr,
 			goto end_rcu_unlock;
 		}
 		stream->total_index_received++;
+		stream->indexes_in_flight--;
+		assert(stream->indexes_in_flight >= 0);
 	}
 
 end_rcu_unlock:
@@ -2143,6 +2163,7 @@ static int handle_index_data(struct relay_stream *stream, uint64_t net_seq_num,
 			goto error;
 		}
 		index_created = 1;
+		stream->indexes_in_flight++;
 	}
 
 	if (rotate_index || stream->index_fd < 0) {
@@ -2185,6 +2206,8 @@ static int handle_index_data(struct relay_stream *stream, uint64_t net_seq_num,
 			goto error;
 		}
 		stream->total_index_received++;
+		stream->indexes_in_flight--;
+		assert(stream->indexes_in_flight >= 0);
 	}
 
 error:
@@ -2386,7 +2409,7 @@ static void destroy_connection(struct lttng_ht *relay_connections_ht,
 	connection_delete(relay_connections_ht, conn);
 
 	/* For the control socket, we try to destroy the session. */
-	if (conn->type == RELAY_CONTROL) {
+	if (conn->type == RELAY_CONTROL && conn->session) {
 		destroy_session(conn->session, conn->sessions_ht);
 	}
 
@@ -2408,6 +2431,7 @@ void *relay_thread_worker(void *data)
 	struct lttcomm_relayd_hdr recv_hdr;
 	struct relay_local_data *relay_ctx = (struct relay_local_data *) data;
 	struct lttng_ht *sessions_ht = relay_ctx->sessions_ht;
+	struct relay_index *index;
 
 	DBG("[thread] Relay worker started");
 
@@ -2639,6 +2663,14 @@ error:
 	}
 	rcu_read_unlock();
 error_poll_create:
+	rcu_read_lock();
+	cds_lfht_for_each_entry(indexes_ht->ht, &iter.iter, index,
+			index_n.node) {
+		health_code_update();
+		relay_index_delete(index);
+		relay_index_free_safe(index);
+	}
+	rcu_read_unlock();
 	lttng_ht_destroy(indexes_ht);
 indexes_ht_error:
 	lttng_ht_destroy(relay_connections_ht);