1 #ifndef _LTTNG_RING_BUFFER_FRONTEND_INTERNAL_H
2 #define _LTTNG_RING_BUFFER_FRONTEND_INTERNAL_H
5 * libringbuffer/frontend_internal.h
7 * Ring Buffer Library Synchronization Header (internal helpers).
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
27 * Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
29 * See ring_buffer_frontend.c for more information on wait-free algorithms.
31 * Dual LGPL v2.1/GPL v2 license.
34 #include <urcu/compiler.h>
35 #include <urcu/tls-compat.h>
39 #include <lttng/ringbuffer-config.h>
40 #include "backend_types.h"
41 #include "frontend_types.h"
44 /* Buffer offset macros */
46 /* buf_trunc mask selects only the buffer number. */
48 unsigned long buf_trunc(unsigned long offset
, struct channel
*chan
)
50 return offset
& ~(chan
->backend
.buf_size
- 1);
54 /* Select the buffer number value (counter). */
56 unsigned long buf_trunc_val(unsigned long offset
, struct channel
*chan
)
58 return buf_trunc(offset
, chan
) >> chan
->backend
.buf_size_order
;
61 /* buf_offset mask selects only the offset within the current buffer. */
63 unsigned long buf_offset(unsigned long offset
, struct channel
*chan
)
65 return offset
& (chan
->backend
.buf_size
- 1);
68 /* subbuf_offset mask selects the offset within the current subbuffer. */
70 unsigned long subbuf_offset(unsigned long offset
, struct channel
*chan
)
72 return offset
& (chan
->backend
.subbuf_size
- 1);
75 /* subbuf_trunc mask selects the subbuffer number. */
77 unsigned long subbuf_trunc(unsigned long offset
, struct channel
*chan
)
79 return offset
& ~(chan
->backend
.subbuf_size
- 1);
82 /* subbuf_align aligns the offset to the next subbuffer. */
84 unsigned long subbuf_align(unsigned long offset
, struct channel
*chan
)
86 return (offset
+ chan
->backend
.subbuf_size
)
87 & ~(chan
->backend
.subbuf_size
- 1);
90 /* subbuf_index returns the index of the current subbuffer within the buffer. */
92 unsigned long subbuf_index(unsigned long offset
, struct channel
*chan
)
94 return buf_offset(offset
, chan
) >> chan
->backend
.subbuf_size_order
;
98 * Last TSC comparison functions. Check if the current TSC overflows tsc_bits
99 * bits from the last TSC read. When overflows are detected, the full 64-bit
100 * timestamp counter should be written in the record header. Reads and writes
101 * last_tsc atomically.
104 #if (CAA_BITS_PER_LONG == 32)
106 void save_last_tsc(const struct lttng_ust_lib_ring_buffer_config
*config
,
107 struct lttng_ust_lib_ring_buffer
*buf
, uint64_t tsc
)
109 if (config
->tsc_bits
== 0 || config
->tsc_bits
== 64)
113 * Ensure the compiler performs this update in a single instruction.
115 v_set(config
, &buf
->last_tsc
, (unsigned long)(tsc
>> config
->tsc_bits
));
119 int last_tsc_overflow(const struct lttng_ust_lib_ring_buffer_config
*config
,
120 struct lttng_ust_lib_ring_buffer
*buf
, uint64_t tsc
)
122 unsigned long tsc_shifted
;
124 if (config
->tsc_bits
== 0 || config
->tsc_bits
== 64)
127 tsc_shifted
= (unsigned long)(tsc
>> config
->tsc_bits
);
128 if (caa_unlikely(tsc_shifted
129 - (unsigned long)v_read(config
, &buf
->last_tsc
)))
136 void save_last_tsc(const struct lttng_ust_lib_ring_buffer_config
*config
,
137 struct lttng_ust_lib_ring_buffer
*buf
, uint64_t tsc
)
139 if (config
->tsc_bits
== 0 || config
->tsc_bits
== 64)
142 v_set(config
, &buf
->last_tsc
, (unsigned long)tsc
);
146 int last_tsc_overflow(const struct lttng_ust_lib_ring_buffer_config
*config
,
147 struct lttng_ust_lib_ring_buffer
*buf
, uint64_t tsc
)
149 if (config
->tsc_bits
== 0 || config
->tsc_bits
== 64)
152 if (caa_unlikely((tsc
- v_read(config
, &buf
->last_tsc
))
153 >> config
->tsc_bits
))
161 int lib_ring_buffer_reserve_slow(struct lttng_ust_lib_ring_buffer_ctx
*ctx
,
165 void lib_ring_buffer_switch_slow(struct lttng_ust_lib_ring_buffer
*buf
,
166 enum switch_mode mode
,
167 struct lttng_ust_shm_handle
*handle
);
169 void lib_ring_buffer_check_deliver_slow(const struct lttng_ust_lib_ring_buffer_config
*config
,
170 struct lttng_ust_lib_ring_buffer
*buf
,
171 struct channel
*chan
,
172 unsigned long offset
,
173 unsigned long commit_count
,
175 struct lttng_ust_shm_handle
*handle
,
178 /* Buffer write helpers */
181 void lib_ring_buffer_reserve_push_reader(struct lttng_ust_lib_ring_buffer
*buf
,
182 struct channel
*chan
,
183 unsigned long offset
)
185 unsigned long consumed_old
, consumed_new
;
188 consumed_old
= uatomic_read(&buf
->consumed
);
190 * If buffer is in overwrite mode, push the reader consumed
191 * count if the write position has reached it and we are not
192 * at the first iteration (don't push the reader farther than
193 * the writer). This operation can be done concurrently by many
194 * writers in the same buffer, the writer being at the farthest
195 * write position sub-buffer index in the buffer being the one
196 * which will win this loop.
198 if (caa_unlikely(subbuf_trunc(offset
, chan
)
199 - subbuf_trunc(consumed_old
, chan
)
200 >= chan
->backend
.buf_size
))
201 consumed_new
= subbuf_align(consumed_old
, chan
);
204 } while (caa_unlikely(uatomic_cmpxchg(&buf
->consumed
, consumed_old
,
205 consumed_new
) != consumed_old
));
209 int lib_ring_buffer_pending_data(const struct lttng_ust_lib_ring_buffer_config
*config
,
210 struct lttng_ust_lib_ring_buffer
*buf
,
211 struct channel
*chan
)
213 return !!subbuf_offset(v_read(config
, &buf
->offset
), chan
);
217 unsigned long lib_ring_buffer_get_data_size(const struct lttng_ust_lib_ring_buffer_config
*config
,
218 struct lttng_ust_lib_ring_buffer
*buf
,
220 struct lttng_ust_shm_handle
*handle
)
222 return subbuffer_get_data_size(config
, &buf
->backend
, idx
, handle
);
226 * Check if all space reservation in a buffer have been committed. This helps
227 * knowing if an execution context is nested (for per-cpu buffers only).
228 * This is a very specific ftrace use-case, so we keep this as "internal" API.
231 int lib_ring_buffer_reserve_committed(const struct lttng_ust_lib_ring_buffer_config
*config
,
232 struct lttng_ust_lib_ring_buffer
*buf
,
233 struct channel
*chan
,
234 struct lttng_ust_shm_handle
*handle
)
236 unsigned long offset
, idx
, commit_count
;
237 struct commit_counters_hot
*cc_hot
;
239 CHAN_WARN_ON(chan
, config
->alloc
!= RING_BUFFER_ALLOC_PER_CPU
);
240 CHAN_WARN_ON(chan
, config
->sync
!= RING_BUFFER_SYNC_PER_CPU
);
243 * Read offset and commit count in a loop so they are both read
244 * atomically wrt interrupts. By deal with interrupt concurrency by
245 * restarting both reads if the offset has been pushed. Note that given
246 * we only have to deal with interrupt concurrency here, an interrupt
247 * modifying the commit count will also modify "offset", so it is safe
248 * to only check for offset modifications.
251 offset
= v_read(config
, &buf
->offset
);
252 idx
= subbuf_index(offset
, chan
);
253 cc_hot
= shmp_index(handle
, buf
->commit_hot
, idx
);
254 if (caa_unlikely(!cc_hot
))
256 commit_count
= v_read(config
, &cc_hot
->cc
);
257 } while (offset
!= v_read(config
, &buf
->offset
));
259 return ((buf_trunc(offset
, chan
) >> chan
->backend
.num_subbuf_order
)
260 - (commit_count
& chan
->commit_count_mask
) == 0);
264 * Receive end of subbuffer TSC as parameter. It has been read in the
265 * space reservation loop of either reserve or switch, which ensures it
266 * progresses monotonically with event records in the buffer. Therefore,
267 * it ensures that the end timestamp of a subbuffer is <= begin
268 * timestamp of the following subbuffers.
271 void lib_ring_buffer_check_deliver(const struct lttng_ust_lib_ring_buffer_config
*config
,
272 struct lttng_ust_lib_ring_buffer
*buf
,
273 struct channel
*chan
,
274 unsigned long offset
,
275 unsigned long commit_count
,
277 struct lttng_ust_shm_handle
*handle
,
280 unsigned long old_commit_count
= commit_count
281 - chan
->backend
.subbuf_size
;
283 /* Check if all commits have been done */
284 if (caa_unlikely((buf_trunc(offset
, chan
) >> chan
->backend
.num_subbuf_order
)
285 - (old_commit_count
& chan
->commit_count_mask
) == 0))
286 lib_ring_buffer_check_deliver_slow(config
, buf
, chan
, offset
,
287 commit_count
, idx
, handle
, tsc
);
291 * lib_ring_buffer_write_commit_counter
293 * For flight recording. must be called after commit.
294 * This function increments the subbuffer's commit_seq counter each time the
295 * commit count reaches back the reserve offset (modulo subbuffer size). It is
296 * useful for crash dump.
299 void lib_ring_buffer_write_commit_counter(const struct lttng_ust_lib_ring_buffer_config
*config
,
300 struct lttng_ust_lib_ring_buffer
*buf
,
301 struct channel
*chan
,
302 unsigned long buf_offset
,
303 unsigned long commit_count
,
304 struct lttng_ust_shm_handle
*handle
,
305 struct commit_counters_hot
*cc_hot
)
307 unsigned long commit_seq_old
;
309 if (config
->oops
!= RING_BUFFER_OOPS_CONSISTENCY
)
313 * subbuf_offset includes commit_count_mask. We can simply
314 * compare the offsets within the subbuffer without caring about
315 * buffer full/empty mismatch because offset is never zero here
316 * (subbuffer header and record headers have non-zero length).
318 if (caa_unlikely(subbuf_offset(buf_offset
- commit_count
, chan
)))
321 commit_seq_old
= v_read(config
, &cc_hot
->seq
);
322 if (caa_likely((long) (commit_seq_old
- commit_count
) < 0))
323 v_set(config
, &cc_hot
->seq
, commit_count
);
326 extern int lib_ring_buffer_create(struct lttng_ust_lib_ring_buffer
*buf
,
327 struct channel_backend
*chanb
, int cpu
,
328 struct lttng_ust_shm_handle
*handle
,
329 struct shm_object
*shmobj
);
330 extern void lib_ring_buffer_free(struct lttng_ust_lib_ring_buffer
*buf
,
331 struct lttng_ust_shm_handle
*handle
);
333 /* Keep track of trap nesting inside ring buffer code */
334 extern DECLARE_URCU_TLS(unsigned int, lib_ring_buffer_nesting
);
336 #endif /* _LTTNG_RING_BUFFER_FRONTEND_INTERNAL_H */