1 #ifndef _LIB_RING_BUFFER_FRONTEND_API_H
2 #define _LIB_RING_BUFFER_FRONTEND_API_H
5 * lib/ringbuffer/frontend_api.h
7 * Ring Buffer Library Synchronization Header (buffer write API).
9 * Copyright (C) 2005-2012 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
11 * This library is free software; you can redistribute it and/or
12 * modify it under the terms of the GNU Lesser General Public
13 * License as published by the Free Software Foundation; only
14 * version 2.1 of the License.
16 * This library is distributed in the hope that it will be useful,
17 * but WITHOUT ANY WARRANTY; without even the implied warranty of
18 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
19 * Lesser General Public License for more details.
21 * You should have received a copy of the GNU Lesser General Public
22 * License along with this library; if not, write to the Free Software
23 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
26 * Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
28 * See ring_buffer_frontend.c for more information on wait-free algorithms.
29 * See linux/ringbuffer/frontend.h for channel allocation and read-side API.
32 #include "../../wrapper/ringbuffer/frontend.h"
33 #include <linux/errno.h>
34 #include <linux/prefetch.h>
37 * lib_ring_buffer_get_cpu - Precedes ring buffer reserve/commit.
39 * Disables preemption (acts as a RCU read-side critical section) and keeps a
40 * ring buffer nesting count as supplementary safety net to ensure tracer client
41 * code will never trigger an endless recursion. Returns the processor ID on
42 * success, -EPERM on failure (nesting count too high).
44 * asm volatile and "memory" clobber prevent the compiler from moving
45 * instructions out of the ring buffer nesting count. This is required to ensure
46 * that probe side-effects which can cause recursion (e.g. unforeseen traps,
47 * divisions by 0, ...) are triggered within the incremented nesting count
51 int lib_ring_buffer_get_cpu(const struct lib_ring_buffer_config
*config
)
55 rcu_read_lock_sched_notrace();
56 cpu
= smp_processor_id();
57 nesting
= ++per_cpu(lib_ring_buffer_nesting
, cpu
);
60 if (unlikely(nesting
> 4)) {
62 per_cpu(lib_ring_buffer_nesting
, cpu
)--;
63 rcu_read_unlock_sched_notrace();
70 * lib_ring_buffer_put_cpu - Follows ring buffer reserve/commit.
73 void lib_ring_buffer_put_cpu(const struct lib_ring_buffer_config
*config
)
76 __get_cpu_var(lib_ring_buffer_nesting
)--;
77 rcu_read_unlock_sched_notrace();
81 * lib_ring_buffer_try_reserve is called by lib_ring_buffer_reserve(). It is not
82 * part of the API per se.
84 * returns 0 if reserve ok, or 1 if the slow path must be taken.
87 int lib_ring_buffer_try_reserve(const struct lib_ring_buffer_config
*config
,
88 struct lib_ring_buffer_ctx
*ctx
,
89 unsigned long *o_begin
, unsigned long *o_end
,
90 unsigned long *o_old
, size_t *before_hdr_pad
)
92 struct channel
*chan
= ctx
->chan
;
93 struct lib_ring_buffer
*buf
= ctx
->buf
;
94 *o_begin
= v_read(config
, &buf
->offset
);
97 ctx
->tsc
= lib_ring_buffer_clock_read(chan
);
98 if ((int64_t) ctx
->tsc
== -EIO
)
102 * Prefetch cacheline for read because we have to read the previous
103 * commit counter to increment it and commit seq value to compare it to
104 * the commit counter.
106 prefetch(&buf
->commit_hot
[subbuf_index(*o_begin
, chan
)]);
108 if (last_tsc_overflow(config
, buf
, ctx
->tsc
))
109 ctx
->rflags
|= RING_BUFFER_RFLAG_FULL_TSC
;
111 if (unlikely(subbuf_offset(*o_begin
, chan
) == 0))
114 ctx
->slot_size
= record_header_size(config
, chan
, *o_begin
,
115 before_hdr_pad
, ctx
);
117 lib_ring_buffer_align(*o_begin
+ ctx
->slot_size
,
118 ctx
->largest_align
) + ctx
->data_size
;
119 if (unlikely((subbuf_offset(*o_begin
, chan
) + ctx
->slot_size
)
120 > chan
->backend
.subbuf_size
))
124 * Record fits in the current buffer and we are not on a switch
125 * boundary. It's safe to write.
127 *o_end
= *o_begin
+ ctx
->slot_size
;
132 * lib_ring_buffer_reserve - Reserve space in a ring buffer.
133 * @config: ring buffer instance configuration.
134 * @ctx: ring buffer context. (input and output) Must be already initialized.
136 * Atomic wait-free slot reservation. The reserved space starts at the context
137 * "pre_offset". Its length is "slot_size". The associated time-stamp is "tsc".
141 * -EAGAIN if channel is disabled.
142 * -ENOSPC if event size is too large for packet.
143 * -ENOBUFS if there is currently not enough space in buffer for the event.
144 * -EIO if data cannot be written into the buffer for any other reason.
148 int lib_ring_buffer_reserve(const struct lib_ring_buffer_config
*config
,
149 struct lib_ring_buffer_ctx
*ctx
)
151 struct channel
*chan
= ctx
->chan
;
152 struct lib_ring_buffer
*buf
;
153 unsigned long o_begin
, o_end
, o_old
;
154 size_t before_hdr_pad
= 0;
156 if (atomic_read(&chan
->record_disabled
))
159 if (config
->alloc
== RING_BUFFER_ALLOC_PER_CPU
)
160 buf
= per_cpu_ptr(chan
->backend
.buf
, ctx
->cpu
);
162 buf
= chan
->backend
.buf
;
163 if (atomic_read(&buf
->record_disabled
))
168 * Perform retryable operations.
170 if (unlikely(lib_ring_buffer_try_reserve(config
, ctx
, &o_begin
,
171 &o_end
, &o_old
, &before_hdr_pad
)))
174 if (unlikely(v_cmpxchg(config
, &ctx
->buf
->offset
, o_old
, o_end
)
179 * Atomically update last_tsc. This update races against concurrent
180 * atomic updates, but the race will always cause supplementary full TSC
181 * record headers, never the opposite (missing a full TSC record header
182 * when it would be needed).
184 save_last_tsc(config
, ctx
->buf
, ctx
->tsc
);
187 * Push the reader if necessary
189 lib_ring_buffer_reserve_push_reader(ctx
->buf
, chan
, o_end
- 1);
192 * Clear noref flag for this subbuffer.
194 lib_ring_buffer_clear_noref(config
, &ctx
->buf
->backend
,
195 subbuf_index(o_end
- 1, chan
));
197 ctx
->pre_offset
= o_begin
;
198 ctx
->buf_offset
= o_begin
+ before_hdr_pad
;
201 return lib_ring_buffer_reserve_slow(ctx
);
205 * lib_ring_buffer_switch - Perform a sub-buffer switch for a per-cpu buffer.
206 * @config: ring buffer instance configuration.
208 * @mode: buffer switch mode (SWITCH_ACTIVE or SWITCH_FLUSH)
210 * This operation is completely reentrant : can be called while tracing is
211 * active with absolutely no lock held.
213 * Note, however, that as a v_cmpxchg is used for some atomic operations and
214 * requires to be executed locally for per-CPU buffers, this function must be
215 * called from the CPU which owns the buffer for a ACTIVE flush, with preemption
216 * disabled, for RING_BUFFER_SYNC_PER_CPU configuration.
219 void lib_ring_buffer_switch(const struct lib_ring_buffer_config
*config
,
220 struct lib_ring_buffer
*buf
, enum switch_mode mode
)
222 lib_ring_buffer_switch_slow(buf
, mode
);
225 /* See ring_buffer_frontend_api.h for lib_ring_buffer_reserve(). */
228 * lib_ring_buffer_commit - Commit an record.
229 * @config: ring buffer instance configuration.
230 * @ctx: ring buffer context. (input arguments only)
232 * Atomic unordered slot commit. Increments the commit count in the
233 * specified sub-buffer, and delivers it if necessary.
236 void lib_ring_buffer_commit(const struct lib_ring_buffer_config
*config
,
237 const struct lib_ring_buffer_ctx
*ctx
)
239 struct channel
*chan
= ctx
->chan
;
240 struct lib_ring_buffer
*buf
= ctx
->buf
;
241 unsigned long offset_end
= ctx
->buf_offset
;
242 unsigned long endidx
= subbuf_index(offset_end
- 1, chan
);
243 unsigned long commit_count
;
246 * Must count record before incrementing the commit count.
248 subbuffer_count_record(config
, &buf
->backend
, endidx
);
251 * Order all writes to buffer before the commit count update that will
252 * determine that the subbuffer is full.
254 if (config
->ipi
== RING_BUFFER_IPI_BARRIER
) {
256 * Must write slot data before incrementing commit count. This
257 * compiler barrier is upgraded into a smp_mb() by the IPI sent
264 v_add(config
, ctx
->slot_size
, &buf
->commit_hot
[endidx
].cc
);
267 * commit count read can race with concurrent OOO commit count updates.
268 * This is only needed for lib_ring_buffer_check_deliver (for
269 * non-polling delivery only) and for
270 * lib_ring_buffer_write_commit_counter. The race can only cause the
271 * counter to be read with the same value more than once, which could
273 * - Multiple delivery for the same sub-buffer (which is handled
274 * gracefully by the reader code) if the value is for a full
275 * sub-buffer. It's important that we can never miss a sub-buffer
276 * delivery. Re-reading the value after the v_add ensures this.
277 * - Reading a commit_count with a higher value that what was actually
278 * added to it for the lib_ring_buffer_write_commit_counter call
279 * (again caused by a concurrent committer). It does not matter,
280 * because this function is interested in the fact that the commit
281 * count reaches back the reserve offset for a specific sub-buffer,
282 * which is completely independent of the order.
284 commit_count
= v_read(config
, &buf
->commit_hot
[endidx
].cc
);
286 lib_ring_buffer_check_deliver(config
, buf
, chan
, offset_end
- 1,
287 commit_count
, endidx
);
289 * Update used size at each commit. It's needed only for extracting
290 * ring_buffer buffers from vmcore, after crash.
292 lib_ring_buffer_write_commit_counter(config
, buf
, chan
, endidx
,
293 ctx
->buf_offset
, commit_count
,
298 * lib_ring_buffer_try_discard_reserve - Try discarding a record.
299 * @config: ring buffer instance configuration.
300 * @ctx: ring buffer context. (input arguments only)
302 * Only succeeds if no other record has been written after the record to
303 * discard. If discard fails, the record must be committed to the buffer.
305 * Returns 0 upon success, -EPERM if the record cannot be discarded.
308 int lib_ring_buffer_try_discard_reserve(const struct lib_ring_buffer_config
*config
,
309 const struct lib_ring_buffer_ctx
*ctx
)
311 struct lib_ring_buffer
*buf
= ctx
->buf
;
312 unsigned long end_offset
= ctx
->pre_offset
+ ctx
->slot_size
;
315 * We need to ensure that if the cmpxchg succeeds and discards the
316 * record, the next record will record a full TSC, because it cannot
317 * rely on the last_tsc associated with the discarded record to detect
318 * overflows. The only way to ensure this is to set the last_tsc to 0
319 * (assuming no 64-bit TSC overflow), which forces to write a 64-bit
320 * timestamp in the next record.
322 * Note: if discard fails, we must leave the TSC in the record header.
323 * It is needed to keep track of TSC overflows for the following
326 save_last_tsc(config
, buf
, 0ULL);
328 if (likely(v_cmpxchg(config
, &buf
->offset
, end_offset
, ctx
->pre_offset
)
336 void channel_record_disable(const struct lib_ring_buffer_config
*config
,
337 struct channel
*chan
)
339 atomic_inc(&chan
->record_disabled
);
343 void channel_record_enable(const struct lib_ring_buffer_config
*config
,
344 struct channel
*chan
)
346 atomic_dec(&chan
->record_disabled
);
350 void lib_ring_buffer_record_disable(const struct lib_ring_buffer_config
*config
,
351 struct lib_ring_buffer
*buf
)
353 atomic_inc(&buf
->record_disabled
);
357 void lib_ring_buffer_record_enable(const struct lib_ring_buffer_config
*config
,
358 struct lib_ring_buffer
*buf
)
360 atomic_dec(&buf
->record_disabled
);
363 #endif /* _LIB_RING_BUFFER_FRONTEND_API_H */