X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer-timer.c;h=c190d3b627d0e8bc2ffc079bcc778456d3d22da4;hp=60ed94083e543a4d082e6f4e700f1ca57662120a;hb=e2d1190b9ea09c54e5d7373643d62e2034bc1531;hpb=873dda4e25e5199d434d322fb7d64697cd9868d3 diff --git a/src/common/consumer/consumer-timer.c b/src/common/consumer/consumer-timer.c index 60ed94083..c190d3b62 100644 --- a/src/common/consumer/consumer-timer.c +++ b/src/common/consumer/consumer-timer.c @@ -1,19 +1,9 @@ /* - * Copyright (C) 2012 - Julien Desfossez - * David Goulet + * Copyright (C) 2012 Julien Desfossez + * Copyright (C) 2012 David Goulet * - * 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. + * SPDX-License-Identifier: GPL-2.0-only * - * 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., 51 - * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ #define _LGPL_SOURCE @@ -21,7 +11,6 @@ #include #include -#include #include #include #include @@ -37,6 +26,7 @@ typedef int (*get_consumed_cb)(struct lttng_consumer_stream *stream, unsigned long *consumed); typedef int (*get_produced_cb)(struct lttng_consumer_stream *stream, unsigned long *produced); +typedef int (*flush_index_cb)(struct lttng_consumer_stream *stream); static struct timer_signal_data timer_signal = { .tid = 0, @@ -72,6 +62,10 @@ static void setmask(sigset_t *mask) if (ret) { PERROR("sigaddset monitor"); } + ret = sigaddset(mask, LTTNG_CONSUMER_SIG_EXIT); + if (ret) { + PERROR("sigaddset exit"); + } } static int channel_monitor_pipe = -1; @@ -84,7 +78,7 @@ static int channel_monitor_pipe = -1; * deadlocks. */ static void metadata_switch_timer(struct lttng_consumer_local_data *ctx, - int sig, siginfo_t *si) + siginfo_t *si) { int ret; struct lttng_consumer_channel *channel; @@ -182,7 +176,8 @@ end: return ret; } -static int check_kernel_stream(struct lttng_consumer_stream *stream) +static int check_stream(struct lttng_consumer_stream *stream, + flush_index_cb flush_index) { int ret; @@ -221,7 +216,7 @@ static int check_kernel_stream(struct lttng_consumer_stream *stream) } break; } - ret = consumer_flush_kernel_index(stream); + ret = flush_index(stream); pthread_mutex_unlock(&stream->lock); end: return ret; @@ -266,64 +261,21 @@ end: return ret; } -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; -} - /* * Execute action on a live timer */ static void live_timer(struct lttng_consumer_local_data *ctx, - int sig, siginfo_t *si) + siginfo_t *si) { int ret; struct lttng_consumer_channel *channel; struct lttng_consumer_stream *stream; - struct lttng_ht *ht; struct lttng_ht_iter iter; + const struct lttng_ht *ht = consumer_data.stream_per_chan_id_ht; + const flush_index_cb flush_index = + ctx->type == LTTNG_CONSUMER_KERNEL ? + consumer_flush_kernel_index : + consumer_flush_ust_index; channel = si->si_value.sival_ptr; assert(channel); @@ -331,38 +283,18 @@ static void live_timer(struct lttng_consumer_local_data *ctx, if (channel->switch_timer_error) { goto error; } - ht = consumer_data.stream_per_chan_id_ht; DBG("Live timer for channel %" PRIu64, channel->key); rcu_read_lock(); - switch (ctx->type) { - case LTTNG_CONSUMER32_UST: - case LTTNG_CONSUMER64_UST: - cds_lfht_for_each_entry_duplicate(ht->ht, - ht->hash_fct(&channel->key, lttng_ht_seed), - ht->match_fct, &channel->key, &iter.iter, - stream, node_channel_id.node) { - ret = check_ust_stream(stream); - if (ret < 0) { - goto error_unlock; - } - } - break; - case LTTNG_CONSUMER_KERNEL: - cds_lfht_for_each_entry_duplicate(ht->ht, - ht->hash_fct(&channel->key, lttng_ht_seed), - ht->match_fct, &channel->key, &iter.iter, - stream, node_channel_id.node) { - ret = check_kernel_stream(stream); - if (ret < 0) { - goto error_unlock; - } + cds_lfht_for_each_entry_duplicate(ht->ht, + ht->hash_fct(&channel->key, lttng_ht_seed), + ht->match_fct, &channel->key, &iter.iter, + stream, node_channel_id.node) { + ret = check_stream(stream, flush_index); + if (ret < 0) { + goto error_unlock; } - break; - case LTTNG_CONSUMER_UNKNOWN: - assert(0); - break; } error_unlock: @@ -394,7 +326,7 @@ void consumer_timer_signal_thread_qs(unsigned int signr) if (ret == -1) { PERROR("sigpending"); } - if (!sigismember(&pending_set, LTTNG_CONSUMER_SIG_SWITCH)) { + if (!sigismember(&pending_set, signr)) { break; } caa_cpu_relax(); @@ -629,17 +561,19 @@ int consumer_signal_init(void) static int sample_channel_positions(struct lttng_consumer_channel *channel, - uint64_t *_highest_use, uint64_t *_lowest_use, + uint64_t *_highest_use, uint64_t *_lowest_use, uint64_t *_total_consumed, sample_positions_cb sample, get_consumed_cb get_consumed, get_produced_cb get_produced) { - int ret; + int ret = 0; struct lttng_ht_iter iter; struct lttng_consumer_stream *stream; bool empty_channel = true; uint64_t high = 0, low = UINT64_MAX; struct lttng_ht *ht = consumer_data.stream_per_chan_id_ht; + *_total_consumed = 0; + rcu_read_lock(); cds_lfht_for_each_entry_duplicate(ht->ht, @@ -677,6 +611,15 @@ int sample_channel_positions(struct lttng_consumer_channel *channel, usage = produced - consumed; high = (usage > high) ? usage : high; low = (usage < low) ? usage : low; + + /* + * We don't use consumed here for 2 reasons: + * - output_written takes into account the padding written in the + * tracefiles when we stop the session; + * - the consumed position is not the accurate representation of what + * was extracted from a buffer in overwrite mode. + */ + *_total_consumed += stream->output_written; next: pthread_mutex_unlock(&stream->lock); } @@ -695,8 +638,7 @@ end: * Execute action on a monitor timer. */ static -void monitor_timer(struct lttng_consumer_local_data *ctx, - struct lttng_consumer_channel *channel) +void monitor_timer(struct lttng_consumer_channel *channel) { int ret; int channel_monitor_pipe = @@ -707,6 +649,7 @@ void monitor_timer(struct lttng_consumer_local_data *ctx, sample_positions_cb sample; get_consumed_cb get_consumed; get_produced_cb get_produced; + uint64_t lowest = 0, highest = 0, total_consumed = 0; assert(channel); @@ -730,11 +673,14 @@ void monitor_timer(struct lttng_consumer_local_data *ctx, abort(); } - ret = sample_channel_positions(channel, &msg.highest, &msg.lowest, - sample, get_consumed, get_produced); + ret = sample_channel_positions(channel, &highest, &lowest, + &total_consumed, sample, get_consumed, get_produced); if (ret) { return; } + msg.highest = highest; + msg.lowest = lowest; + msg.total_consumed = total_consumed; /* * Writes performed here are assumed to be atomic which is only @@ -782,7 +728,7 @@ end: /* * This thread is the sighandler for signals LTTNG_CONSUMER_SIG_SWITCH, * LTTNG_CONSUMER_SIG_TEARDOWN, LTTNG_CONSUMER_SIG_LIVE, and - * LTTNG_CONSUMER_SIG_MONITOR. + * LTTNG_CONSUMER_SIG_MONITOR, LTTNG_CONSUMER_SIG_EXIT. */ void *consumer_timer_thread(void *data) { @@ -823,19 +769,22 @@ void *consumer_timer_thread(void *data) } continue; } else if (signr == LTTNG_CONSUMER_SIG_SWITCH) { - metadata_switch_timer(ctx, info.si_signo, &info); + metadata_switch_timer(ctx, &info); } else if (signr == LTTNG_CONSUMER_SIG_TEARDOWN) { cmm_smp_mb(); CMM_STORE_SHARED(timer_signal.qs_done, 1); cmm_smp_mb(); DBG("Signal timer metadata thread teardown"); } else if (signr == LTTNG_CONSUMER_SIG_LIVE) { - live_timer(ctx, info.si_signo, &info); + live_timer(ctx, &info); } else if (signr == LTTNG_CONSUMER_SIG_MONITOR) { struct lttng_consumer_channel *channel; channel = info.si_value.sival_ptr; - monitor_timer(ctx, channel); + monitor_timer(channel); + } else if (signr == LTTNG_CONSUMER_SIG_EXIT) { + assert(CMM_LOAD_SHARED(consumer_quit)); + goto end; } else { ERR("Unexpected signal %d\n", info.si_signo); } @@ -844,10 +793,8 @@ void *consumer_timer_thread(void *data) error_testpoint: /* Only reached in testpoint error */ health_error(); +end: health_unregister(health_consumerd); - rcu_unregister_thread(); - - /* Never return */ return NULL; }