From: Mathieu Desnoyers 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: https://git.liburcu.org/?a=commitdiff_plain;h=4078b776c9382a540125d810bcd7cca3a8c84bc8;hp=87dc6a9c2c936cf4386043083412c695a914cb36;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 --- 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) {