From: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
Date: Thu, 1 Mar 2012 03:30:17 +0000 (-0500)
Subject: Fix: make ust consumer posix compliant for poll flags
X-Git-Tag: v2.0.0-rc3~23
X-Git-Url: http://git.lttng.org/?a=commitdiff_plain;h=4078b776c9382a540125d810bcd7cca3a8c84bc8;p=lttng-tools.git

Fix: make ust consumer posix compliant for poll flags

poll flags can have both POLLHUP and POLLIN set, which is not expected
by the consumer. Do not depend on this implementation-specific behavior.

Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
---

diff --git a/src/common/consumer.c b/src/common/consumer.c
index 2e5ec5c35..ae59b6b60 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -646,7 +646,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),
@@ -734,7 +734,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)
 {
@@ -757,7 +757,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)
 {
@@ -903,6 +903,10 @@ 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);
@@ -923,11 +927,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
@@ -946,69 +945,90 @@ 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;
+
+				assert(!(pollfd[i].revents & POLLERR));
+				assert(!(pollfd[i].revents & POLLNVAL));
+				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) {
+					rcu_read_lock();
+					consumer_del_stream_rcu(&local_stream[i]->node.head);
+					rcu_read_unlock();
+					num_hup++;
+				}
+			} else if (pollfd[i].revents & POLLERR) {
+				ERR("Error returned in polling fd %d.", pollfd[i].fd);
+				if (!local_stream[i]->data_read) {
+					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);
+				if (!local_stream[i]->data_read) {
+					rcu_read_lock();
+					consumer_del_stream_rcu(&local_stream[i]->node.head);
+					rcu_read_unlock();
+					num_hup++;
 				}
 			}
+			local_stream[i]->data_read = 0;
 		}
 	}
 end:
@@ -1140,7 +1160,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) {
diff --git a/src/common/consumer.h b/src/common/consumer.h
index 35a72b50f..71ae39903 100644
--- a/src/common/consumer.h
+++ b/src/common/consumer.h
@@ -114,6 +114,7 @@ struct lttng_consumer_stream {
 	/* For UST */
 	struct lttng_ust_lib_ring_buffer *buf;
 	int cpu;
+	int data_read;
 	int hangup_flush_done;
 	/* UID/GID of the user owning the session to which stream belongs */
 	uid_t uid;
@@ -125,8 +126,11 @@ struct lttng_consumer_stream {
  * process.
  */
 struct lttng_consumer_local_data {
-	/* function to call when data is available on a buffer */
-	int (*on_buffer_ready)(struct lttng_consumer_stream *stream,
+	/*
+	 * Function to call when data is available on a buffer.
+	 * Returns the number of bytes read, or negative error value.
+	 */
+	ssize_t (*on_buffer_ready)(struct lttng_consumer_stream *stream,
 			struct lttng_consumer_local_data *ctx);
 	/*
 	 * function to call when we receive a new channel, it receives a
@@ -285,16 +289,16 @@ int consumer_add_channel(struct lttng_consumer_channel *channel);
 
 extern 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),
 		int (*update_stream)(int sessiond_key, uint32_t state));
 extern void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx);
-extern int lttng_consumer_on_read_subbuffer_mmap(
+extern ssize_t lttng_consumer_on_read_subbuffer_mmap(
 		struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_stream *stream, unsigned long len);
-extern int lttng_consumer_on_read_subbuffer_splice(
+extern ssize_t lttng_consumer_on_read_subbuffer_splice(
 		struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_stream *stream, unsigned long len);
 extern int lttng_consumer_take_snapshot(struct lttng_consumer_local_data *ctx,
@@ -308,7 +312,7 @@ extern void *lttng_consumer_thread_receive_fds(void *data);
 extern int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		int sock, struct pollfd *consumer_sockpoll);
 
-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);
 int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream);
 
diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c
index c1ba1405e..608265f96 100644
--- a/src/common/kernel-consumer/kernel-consumer.c
+++ b/src/common/kernel-consumer/kernel-consumer.c
@@ -45,12 +45,12 @@ extern volatile int consumer_quit;
  *
  * Returns the number of bytes written
  */
-int lttng_kconsumer_on_read_subbuffer_mmap(
+ssize_t lttng_kconsumer_on_read_subbuffer_mmap(
 		struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_stream *stream, unsigned long len)
 {
 	unsigned long mmap_offset;
-	long ret = 0;
+	ssize_t ret = 0;
 	off_t orig_offset = stream->out_fd_offset;
 	int fd = stream->wait_fd;
 	int outfd = stream->out_fd;
@@ -91,11 +91,11 @@ end:
  *
  * Returns the number of bytes spliced.
  */
-int lttng_kconsumer_on_read_subbuffer_splice(
+ssize_t lttng_kconsumer_on_read_subbuffer_splice(
 		struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_stream *stream, unsigned long len)
 {
-	long ret = 0;
+	ssize_t ret = 0;
 	loff_t offset = 0;
 	off_t orig_offset = stream->out_fd_offset;
 	int fd = stream->wait_fd;
@@ -308,12 +308,12 @@ end_nosignal:
 /*
  * Consume data on a file descriptor and write it on a trace file.
  */
-int lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
+ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
 		struct lttng_consumer_local_data *ctx)
 {
 	unsigned long len;
 	int err;
-	long ret = 0;
+	ssize_t ret = 0;
 	int infd = stream->wait_fd;
 
 	DBG("In read_subbuffer (infd : %d)", infd);
diff --git a/src/common/kernel-consumer/kernel-consumer.h b/src/common/kernel-consumer/kernel-consumer.h
index 6e820044b..f4f46169d 100644
--- a/src/common/kernel-consumer/kernel-consumer.h
+++ b/src/common/kernel-consumer/kernel-consumer.h
@@ -25,18 +25,19 @@
 /*
  * Mmap the ring buffer, read it and write the data to the tracefile.
  *
- * Returns the number of bytes written.
+ * Returns the number of bytes written, or negative value on error.
  */
-extern int lttng_kconsumer_on_read_subbuffer_mmap(
+extern ssize_t lttng_kconsumer_on_read_subbuffer_mmap(
 		struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_stream *stream, unsigned long len);
 
 /*
  * Splice the data from the ring buffer to the tracefile.
  *
- * Returns the number of bytes spliced.
+ * Returns the number of bytes spliced, or negative error value on
+ * error.
  */
-extern int lttng_kconsumer_on_read_subbuffer_splice(
+extern ssize_t lttng_kconsumer_on_read_subbuffer_splice(
 		struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_stream *stream, unsigned long len);
 
@@ -62,7 +63,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		int sock, struct pollfd *consumer_sockpoll);
 
 
-int lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
+ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
 		struct lttng_consumer_local_data *ctx);
 int lttng_kconsumer_on_recv_stream(struct lttng_consumer_stream *stream);
 
diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
index 113682384..c8ba08460 100644
--- a/src/common/ust-consumer/ust-consumer.c
+++ b/src/common/ust-consumer/ust-consumer.c
@@ -43,9 +43,9 @@ extern volatile int consumer_quit;
 /*
  * Mmap the ring buffer, read it and write the data to the tracefile.
  *
- * Returns the number of bytes written
+ * Returns the number of bytes written, else negative value on error.
  */
-int lttng_ustconsumer_on_read_subbuffer_mmap(
+ssize_t lttng_ustconsumer_on_read_subbuffer_mmap(
 		struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_stream *stream, unsigned long len)
 {
@@ -90,7 +90,7 @@ end:
  *
  * Returns the number of bytes spliced.
  */
-int lttng_ustconsumer_on_read_subbuffer_splice(
+ssize_t lttng_ustconsumer_on_read_subbuffer_splice(
 		struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_stream *stream, unsigned long len)
 {
diff --git a/src/common/ust-consumer/ust-consumer.h b/src/common/ust-consumer/ust-consumer.h
index 76b4eeece..0089ddf17 100644
--- a/src/common/ust-consumer/ust-consumer.h
+++ b/src/common/ust-consumer/ust-consumer.h
@@ -30,14 +30,14 @@
 /*
  * Mmap the ring buffer, read it and write the data to the tracefile.
  *
- * Returns the number of bytes written.
+ * Returns the number of bytes written, else negative value on error.
  */
-extern int lttng_ustconsumer_on_read_subbuffer_mmap(
+extern ssize_t lttng_ustconsumer_on_read_subbuffer_mmap(
 		struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_stream *stream, unsigned long len);
 
 /* Not implemented */
-extern int lttng_ustconsumer_on_read_subbuffer_splice(
+extern ssize_t lttng_ustconsumer_on_read_subbuffer_splice(
 		struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_stream *stream, unsigned long len);
 
@@ -76,7 +76,7 @@ void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream);
 #else /* HAVE_LIBLTTNG_UST_CTL */
 
 static inline
-int lttng_ustconsumer_on_read_subbuffer_mmap(
+ssize_t lttng_ustconsumer_on_read_subbuffer_mmap(
 		struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_stream *stream, unsigned long len)
 {
@@ -84,7 +84,7 @@ int lttng_ustconsumer_on_read_subbuffer_mmap(
 }
 
 static inline
-int lttng_ustconsumer_on_read_subbuffer_splice(
+ssize_t lttng_ustconsumer_on_read_subbuffer_splice(
 		struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_stream *uststream, unsigned long len)
 {