X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=b1057aaeb139f1fda7681b076a96643f5604ff5f;hp=b605591e478a1c2a40dd13a25228bd2ad80bc684;hb=ca2eb7f43cf00d12f563905d741a6789c3d130ee;hpb=ca4537d385628d3568a918409d909d3042ca7a38

diff --git a/src/common/consumer.c b/src/common/consumer.c
index b605591e4..b1057aaeb 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -19,7 +19,6 @@
 
 #define _GNU_SOURCE
 #include <assert.h>
-#include <fcntl.h>
 #include <poll.h>
 #include <pthread.h>
 #include <stdlib.h>
@@ -86,9 +85,18 @@ static void consumer_steal_stream_key(int key)
 {
 	struct lttng_consumer_stream *stream;
 
+	rcu_read_lock();
 	stream = consumer_find_stream(key);
-	if (stream)
+	if (stream) {
 		stream->key = -1;
+		/*
+		 * We don't want the lookup to match, but we still need
+		 * to iterate on this stream when iterating over the hash table. Just
+		 * change the node key.
+		 */
+		stream->node.key = -1;
+	}
+	rcu_read_unlock();
 }
 
 static struct lttng_consumer_channel *consumer_find_channel(int key)
@@ -119,9 +127,29 @@ static void consumer_steal_channel_key(int key)
 {
 	struct lttng_consumer_channel *channel;
 
+	rcu_read_lock();
 	channel = consumer_find_channel(key);
-	if (channel)
+	if (channel) {
 		channel->key = -1;
+		/*
+		 * We don't want the lookup to match, but we still need
+		 * to iterate on this channel when iterating over the hash table. Just
+		 * change the node key.
+		 */
+		channel->node.key = -1;
+	}
+	rcu_read_unlock();
+}
+
+static
+void consumer_free_stream(struct rcu_head *head)
+{
+	struct lttng_ht_node_ulong *node =
+		caa_container_of(head, struct lttng_ht_node_ulong, head);
+	struct lttng_consumer_stream *stream =
+		caa_container_of(node, struct lttng_consumer_stream, node);
+
+	free(stream);
 }
 
 /*
@@ -156,11 +184,7 @@ void consumer_del_stream(struct lttng_consumer_stream *stream)
 	}
 
 	rcu_read_lock();
-
-	/* Get stream node from hash table */
-	lttng_ht_lookup(consumer_data.stream_ht,
-			(void *)((unsigned long) stream->key), &iter);
-	/* Remove stream node from hash table */
+	iter.iter.node = &stream->node.node;
 	ret = lttng_ht_del(consumer_data.stream_ht, &iter);
 	assert(!ret);
 
@@ -174,17 +198,27 @@ void consumer_del_stream(struct lttng_consumer_stream *stream)
 		goto end;
 	}
 	if (stream->out_fd >= 0) {
-		close(stream->out_fd);
+		ret = close(stream->out_fd);
+		if (ret) {
+			PERROR("close");
+		}
 	}
 	if (stream->wait_fd >= 0 && !stream->wait_fd_is_copy) {
-		close(stream->wait_fd);
+		ret = close(stream->wait_fd);
+		if (ret) {
+			PERROR("close");
+		}
 	}
 	if (stream->shm_fd >= 0 && stream->wait_fd != stream->shm_fd) {
-		close(stream->shm_fd);
+		ret = close(stream->shm_fd);
+		if (ret) {
+			PERROR("close");
+		}
 	}
 	if (!--stream->chan->refcount)
 		free_chan = stream->chan;
-	free(stream);
+
+	call_rcu(&stream->node.head, consumer_free_stream);
 end:
 	consumer_data.need_update = 1;
 	pthread_mutex_unlock(&consumer_data.lock);
@@ -193,16 +227,6 @@ end:
 		consumer_del_channel(free_chan);
 }
 
-static void consumer_del_stream_rcu(struct rcu_head *head)
-{
-	struct lttng_ht_node_ulong *node =
-		caa_container_of(head, struct lttng_ht_node_ulong, head);
-	struct lttng_consumer_stream *stream =
-		caa_container_of(node, struct lttng_consumer_stream, node);
-
-	consumer_del_stream(stream);
-}
-
 struct lttng_consumer_stream *consumer_allocate_stream(
 		int channel_key, int stream_key,
 		int shm_fd, int wait_fd,
@@ -301,6 +325,7 @@ int consumer_add_stream(struct lttng_consumer_stream *stream)
 
 end:
 	pthread_mutex_unlock(&consumer_data.lock);
+
 	return ret;
 }
 
@@ -321,6 +346,17 @@ void consumer_change_stream_state(int stream_key,
 	pthread_mutex_unlock(&consumer_data.lock);
 }
 
+static
+void consumer_free_channel(struct rcu_head *head)
+{
+	struct lttng_ht_node_ulong *node =
+		caa_container_of(head, struct lttng_ht_node_ulong, head);
+	struct lttng_consumer_channel *channel =
+		caa_container_of(node, struct lttng_consumer_channel, node);
+
+	free(channel);
+}
+
 /*
  * Remove a channel from the global list protected by a mutex. This
  * function is also responsible for freeing its data structures.
@@ -346,12 +382,9 @@ void consumer_del_channel(struct lttng_consumer_channel *channel)
 	}
 
 	rcu_read_lock();
-
-	lttng_ht_lookup(consumer_data.channel_ht,
-			(void *)((unsigned long) channel->key), &iter);
+	iter.iter.node = &channel->node.node;
 	ret = lttng_ht_del(consumer_data.channel_ht, &iter);
 	assert(!ret);
-
 	rcu_read_unlock();
 
 	if (channel->mmap_base != NULL) {
@@ -361,26 +394,23 @@ void consumer_del_channel(struct lttng_consumer_channel *channel)
 		}
 	}
 	if (channel->wait_fd >= 0 && !channel->wait_fd_is_copy) {
-		close(channel->wait_fd);
+		ret = close(channel->wait_fd);
+		if (ret) {
+			PERROR("close");
+		}
 	}
 	if (channel->shm_fd >= 0 && channel->wait_fd != channel->shm_fd) {
-		close(channel->shm_fd);
+		ret = close(channel->shm_fd);
+		if (ret) {
+			PERROR("close");
+		}
 	}
-	free(channel);
+
+	call_rcu(&channel->node.head, consumer_free_channel);
 end:
 	pthread_mutex_unlock(&consumer_data.lock);
 }
 
-static void consumer_del_channel_rcu(struct rcu_head *head)
-{
-	struct lttng_ht_node_ulong *node =
-		caa_container_of(head, struct lttng_ht_node_ulong, head);
-	struct lttng_consumer_channel *channel=
-		caa_container_of(node, struct lttng_consumer_channel, node);
-
-	consumer_del_channel(channel);
-}
-
 struct lttng_consumer_channel *consumer_allocate_channel(
 		int channel_key,
 		int shm_fd, int wait_fd,
@@ -444,6 +474,7 @@ int consumer_add_channel(struct lttng_consumer_channel *channel)
 	lttng_ht_add_unique_ulong(consumer_data.channel_ht, &channel->node);
 	rcu_read_unlock();
 	pthread_mutex_unlock(&consumer_data.lock);
+
 	return 0;
 }
 
@@ -463,6 +494,7 @@ int consumer_update_poll_array(
 	struct lttng_consumer_stream *stream;
 
 	DBG("Updating poll fd array");
+	rcu_read_lock();
 	cds_lfht_for_each_entry(consumer_data.stream_ht->ht, &iter.iter, stream,
 			node.node) {
 		if (stream->state != LTTNG_CONSUMER_ACTIVE_STREAM) {
@@ -474,6 +506,7 @@ int consumer_update_poll_array(
 		local_stream[i] = stream;
 		i++;
 	}
+	rcu_read_unlock();
 
 	/*
 	 * Insert the consumer_poll_pipe at the end of the array and don't
@@ -492,8 +525,15 @@ int lttng_consumer_poll_socket(struct pollfd *consumer_sockpoll)
 {
 	int num_rdy;
 
+restart:
 	num_rdy = poll(consumer_sockpoll, 2, -1);
 	if (num_rdy == -1) {
+		/*
+		 * Restart interrupted system call.
+		 */
+		if (errno == EINTR) {
+			goto restart;
+		}
 		perror("Poll error");
 		goto exit;
 	}
@@ -547,7 +587,6 @@ int lttng_consumer_send_error(
  */
 void lttng_consumer_cleanup(void)
 {
-	int ret;
 	struct lttng_ht_iter iter;
 	struct lttng_ht_node_ulong *node;
 
@@ -559,16 +598,16 @@ void lttng_consumer_cleanup(void)
 	 */
 	cds_lfht_for_each_entry(consumer_data.stream_ht->ht, &iter.iter, node,
 			node) {
-		ret = lttng_ht_del(consumer_data.stream_ht, &iter);
-		assert(!ret);
-		call_rcu(&node->head, consumer_del_stream_rcu);
+		struct lttng_consumer_stream *stream =
+			caa_container_of(node, struct lttng_consumer_stream, node);
+		consumer_del_stream(stream);
 	}
 
 	cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter, node,
 			node) {
-		ret = lttng_ht_del(consumer_data.channel_ht, &iter);
-		assert(!ret);
-		call_rcu(&node->head, consumer_del_channel_rcu);
+		struct lttng_consumer_channel *channel =
+			caa_container_of(node, struct lttng_consumer_channel, node);
+		consumer_del_channel(channel);
 	}
 
 	rcu_read_unlock();
@@ -601,7 +640,7 @@ void lttng_consumer_sync_trace_file(
 	if (orig_offset < stream->chan->max_sb_size) {
 		return;
 	}
-	sync_file_range(outfd, orig_offset - stream->chan->max_sb_size,
+	lttng_sync_file_range(outfd, orig_offset - stream->chan->max_sb_size,
 			stream->chan->max_sb_size,
 			SYNC_FILE_RANGE_WAIT_BEFORE
 			| SYNC_FILE_RANGE_WRITE
@@ -640,7 +679,7 @@ void lttng_consumer_sync_trace_file(
  */
 struct lttng_consumer_local_data *lttng_consumer_create(
 		enum lttng_consumer_type type,
-		int (*buffer_ready)(struct lttng_consumer_stream *stream,
+		ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream,
 			struct lttng_consumer_local_data *ctx),
 		int (*recv_channel)(struct lttng_consumer_channel *channel),
 		int (*recv_stream)(struct lttng_consumer_stream *stream),
@@ -692,14 +731,18 @@ error_thread_pipe:
 		int err;
 
 		err = close(ctx->consumer_should_quit[i]);
-		assert(!err);
+		if (err) {
+			PERROR("close");
+		}
 	}
 error_quit_pipe:
 	for (i = 0; i < 2; i++) {
 		int err;
 
 		err = close(ctx->consumer_poll_pipe[i]);
-		assert(!err);
+		if (err) {
+			PERROR("close");
+		}
 	}
 error_poll_pipe:
 	free(ctx);
@@ -712,13 +755,36 @@ error:
  */
 void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
 {
-	close(ctx->consumer_error_socket);
-	close(ctx->consumer_thread_pipe[0]);
-	close(ctx->consumer_thread_pipe[1]);
-	close(ctx->consumer_poll_pipe[0]);
-	close(ctx->consumer_poll_pipe[1]);
-	close(ctx->consumer_should_quit[0]);
-	close(ctx->consumer_should_quit[1]);
+	int ret;
+
+	ret = close(ctx->consumer_error_socket);
+	if (ret) {
+		PERROR("close");
+	}
+	ret = close(ctx->consumer_thread_pipe[0]);
+	if (ret) {
+		PERROR("close");
+	}
+	ret = close(ctx->consumer_thread_pipe[1]);
+	if (ret) {
+		PERROR("close");
+	}
+	ret = close(ctx->consumer_poll_pipe[0]);
+	if (ret) {
+		PERROR("close");
+	}
+	ret = close(ctx->consumer_poll_pipe[1]);
+	if (ret) {
+		PERROR("close");
+	}
+	ret = close(ctx->consumer_should_quit[0]);
+	if (ret) {
+		PERROR("close");
+	}
+	ret = close(ctx->consumer_should_quit[1]);
+	if (ret) {
+		PERROR("close");
+	}
 	unlink(ctx->consumer_command_sock_path);
 	free(ctx);
 }
@@ -728,7 +794,7 @@ void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
  *
  * Returns the number of bytes written
  */
-int lttng_consumer_on_read_subbuffer_mmap(
+ssize_t lttng_consumer_on_read_subbuffer_mmap(
 		struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_stream *stream, unsigned long len)
 {
@@ -742,6 +808,8 @@ int lttng_consumer_on_read_subbuffer_mmap(
 		ERR("Unknown consumer_data type");
 		assert(0);
 	}
+
+	return 0;
 }
 
 /*
@@ -749,7 +817,7 @@ int lttng_consumer_on_read_subbuffer_mmap(
  *
  * Returns the number of bytes spliced.
  */
-int lttng_consumer_on_read_subbuffer_splice(
+ssize_t lttng_consumer_on_read_subbuffer_splice(
 		struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_stream *stream, unsigned long len)
 {
@@ -895,11 +963,22 @@ void *lttng_consumer_thread_poll_fds(void *data)
 		}
 		pthread_mutex_unlock(&consumer_data.lock);
 
+		/* No FDs and consumer_quit, consumer_cleanup the thread */
+		if (nb_fd == 0 && consumer_quit == 1) {
+			goto end;
+		}
 		/* poll on the array of fds */
+	restart:
 		DBG("polling on %d fd", nb_fd + 1);
 		num_rdy = poll(pollfd, nb_fd + 1, consumer_poll_timeout);
 		DBG("poll num_rdy : %d", num_rdy);
 		if (num_rdy == -1) {
+			/*
+			 * Restart interrupted system call.
+			 */
+			if (errno == EINTR) {
+				goto restart;
+			}
 			perror("Poll error");
 			lttng_consumer_send_error(ctx, CONSUMERD_POLL_ERROR);
 			goto end;
@@ -908,11 +987,6 @@ void *lttng_consumer_thread_poll_fds(void *data)
 			goto end;
 		}
 
-		/* No FDs and consumer_quit, consumer_cleanup the thread */
-		if (nb_fd == 0 && consumer_quit == 1) {
-			goto end;
-		}
-
 		/*
 		 * If the consumer_poll_pipe triggered poll go
 		 * directly to the beginning of the loop to update the
@@ -931,69 +1005,82 @@ void *lttng_consumer_thread_poll_fds(void *data)
 		/* Take care of high priority channels first. */
 		for (i = 0; i < nb_fd; i++) {
 			if (pollfd[i].revents & POLLPRI) {
+				ssize_t len;
+
 				DBG("Urgent read on fd %d", pollfd[i].fd);
 				high_prio = 1;
-				ret = ctx->on_buffer_ready(local_stream[i], ctx);
+				len = ctx->on_buffer_ready(local_stream[i], ctx);
 				/* it's ok to have an unavailable sub-buffer */
-				if (ret == EAGAIN) {
-					ret = 0;
+				if (len < 0 && len != -EAGAIN) {
+					goto end;
+				} else if (len > 0) {
+					local_stream[i]->data_read = 1;
 				}
-			} else if (pollfd[i].revents & POLLERR) {
-				ERR("Error returned in polling fd %d.", pollfd[i].fd);
-				rcu_read_lock();
-				consumer_del_stream_rcu(&local_stream[i]->node.head);
-				rcu_read_unlock();
-				num_hup++;
-			} else if (pollfd[i].revents & POLLNVAL) {
-				ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
-				rcu_read_lock();
-				consumer_del_stream_rcu(&local_stream[i]->node.head);
-				rcu_read_unlock();
-				num_hup++;
-			} else if ((pollfd[i].revents & POLLHUP) &&
-					!(pollfd[i].revents & POLLIN)) {
-				if (consumer_data.type == LTTNG_CONSUMER32_UST
-						|| consumer_data.type == LTTNG_CONSUMER64_UST) {
-					DBG("Polling fd %d tells it has hung up. Attempting flush and read.",
-						pollfd[i].fd);
-					if (!local_stream[i]->hangup_flush_done) {
-						lttng_ustconsumer_on_stream_hangup(local_stream[i]);
-						/* read after flush */
-						do {
-							ret = ctx->on_buffer_ready(local_stream[i], ctx);
-						} while (ret == EAGAIN);
-					}
-				} else {
-					DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
-				}
-				rcu_read_lock();
-				consumer_del_stream_rcu(&local_stream[i]->node.head);
-				rcu_read_unlock();
-				num_hup++;
 			}
 		}
 
-		/* If every buffer FD has hung up, we end the read loop here */
-		if (nb_fd > 0 && num_hup == nb_fd) {
-			DBG("every buffer FD has hung up\n");
-			if (consumer_quit == 1) {
-				goto end;
-			}
+		/*
+		 * If we read high prio channel in this loop, try again
+		 * for more high prio data.
+		 */
+		if (high_prio) {
 			continue;
 		}
 
 		/* Take care of low priority channels. */
-		if (high_prio == 0) {
-			for (i = 0; i < nb_fd; i++) {
-				if (pollfd[i].revents & POLLIN) {
-					DBG("Normal read on fd %d", pollfd[i].fd);
-					ret = ctx->on_buffer_ready(local_stream[i], ctx);
-					/* it's ok to have an unavailable subbuffer */
-					if (ret == EAGAIN) {
-						ret = 0;
-					}
+		for (i = 0; i < nb_fd; i++) {
+			if ((pollfd[i].revents & POLLIN) ||
+					local_stream[i]->hangup_flush_done) {
+				ssize_t len;
+
+				DBG("Normal read on fd %d", pollfd[i].fd);
+				len = ctx->on_buffer_ready(local_stream[i], ctx);
+				/* it's ok to have an unavailable sub-buffer */
+				if (len < 0 && len != -EAGAIN) {
+					goto end;
+				} else if (len > 0) {
+					local_stream[i]->data_read = 1;
+				}
+			}
+		}
+
+		/* Handle hangup and errors */
+		for (i = 0; i < nb_fd; i++) {
+			if (!local_stream[i]->hangup_flush_done
+					&& (pollfd[i].revents & (POLLHUP | POLLERR | POLLNVAL))
+					&& (consumer_data.type == LTTNG_CONSUMER32_UST
+						|| consumer_data.type == LTTNG_CONSUMER64_UST)) {
+				DBG("fd %d is hup|err|nval. Attempting flush and read.",
+					pollfd[i].fd);
+				lttng_ustconsumer_on_stream_hangup(local_stream[i]);
+				/* Attempt read again, for the data we just flushed. */
+				local_stream[i]->data_read = 1;
+			}
+			/*
+			 * If the poll flag is HUP/ERR/NVAL and we have
+			 * read no data in this pass, we can remove the
+			 * stream from its hash table.
+			 */
+			if ((pollfd[i].revents & POLLHUP)) {
+				DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
+				if (!local_stream[i]->data_read) {
+					consumer_del_stream(local_stream[i]);
+					num_hup++;
+				}
+			} else if (pollfd[i].revents & POLLERR) {
+				ERR("Error returned in polling fd %d.", pollfd[i].fd);
+				if (!local_stream[i]->data_read) {
+					consumer_del_stream(local_stream[i]);
+					num_hup++;
+				}
+			} else if (pollfd[i].revents & POLLNVAL) {
+				ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
+				if (!local_stream[i]->data_read) {
+					consumer_del_stream(local_stream[i]);
+					num_hup++;
 				}
 			}
+			local_stream[i]->data_read = 0;
 		}
 	}
 end:
@@ -1125,7 +1212,7 @@ end:
 	return NULL;
 }
 
-int lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
+ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
 		struct lttng_consumer_local_data *ctx)
 {
 	switch (consumer_data.type) {