2 * SPDX-License-Identifier: LGPL-2.1-only
4 * Copyright (C) 2005-2012 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
6 * See ring_buffer_frontend.c for more information on wait-free
8 * See frontend.h for channel allocation and read-side API.
11 #ifndef _LTTNG_RING_BUFFER_FRONTEND_API_H
12 #define _LTTNG_RING_BUFFER_FRONTEND_API_H
16 #include <urcu/compiler.h>
18 #include "common/getcpu.h"
22 * lib_ring_buffer_nesting_inc - Ring buffer recursive use protection.
24 * The rint buffer buffer nesting count is a safety net to ensure tracer
25 * client code will never trigger an endless recursion.
26 * Returns a nesting level >= 0 on success, -EPERM on failure (nesting
29 * __asm__ __volatile__ and "memory" clobber prevent the compiler from moving
30 * instructions out of the ring buffer nesting count. This is required to ensure
31 * that probe side-effects which can cause recursion (e.g. unforeseen traps,
32 * divisions by 0, ...) are triggered within the incremented nesting count
36 int lib_ring_buffer_nesting_inc(
37 const struct lttng_ust_ring_buffer_config
*config
__attribute__((unused
)))
41 nesting
= ++URCU_TLS(lib_ring_buffer_nesting
);
43 if (caa_unlikely(nesting
>= LIB_RING_BUFFER_MAX_NESTING
)) {
45 URCU_TLS(lib_ring_buffer_nesting
)--;
52 int lib_ring_buffer_nesting_count(
53 const struct lttng_ust_ring_buffer_config
*config
__attribute__((unused
)))
55 return URCU_TLS(lib_ring_buffer_nesting
);
59 void lib_ring_buffer_nesting_dec(
60 const struct lttng_ust_ring_buffer_config
*config
__attribute__((unused
)))
63 URCU_TLS(lib_ring_buffer_nesting
)--; /* TLS */
67 * lib_ring_buffer_try_reserve is called by lib_ring_buffer_reserve(). It is not
68 * part of the API per se.
70 * returns 0 if reserve ok, or 1 if the slow path must be taken.
73 int lib_ring_buffer_try_reserve(const struct lttng_ust_ring_buffer_config
*config
,
74 struct lttng_ust_ring_buffer_ctx
*ctx
,
76 unsigned long *o_begin
, unsigned long *o_end
,
77 unsigned long *o_old
, size_t *before_hdr_pad
)
79 struct lttng_ust_ring_buffer_ctx_private
*ctx_private
= ctx
->priv
;
80 struct lttng_ust_ring_buffer_channel
*chan
= ctx_private
->chan
;
81 struct lttng_ust_ring_buffer
*buf
= ctx_private
->buf
;
82 *o_begin
= v_read(config
, &buf
->offset
);
85 ctx_private
->timestamp
= lib_ring_buffer_clock_read(chan
);
86 if ((int64_t) ctx_private
->timestamp
== -EIO
)
90 * Prefetch cacheline for read because we have to read the previous
91 * commit counter to increment it and commit seq value to compare it to
94 //prefetch(&buf->commit_hot[subbuf_index(*o_begin, chan)]);
96 if (last_timestamp_overflow(config
, buf
, ctx_private
->timestamp
))
97 ctx_private
->rflags
|= RING_BUFFER_RFLAG_FULL_TIMESTAMP
;
99 if (caa_unlikely(subbuf_offset(*o_begin
, chan
) == 0))
102 ctx_private
->slot_size
= record_header_size(config
, chan
, *o_begin
,
103 before_hdr_pad
, ctx
, client_ctx
);
104 ctx_private
->slot_size
+=
105 lttng_ust_ring_buffer_align(*o_begin
+ ctx_private
->slot_size
,
106 ctx
->largest_align
) + ctx
->data_size
;
107 if (caa_unlikely((subbuf_offset(*o_begin
, chan
) + ctx_private
->slot_size
)
108 > chan
->backend
.subbuf_size
))
112 * Record fits in the current buffer and we are not on a switch
113 * boundary. It's safe to write.
115 *o_end
= *o_begin
+ ctx_private
->slot_size
;
117 if (caa_unlikely((subbuf_offset(*o_end
, chan
)) == 0))
119 * The offset_end will fall at the very beginning of the next
128 * lib_ring_buffer_reserve - Reserve space in a ring buffer.
129 * @config: ring buffer instance configuration.
130 * @ctx: ring buffer context. (input and output) Must be already initialized.
132 * Atomic wait-free slot reservation. The reserved space starts at the context
133 * "pre_offset". Its length is "slot_size". The associated time-stamp is
138 * -EAGAIN if channel is disabled.
139 * -ENOSPC if event size is too large for packet.
140 * -ENOBUFS if there is currently not enough space in buffer for the event.
141 * -EIO if data cannot be written into the buffer for any other reason.
145 int lib_ring_buffer_reserve(const struct lttng_ust_ring_buffer_config
*config
,
146 struct lttng_ust_ring_buffer_ctx
*ctx
,
149 struct lttng_ust_ring_buffer_ctx_private
*ctx_private
= ctx
->priv
;
150 struct lttng_ust_ring_buffer_channel
*chan
= ctx_private
->chan
;
151 struct lttng_ust_shm_handle
*handle
= chan
->handle
;
152 struct lttng_ust_ring_buffer
*buf
;
153 unsigned long o_begin
, o_end
, o_old
;
154 size_t before_hdr_pad
= 0;
156 if (caa_unlikely(uatomic_read(&chan
->record_disabled
)))
159 if (config
->alloc
== RING_BUFFER_ALLOC_PER_CPU
) {
160 ctx_private
->reserve_cpu
= lttng_ust_get_cpu();
161 buf
= shmp(handle
, chan
->backend
.buf
[ctx_private
->reserve_cpu
].shmp
);
163 buf
= shmp(handle
, chan
->backend
.buf
[0].shmp
);
165 if (caa_unlikely(!buf
))
167 if (caa_unlikely(uatomic_read(&buf
->record_disabled
)))
169 ctx_private
->buf
= buf
;
172 * Perform retryable operations.
174 if (caa_unlikely(lib_ring_buffer_try_reserve(config
, ctx
, client_ctx
, &o_begin
,
175 &o_end
, &o_old
, &before_hdr_pad
)))
178 if (caa_unlikely(v_cmpxchg(config
, &buf
->offset
, o_old
, o_end
)
183 * Atomically update last_timestamp. This update races against concurrent
184 * atomic updates, but the race will always cause supplementary full
185 * timestamp record headers, never the opposite (missing a full
186 * timestamp record header when it would be needed).
188 save_last_timestamp(config
, buf
, ctx_private
->timestamp
);
191 * Push the reader if necessary
193 lib_ring_buffer_reserve_push_reader(buf
, chan
, o_end
- 1);
196 * Clear noref flag for this subbuffer.
198 lib_ring_buffer_clear_noref(config
, &buf
->backend
,
199 subbuf_index(o_end
- 1, chan
), handle
);
201 ctx_private
->pre_offset
= o_begin
;
202 ctx_private
->buf_offset
= o_begin
+ before_hdr_pad
;
205 return lib_ring_buffer_reserve_slow(ctx
, client_ctx
);
209 * lib_ring_buffer_switch - Perform a sub-buffer switch for a per-cpu buffer.
210 * @config: ring buffer instance configuration.
212 * @mode: buffer switch mode (SWITCH_ACTIVE or SWITCH_FLUSH)
214 * This operation is completely reentrant : can be called while tracing is
215 * active with absolutely no lock held.
217 * Note, however, that as a v_cmpxchg is used for some atomic operations and
218 * requires to be executed locally for per-CPU buffers, this function must be
219 * called from the CPU which owns the buffer for a ACTIVE flush, with preemption
220 * disabled, for RING_BUFFER_SYNC_PER_CPU configuration.
223 void lib_ring_buffer_switch(
224 const struct lttng_ust_ring_buffer_config
*config
__attribute__((unused
)),
225 struct lttng_ust_ring_buffer
*buf
, enum switch_mode mode
,
226 struct lttng_ust_shm_handle
*handle
)
228 lib_ring_buffer_switch_slow(buf
, mode
, handle
);
231 /* See ring_buffer_frontend_api.h for lib_ring_buffer_reserve(). */
234 * lib_ring_buffer_commit - Commit an record.
235 * @config: ring buffer instance configuration.
236 * @ctx: ring buffer context. (input arguments only)
238 * Atomic unordered slot commit. Increments the commit count in the
239 * specified sub-buffer, and delivers it if necessary.
242 void lib_ring_buffer_commit(const struct lttng_ust_ring_buffer_config
*config
,
243 const struct lttng_ust_ring_buffer_ctx
*ctx
)
245 struct lttng_ust_ring_buffer_ctx_private
*ctx_private
= ctx
->priv
;
246 struct lttng_ust_ring_buffer_channel
*chan
= ctx_private
->chan
;
247 struct lttng_ust_shm_handle
*handle
= chan
->handle
;
248 struct lttng_ust_ring_buffer
*buf
= ctx_private
->buf
;
249 unsigned long offset_end
= ctx_private
->buf_offset
;
250 unsigned long endidx
= subbuf_index(offset_end
- 1, chan
);
251 unsigned long commit_count
;
252 struct commit_counters_hot
*cc_hot
= shmp_index(handle
,
253 buf
->commit_hot
, endidx
);
255 if (caa_unlikely(!cc_hot
))
259 * Must count record before incrementing the commit count.
261 subbuffer_count_record(config
, ctx
);
264 * Order all writes to buffer before the commit count update that will
265 * determine that the subbuffer is full.
269 v_add(config
, ctx_private
->slot_size
, &cc_hot
->cc
);
272 * commit count read can race with concurrent OOO commit count updates.
273 * This is only needed for lib_ring_buffer_check_deliver (for
274 * non-polling delivery only) and for
275 * lib_ring_buffer_write_commit_counter. The race can only cause the
276 * counter to be read with the same value more than once, which could
278 * - Multiple delivery for the same sub-buffer (which is handled
279 * gracefully by the reader code) if the value is for a full
280 * sub-buffer. It's important that we can never miss a sub-buffer
281 * delivery. Re-reading the value after the v_add ensures this.
282 * - Reading a commit_count with a higher value that what was actually
283 * added to it for the lib_ring_buffer_write_commit_counter call
284 * (again caused by a concurrent committer). It does not matter,
285 * because this function is interested in the fact that the commit
286 * count reaches back the reserve offset for a specific sub-buffer,
287 * which is completely independent of the order.
289 commit_count
= v_read(config
, &cc_hot
->cc
);
291 lib_ring_buffer_check_deliver(config
, buf
, chan
, offset_end
- 1,
292 commit_count
, endidx
, handle
, ctx
);
294 * Update used size at each commit. It's needed only for extracting
295 * ring_buffer buffers from vmcore, after crash.
297 lib_ring_buffer_write_commit_counter(config
, buf
, chan
,
298 offset_end
, commit_count
, handle
, cc_hot
);
302 * lib_ring_buffer_try_discard_reserve - Try discarding a record.
303 * @config: ring buffer instance configuration.
304 * @ctx: ring buffer context. (input arguments only)
306 * Only succeeds if no other record has been written after the record to
307 * discard. If discard fails, the record must be committed to the buffer.
309 * Returns 0 upon success, -EPERM if the record cannot be discarded.
312 int lib_ring_buffer_try_discard_reserve(const struct lttng_ust_ring_buffer_config
*config
,
313 const struct lttng_ust_ring_buffer_ctx
*ctx
)
315 struct lttng_ust_ring_buffer_ctx_private
*ctx_private
= ctx
->priv
;
316 struct lttng_ust_ring_buffer
*buf
= ctx_private
->buf
;
317 unsigned long end_offset
= ctx_private
->pre_offset
+ ctx_private
->slot_size
;
320 * We need to ensure that if the cmpxchg succeeds and discards the
321 * record, the next record will record a full timestamp, because it cannot
322 * rely on the last_timestamp associated with the discarded record to detect
323 * overflows. The only way to ensure this is to set the last_timestamp to 0
324 * (assuming no 64-bit timestamp overflow), which forces to write a 64-bit
325 * timestamp in the next record.
327 * Note: if discard fails, we must leave the timestamp in the record header.
328 * It is needed to keep track of timestamp overflows for the following
331 save_last_timestamp(config
, buf
, 0ULL);
333 if (caa_likely(v_cmpxchg(config
, &buf
->offset
, end_offset
, ctx_private
->pre_offset
)
341 void channel_record_disable(
342 const struct lttng_ust_ring_buffer_config
*config
__attribute__((unused
)),
343 struct lttng_ust_ring_buffer_channel
*chan
)
345 uatomic_inc(&chan
->record_disabled
);
349 void channel_record_enable(
350 const struct lttng_ust_ring_buffer_config
*config
__attribute__((unused
)),
351 struct lttng_ust_ring_buffer_channel
*chan
)
353 uatomic_dec(&chan
->record_disabled
);
357 void lib_ring_buffer_record_disable(
358 const struct lttng_ust_ring_buffer_config
*config
__attribute__((unused
)),
359 struct lttng_ust_ring_buffer
*buf
)
361 uatomic_inc(&buf
->record_disabled
);
365 void lib_ring_buffer_record_enable(
366 const struct lttng_ust_ring_buffer_config
*config
__attribute__((unused
)),
367 struct lttng_ust_ring_buffer
*buf
)
369 uatomic_dec(&buf
->record_disabled
);
372 #endif /* _LTTNG_RING_BUFFER_FRONTEND_API_H */