X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer-timer.c;h=5764b13339cafd0366d68e7120519c4333ef3057;hp=d36e2bccfdd08cb304d2d362403325db85c4c3af;hb=ea26306076d26c40ed31e4f9170dc2852bda502f;hpb=528f2ffaebbc88b3fd541fa404b567b878aa5255 diff --git a/src/common/consumer-timer.c b/src/common/consumer-timer.c index d36e2bccf..5764b1333 100644 --- a/src/common/consumer-timer.c +++ b/src/common/consumer-timer.c @@ -16,7 +16,7 @@ * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ -#define _GNU_SOURCE +#define _LGPL_SOURCE #include #include #include @@ -27,7 +27,6 @@ #include #include #include -#include #include "consumer-timer.h" #include "consumer-testpoint.h" @@ -133,78 +132,103 @@ error: return ret; } -static int check_kernel_stream(struct lttng_consumer_stream *stream) +int consumer_flush_kernel_index(struct lttng_consumer_stream *stream) { uint64_t ts, stream_id; int ret; - /* - * While holding the stream mutex, try to take a snapshot, if it - * succeeds, it means that data is ready to be sent, just let the data - * thread handle that. Otherwise, if the snapshot returns EAGAIN, it - * means that there is no data to read after the flush, so we can - * safely send the empty index. - */ - pthread_mutex_lock(&stream->lock); ret = kernctl_get_current_timestamp(stream->wait_fd, &ts); if (ret < 0) { ERR("Failed to get the current timestamp"); - goto error_unlock; + goto end; } ret = kernctl_buffer_flush(stream->wait_fd); if (ret < 0) { ERR("Failed to flush kernel stream"); - goto error_unlock; + goto end; } ret = kernctl_snapshot(stream->wait_fd); if (ret < 0) { if (errno != EAGAIN && errno != ENODATA) { PERROR("live timer kernel snapshot"); ret = -1; - goto error_unlock; + goto end; } ret = kernctl_get_stream_id(stream->wait_fd, &stream_id); if (ret < 0) { PERROR("kernctl_get_stream_id"); - goto error_unlock; + goto end; } DBG("Stream %" PRIu64 " empty, sending beacon", stream->key); ret = send_empty_index(stream, ts, stream_id); if (ret < 0) { - goto error_unlock; + goto end; } } ret = 0; - -error_unlock: - pthread_mutex_unlock(&stream->lock); +end: return ret; } -static int check_ust_stream(struct lttng_consumer_stream *stream) +static int check_kernel_stream(struct lttng_consumer_stream *stream) { - uint64_t ts, stream_id; int ret; - assert(stream); - assert(stream->ustream); /* * While holding the stream mutex, try to take a snapshot, if it * succeeds, it means that data is ready to be sent, just let the data * thread handle that. Otherwise, if the snapshot returns EAGAIN, it * means that there is no data to read after the flush, so we can * safely send the empty index. + * + * Doing a trylock and checking if waiting on metadata if + * trylock fails. Bail out of the stream is indeed waiting for + * metadata to be pushed. Busy wait on trylock otherwise. */ - pthread_mutex_lock(&stream->lock); + for (;;) { + ret = pthread_mutex_trylock(&stream->lock); + switch (ret) { + case 0: + break; /* We have the lock. */ + case EBUSY: + pthread_mutex_lock(&stream->metadata_timer_lock); + if (stream->waiting_on_metadata) { + ret = 0; + stream->missed_metadata_flush = true; + pthread_mutex_unlock(&stream->metadata_timer_lock); + goto end; /* Bail out. */ + } + pthread_mutex_unlock(&stream->metadata_timer_lock); + /* Try again. */ + caa_cpu_relax(); + continue; + default: + ERR("Unexpected pthread_mutex_trylock error %d", ret); + ret = -1; + goto end; + } + break; + } + ret = consumer_flush_kernel_index(stream); + pthread_mutex_unlock(&stream->lock); +end: + return ret; +} + +int consumer_flush_ust_index(struct lttng_consumer_stream *stream) +{ + uint64_t ts, stream_id; + int ret; + ret = cds_lfht_is_node_deleted(&stream->node.node); if (ret) { - goto error_unlock; + goto end; } ret = lttng_ustconsumer_get_current_timestamp(stream, &ts); if (ret < 0) { ERR("Failed to get the current timestamp"); - goto error_unlock; + goto end; } lttng_ustconsumer_flush_buffer(stream, 1); ret = lttng_ustconsumer_take_snapshot(stream); @@ -212,23 +236,68 @@ static int check_ust_stream(struct lttng_consumer_stream *stream) if (ret != -EAGAIN) { ERR("Taking UST snapshot"); ret = -1; - goto error_unlock; + goto end; } - ret = ustctl_get_stream_id(stream->ustream, &stream_id); + ret = lttng_ustconsumer_get_stream_id(stream, &stream_id); if (ret < 0) { PERROR("ustctl_get_stream_id"); - goto error_unlock; + goto end; } DBG("Stream %" PRIu64 " empty, sending beacon", stream->key); ret = send_empty_index(stream, ts, stream_id); if (ret < 0) { - goto error_unlock; + goto end; } } ret = 0; +end: + return ret; +} -error_unlock: +static int check_ust_stream(struct lttng_consumer_stream *stream) +{ + int ret; + + assert(stream); + assert(stream->ustream); + /* + * While holding the stream mutex, try to take a snapshot, if it + * succeeds, it means that data is ready to be sent, just let the data + * thread handle that. Otherwise, if the snapshot returns EAGAIN, it + * means that there is no data to read after the flush, so we can + * safely send the empty index. + * + * Doing a trylock and checking if waiting on metadata if + * trylock fails. Bail out of the stream is indeed waiting for + * metadata to be pushed. Busy wait on trylock otherwise. + */ + for (;;) { + ret = pthread_mutex_trylock(&stream->lock); + switch (ret) { + case 0: + break; /* We have the lock. */ + case EBUSY: + pthread_mutex_lock(&stream->metadata_timer_lock); + if (stream->waiting_on_metadata) { + ret = 0; + stream->missed_metadata_flush = true; + pthread_mutex_unlock(&stream->metadata_timer_lock); + goto end; /* Bail out. */ + } + pthread_mutex_unlock(&stream->metadata_timer_lock); + /* Try again. */ + caa_cpu_relax(); + continue; + default: + ERR("Unexpected pthread_mutex_trylock error %d", ret); + ret = -1; + goto end; + } + break; + } + ret = consumer_flush_ust_index(stream); pthread_mutex_unlock(&stream->lock); +end: return ret; } @@ -460,7 +529,7 @@ void consumer_timer_live_stop(struct lttng_consumer_channel *channel) * Block the RT signals for the entire process. It must be called from the * consumer main before creating the threads */ -void consumer_signal_init(void) +int consumer_signal_init(void) { int ret; sigset_t mask; @@ -471,7 +540,9 @@ void consumer_signal_init(void) if (ret) { errno = ret; PERROR("pthread_sigmask"); + return -1; } + return 0; } /* @@ -485,6 +556,8 @@ void *consumer_timer_thread(void *data) siginfo_t info; struct lttng_consumer_local_data *ctx = data; + rcu_register_thread(); + health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_METADATA_TIMER); if (testpoint(consumerd_thread_metadata_timer)) { @@ -527,6 +600,8 @@ error_testpoint: health_error(); health_unregister(health_consumerd); + rcu_unregister_thread(); + /* Never return */ return NULL; }