X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=36ec02479edc3b74ccf70df1cc8f6a8a023770fe;hb=9b5e086337f67aaf99375fc39d911d167fd8c778;hp=2f20ffb5d1d97fb585803d1721bdfeb7e494ffb9;hpb=9ce5646a7ef9b8d7936c46649a21ee546fadd538;p=lttng-tools.git

diff --git a/src/common/consumer.c b/src/common/consumer.c
index 2f20ffb5d..36ec02479 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -30,6 +30,7 @@
 #include <inttypes.h>
 #include <signal.h>
 
+#include <bin/lttng-consumerd/health-consumerd.h>
 #include <common/common.h>
 #include <common/utils.h>
 #include <common/compat/poll.h>
@@ -44,7 +45,7 @@
 
 #include "consumer.h"
 #include "consumer-stream.h"
-#include "../bin/lttng-consumerd/health-consumerd.h"
+#include "consumer-testpoint.h"
 
 struct lttng_consumer_global_data consumer_data = {
 	.stream_count = 0,
@@ -94,22 +95,33 @@ static void notify_thread_lttng_pipe(struct lttng_pipe *pipe)
 	(void) lttng_pipe_write(pipe, &null_stream, sizeof(null_stream));
 }
 
+static void notify_health_quit_pipe(int *pipe)
+{
+	ssize_t ret;
+
+	ret = lttng_write(pipe[1], "4", 1);
+	if (ret < 1) {
+		PERROR("write consumer health quit");
+	}
+}
+
 static void notify_channel_pipe(struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_channel *chan,
 		uint64_t key,
 		enum consumer_channel_action action)
 {
 	struct consumer_channel_msg msg;
-	int ret;
+	ssize_t ret;
 
 	memset(&msg, 0, sizeof(msg));
 
 	msg.action = action;
 	msg.chan = chan;
 	msg.key = key;
-	do {
-		ret = write(ctx->consumer_channel_pipe[1], &msg, sizeof(msg));
-	} while (ret < 0 && errno == EINTR);
+	ret = lttng_write(ctx->consumer_channel_pipe[1], &msg, sizeof(msg));
+	if (ret < sizeof(msg)) {
+		PERROR("notify_channel_pipe write error");
+	}
 }
 
 void notify_thread_del_channel(struct lttng_consumer_local_data *ctx,
@@ -124,17 +136,18 @@ static int read_channel_pipe(struct lttng_consumer_local_data *ctx,
 		enum consumer_channel_action *action)
 {
 	struct consumer_channel_msg msg;
-	int ret;
+	ssize_t ret;
 
-	do {
-		ret = read(ctx->consumer_channel_pipe[0], &msg, sizeof(msg));
-	} while (ret < 0 && errno == EINTR);
-	if (ret > 0) {
-		*action = msg.action;
-		*chan = msg.chan;
-		*key = msg.key;
+	ret = lttng_read(ctx->consumer_channel_pipe[0], &msg, sizeof(msg));
+	if (ret < sizeof(msg)) {
+		ret = -1;
+		goto error;
 	}
-	return ret;
+	*action = msg.action;
+	*chan = msg.chan;
+	*key = msg.key;
+error:
+	return (int) ret;
 }
 
 /*
@@ -756,6 +769,44 @@ end:
 	return ret;
 }
 
+/*
+ * Find a relayd and send the streams sent message
+ *
+ * Returns 0 on success, < 0 on error
+ */
+int consumer_send_relayd_streams_sent(uint64_t net_seq_idx)
+{
+	int ret = 0;
+	struct consumer_relayd_sock_pair *relayd;
+
+	assert(net_seq_idx != -1ULL);
+
+	/* The stream is not metadata. Get relayd reference if exists. */
+	rcu_read_lock();
+	relayd = consumer_find_relayd(net_seq_idx);
+	if (relayd != NULL) {
+		/* Add stream on the relayd */
+		pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+		ret = relayd_streams_sent(&relayd->control_sock);
+		pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
+		if (ret < 0) {
+			goto end;
+		}
+	} else {
+		ERR("Relayd ID %" PRIu64 " unknown. Can't send streams_sent.",
+				net_seq_idx);
+		ret = -1;
+		goto end;
+	}
+
+	ret = 0;
+	DBG("All streams sent relayd id %" PRIu64, net_seq_idx);
+
+end:
+	rcu_read_unlock();
+	return ret;
+}
+
 /*
  * Find a relayd and close the stream
  */
@@ -867,7 +918,6 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
 	channel->uid = uid;
 	channel->gid = gid;
 	channel->relayd_id = relayd_id;
-	channel->output = output;
 	channel->tracefile_size = tracefile_size;
 	channel->tracefile_count = tracefile_count;
 	channel->monitor = monitor;
@@ -875,6 +925,20 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
 	pthread_mutex_init(&channel->lock, NULL);
 	pthread_mutex_init(&channel->timer_lock, NULL);
 
+	switch (output) {
+	case LTTNG_EVENT_SPLICE:
+		channel->output = CONSUMER_CHANNEL_SPLICE;
+		break;
+	case LTTNG_EVENT_MMAP:
+		channel->output = CONSUMER_CHANNEL_MMAP;
+		break;
+	default:
+		assert(0);
+		free(channel);
+		channel = NULL;
+		goto end;
+	}
+
 	/*
 	 * In monitor mode, the streams associated with the channel will be put in
 	 * a special list ONLY owned by this channel. So, the refcount is set to 1
@@ -1105,12 +1169,11 @@ void lttng_consumer_cleanup(void)
  */
 void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx)
 {
-	int ret;
+	ssize_t ret;
+
 	consumer_quit = 1;
-	do {
-		ret = write(ctx->consumer_should_quit[1], "4", 1);
-	} while (ret < 0 && errno == EINTR);
-	if (ret < 0 || ret != 1) {
+	ret = lttng_write(ctx->consumer_should_quit[1], "4", 1);
+	if (ret < 1) {
 		PERROR("write consumer quit");
 	}
 
@@ -1249,6 +1312,57 @@ error:
 	return NULL;
 }
 
+/*
+ * Iterate over all streams of the hashtable and free them properly.
+ */
+static void destroy_data_stream_ht(struct lttng_ht *ht)
+{
+	struct lttng_ht_iter iter;
+	struct lttng_consumer_stream *stream;
+
+	if (ht == NULL) {
+		return;
+	}
+
+	rcu_read_lock();
+	cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
+		/*
+		 * Ignore return value since we are currently cleaning up so any error
+		 * can't be handled.
+		 */
+		(void) consumer_del_stream(stream, ht);
+	}
+	rcu_read_unlock();
+
+	lttng_ht_destroy(ht);
+}
+
+/*
+ * Iterate over all streams of the metadata hashtable and free them
+ * properly.
+ */
+static void destroy_metadata_stream_ht(struct lttng_ht *ht)
+{
+	struct lttng_ht_iter iter;
+	struct lttng_consumer_stream *stream;
+
+	if (ht == NULL) {
+		return;
+	}
+
+	rcu_read_lock();
+	cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
+		/*
+		 * Ignore return value since we are currently cleaning up so any error
+		 * can't be handled.
+		 */
+		(void) consumer_del_metadata_stream(stream, ht);
+	}
+	rcu_read_unlock();
+
+	lttng_ht_destroy(ht);
+}
+
 /*
  * Close all fds associated with the instance and free the context.
  */
@@ -1258,6 +1372,9 @@ void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
 
 	DBG("Consumer destroying it. Closing everything.");
 
+	destroy_data_stream_ht(data_ht);
+	destroy_metadata_stream_ht(metadata_ht);
+
 	ret = close(ctx->consumer_error_socket);
 	if (ret) {
 		PERROR("close");
@@ -1284,15 +1401,13 @@ static int write_relayd_metadata_id(int fd,
 		struct lttng_consumer_stream *stream,
 		struct consumer_relayd_sock_pair *relayd, unsigned long padding)
 {
-	int ret;
+	ssize_t ret;
 	struct lttcomm_relayd_metadata_payload hdr;
 
 	hdr.stream_id = htobe64(stream->relayd_stream_id);
 	hdr.padding_size = htobe32(padding);
-	do {
-		ret = write(fd, (void *) &hdr, sizeof(hdr));
-	} while (ret < 0 && errno == EINTR);
-	if (ret < 0 || ret != sizeof(hdr)) {
+	ret = lttng_write(fd, (void *) &hdr, sizeof(hdr));
+	if (ret < sizeof(hdr)) {
 		/*
 		 * This error means that the fd's end is closed so ignore the perror
 		 * not to clubber the error output since this can happen in a normal
@@ -1314,7 +1429,7 @@ static int write_relayd_metadata_id(int fd,
 			stream->relayd_stream_id, padding);
 
 end:
-	return ret;
+	return (int) ret;
 }
 
 /*
@@ -1332,7 +1447,7 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 		struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_stream *stream, unsigned long len,
 		unsigned long padding,
-		struct lttng_packet_index *index)
+		struct ctf_packet_index *index)
 {
 	unsigned long mmap_offset;
 	void *mmap_base;
@@ -1470,11 +1585,9 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 	}
 
 	while (len > 0) {
-		do {
-			ret = write(outfd, mmap_base + mmap_offset, len);
-		} while (ret < 0 && errno == EINTR);
+		ret = lttng_write(outfd, mmap_base + mmap_offset, len);
 		DBG("Consumer mmap write() ret %zd (len %lu)", ret, len);
-		if (ret < 0) {
+		if (ret < len) {
 			/*
 			 * This is possible if the fd is closed on the other side (outfd)
 			 * or any write problem. It can be verbose a bit for a normal
@@ -1542,7 +1655,7 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
 		struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_stream *stream, unsigned long len,
 		unsigned long padding,
-		struct lttng_packet_index *index)
+		struct ctf_packet_index *index)
 {
 	ssize_t ret = 0, written = 0, ret_splice = 0;
 	loff_t offset = 0;
@@ -1836,60 +1949,6 @@ int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 	}
 }
 
-/*
- * Iterate over all streams of the hashtable and free them properly.
- *
- * WARNING: *MUST* be used with data stream only.
- */
-static void destroy_data_stream_ht(struct lttng_ht *ht)
-{
-	struct lttng_ht_iter iter;
-	struct lttng_consumer_stream *stream;
-
-	if (ht == NULL) {
-		return;
-	}
-
-	rcu_read_lock();
-	cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
-		/*
-		 * Ignore return value since we are currently cleaning up so any error
-		 * can't be handled.
-		 */
-		(void) consumer_del_stream(stream, ht);
-	}
-	rcu_read_unlock();
-
-	lttng_ht_destroy(ht);
-}
-
-/*
- * Iterate over all streams of the hashtable and free them properly.
- *
- * XXX: Should not be only for metadata stream or else use an other name.
- */
-static void destroy_stream_ht(struct lttng_ht *ht)
-{
-	struct lttng_ht_iter iter;
-	struct lttng_consumer_stream *stream;
-
-	if (ht == NULL) {
-		return;
-	}
-
-	rcu_read_lock();
-	cds_lfht_for_each_entry(ht->ht, &iter.iter, stream, node.node) {
-		/*
-		 * Ignore return value since we are currently cleaning up so any error
-		 * can't be handled.
-		 */
-		(void) consumer_del_metadata_stream(stream, ht);
-	}
-	rcu_read_unlock();
-
-	lttng_ht_destroy(ht);
-}
-
 void lttng_consumer_close_metadata(void)
 {
 	switch (consumer_data.type) {
@@ -2196,14 +2255,12 @@ void *consumer_thread_metadata_poll(void *data)
 
 	health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_METADATA);
 
-	health_code_update();
-
-	metadata_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
-	if (!metadata_ht) {
-		/* ENOMEM at this point. Better to bail out. */
-		goto end_ht;
+	if (testpoint(consumerd_thread_metadata)) {
+		goto error_testpoint;
 	}
 
+	health_code_update();
+
 	DBG("Thread metadata poll started");
 
 	/* Size is set to 1 for the consumer_metadata pipe */
@@ -2270,8 +2327,8 @@ restart:
 
 					pipe_len = lttng_pipe_read(ctx->consumer_metadata_pipe,
 							&stream, sizeof(stream));
-					if (pipe_len < 0) {
-						ERR("read metadata stream, ret: %zd", pipe_len);
+					if (pipe_len < sizeof(stream)) {
+						PERROR("read metadata stream");
 						/*
 						 * Continue here to handle the rest of the streams.
 						 */
@@ -2376,8 +2433,7 @@ end:
 
 	lttng_poll_clean(&events);
 end_poll:
-	destroy_stream_ht(metadata_ht);
-end_ht:
+error_testpoint:
 	if (err) {
 		health_error();
 		ERR("Health error occurred in %s", __func__);
@@ -2406,14 +2462,12 @@ void *consumer_thread_data_poll(void *data)
 
 	health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_DATA);
 
-	health_code_update();
-
-	data_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
-	if (data_ht == NULL) {
-		/* ENOMEM at this point. Better to bail out. */
-		goto end;
+	if (testpoint(consumerd_thread_data)) {
+		goto error_testpoint;
 	}
 
+	health_code_update();
+
 	local_stream = zmalloc(sizeof(struct lttng_consumer_stream *));
 	if (local_stream == NULL) {
 		PERROR("local_stream malloc");
@@ -2505,8 +2559,8 @@ void *consumer_thread_data_poll(void *data)
 			DBG("consumer_data_pipe wake up");
 			pipe_readlen = lttng_pipe_read(ctx->consumer_data_pipe,
 					&new_stream, sizeof(new_stream));
-			if (pipe_readlen < 0) {
-				ERR("Consumer data pipe ret %zd", pipe_readlen);
+			if (pipe_readlen < sizeof(new_stream)) {
+				PERROR("Consumer data pipe");
 				/* Continue so we can at least handle the current stream(s). */
 				continue;
 			}
@@ -2643,8 +2697,7 @@ end:
 	 */
 	(void) lttng_pipe_write_close(ctx->consumer_metadata_pipe);
 
-	destroy_data_stream_ht(data_ht);
-
+error_testpoint:
 	if (err) {
 		health_error();
 		ERR("Health error occurred in %s", __func__);
@@ -2745,6 +2798,10 @@ void *consumer_thread_channel_poll(void *data)
 
 	health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_CHANNEL);
 
+	if (testpoint(consumerd_thread_channel)) {
+		goto error_testpoint;
+	}
+
 	health_code_update();
 
 	channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
@@ -2946,6 +3003,7 @@ end:
 end_poll:
 	destroy_channel_ht(channel_ht);
 end_ht:
+error_testpoint:
 	DBG("Channel poll thread exiting");
 	if (err) {
 		health_error();
@@ -3001,6 +3059,10 @@ void *consumer_thread_sessiond_poll(void *data)
 
 	health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_SESSIOND);
 
+	if (testpoint(consumerd_thread_sessiond)) {
+		goto error_testpoint;
+	}
+
 	health_code_update();
 
 	DBG("Creating command socket %s", ctx->consumer_command_sock_path);
@@ -3121,6 +3183,8 @@ end:
 
 	notify_channel_pipe(ctx, NULL, -1, CONSUMER_CHANNEL_QUIT);
 
+	notify_health_quit_pipe(health_quit_pipe);
+
 	/* Cleaning up possibly open sockets. */
 	if (sock >= 0) {
 		ret = close(sock);
@@ -3135,6 +3199,7 @@ end:
 		}
 	}
 
+error_testpoint:
 	if (err) {
 		health_error();
 		ERR("Health error occurred in %s", __func__);
@@ -3196,12 +3261,42 @@ int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
 /*
  * Allocate and set consumer data hash tables.
  */
-void lttng_consumer_init(void)
+int lttng_consumer_init(void)
 {
 	consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+	if (!consumer_data.channel_ht) {
+		goto error;
+	}
+
 	consumer_data.relayd_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+	if (!consumer_data.relayd_ht) {
+		goto error;
+	}
+
 	consumer_data.stream_list_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+	if (!consumer_data.stream_list_ht) {
+		goto error;
+	}
+
 	consumer_data.stream_per_chan_id_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+	if (!consumer_data.stream_per_chan_id_ht) {
+		goto error;
+	}
+
+	data_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+	if (!data_ht) {
+		goto error;
+	}
+
+	metadata_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+	if (!metadata_ht) {
+		goto error;
+	}
+
+	return 0;
+
+error:
+	return -1;
 }
 
 /*
@@ -3217,7 +3312,7 @@ int consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
 		uint64_t relayd_session_id)
 {
 	int fd = -1, ret = -1, relayd_created = 0;
-	enum lttng_error_code ret_code = LTTNG_OK;
+	enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
 	struct consumer_relayd_sock_pair *relayd = NULL;
 
 	assert(ctx);
@@ -3253,7 +3348,7 @@ int consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
 	}
 
 	/* First send a status message before receiving the fds. */
-	ret = consumer_send_status_msg(sock, LTTNG_OK);
+	ret = consumer_send_status_msg(sock, LTTCOMM_CONSUMERD_SUCCESS);
 	if (ret < 0) {
 		/* Somehow, the session daemon is not responding anymore. */
 		lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_FATAL);
@@ -3614,9 +3709,9 @@ int consumer_send_status_channel(int sock,
 	assert(sock >= 0);
 
 	if (!channel) {
-		msg.ret_code = -LTTNG_ERR_UST_CHAN_FAIL;
+		msg.ret_code = LTTCOMM_CONSUMERD_CHANNEL_FAIL;
 	} else {
-		msg.ret_code = LTTNG_OK;
+		msg.ret_code = LTTCOMM_CONSUMERD_SUCCESS;
 		msg.key = channel->key;
 		msg.stream_count = channel->streams.count;
 	}