From: Mathieu Desnoyers Date: Wed, 9 Mar 2022 21:42:15 +0000 (-0500) Subject: Fix: sample discarded events count before reserve X-Git-Url: https://git.lttng.org/?p=lttng-ust.git;a=commitdiff_plain;h=f27b0770c5fe934bd90c4b20103c59f9e7f96a5b Fix: sample discarded events count before reserve Sampling the discarded events count in the buffer_end callback is done out of order, and may therefore include increments performed by following events (in following packets) if the thread doing the end-of-packet event write is preempted for a long time. Sampling the event discarded counts before reserving space for the last event in a packet, and keeping this as part of the private ring buffer context, should fix this race. Signed-off-by: Mathieu Desnoyers Change-Id: Ib59b634bbaefd2444751547d20a891c9dd93cd73 --- diff --git a/src/common/ringbuffer-clients/metadata-template.h b/src/common/ringbuffer-clients/metadata-template.h index 1f0e9fe4..56d95516 100644 --- a/src/common/ringbuffer-clients/metadata-template.h +++ b/src/common/ringbuffer-clients/metadata-template.h @@ -127,7 +127,8 @@ static void client_buffer_begin(struct lttng_ust_ring_buffer *buf, static void client_buffer_end(struct lttng_ust_ring_buffer *buf, uint64_t tsc __attribute__((unused)), unsigned int subbuf_idx, unsigned long data_size, - struct lttng_ust_shm_handle *handle) + struct lttng_ust_shm_handle *handle, + const struct lttng_ust_ring_buffer_ctx *ctx) { struct lttng_ust_ring_buffer_channel *chan = shmp(handle, buf->backend.chan); struct metadata_packet_header *header = @@ -146,9 +147,9 @@ static void client_buffer_end(struct lttng_ust_ring_buffer *buf, * We do not care about the records lost count, because the metadata * channel waits and retry. */ - (void) lib_ring_buffer_get_records_lost_full(&client_config, buf); - records_lost += lib_ring_buffer_get_records_lost_wrap(&client_config, buf); - records_lost += lib_ring_buffer_get_records_lost_big(&client_config, buf); + (void) lib_ring_buffer_get_records_lost_full(&client_config, ctx); + records_lost += lib_ring_buffer_get_records_lost_wrap(&client_config, ctx); + records_lost += lib_ring_buffer_get_records_lost_big(&client_config, ctx); WARN_ON_ONCE(records_lost != 0); } diff --git a/src/common/ringbuffer-clients/template.h b/src/common/ringbuffer-clients/template.h index b0a5fdd7..824c344d 100644 --- a/src/common/ringbuffer-clients/template.h +++ b/src/common/ringbuffer-clients/template.h @@ -399,7 +399,8 @@ static void client_buffer_begin(struct lttng_ust_ring_buffer *buf, uint64_t tsc, */ static void client_buffer_end(struct lttng_ust_ring_buffer *buf, uint64_t tsc, unsigned int subbuf_idx, unsigned long data_size, - struct lttng_ust_shm_handle *handle) + struct lttng_ust_shm_handle *handle, + const struct lttng_ust_ring_buffer_ctx *ctx) { struct lttng_ust_ring_buffer_channel *chan = shmp(handle, buf->backend.chan); struct packet_header *header = @@ -418,9 +419,9 @@ static void client_buffer_end(struct lttng_ust_ring_buffer *buf, uint64_t tsc, header->ctx.packet_size = (uint64_t) LTTNG_UST_PAGE_ALIGN(data_size) * CHAR_BIT; /* in bits */ - records_lost += lib_ring_buffer_get_records_lost_full(&client_config, buf); - records_lost += lib_ring_buffer_get_records_lost_wrap(&client_config, buf); - records_lost += lib_ring_buffer_get_records_lost_big(&client_config, buf); + records_lost += lib_ring_buffer_get_records_lost_full(&client_config, ctx); + records_lost += lib_ring_buffer_get_records_lost_wrap(&client_config, ctx); + records_lost += lib_ring_buffer_get_records_lost_big(&client_config, ctx); header->ctx.events_discarded = records_lost; } diff --git a/src/common/ringbuffer/frontend.h b/src/common/ringbuffer/frontend.h index ed0cc7d4..9850d710 100644 --- a/src/common/ringbuffer/frontend.h +++ b/src/common/ringbuffer/frontend.h @@ -265,26 +265,26 @@ unsigned long lib_ring_buffer_get_records_overrun( static inline unsigned long lib_ring_buffer_get_records_lost_full( - const struct lttng_ust_ring_buffer_config *config, - struct lttng_ust_ring_buffer *buf) + const struct lttng_ust_ring_buffer_config *config __attribute__((unused)), + const struct lttng_ust_ring_buffer_ctx *ctx) { - return v_read(config, &buf->records_lost_full); + return ctx->priv->records_lost_full; } static inline unsigned long lib_ring_buffer_get_records_lost_wrap( - const struct lttng_ust_ring_buffer_config *config, - struct lttng_ust_ring_buffer *buf) + const struct lttng_ust_ring_buffer_config *config __attribute__((unused)), + const struct lttng_ust_ring_buffer_ctx *ctx) { - return v_read(config, &buf->records_lost_wrap); + return ctx->priv->records_lost_wrap; } static inline unsigned long lib_ring_buffer_get_records_lost_big( - const struct lttng_ust_ring_buffer_config *config, - struct lttng_ust_ring_buffer *buf) + const struct lttng_ust_ring_buffer_config *config __attribute__((unused)), + const struct lttng_ust_ring_buffer_ctx *ctx) { - return v_read(config, &buf->records_lost_big); + return ctx->priv->records_lost_big; } static inline diff --git a/src/common/ringbuffer/frontend_api.h b/src/common/ringbuffer/frontend_api.h index b306986e..a697ed2e 100644 --- a/src/common/ringbuffer/frontend_api.h +++ b/src/common/ringbuffer/frontend_api.h @@ -288,7 +288,7 @@ void lib_ring_buffer_commit(const struct lttng_ust_ring_buffer_config *config, commit_count = v_read(config, &cc_hot->cc); lib_ring_buffer_check_deliver(config, buf, chan, offset_end - 1, - commit_count, endidx, handle, ctx_private->tsc); + commit_count, endidx, handle, ctx); /* * Update used size at each commit. It's needed only for extracting * ring_buffer buffers from vmcore, after crash. diff --git a/src/common/ringbuffer/frontend_internal.h b/src/common/ringbuffer/frontend_internal.h index 9655a19f..1dc816a3 100644 --- a/src/common/ringbuffer/frontend_internal.h +++ b/src/common/ringbuffer/frontend_internal.h @@ -165,7 +165,7 @@ void lib_ring_buffer_check_deliver_slow(const struct lttng_ust_ring_buffer_confi unsigned long commit_count, unsigned long idx, struct lttng_ust_shm_handle *handle, - uint64_t tsc) + const struct lttng_ust_ring_buffer_ctx *ctx) __attribute__((visibility("hidden"))); /* Buffer write helpers */ @@ -301,7 +301,7 @@ void lib_ring_buffer_check_deliver(const struct lttng_ust_ring_buffer_config *co unsigned long commit_count, unsigned long idx, struct lttng_ust_shm_handle *handle, - uint64_t tsc) + const struct lttng_ust_ring_buffer_ctx *ctx) { unsigned long old_commit_count = commit_count - chan->backend.subbuf_size; @@ -310,7 +310,7 @@ void lib_ring_buffer_check_deliver(const struct lttng_ust_ring_buffer_config *co if (caa_unlikely((buf_trunc(offset, chan) >> chan->backend.num_subbuf_order) - (old_commit_count & chan->commit_count_mask) == 0)) lib_ring_buffer_check_deliver_slow(config, buf, chan, offset, - commit_count, idx, handle, tsc); + commit_count, idx, handle, ctx); } /* diff --git a/src/common/ringbuffer/frontend_types.h b/src/common/ringbuffer/frontend_types.h index d1f70c3a..1b0e1a08 100644 --- a/src/common/ringbuffer/frontend_types.h +++ b/src/common/ringbuffer/frontend_types.h @@ -253,12 +253,19 @@ struct lttng_ust_ring_buffer_ctx_private { */ uint64_t tsc; /* time-stamp counter value */ unsigned int rflags; /* reservation flags */ - struct lttng_ust_ring_buffer *buf; /* * buffer corresponding to processor id * for this channel */ struct lttng_ust_ring_buffer_backend_pages *backend_pages; + + /* + * Records lost counts are only loaded into these fields before + * reserving the last bytes from the ring buffer. + */ + unsigned long records_lost_full; + unsigned long records_lost_wrap; + unsigned long records_lost_big; }; static inline diff --git a/src/common/ringbuffer/ring_buffer_frontend.c b/src/common/ringbuffer/ring_buffer_frontend.c index 4d06bf63..91c69b76 100644 --- a/src/common/ringbuffer/ring_buffer_frontend.c +++ b/src/common/ringbuffer/ring_buffer_frontend.c @@ -1763,7 +1763,7 @@ static void lib_ring_buffer_switch_old_start(struct lttng_ust_ring_buffer *buf, struct lttng_ust_ring_buffer_channel *chan, struct switch_offsets *offsets, - uint64_t tsc, + const struct lttng_ust_ring_buffer_ctx *ctx, struct lttng_ust_shm_handle *handle) { const struct lttng_ust_ring_buffer_config *config = &chan->backend.config; @@ -1771,7 +1771,7 @@ void lib_ring_buffer_switch_old_start(struct lttng_ust_ring_buffer *buf, unsigned long commit_count; struct commit_counters_hot *cc_hot; - config->cb.buffer_begin(buf, tsc, oldidx, handle); + config->cb.buffer_begin(buf, ctx->priv->tsc, oldidx, handle); /* * Order all writes to buffer before the commit count update that will @@ -1786,7 +1786,7 @@ void lib_ring_buffer_switch_old_start(struct lttng_ust_ring_buffer *buf, commit_count = v_read(config, &cc_hot->cc); /* Check if the written buffer has to be delivered */ lib_ring_buffer_check_deliver(config, buf, chan, offsets->old, - commit_count, oldidx, handle, tsc); + commit_count, oldidx, handle, ctx); lib_ring_buffer_write_commit_counter(config, buf, chan, offsets->old + config->cb.subbuffer_header_size(), commit_count, handle, cc_hot); @@ -1804,7 +1804,7 @@ static void lib_ring_buffer_switch_old_end(struct lttng_ust_ring_buffer *buf, struct lttng_ust_ring_buffer_channel *chan, struct switch_offsets *offsets, - uint64_t tsc, + const struct lttng_ust_ring_buffer_ctx *ctx, struct lttng_ust_shm_handle *handle) { const struct lttng_ust_ring_buffer_config *config = &chan->backend.config; @@ -1829,7 +1829,7 @@ void lib_ring_buffer_switch_old_end(struct lttng_ust_ring_buffer *buf, * postponed until the commit counter is incremented for the * current space reservation. */ - *ts_end = tsc; + *ts_end = ctx->priv->tsc; /* * Order all writes to buffer and store to ts_end before the commit @@ -1842,7 +1842,7 @@ void lib_ring_buffer_switch_old_end(struct lttng_ust_ring_buffer *buf, v_add(config, padding_size, &cc_hot->cc); commit_count = v_read(config, &cc_hot->cc); lib_ring_buffer_check_deliver(config, buf, chan, offsets->old - 1, - commit_count, oldidx, handle, tsc); + commit_count, oldidx, handle, ctx); lib_ring_buffer_write_commit_counter(config, buf, chan, offsets->old + padding_size, commit_count, handle, cc_hot); @@ -1859,7 +1859,7 @@ static void lib_ring_buffer_switch_new_start(struct lttng_ust_ring_buffer *buf, struct lttng_ust_ring_buffer_channel *chan, struct switch_offsets *offsets, - uint64_t tsc, + const struct lttng_ust_ring_buffer_ctx *ctx, struct lttng_ust_shm_handle *handle) { const struct lttng_ust_ring_buffer_config *config = &chan->backend.config; @@ -1867,7 +1867,7 @@ void lib_ring_buffer_switch_new_start(struct lttng_ust_ring_buffer *buf, unsigned long commit_count; struct commit_counters_hot *cc_hot; - config->cb.buffer_begin(buf, tsc, beginidx, handle); + config->cb.buffer_begin(buf, ctx->priv->tsc, beginidx, handle); /* * Order all writes to buffer before the commit count update that will @@ -1881,7 +1881,7 @@ void lib_ring_buffer_switch_new_start(struct lttng_ust_ring_buffer *buf, commit_count = v_read(config, &cc_hot->cc); /* Check if the written buffer has to be delivered */ lib_ring_buffer_check_deliver(config, buf, chan, offsets->begin, - commit_count, beginidx, handle, tsc); + commit_count, beginidx, handle, ctx); lib_ring_buffer_write_commit_counter(config, buf, chan, offsets->begin + config->cb.subbuffer_header_size(), commit_count, handle, cc_hot); @@ -1899,7 +1899,7 @@ static void lib_ring_buffer_switch_new_end(struct lttng_ust_ring_buffer *buf, struct lttng_ust_ring_buffer_channel *chan, struct switch_offsets *offsets, - uint64_t tsc, + const struct lttng_ust_ring_buffer_ctx *ctx, struct lttng_ust_shm_handle *handle) { const struct lttng_ust_ring_buffer_config *config = &chan->backend.config; @@ -1921,7 +1921,7 @@ void lib_ring_buffer_switch_new_end(struct lttng_ust_ring_buffer *buf, * postponed until the commit counter is incremented for the * current space reservation. */ - *ts_end = tsc; + *ts_end = ctx->priv->tsc; } /* @@ -1934,7 +1934,7 @@ int lib_ring_buffer_try_switch_slow(enum switch_mode mode, struct lttng_ust_ring_buffer *buf, struct lttng_ust_ring_buffer_channel *chan, struct switch_offsets *offsets, - uint64_t *tsc, + struct lttng_ust_ring_buffer_ctx *ctx, struct lttng_ust_shm_handle *handle) { const struct lttng_ust_ring_buffer_config *config = &chan->backend.config; @@ -1945,7 +1945,7 @@ int lib_ring_buffer_try_switch_slow(enum switch_mode mode, offsets->switch_old_start = 0; off = subbuf_offset(offsets->begin, chan); - *tsc = config->cb.ring_buffer_clock_read(chan); + ctx->priv->tsc = config->cb.ring_buffer_clock_read(chan); /* * Ensure we flush the header of an empty subbuffer when doing the @@ -2032,6 +2032,13 @@ int lib_ring_buffer_try_switch_slow(enum switch_mode mode, offsets->begin = subbuf_align(offsets->begin, chan); /* Note: old points to the next subbuf at offset 0 */ offsets->end = offsets->begin; + /* + * Populate the records lost counters prior to performing a + * sub-buffer switch. + */ + ctx->priv->records_lost_full = v_read(config, &buf->records_lost_full); + ctx->priv->records_lost_wrap = v_read(config, &buf->records_lost_wrap); + ctx->priv->records_lost_big = v_read(config, &buf->records_lost_big); return 0; } @@ -2050,10 +2057,12 @@ void lib_ring_buffer_switch_slow(struct lttng_ust_ring_buffer *buf, enum switch_ { struct lttng_ust_ring_buffer_channel *chan; const struct lttng_ust_ring_buffer_config *config; + struct lttng_ust_ring_buffer_ctx_private ctx_priv; + struct lttng_ust_ring_buffer_ctx ctx; struct switch_offsets offsets; unsigned long oldidx; - uint64_t tsc; + ctx.priv = &ctx_priv; chan = shmp(handle, buf->backend.chan); if (!chan) return; @@ -2066,7 +2075,7 @@ void lib_ring_buffer_switch_slow(struct lttng_ust_ring_buffer *buf, enum switch_ */ do { if (lib_ring_buffer_try_switch_slow(mode, buf, chan, &offsets, - &tsc, handle)) + &ctx, handle)) return; /* Switch not needed */ } while (v_cmpxchg(config, &buf->offset, offsets.old, offsets.end) != offsets.old); @@ -2077,7 +2086,7 @@ void lib_ring_buffer_switch_slow(struct lttng_ust_ring_buffer *buf, enum switch_ * records, never the opposite (missing a full TSC record when it would * be needed). */ - save_last_tsc(config, buf, tsc); + save_last_tsc(config, buf, ctx.priv->tsc); /* * Push the reader if necessary @@ -2091,14 +2100,14 @@ void lib_ring_buffer_switch_slow(struct lttng_ust_ring_buffer *buf, enum switch_ * May need to populate header start on SWITCH_FLUSH. */ if (offsets.switch_old_start) { - lib_ring_buffer_switch_old_start(buf, chan, &offsets, tsc, handle); + lib_ring_buffer_switch_old_start(buf, chan, &offsets, &ctx, handle); offsets.old += config->cb.subbuffer_header_size(); } /* * Switch old subbuffer. */ - lib_ring_buffer_switch_old_end(buf, chan, &offsets, tsc, handle); + lib_ring_buffer_switch_old_end(buf, chan, &offsets, &ctx, handle); } static @@ -2308,6 +2317,15 @@ retry: */ offsets->switch_new_end = 1; /* For offsets->begin */ } + /* + * Populate the records lost counters when the space reservation + * may cause a sub-buffer switch. + */ + if (offsets->switch_new_end || offsets->switch_old_end) { + ctx_private->records_lost_full = v_read(config, &buf->records_lost_full); + ctx_private->records_lost_wrap = v_read(config, &buf->records_lost_wrap); + ctx_private->records_lost_big = v_read(config, &buf->records_lost_big); + } return 0; } @@ -2376,17 +2394,17 @@ int lib_ring_buffer_reserve_slow(struct lttng_ust_ring_buffer_ctx *ctx, lib_ring_buffer_clear_noref(config, &buf->backend, subbuf_index(offsets.old - 1, chan), handle); - lib_ring_buffer_switch_old_end(buf, chan, &offsets, ctx_private->tsc, handle); + lib_ring_buffer_switch_old_end(buf, chan, &offsets, ctx, handle); } /* * Populate new subbuffer. */ if (caa_unlikely(offsets.switch_new_start)) - lib_ring_buffer_switch_new_start(buf, chan, &offsets, ctx_private->tsc, handle); + lib_ring_buffer_switch_new_start(buf, chan, &offsets, ctx, handle); if (caa_unlikely(offsets.switch_new_end)) - lib_ring_buffer_switch_new_end(buf, chan, &offsets, ctx_private->tsc, handle); + lib_ring_buffer_switch_new_end(buf, chan, &offsets, ctx, handle); ctx_private->slot_size = offsets.size; ctx_private->pre_offset = offsets.begin; @@ -2447,7 +2465,7 @@ void lib_ring_buffer_check_deliver_slow(const struct lttng_ust_ring_buffer_confi unsigned long commit_count, unsigned long idx, struct lttng_ust_shm_handle *handle, - uint64_t tsc __attribute__((unused))) + const struct lttng_ust_ring_buffer_ctx *ctx) { unsigned long old_commit_count = commit_count - chan->backend.subbuf_size; @@ -2515,7 +2533,7 @@ void lib_ring_buffer_check_deliver_slow(const struct lttng_ust_ring_buffer_confi buf, idx, handle), - handle); + handle, ctx); /* * Increment the packet counter while we have exclusive diff --git a/src/common/ringbuffer/ringbuffer-config.h b/src/common/ringbuffer/ringbuffer-config.h index 94f283ff..61386174 100644 --- a/src/common/ringbuffer/ringbuffer-config.h +++ b/src/common/ringbuffer/ringbuffer-config.h @@ -51,7 +51,8 @@ struct lttng_ust_ring_buffer_client_cb { struct lttng_ust_shm_handle *handle); void (*buffer_end) (struct lttng_ust_ring_buffer *buf, uint64_t tsc, unsigned int subbuf_idx, unsigned long data_size, - struct lttng_ust_shm_handle *handle); + struct lttng_ust_shm_handle *handle, + const struct lttng_ust_ring_buffer_ctx *ctx); /* Optional callbacks (can be set to NULL) */