1 /* SPDX-License-Identifier: (GPL-2.0-only OR LGPL-2.1-only)
3 * ringbuffer/frontend_api.h
5 * Ring Buffer Library Synchronization Header (buffer write API).
7 * Copyright (C) 2005-2012 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
9 * See ring_buffer_frontend.c for more information on wait-free algorithms.
10 * See linux/ringbuffer/frontend.h for channel allocation and read-side API.
13 #ifndef _LIB_RING_BUFFER_FRONTEND_API_H
14 #define _LIB_RING_BUFFER_FRONTEND_API_H
16 #include <ringbuffer/frontend.h>
17 #include <linux/percpu-defs.h>
18 #include <linux/errno.h>
19 #include <linux/prefetch.h>
22 * lib_ring_buffer_get_cpu - Precedes ring buffer reserve/commit.
24 * Disables preemption (acts as a RCU read-side critical section) and keeps a
25 * ring buffer nesting count as supplementary safety net to ensure tracer client
26 * code will never trigger an endless recursion. Returns the processor ID on
27 * success, -EPERM on failure (nesting count too high).
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_get_cpu(const struct lttng_kernel_ring_buffer_config
*config
)
40 rcu_read_lock_sched_notrace();
41 cpu
= smp_processor_id();
42 nesting
= ++per_cpu(lib_ring_buffer_nesting
, cpu
);
45 if (unlikely(nesting
> RING_BUFFER_MAX_NESTING
)) {
47 per_cpu(lib_ring_buffer_nesting
, cpu
)--;
48 rcu_read_unlock_sched_notrace();
55 * lib_ring_buffer_put_cpu - Follows ring buffer reserve/commit.
58 void lib_ring_buffer_put_cpu(const struct lttng_kernel_ring_buffer_config
*config
)
61 (*this_cpu_ptr(&lib_ring_buffer_nesting
))--;
62 rcu_read_unlock_sched_notrace();
66 * lib_ring_buffer_try_reserve is called by lib_ring_buffer_reserve(). It is not
67 * part of the API per se.
69 * returns 0 if reserve ok, or 1 if the slow path must be taken.
72 int lib_ring_buffer_try_reserve(const struct lttng_kernel_ring_buffer_config
*config
,
73 struct lttng_kernel_ring_buffer_ctx
*ctx
,
75 unsigned long *o_begin
, unsigned long *o_end
,
76 unsigned long *o_old
, size_t *before_hdr_pad
)
78 struct lttng_kernel_ring_buffer_channel
*chan
= ctx
->priv
.chan
;
79 struct lttng_kernel_ring_buffer
*buf
= ctx
->priv
.buf
;
80 *o_begin
= v_read(config
, &buf
->offset
);
83 ctx
->priv
.timestamp
= lib_ring_buffer_clock_read(chan
);
84 if ((int64_t) ctx
->priv
.timestamp
== -EIO
)
88 * Prefetch cacheline for read because we have to read the previous
89 * commit counter to increment it and commit seq value to compare it to
92 prefetch(&buf
->commit_hot
[subbuf_index(*o_begin
, chan
)]);
94 if (last_timestamp_overflow(config
, buf
, ctx
->priv
.timestamp
))
95 ctx
->priv
.rflags
|= RING_BUFFER_RFLAG_FULL_TIMESTAMP
;
97 if (unlikely(subbuf_offset(*o_begin
, chan
) == 0))
100 ctx
->priv
.slot_size
= record_header_size(config
, chan
, *o_begin
,
101 before_hdr_pad
, ctx
, client_ctx
);
102 ctx
->priv
.slot_size
+=
103 lib_ring_buffer_align(*o_begin
+ ctx
->priv
.slot_size
,
104 ctx
->largest_align
) + ctx
->data_size
;
105 if (unlikely((subbuf_offset(*o_begin
, chan
) + ctx
->priv
.slot_size
)
106 > chan
->backend
.subbuf_size
))
110 * Record fits in the current buffer and we are not on a switch
111 * boundary. It's safe to write.
113 *o_end
= *o_begin
+ ctx
->priv
.slot_size
;
115 if (unlikely((subbuf_offset(*o_end
, chan
)) == 0))
117 * The offset_end will fall at the very beginning of the next
126 * lib_ring_buffer_reserve - Reserve space in a ring buffer.
127 * @config: ring buffer instance configuration.
128 * @ctx: ring buffer context. (input and output) Must be already initialized.
130 * Atomic wait-free slot reservation. The reserved space starts at the context
131 * "pre_offset". Its length is "slot_size". The associated time-stamp is
136 * -EAGAIN if channel is disabled.
137 * -ENOSPC if event size is too large for packet.
138 * -ENOBUFS if there is currently not enough space in buffer for the event.
139 * -EIO if data cannot be written into the buffer for any other reason.
143 int lib_ring_buffer_reserve(const struct lttng_kernel_ring_buffer_config
*config
,
144 struct lttng_kernel_ring_buffer_ctx
*ctx
,
147 struct lttng_kernel_ring_buffer_channel
*chan
= ctx
->priv
.chan
;
148 struct lttng_kernel_ring_buffer
*buf
;
149 unsigned long o_begin
, o_end
, o_old
;
150 size_t before_hdr_pad
= 0;
152 if (unlikely(atomic_read(&chan
->record_disabled
)))
155 if (config
->alloc
== RING_BUFFER_ALLOC_PER_CPU
)
156 buf
= per_cpu_ptr(chan
->backend
.buf
, ctx
->priv
.reserve_cpu
);
158 buf
= chan
->backend
.buf
;
159 if (unlikely(atomic_read(&buf
->record_disabled
)))
164 * Perform retryable operations.
166 if (unlikely(lib_ring_buffer_try_reserve(config
, ctx
, client_ctx
, &o_begin
,
167 &o_end
, &o_old
, &before_hdr_pad
)))
170 if (unlikely(v_cmpxchg(config
, &ctx
->priv
.buf
->offset
, o_old
, o_end
)
175 * Atomically update last_timestamp. This update races against concurrent
176 * atomic updates, but the race will always cause supplementary
177 * full timestamp record headers, never the opposite (missing a
178 * full timestamp record header when it would be needed).
180 save_last_timestamp(config
, ctx
->priv
.buf
, ctx
->priv
.timestamp
);
183 * Push the reader if necessary
185 lib_ring_buffer_reserve_push_reader(ctx
->priv
.buf
, chan
, o_end
- 1);
188 * Clear noref flag for this subbuffer.
190 lib_ring_buffer_clear_noref(config
, &ctx
->priv
.buf
->backend
,
191 subbuf_index(o_end
- 1, chan
));
193 ctx
->priv
.pre_offset
= o_begin
;
194 ctx
->priv
.buf_offset
= o_begin
+ before_hdr_pad
;
197 return lib_ring_buffer_reserve_slow(ctx
, client_ctx
);
201 * lib_ring_buffer_switch - Perform a sub-buffer switch for a per-cpu buffer.
202 * @config: ring buffer instance configuration.
204 * @mode: buffer switch mode (SWITCH_ACTIVE or SWITCH_FLUSH)
206 * This operation is completely reentrant : can be called while tracing is
207 * active with absolutely no lock held.
209 * Note, however, that as a v_cmpxchg is used for some atomic operations and
210 * requires to be executed locally for per-CPU buffers, this function must be
211 * called from the CPU which owns the buffer for a ACTIVE flush, with preemption
212 * disabled, for RING_BUFFER_SYNC_PER_CPU configuration.
215 void lib_ring_buffer_switch(const struct lttng_kernel_ring_buffer_config
*config
,
216 struct lttng_kernel_ring_buffer
*buf
, enum switch_mode mode
)
218 lib_ring_buffer_switch_slow(buf
, mode
);
221 /* See ring_buffer_frontend_api.h for lib_ring_buffer_reserve(). */
224 * lib_ring_buffer_commit - Commit an record.
225 * @config: ring buffer instance configuration.
226 * @ctx: ring buffer context. (input arguments only)
228 * Atomic unordered slot commit. Increments the commit count in the
229 * specified sub-buffer, and delivers it if necessary.
232 void lib_ring_buffer_commit(const struct lttng_kernel_ring_buffer_config
*config
,
233 const struct lttng_kernel_ring_buffer_ctx
*ctx
)
235 struct lttng_kernel_ring_buffer_channel
*chan
= ctx
->priv
.chan
;
236 struct lttng_kernel_ring_buffer
*buf
= ctx
->priv
.buf
;
237 unsigned long offset_end
= ctx
->priv
.buf_offset
;
238 unsigned long endidx
= subbuf_index(offset_end
- 1, chan
);
239 unsigned long commit_count
;
240 struct commit_counters_hot
*cc_hot
= &buf
->commit_hot
[endidx
];
243 * Must count record before incrementing the commit count.
245 subbuffer_count_record(config
, &buf
->backend
, endidx
);
248 * Order all writes to buffer before the commit count update that will
249 * determine that the subbuffer is full.
251 if (config
->ipi
== RING_BUFFER_IPI_BARRIER
) {
253 * Must write slot data before incrementing commit count. This
254 * compiler barrier is upgraded into a smp_mb() by the IPI sent
261 v_add(config
, ctx
->priv
.slot_size
, &cc_hot
->cc
);
264 * commit count read can race with concurrent OOO commit count updates.
265 * This is only needed for lib_ring_buffer_check_deliver (for
266 * non-polling delivery only) and for
267 * lib_ring_buffer_write_commit_counter. The race can only cause the
268 * counter to be read with the same value more than once, which could
270 * - Multiple delivery for the same sub-buffer (which is handled
271 * gracefully by the reader code) if the value is for a full
272 * sub-buffer. It's important that we can never miss a sub-buffer
273 * delivery. Re-reading the value after the v_add ensures this.
274 * - Reading a commit_count with a higher value that what was actually
275 * added to it for the lib_ring_buffer_write_commit_counter call
276 * (again caused by a concurrent committer). It does not matter,
277 * because this function is interested in the fact that the commit
278 * count reaches back the reserve offset for a specific sub-buffer,
279 * which is completely independent of the order.
281 commit_count
= v_read(config
, &cc_hot
->cc
);
283 lib_ring_buffer_check_deliver(config
, buf
, chan
, offset_end
- 1,
284 commit_count
, endidx
, ctx
);
286 * Update used size at each commit. It's needed only for extracting
287 * ring_buffer buffers from vmcore, after crash.
289 lib_ring_buffer_write_commit_counter(config
, buf
, chan
,
290 offset_end
, commit_count
, cc_hot
);
294 * lib_ring_buffer_try_discard_reserve - Try discarding a record.
295 * @config: ring buffer instance configuration.
296 * @ctx: ring buffer context. (input arguments only)
298 * Only succeeds if no other record has been written after the record to
299 * discard. If discard fails, the record must be committed to the buffer.
301 * Returns 0 upon success, -EPERM if the record cannot be discarded.
304 int lib_ring_buffer_try_discard_reserve(const struct lttng_kernel_ring_buffer_config
*config
,
305 const struct lttng_kernel_ring_buffer_ctx
*ctx
)
307 struct lttng_kernel_ring_buffer
*buf
= ctx
->priv
.buf
;
308 unsigned long end_offset
= ctx
->priv
.pre_offset
+ ctx
->priv
.slot_size
;
311 * We need to ensure that if the cmpxchg succeeds and discards the
312 * record, the next record will record a full timestamp, because
313 * it cannot rely on the last_timestamp associated with the
314 * discarded record to detect overflows. The only way to ensure
315 * this is to set the last_timestamp to 0 (assuming no 64-bit
316 * timestamp overflow), which forces to write a 64-bit timestamp in
319 * Note: if discard fails, we must leave the timestamp in the
320 * record header. It is needed to keep track of timestamp
321 * overflows for the following records.
323 save_last_timestamp(config
, buf
, 0ULL);
325 if (likely(v_cmpxchg(config
, &buf
->offset
, end_offset
, ctx
->priv
.pre_offset
)
333 void channel_record_disable(const struct lttng_kernel_ring_buffer_config
*config
,
334 struct lttng_kernel_ring_buffer_channel
*chan
)
336 atomic_inc(&chan
->record_disabled
);
340 void channel_record_enable(const struct lttng_kernel_ring_buffer_config
*config
,
341 struct lttng_kernel_ring_buffer_channel
*chan
)
343 atomic_dec(&chan
->record_disabled
);
347 void lib_ring_buffer_record_disable(const struct lttng_kernel_ring_buffer_config
*config
,
348 struct lttng_kernel_ring_buffer
*buf
)
350 atomic_inc(&buf
->record_disabled
);
354 void lib_ring_buffer_record_enable(const struct lttng_kernel_ring_buffer_config
*config
,
355 struct lttng_kernel_ring_buffer
*buf
)
357 atomic_dec(&buf
->record_disabled
);
360 #endif /* _LIB_RING_BUFFER_FRONTEND_API_H */