X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fmain.c;h=f7c24b896ee24e54db183928992b9c4f88d494a1;hb=fe5b36ea157526c6a118c31b77f18eff48f46feb;hp=a3b8016db46d157dc44f478553011f489b60c591;hpb=8bdee6e2bac74a577147046126628ff3b1b34930;p=lttng-tools.git

diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c
index a3b8016db..f7c24b896 100644
--- a/src/bin/lttng-relayd/main.c
+++ b/src/bin/lttng-relayd/main.c
@@ -18,6 +18,7 @@
  */
 
 #define _GNU_SOURCE
+#define _LGPL_SOURCE
 #include <getopt.h>
 #include <grp.h>
 #include <limits.h>
@@ -233,6 +234,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':
@@ -1345,7 +1351,6 @@ int relay_close_stream(struct lttcomm_relayd_hdr *recv_hdr,
 	stream->last_net_seq_num = be64toh(stream_info.last_net_seq_num);
 	stream->close_flag = 1;
 	session->stream_count--;
-	assert(session->stream_count >= 0);
 
 	/* Check if we can close it or else the data will do it. */
 	try_close_stream(session, stream);
@@ -1944,9 +1949,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;
@@ -1963,6 +1969,7 @@ 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);
@@ -1993,6 +2000,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:
@@ -2156,6 +2165,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) {
@@ -2198,6 +2208,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:
@@ -2421,6 +2433,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");
 
@@ -2609,6 +2622,7 @@ restart:
 
 				if (revents & LPOLLIN) {
 					if (conn->type != RELAY_DATA) {
+						rcu_read_unlock();
 						continue;
 					}
 
@@ -2652,6 +2666,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);