X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=dd8c621f177d56968306fb500ce4546870b5bd2d;hp=05bf85b3dc8d8e73e0bbe62497a64bf0f0bc3e26;hb=5c827ce0dce140b121032837510f89cb70d1650d;hpb=a186a15913c34e8adc83ea71565d3b0eec296774 diff --git a/src/common/consumer.c b/src/common/consumer.c index 05bf85b3d..dd8c621f1 100644 --- a/src/common/consumer.c +++ b/src/common/consumer.c @@ -2,19 +2,18 @@ * Copyright (C) 2011 - Julien Desfossez * Mathieu Desnoyers * - * This program is free software; you can redistribute it and/or - * modify it under the terms of the GNU General Public License - * as published by the Free Software Foundation; only version 2 - * of the License. + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License, version 2 only, + * as published by the Free Software Foundation. * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU General Public License for more details. + * This program is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for + * more details. * - * You should have received a copy of the GNU General Public License - * along with this program; if not, write to the Free Software - * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. + * You should have received a copy of the GNU General Public License along + * with this program; if not, write to the Free Software Foundation, Inc., + * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ #define _GNU_SOURCE @@ -85,9 +84,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) @@ -118,9 +126,18 @@ 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 @@ -166,15 +183,9 @@ 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. It can fail if it's been - * replaced due to key reuse. - */ - (void) lttng_ht_del(consumer_data.stream_ht, &iter); + iter.iter.node = &stream->node.node; + ret = lttng_ht_del(consumer_data.stream_ht, &iter); + assert(!ret); rcu_read_unlock(); @@ -287,17 +298,24 @@ end: int consumer_add_stream(struct lttng_consumer_stream *stream) { int ret = 0; + struct lttng_ht_node_ulong *node; + struct lttng_ht_iter iter; pthread_mutex_lock(&consumer_data.lock); /* Steal stream identifier, for UST */ consumer_steal_stream_key(stream->key); rcu_read_lock(); - /* - * We simply remove the old channel from the hash table. It's - * ok, since we know for sure the sessiond wants to replace it - * with the new version, because the key has been reused. - */ - (void) lttng_ht_add_replace_ulong(consumer_data.stream_ht, &stream->node); + + lttng_ht_lookup(consumer_data.stream_ht, + (void *)((unsigned long) stream->key), &iter); + node = lttng_ht_iter_get_node_ulong(&iter); + if (node != NULL) { + rcu_read_unlock(); + /* Stream already exist. Ignore the insertion */ + goto end; + } + + lttng_ht_add_unique_ulong(consumer_data.stream_ht, &stream->node); rcu_read_unlock(); consumer_data.stream_count++; consumer_data.need_update = 1; @@ -375,16 +393,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); - - /* - * Remove channel node from hash table. It can fail if it's been - * replaced due to key reuse. - */ - (void) lttng_ht_del(consumer_data.channel_ht, &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) { @@ -467,16 +478,25 @@ end: */ int consumer_add_channel(struct lttng_consumer_channel *channel) { + struct lttng_ht_node_ulong *node; + struct lttng_ht_iter iter; + pthread_mutex_lock(&consumer_data.lock); /* Steal channel identifier, for UST */ consumer_steal_channel_key(channel->key); rcu_read_lock(); - /* - * We simply remove the old channel from the hash table. It's - * ok, since we know for sure the sessiond wants to replace it - * with the new version, because the key has been reused. - */ - (void) lttng_ht_add_replace_ulong(consumer_data.channel_ht, &channel->node); + + lttng_ht_lookup(consumer_data.channel_ht, + (void *)((unsigned long) channel->key), &iter); + node = lttng_ht_iter_get_node_ulong(&iter); + if (node != NULL) { + /* Channel already exist. Ignore the insertion */ + goto end; + } + + lttng_ht_add_unique_ulong(consumer_data.channel_ht, &channel->node); + +end: rcu_read_unlock(); pthread_mutex_unlock(&consumer_data.lock); @@ -499,6 +519,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) { @@ -510,13 +531,14 @@ 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 * increment i so nb_fd is the number of real FD. */ (*pollfd)[i].fd = ctx->consumer_poll_pipe[0]; - (*pollfd)[i].events = POLLIN; + (*pollfd)[i].events = POLLIN | POLLPRI; return i; } @@ -540,7 +562,7 @@ restart: perror("Poll error"); goto exit; } - if (consumer_sockpoll[0].revents == POLLIN) { + if (consumer_sockpoll[0].revents & (POLLIN | POLLPRI)) { DBG("consumer_should_quit wake up"); goto exit; } @@ -614,6 +636,9 @@ void lttng_consumer_cleanup(void) } rcu_read_unlock(); + + lttng_ht_destroy(consumer_data.stream_ht); + lttng_ht_destroy(consumer_data.channel_ht); } /* @@ -714,6 +739,20 @@ struct lttng_consumer_local_data *lttng_consumer_create( goto error_poll_pipe; } + /* set read end of the pipe to non-blocking */ + ret = fcntl(ctx->consumer_poll_pipe[0], F_SETFL, O_NONBLOCK); + if (ret < 0) { + perror("fcntl O_NONBLOCK"); + goto error_poll_fcntl; + } + + /* set write end of the pipe to non-blocking */ + ret = fcntl(ctx->consumer_poll_pipe[1], F_SETFL, O_NONBLOCK); + if (ret < 0) { + perror("fcntl O_NONBLOCK"); + goto error_poll_fcntl; + } + ret = pipe(ctx->consumer_should_quit); if (ret < 0) { perror("Error creating recv pipe"); @@ -738,6 +777,7 @@ error_thread_pipe: PERROR("close"); } } +error_poll_fcntl: error_quit_pipe: for (i = 0; i < 2; i++) { int err; @@ -911,8 +951,6 @@ void *lttng_consumer_thread_poll_fds(void *data) struct lttng_consumer_stream **local_stream = NULL; /* local view of consumer_data.fds_count */ int nb_fd = 0; - char tmp; - int tmp2; struct lttng_consumer_local_data *ctx = data; rcu_register_thread(); @@ -996,12 +1034,15 @@ void *lttng_consumer_thread_poll_fds(void *data) * array. We want to prioritize array update over * low-priority reads. */ - if (pollfd[nb_fd].revents & POLLIN) { + if (pollfd[nb_fd].revents & (POLLIN | POLLPRI)) { + size_t pipe_readlen; + char tmp; + DBG("consumer_poll_pipe wake up"); - tmp2 = read(ctx->consumer_poll_pipe[0], &tmp, 1); - if (tmp2 < 0) { - perror("read consumer poll"); - } + /* Consume 1 byte of pipe data */ + do { + pipe_readlen = read(ctx->consumer_poll_pipe[0], &tmp, 1); + } while (pipe_readlen == -1 && errno == EINTR); continue; } @@ -1036,8 +1077,6 @@ void *lttng_consumer_thread_poll_fds(void *data) 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 */ @@ -1208,11 +1247,20 @@ end: */ consumer_poll_timeout = LTTNG_CONSUMER_POLL_TIMEOUT; - /* wake up the polling thread */ - ret = write(ctx->consumer_poll_pipe[1], "4", 1); - if (ret < 0) { - perror("poll pipe write"); - } + /* + * Wake-up the other end by writing a null byte in the pipe + * (non-blocking). Important note: Because writing into the + * pipe is non-blocking (and therefore we allow dropping wakeup + * data, as long as there is wakeup data present in the pipe + * buffer to wake up the other end), the other end should + * perform the following sequence for waiting: + * 1) empty the pipe (reads). + * 2) perform update operation. + * 3) wait on the pipe (poll). + */ + do { + ret = write(ctx->consumer_poll_pipe[1], "", 1); + } while (ret == -1UL && errno == EINTR); rcu_unregister_thread(); return NULL; }