X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;ds=sidebyside;f=src%2Fcommon%2Fconsumer%2Fconsumer-timer.cpp;h=fa496336d8643e84018e2ca286018b70078fad50;hb=28ab034a2c3582d07d3423d2d746731f87d3969f;hp=51fa1687e9359a2d0c9a09e77b4573e8dce6e0ce;hpb=1c9a0b0e83c7e073c4e576c0bed95de335b0e502;p=lttng-tools.git diff --git a/src/common/consumer/consumer-timer.cpp b/src/common/consumer/consumer-timer.cpp index 51fa1687e..fa496336d 100644 --- a/src/common/consumer/consumer-timer.cpp +++ b/src/common/consumer/consumer-timer.cpp @@ -7,24 +7,22 @@ */ #define _LGPL_SOURCE +#include +#include +#include +#include +#include +#include +#include +#include + +#include #include #include -#include -#include -#include -#include -#include -#include -#include -#include -#include - typedef int (*sample_positions_cb)(struct lttng_consumer_stream *stream); -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 (*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 = { @@ -76,8 +74,7 @@ static int the_channel_monitor_pipe = -1; * while consumer_timer_switch_stop() is called. It would result in * deadlocks. */ -static void metadata_switch_timer(struct lttng_consumer_local_data *ctx, - siginfo_t *si) +static void metadata_switch_timer(struct lttng_consumer_local_data *ctx, siginfo_t *si) { int ret; struct lttng_consumer_channel *channel; @@ -119,8 +116,7 @@ static void metadata_switch_timer(struct lttng_consumer_local_data *ctx, } } -static int send_empty_index(struct lttng_consumer_stream *stream, uint64_t ts, - uint64_t stream_id) +static int send_empty_index(struct lttng_consumer_stream *stream, uint64_t ts, uint64_t stream_id) { int ret; struct ctf_packet_index index; @@ -175,8 +171,7 @@ end: return ret; } -static int check_stream(struct lttng_consumer_stream *stream, - flush_index_cb flush_index) +static int check_stream(struct lttng_consumer_stream *stream, flush_index_cb flush_index) { int ret; @@ -195,14 +190,14 @@ static int check_stream(struct lttng_consumer_stream *stream, ret = pthread_mutex_trylock(&stream->lock); switch (ret) { case 0: - break; /* We have the lock. */ + 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. */ + goto end; /* Bail out. */ } pthread_mutex_unlock(&stream->metadata_timer_lock); /* Try again. */ @@ -267,18 +262,16 @@ end: /* * Execute action on a live timer */ -static void live_timer(struct lttng_consumer_local_data *ctx, - siginfo_t *si) +static void live_timer(struct lttng_consumer_local_data *ctx, siginfo_t *si) { int ret; struct lttng_consumer_channel *channel; struct lttng_consumer_stream *stream; struct lttng_ht_iter iter; const struct lttng_ht *ht = the_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; + const flush_index_cb flush_index = ctx->type == LTTNG_CONSUMER_KERNEL ? + consumer_flush_kernel_index : + consumer_flush_ust_index; channel = (lttng_consumer_channel *) si->si_value.sival_ptr; LTTNG_ASSERT(channel); @@ -291,9 +284,13 @@ static void live_timer(struct lttng_consumer_local_data *ctx, rcu_read_lock(); 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) { + 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; @@ -307,8 +304,7 @@ error: return; } -static -void consumer_timer_signal_thread_qs(unsigned int signr) +static void consumer_timer_signal_thread_qs(unsigned int signr) { sigset_t pending_set; int ret; @@ -365,10 +361,10 @@ void consumer_timer_signal_thread_qs(unsigned int signr) * Returns a negative value on error, 0 if a timer was created, and * a positive value if no timer was created (not an error). */ -static -int consumer_channel_timer_start(timer_t *timer_id, - struct lttng_consumer_channel *channel, - unsigned int timer_interval_us, int signal) +static int consumer_channel_timer_start(timer_t *timer_id, + struct lttng_consumer_channel *channel, + unsigned int timer_interval_us, + int signal) { int ret = 0, delete_ret; struct sigevent sev = {}; @@ -412,8 +408,7 @@ error_destroy_timer: goto end; } -static -int consumer_channel_timer_stop(timer_t *timer_id, int signal) +static int consumer_channel_timer_stop(timer_t *timer_id, int signal) { int ret = 0; @@ -433,15 +428,17 @@ end: * Set the channel's switch timer. */ void consumer_timer_switch_start(struct lttng_consumer_channel *channel, - unsigned int switch_timer_interval_us) + unsigned int switch_timer_interval_us) { int ret; LTTNG_ASSERT(channel); LTTNG_ASSERT(channel->key); - ret = consumer_channel_timer_start(&channel->switch_timer, channel, - switch_timer_interval_us, LTTNG_CONSUMER_SIG_SWITCH); + ret = consumer_channel_timer_start(&channel->switch_timer, + channel, + switch_timer_interval_us, + LTTNG_CONSUMER_SIG_SWITCH); channel->switch_timer_enabled = !!(ret == 0); } @@ -455,8 +452,7 @@ void consumer_timer_switch_stop(struct lttng_consumer_channel *channel) LTTNG_ASSERT(channel); - ret = consumer_channel_timer_stop(&channel->switch_timer, - LTTNG_CONSUMER_SIG_SWITCH); + ret = consumer_channel_timer_stop(&channel->switch_timer, LTTNG_CONSUMER_SIG_SWITCH); if (ret == -1) { ERR("Failed to stop switch timer"); } @@ -468,15 +464,15 @@ void consumer_timer_switch_stop(struct lttng_consumer_channel *channel) * Set the channel's live timer. */ void consumer_timer_live_start(struct lttng_consumer_channel *channel, - unsigned int live_timer_interval_us) + unsigned int live_timer_interval_us) { int ret; LTTNG_ASSERT(channel); LTTNG_ASSERT(channel->key); - ret = consumer_channel_timer_start(&channel->live_timer, channel, - live_timer_interval_us, LTTNG_CONSUMER_SIG_LIVE); + ret = consumer_channel_timer_start( + &channel->live_timer, channel, live_timer_interval_us, LTTNG_CONSUMER_SIG_LIVE); channel->live_timer_enabled = !!(ret == 0); } @@ -490,8 +486,7 @@ void consumer_timer_live_stop(struct lttng_consumer_channel *channel) LTTNG_ASSERT(channel); - ret = consumer_channel_timer_stop(&channel->live_timer, - LTTNG_CONSUMER_SIG_LIVE); + ret = consumer_channel_timer_stop(&channel->live_timer, LTTNG_CONSUMER_SIG_LIVE); if (ret == -1) { ERR("Failed to stop live timer"); } @@ -506,7 +501,7 @@ void consumer_timer_live_stop(struct lttng_consumer_channel *channel) * a positive value if no timer was created (not an error). */ int consumer_timer_monitor_start(struct lttng_consumer_channel *channel, - unsigned int monitor_timer_interval_us) + unsigned int monitor_timer_interval_us) { int ret; @@ -514,8 +509,10 @@ int consumer_timer_monitor_start(struct lttng_consumer_channel *channel, LTTNG_ASSERT(channel->key); LTTNG_ASSERT(!channel->monitor_timer_enabled); - ret = consumer_channel_timer_start(&channel->monitor_timer, channel, - monitor_timer_interval_us, LTTNG_CONSUMER_SIG_MONITOR); + ret = consumer_channel_timer_start(&channel->monitor_timer, + channel, + monitor_timer_interval_us, + LTTNG_CONSUMER_SIG_MONITOR); channel->monitor_timer_enabled = !!(ret == 0); return ret; } @@ -530,8 +527,7 @@ int consumer_timer_monitor_stop(struct lttng_consumer_channel *channel) LTTNG_ASSERT(channel); LTTNG_ASSERT(channel->monitor_timer_enabled); - ret = consumer_channel_timer_stop(&channel->monitor_timer, - LTTNG_CONSUMER_SIG_MONITOR); + ret = consumer_channel_timer_stop(&channel->monitor_timer, LTTNG_CONSUMER_SIG_MONITOR); if (ret == -1) { ERR("Failed to stop live timer"); goto end; @@ -562,11 +558,13 @@ int consumer_signal_init(void) return 0; } -static -int sample_channel_positions(struct lttng_consumer_channel *channel, - 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) +static int sample_channel_positions(struct lttng_consumer_channel *channel, + 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 = 0; struct lttng_ht_iter iter; @@ -580,9 +578,13 @@ int sample_channel_positions(struct lttng_consumer_channel *channel, rcu_read_lock(); 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) { + ht->hash_fct(&channel->key, lttng_ht_seed), + ht->match_fct, + &channel->key, + &iter.iter, + stream, + node_channel_id.node) + { unsigned long produced, consumed, usage; empty_channel = false; @@ -594,7 +596,8 @@ int sample_channel_positions(struct lttng_consumer_channel *channel, ret = sample(stream); if (ret) { - ERR("Failed to take buffer position snapshot in monitor timer (ret = %d)", ret); + ERR("Failed to take buffer position snapshot in monitor timer (ret = %d)", + ret); pthread_mutex_unlock(&stream->lock); goto end; } @@ -637,20 +640,17 @@ end: return ret; } -/* - * Execute action on a monitor timer. - */ -static -void monitor_timer(struct lttng_consumer_channel *channel) +/* Sample and send channel buffering statistics to the session daemon. */ +void sample_and_send_channel_buffer_stats(struct lttng_consumer_channel *channel) { int ret; - int channel_monitor_pipe = - consumer_timer_thread_get_channel_monitor_pipe(); + int channel_monitor_pipe = consumer_timer_thread_get_channel_monitor_pipe(); struct lttcomm_consumer_channel_monitor_msg msg = { .key = channel->key, + .session_id = channel->session_id, .lowest = 0, .highest = 0, - .total_consumed = 0, + .consumed_since_last_sample = 0, }; sample_positions_cb sample; get_consumed_cb get_consumed; @@ -679,14 +679,15 @@ void monitor_timer(struct lttng_consumer_channel *channel) abort(); } - ret = sample_channel_positions(channel, &highest, &lowest, - &total_consumed, 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; + msg.consumed_since_last_sample = total_consumed - channel->last_consumed_size_sample_sent; /* * Writes performed here are assumed to be atomic which is only @@ -701,14 +702,17 @@ void monitor_timer(struct lttng_consumer_channel *channel) if (errno == EAGAIN) { /* Not an error, the sample is merely dropped. */ DBG("Channel monitor pipe is full; dropping sample for channel key = %" PRIu64, - channel->key); + channel->key); } else { PERROR("write to the channel monitor pipe"); } } else { DBG("Sent channel monitoring sample for channel key %" PRIu64 - ", (highest = %" PRIu64 ", lowest = %" PRIu64 ")", - channel->key, msg.highest, msg.lowest); + ", (highest = %" PRIu64 ", lowest = %" PRIu64 ")", + channel->key, + msg.highest, + msg.lowest); + channel->last_consumed_size_sample_sent = msg.consumed_since_last_sample; } } @@ -787,7 +791,7 @@ void *consumer_timer_thread(void *data) struct lttng_consumer_channel *channel; channel = (lttng_consumer_channel *) info.si_value.sival_ptr; - monitor_timer(channel); + sample_and_send_channel_buffer_stats(channel); } else if (signr == LTTNG_CONSUMER_SIG_EXIT) { LTTNG_ASSERT(CMM_LOAD_SHARED(consumer_quit)); goto end;