2 * SPDX-License-Identifier: LGPL-2.1-only
4 * Copyright (C) 2005-2012 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
6 * Ring buffer wait-free buffer synchronization. Producer-consumer and flight
7 * recorder (overwrite) modes. See thesis:
9 * Desnoyers, Mathieu (2009), "Low-Impact Operating System Tracing", Ph.D.
10 * dissertation, Ecole Polytechnique de Montreal.
11 * http://www.lttng.org/pub/thesis/desnoyers-dissertation-2009-12.pdf
13 * - Algorithm presentation in Chapter 5:
14 * "Lockless Multi-Core High-Throughput Buffering".
15 * - Algorithm formal verification in Section 8.6:
16 * "Formal verification of LTTng"
19 * Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
21 * Inspired from LTT and RelayFS:
22 * Karim Yaghmour <karim@opersys.com>
23 * Tom Zanussi <zanussi@us.ibm.com>
24 * Bob Wisniewski <bob@watson.ibm.com>
26 * Bob Wisniewski <bob@watson.ibm.com>
28 * Buffer reader semantic :
31 * while buffer is not finalized and empty
33 * - if return value != 0, continue
34 * - splice one subbuffer worth of data to a pipe
35 * - splice the data from pipe to disk/network
40 #include <sys/types.h>
49 #include <urcu/compiler.h>
51 #include <urcu/tls-compat.h>
53 #include <ust-helper.h>
55 #include <lttng/align.h>
57 #include <lttng/ringbuffer-context.h>
58 #include "ringbuffer-config.h"
64 #include "../liblttng-ust/compat.h" /* For ENODATA */
66 /* Print DBG() messages about events lost only every 1048576 hits */
67 #define DBG_PRINT_NR_LOST (1UL << 20)
69 #define LTTNG_UST_RB_SIG_FLUSH SIGRTMIN
70 #define LTTNG_UST_RB_SIG_READ SIGRTMIN + 1
71 #define LTTNG_UST_RB_SIG_TEARDOWN SIGRTMIN + 2
72 #define CLOCKID CLOCK_MONOTONIC
73 #define LTTNG_UST_RING_BUFFER_GET_RETRY 10
74 #define LTTNG_UST_RING_BUFFER_RETRY_DELAY_MS 10
75 #define RETRY_DELAY_MS 100 /* 100 ms. */
78 * Non-static to ensure the compiler does not optimize away the xor.
80 uint8_t lttng_crash_magic_xor
[] = RB_CRASH_DUMP_ABI_MAGIC_XOR
;
83 * Use POSIX SHM: shm_open(3) and shm_unlink(3).
84 * close(2) to close the fd returned by shm_open.
85 * shm_unlink releases the shared memory object name.
86 * ftruncate(2) sets the size of the memory object.
87 * mmap/munmap maps the shared memory obj to a virtual address in the
88 * calling proceess (should be done both in libust and consumer).
89 * See shm_overview(7) for details.
90 * Pass file descriptor returned by shm_open(3) to ltt-sessiond through
93 * Since we don't need to access the object using its name, we can
94 * immediately shm_unlink(3) it, and only keep the handle with its file
99 * Internal structure representing offsets to use at a sub-buffer switch.
101 struct switch_offsets
{
102 unsigned long begin
, end
, old
;
103 size_t pre_header_padding
, size
;
104 unsigned int switch_new_start
:1, switch_new_end
:1, switch_old_start
:1,
108 DEFINE_URCU_TLS(unsigned int, lib_ring_buffer_nesting
);
111 * wakeup_fd_mutex protects wakeup fd use by timer from concurrent
114 static pthread_mutex_t wakeup_fd_mutex
= PTHREAD_MUTEX_INITIALIZER
;
117 void lib_ring_buffer_print_errors(struct lttng_ust_lib_ring_buffer_channel
*chan
,
118 struct lttng_ust_lib_ring_buffer
*buf
, int cpu
,
119 struct lttng_ust_shm_handle
*handle
);
122 * Handle timer teardown race wrt memory free of private data by
123 * ring buffer signals are handled by a single thread, which permits
124 * a synchronization point between handling of each signal.
125 * Protected by the lock within the structure.
127 struct timer_signal_data
{
128 pthread_t tid
; /* thread id managing signals */
131 pthread_mutex_t lock
;
134 static struct timer_signal_data timer_signal
= {
138 .lock
= PTHREAD_MUTEX_INITIALIZER
,
141 static bool lttng_ust_allow_blocking
;
143 void lttng_ust_ringbuffer_set_allow_blocking(void)
145 lttng_ust_allow_blocking
= true;
148 /* Get blocking timeout, in ms */
149 static int lttng_ust_ringbuffer_get_timeout(struct lttng_ust_lib_ring_buffer_channel
*chan
)
151 if (!lttng_ust_allow_blocking
)
153 return chan
->u
.s
.blocking_timeout_ms
;
157 * lib_ring_buffer_reset - Reset ring buffer to initial values.
160 * Effectively empty the ring buffer. Should be called when the buffer is not
161 * used for writing. The ring buffer can be opened for reading, but the reader
162 * should not be using the iterator concurrently with reset. The previous
163 * current iterator record is reset.
165 void lib_ring_buffer_reset(struct lttng_ust_lib_ring_buffer
*buf
,
166 struct lttng_ust_shm_handle
*handle
)
168 struct lttng_ust_lib_ring_buffer_channel
*chan
;
169 const struct lttng_ust_lib_ring_buffer_config
*config
;
172 chan
= shmp(handle
, buf
->backend
.chan
);
175 config
= &chan
->backend
.config
;
177 * Reset iterator first. It will put the subbuffer if it currently holds
180 v_set(config
, &buf
->offset
, 0);
181 for (i
= 0; i
< chan
->backend
.num_subbuf
; i
++) {
182 struct commit_counters_hot
*cc_hot
;
183 struct commit_counters_cold
*cc_cold
;
186 cc_hot
= shmp_index(handle
, buf
->commit_hot
, i
);
189 cc_cold
= shmp_index(handle
, buf
->commit_cold
, i
);
192 ts_end
= shmp_index(handle
, buf
->ts_end
, i
);
195 v_set(config
, &cc_hot
->cc
, 0);
196 v_set(config
, &cc_hot
->seq
, 0);
197 v_set(config
, &cc_cold
->cc_sb
, 0);
200 uatomic_set(&buf
->consumed
, 0);
201 uatomic_set(&buf
->record_disabled
, 0);
202 v_set(config
, &buf
->last_tsc
, 0);
203 lib_ring_buffer_backend_reset(&buf
->backend
, handle
);
204 /* Don't reset number of active readers */
205 v_set(config
, &buf
->records_lost_full
, 0);
206 v_set(config
, &buf
->records_lost_wrap
, 0);
207 v_set(config
, &buf
->records_lost_big
, 0);
208 v_set(config
, &buf
->records_count
, 0);
209 v_set(config
, &buf
->records_overrun
, 0);
214 * channel_reset - Reset channel to initial values.
217 * Effectively empty the channel. Should be called when the channel is not used
218 * for writing. The channel can be opened for reading, but the reader should not
219 * be using the iterator concurrently with reset. The previous current iterator
222 void channel_reset(struct lttng_ust_lib_ring_buffer_channel
*chan
)
225 * Reset iterators first. Will put the subbuffer if held for reading.
227 uatomic_set(&chan
->record_disabled
, 0);
228 /* Don't reset commit_count_mask, still valid */
229 channel_backend_reset(&chan
->backend
);
230 /* Don't reset switch/read timer interval */
231 /* Don't reset notifiers and notifier enable bits */
232 /* Don't reset reader reference count */
236 void init_crash_abi(const struct lttng_ust_lib_ring_buffer_config
*config
,
237 struct lttng_crash_abi
*crash_abi
,
238 struct lttng_ust_lib_ring_buffer
*buf
,
239 struct channel_backend
*chanb
,
240 struct shm_object
*shmobj
,
241 struct lttng_ust_shm_handle
*handle
)
245 for (i
= 0; i
< RB_CRASH_DUMP_ABI_MAGIC_LEN
; i
++)
246 crash_abi
->magic
[i
] = lttng_crash_magic_xor
[i
] ^ 0xFF;
247 crash_abi
->mmap_length
= shmobj
->memory_map_size
;
248 crash_abi
->endian
= RB_CRASH_ENDIAN
;
249 crash_abi
->major
= RB_CRASH_DUMP_ABI_MAJOR
;
250 crash_abi
->minor
= RB_CRASH_DUMP_ABI_MINOR
;
251 crash_abi
->word_size
= sizeof(unsigned long);
252 crash_abi
->layout_type
= LTTNG_CRASH_TYPE_UST
;
254 /* Offset of fields */
255 crash_abi
->offset
.prod_offset
=
256 (uint32_t) ((char *) &buf
->offset
- (char *) buf
);
257 crash_abi
->offset
.consumed_offset
=
258 (uint32_t) ((char *) &buf
->consumed
- (char *) buf
);
259 crash_abi
->offset
.commit_hot_array
=
260 (uint32_t) ((char *) shmp(handle
, buf
->commit_hot
) - (char *) buf
);
261 crash_abi
->offset
.commit_hot_seq
=
262 offsetof(struct commit_counters_hot
, seq
);
263 crash_abi
->offset
.buf_wsb_array
=
264 (uint32_t) ((char *) shmp(handle
, buf
->backend
.buf_wsb
) - (char *) buf
);
265 crash_abi
->offset
.buf_wsb_id
=
266 offsetof(struct lttng_ust_lib_ring_buffer_backend_subbuffer
, id
);
267 crash_abi
->offset
.sb_array
=
268 (uint32_t) ((char *) shmp(handle
, buf
->backend
.array
) - (char *) buf
);
269 crash_abi
->offset
.sb_array_shmp_offset
=
270 offsetof(struct lttng_ust_lib_ring_buffer_backend_pages_shmp
,
272 crash_abi
->offset
.sb_backend_p_offset
=
273 offsetof(struct lttng_ust_lib_ring_buffer_backend_pages
,
277 crash_abi
->length
.prod_offset
= sizeof(buf
->offset
);
278 crash_abi
->length
.consumed_offset
= sizeof(buf
->consumed
);
279 crash_abi
->length
.commit_hot_seq
=
280 sizeof(((struct commit_counters_hot
*) NULL
)->seq
);
281 crash_abi
->length
.buf_wsb_id
=
282 sizeof(((struct lttng_ust_lib_ring_buffer_backend_subbuffer
*) NULL
)->id
);
283 crash_abi
->length
.sb_array_shmp_offset
=
284 sizeof(((struct lttng_ust_lib_ring_buffer_backend_pages_shmp
*) NULL
)->shmp
._ref
.offset
);
285 crash_abi
->length
.sb_backend_p_offset
=
286 sizeof(((struct lttng_ust_lib_ring_buffer_backend_pages
*) NULL
)->p
._ref
.offset
);
289 crash_abi
->stride
.commit_hot_array
=
290 sizeof(struct commit_counters_hot
);
291 crash_abi
->stride
.buf_wsb_array
=
292 sizeof(struct lttng_ust_lib_ring_buffer_backend_subbuffer
);
293 crash_abi
->stride
.sb_array
=
294 sizeof(struct lttng_ust_lib_ring_buffer_backend_pages_shmp
);
296 /* Buffer constants */
297 crash_abi
->buf_size
= chanb
->buf_size
;
298 crash_abi
->subbuf_size
= chanb
->subbuf_size
;
299 crash_abi
->num_subbuf
= chanb
->num_subbuf
;
300 crash_abi
->mode
= (uint32_t) chanb
->config
.mode
;
302 if (config
->cb
.content_size_field
) {
303 size_t offset
, length
;
305 config
->cb
.content_size_field(config
, &offset
, &length
);
306 crash_abi
->offset
.content_size
= offset
;
307 crash_abi
->length
.content_size
= length
;
309 crash_abi
->offset
.content_size
= 0;
310 crash_abi
->length
.content_size
= 0;
312 if (config
->cb
.packet_size_field
) {
313 size_t offset
, length
;
315 config
->cb
.packet_size_field(config
, &offset
, &length
);
316 crash_abi
->offset
.packet_size
= offset
;
317 crash_abi
->length
.packet_size
= length
;
319 crash_abi
->offset
.packet_size
= 0;
320 crash_abi
->length
.packet_size
= 0;
325 * Must be called under cpu hotplug protection.
327 int lib_ring_buffer_create(struct lttng_ust_lib_ring_buffer
*buf
,
328 struct channel_backend
*chanb
, int cpu
,
329 struct lttng_ust_shm_handle
*handle
,
330 struct shm_object
*shmobj
)
332 const struct lttng_ust_lib_ring_buffer_config
*config
= &chanb
->config
;
333 struct lttng_ust_lib_ring_buffer_channel
*chan
= caa_container_of(chanb
,
334 struct lttng_ust_lib_ring_buffer_channel
, backend
);
335 struct lttng_ust_lib_ring_buffer_backend_subbuffer
*wsb
;
336 struct lttng_ust_lib_ring_buffer_channel
*shmp_chan
;
337 struct commit_counters_hot
*cc_hot
;
338 void *priv
= channel_get_private(chan
);
339 size_t subbuf_header_size
;
343 /* Test for cpu hotplug */
344 if (buf
->backend
.allocated
)
347 align_shm(shmobj
, __alignof__(struct commit_counters_hot
));
348 set_shmp(buf
->commit_hot
,
350 sizeof(struct commit_counters_hot
) * chan
->backend
.num_subbuf
));
351 if (!shmp(handle
, buf
->commit_hot
)) {
355 align_shm(shmobj
, __alignof__(struct commit_counters_cold
));
356 set_shmp(buf
->commit_cold
,
358 sizeof(struct commit_counters_cold
) * chan
->backend
.num_subbuf
));
359 if (!shmp(handle
, buf
->commit_cold
)) {
364 align_shm(shmobj
, __alignof__(uint64_t));
365 set_shmp(buf
->ts_end
,
367 sizeof(uint64_t) * chan
->backend
.num_subbuf
));
368 if (!shmp(handle
, buf
->ts_end
)) {
370 goto free_commit_cold
;
374 ret
= lib_ring_buffer_backend_create(&buf
->backend
, &chan
->backend
,
375 cpu
, handle
, shmobj
);
381 * Write the subbuffer header for first subbuffer so we know the total
382 * duration of data gathering.
384 subbuf_header_size
= config
->cb
.subbuffer_header_size();
385 v_set(config
, &buf
->offset
, subbuf_header_size
);
386 wsb
= shmp_index(handle
, buf
->backend
.buf_wsb
, 0);
391 subbuffer_id_clear_noref(config
, &wsb
->id
);
392 shmp_chan
= shmp(handle
, buf
->backend
.chan
);
397 tsc
= config
->cb
.ring_buffer_clock_read(shmp_chan
);
398 config
->cb
.buffer_begin(buf
, tsc
, 0, handle
);
399 cc_hot
= shmp_index(handle
, buf
->commit_hot
, 0);
404 v_add(config
, subbuf_header_size
, &cc_hot
->cc
);
405 v_add(config
, subbuf_header_size
, &cc_hot
->seq
);
407 if (config
->cb
.buffer_create
) {
408 ret
= config
->cb
.buffer_create(buf
, priv
, cpu
, chanb
->name
, handle
);
413 init_crash_abi(config
, &buf
->crash_abi
, buf
, chanb
, shmobj
, handle
);
415 buf
->backend
.allocated
= 1;
420 /* ts_end will be freed by shm teardown */
422 /* commit_cold will be freed by shm teardown */
424 /* commit_hot will be freed by shm teardown */
430 void lib_ring_buffer_channel_switch_timer(int sig
, siginfo_t
*si
, void *uc
)
432 const struct lttng_ust_lib_ring_buffer_config
*config
;
433 struct lttng_ust_shm_handle
*handle
;
434 struct lttng_ust_lib_ring_buffer_channel
*chan
;
437 assert(CMM_LOAD_SHARED(timer_signal
.tid
) == pthread_self());
439 chan
= si
->si_value
.sival_ptr
;
440 handle
= chan
->handle
;
441 config
= &chan
->backend
.config
;
443 DBG("Switch timer for channel %p\n", chan
);
446 * Only flush buffers periodically if readers are active.
448 pthread_mutex_lock(&wakeup_fd_mutex
);
449 if (config
->alloc
== RING_BUFFER_ALLOC_PER_CPU
) {
450 for_each_possible_cpu(cpu
) {
451 struct lttng_ust_lib_ring_buffer
*buf
=
452 shmp(handle
, chan
->backend
.buf
[cpu
].shmp
);
456 if (uatomic_read(&buf
->active_readers
))
457 lib_ring_buffer_switch_slow(buf
, SWITCH_ACTIVE
,
461 struct lttng_ust_lib_ring_buffer
*buf
=
462 shmp(handle
, chan
->backend
.buf
[0].shmp
);
466 if (uatomic_read(&buf
->active_readers
))
467 lib_ring_buffer_switch_slow(buf
, SWITCH_ACTIVE
,
471 pthread_mutex_unlock(&wakeup_fd_mutex
);
476 int lib_ring_buffer_poll_deliver(const struct lttng_ust_lib_ring_buffer_config
*config
,
477 struct lttng_ust_lib_ring_buffer
*buf
,
478 struct lttng_ust_lib_ring_buffer_channel
*chan
,
479 struct lttng_ust_shm_handle
*handle
)
481 unsigned long consumed_old
, consumed_idx
, commit_count
, write_offset
;
482 struct commit_counters_cold
*cc_cold
;
484 consumed_old
= uatomic_read(&buf
->consumed
);
485 consumed_idx
= subbuf_index(consumed_old
, chan
);
486 cc_cold
= shmp_index(handle
, buf
->commit_cold
, consumed_idx
);
489 commit_count
= v_read(config
, &cc_cold
->cc_sb
);
491 * No memory barrier here, since we are only interested
492 * in a statistically correct polling result. The next poll will
493 * get the data is we are racing. The mb() that ensures correct
494 * memory order is in get_subbuf.
496 write_offset
= v_read(config
, &buf
->offset
);
499 * Check that the subbuffer we are trying to consume has been
500 * already fully committed.
503 if (((commit_count
- chan
->backend
.subbuf_size
)
504 & chan
->commit_count_mask
)
505 - (buf_trunc(consumed_old
, chan
)
506 >> chan
->backend
.num_subbuf_order
)
511 * Check that we are not about to read the same subbuffer in
512 * which the writer head is.
514 if (subbuf_trunc(write_offset
, chan
) - subbuf_trunc(consumed_old
, chan
)
522 void lib_ring_buffer_wakeup(struct lttng_ust_lib_ring_buffer
*buf
,
523 struct lttng_ust_shm_handle
*handle
)
525 int wakeup_fd
= shm_get_wakeup_fd(handle
, &buf
->self
._ref
);
526 sigset_t sigpipe_set
, pending_set
, old_set
;
527 int ret
, sigpipe_was_pending
= 0;
533 * Wake-up the other end by writing a null byte in the pipe
534 * (non-blocking). Important note: Because writing into the
535 * pipe is non-blocking (and therefore we allow dropping wakeup
536 * data, as long as there is wakeup data present in the pipe
537 * buffer to wake up the consumer), the consumer should perform
538 * the following sequence for waiting:
539 * 1) empty the pipe (reads).
540 * 2) check if there is data in the buffer.
541 * 3) wait on the pipe (poll).
543 * Discard the SIGPIPE from write(), not disturbing any SIGPIPE
544 * that might be already pending. If a bogus SIGPIPE is sent to
545 * the entire process concurrently by a malicious user, it may
546 * be simply discarded.
548 ret
= sigemptyset(&pending_set
);
551 * sigpending returns the mask of signals that are _both_
552 * blocked for the thread _and_ pending for either the thread or
553 * the entire process.
555 ret
= sigpending(&pending_set
);
557 sigpipe_was_pending
= sigismember(&pending_set
, SIGPIPE
);
559 * If sigpipe was pending, it means it was already blocked, so
560 * no need to block it.
562 if (!sigpipe_was_pending
) {
563 ret
= sigemptyset(&sigpipe_set
);
565 ret
= sigaddset(&sigpipe_set
, SIGPIPE
);
567 ret
= pthread_sigmask(SIG_BLOCK
, &sigpipe_set
, &old_set
);
571 ret
= write(wakeup_fd
, "", 1);
572 } while (ret
== -1L && errno
== EINTR
);
573 if (ret
== -1L && errno
== EPIPE
&& !sigpipe_was_pending
) {
574 struct timespec timeout
= { 0, 0 };
576 ret
= sigtimedwait(&sigpipe_set
, NULL
,
578 } while (ret
== -1L && errno
== EINTR
);
580 if (!sigpipe_was_pending
) {
581 ret
= pthread_sigmask(SIG_SETMASK
, &old_set
, NULL
);
587 void lib_ring_buffer_channel_do_read(struct lttng_ust_lib_ring_buffer_channel
*chan
)
589 const struct lttng_ust_lib_ring_buffer_config
*config
;
590 struct lttng_ust_shm_handle
*handle
;
593 handle
= chan
->handle
;
594 config
= &chan
->backend
.config
;
597 * Only flush buffers periodically if readers are active.
599 pthread_mutex_lock(&wakeup_fd_mutex
);
600 if (config
->alloc
== RING_BUFFER_ALLOC_PER_CPU
) {
601 for_each_possible_cpu(cpu
) {
602 struct lttng_ust_lib_ring_buffer
*buf
=
603 shmp(handle
, chan
->backend
.buf
[cpu
].shmp
);
607 if (uatomic_read(&buf
->active_readers
)
608 && lib_ring_buffer_poll_deliver(config
, buf
,
610 lib_ring_buffer_wakeup(buf
, handle
);
614 struct lttng_ust_lib_ring_buffer
*buf
=
615 shmp(handle
, chan
->backend
.buf
[0].shmp
);
619 if (uatomic_read(&buf
->active_readers
)
620 && lib_ring_buffer_poll_deliver(config
, buf
,
622 lib_ring_buffer_wakeup(buf
, handle
);
626 pthread_mutex_unlock(&wakeup_fd_mutex
);
630 void lib_ring_buffer_channel_read_timer(int sig
, siginfo_t
*si
, void *uc
)
632 struct lttng_ust_lib_ring_buffer_channel
*chan
;
634 assert(CMM_LOAD_SHARED(timer_signal
.tid
) == pthread_self());
635 chan
= si
->si_value
.sival_ptr
;
636 DBG("Read timer for channel %p\n", chan
);
637 lib_ring_buffer_channel_do_read(chan
);
642 void rb_setmask(sigset_t
*mask
)
646 ret
= sigemptyset(mask
);
648 PERROR("sigemptyset");
650 ret
= sigaddset(mask
, LTTNG_UST_RB_SIG_FLUSH
);
654 ret
= sigaddset(mask
, LTTNG_UST_RB_SIG_READ
);
658 ret
= sigaddset(mask
, LTTNG_UST_RB_SIG_TEARDOWN
);
665 void *sig_thread(void *arg
)
671 /* Only self thread will receive signal mask. */
673 CMM_STORE_SHARED(timer_signal
.tid
, pthread_self());
676 signr
= sigwaitinfo(&mask
, &info
);
679 PERROR("sigwaitinfo");
682 if (signr
== LTTNG_UST_RB_SIG_FLUSH
) {
683 lib_ring_buffer_channel_switch_timer(info
.si_signo
,
685 } else if (signr
== LTTNG_UST_RB_SIG_READ
) {
686 lib_ring_buffer_channel_read_timer(info
.si_signo
,
688 } else if (signr
== LTTNG_UST_RB_SIG_TEARDOWN
) {
690 CMM_STORE_SHARED(timer_signal
.qs_done
, 1);
693 ERR("Unexptected signal %d\n", info
.si_signo
);
700 * Ensure only a single thread listens on the timer signal.
703 void lib_ring_buffer_setup_timer_thread(void)
708 pthread_mutex_lock(&timer_signal
.lock
);
709 if (timer_signal
.setup_done
)
712 ret
= pthread_create(&thread
, NULL
, &sig_thread
, NULL
);
715 PERROR("pthread_create");
717 ret
= pthread_detach(thread
);
720 PERROR("pthread_detach");
722 timer_signal
.setup_done
= 1;
724 pthread_mutex_unlock(&timer_signal
.lock
);
728 * Wait for signal-handling thread quiescent state.
731 void lib_ring_buffer_wait_signal_thread_qs(unsigned int signr
)
733 sigset_t pending_set
;
737 * We need to be the only thread interacting with the thread
738 * that manages signals for teardown synchronization.
740 pthread_mutex_lock(&timer_signal
.lock
);
743 * Ensure we don't have any signal queued for this channel.
746 ret
= sigemptyset(&pending_set
);
748 PERROR("sigemptyset");
750 ret
= sigpending(&pending_set
);
752 PERROR("sigpending");
754 if (!sigismember(&pending_set
, signr
))
760 * From this point, no new signal handler will be fired that
761 * would try to access "chan". However, we still need to wait
762 * for any currently executing handler to complete.
765 CMM_STORE_SHARED(timer_signal
.qs_done
, 0);
769 * Kill with LTTNG_UST_RB_SIG_TEARDOWN, so signal management
772 kill(getpid(), LTTNG_UST_RB_SIG_TEARDOWN
);
774 while (!CMM_LOAD_SHARED(timer_signal
.qs_done
))
778 pthread_mutex_unlock(&timer_signal
.lock
);
782 void lib_ring_buffer_channel_switch_timer_start(struct lttng_ust_lib_ring_buffer_channel
*chan
)
785 struct itimerspec its
;
788 if (!chan
->switch_timer_interval
|| chan
->switch_timer_enabled
)
791 chan
->switch_timer_enabled
= 1;
793 lib_ring_buffer_setup_timer_thread();
795 memset(&sev
, 0, sizeof(sev
));
796 sev
.sigev_notify
= SIGEV_SIGNAL
;
797 sev
.sigev_signo
= LTTNG_UST_RB_SIG_FLUSH
;
798 sev
.sigev_value
.sival_ptr
= chan
;
799 ret
= timer_create(CLOCKID
, &sev
, &chan
->switch_timer
);
801 PERROR("timer_create");
804 its
.it_value
.tv_sec
= chan
->switch_timer_interval
/ 1000000;
805 its
.it_value
.tv_nsec
= (chan
->switch_timer_interval
% 1000000) * 1000;
806 its
.it_interval
.tv_sec
= its
.it_value
.tv_sec
;
807 its
.it_interval
.tv_nsec
= its
.it_value
.tv_nsec
;
809 ret
= timer_settime(chan
->switch_timer
, 0, &its
, NULL
);
811 PERROR("timer_settime");
816 void lib_ring_buffer_channel_switch_timer_stop(struct lttng_ust_lib_ring_buffer_channel
*chan
)
820 if (!chan
->switch_timer_interval
|| !chan
->switch_timer_enabled
)
823 ret
= timer_delete(chan
->switch_timer
);
825 PERROR("timer_delete");
828 lib_ring_buffer_wait_signal_thread_qs(LTTNG_UST_RB_SIG_FLUSH
);
830 chan
->switch_timer
= 0;
831 chan
->switch_timer_enabled
= 0;
835 void lib_ring_buffer_channel_read_timer_start(struct lttng_ust_lib_ring_buffer_channel
*chan
)
837 const struct lttng_ust_lib_ring_buffer_config
*config
= &chan
->backend
.config
;
839 struct itimerspec its
;
842 if (config
->wakeup
!= RING_BUFFER_WAKEUP_BY_TIMER
843 || !chan
->read_timer_interval
|| chan
->read_timer_enabled
)
846 chan
->read_timer_enabled
= 1;
848 lib_ring_buffer_setup_timer_thread();
850 sev
.sigev_notify
= SIGEV_SIGNAL
;
851 sev
.sigev_signo
= LTTNG_UST_RB_SIG_READ
;
852 sev
.sigev_value
.sival_ptr
= chan
;
853 ret
= timer_create(CLOCKID
, &sev
, &chan
->read_timer
);
855 PERROR("timer_create");
858 its
.it_value
.tv_sec
= chan
->read_timer_interval
/ 1000000;
859 its
.it_value
.tv_nsec
= (chan
->read_timer_interval
% 1000000) * 1000;
860 its
.it_interval
.tv_sec
= its
.it_value
.tv_sec
;
861 its
.it_interval
.tv_nsec
= its
.it_value
.tv_nsec
;
863 ret
= timer_settime(chan
->read_timer
, 0, &its
, NULL
);
865 PERROR("timer_settime");
870 void lib_ring_buffer_channel_read_timer_stop(struct lttng_ust_lib_ring_buffer_channel
*chan
)
872 const struct lttng_ust_lib_ring_buffer_config
*config
= &chan
->backend
.config
;
875 if (config
->wakeup
!= RING_BUFFER_WAKEUP_BY_TIMER
876 || !chan
->read_timer_interval
|| !chan
->read_timer_enabled
)
879 ret
= timer_delete(chan
->read_timer
);
881 PERROR("timer_delete");
885 * do one more check to catch data that has been written in the last
888 lib_ring_buffer_channel_do_read(chan
);
890 lib_ring_buffer_wait_signal_thread_qs(LTTNG_UST_RB_SIG_READ
);
892 chan
->read_timer
= 0;
893 chan
->read_timer_enabled
= 0;
896 static void channel_unregister_notifiers(struct lttng_ust_lib_ring_buffer_channel
*chan
,
897 struct lttng_ust_shm_handle
*handle
)
899 lib_ring_buffer_channel_switch_timer_stop(chan
);
900 lib_ring_buffer_channel_read_timer_stop(chan
);
903 static void channel_print_errors(struct lttng_ust_lib_ring_buffer_channel
*chan
,
904 struct lttng_ust_shm_handle
*handle
)
906 const struct lttng_ust_lib_ring_buffer_config
*config
=
907 &chan
->backend
.config
;
910 if (config
->alloc
== RING_BUFFER_ALLOC_PER_CPU
) {
911 for_each_possible_cpu(cpu
) {
912 struct lttng_ust_lib_ring_buffer
*buf
=
913 shmp(handle
, chan
->backend
.buf
[cpu
].shmp
);
915 lib_ring_buffer_print_errors(chan
, buf
, cpu
, handle
);
918 struct lttng_ust_lib_ring_buffer
*buf
=
919 shmp(handle
, chan
->backend
.buf
[0].shmp
);
922 lib_ring_buffer_print_errors(chan
, buf
, -1, handle
);
926 static void channel_free(struct lttng_ust_lib_ring_buffer_channel
*chan
,
927 struct lttng_ust_shm_handle
*handle
,
930 channel_backend_free(&chan
->backend
, handle
);
931 /* chan is freed by shm teardown */
932 shm_object_table_destroy(handle
->table
, consumer
);
937 * channel_create - Create channel.
938 * @config: ring buffer instance configuration
939 * @name: name of the channel
940 * @priv_data: ring buffer client private data area pointer (output)
941 * @priv_data_size: length, in bytes, of the private data area.
942 * @priv_data_init: initialization data for private data.
943 * @buf_addr: pointer the the beginning of the preallocated buffer contiguous
944 * address mapping. It is used only by RING_BUFFER_STATIC
945 * configuration. It can be set to NULL for other backends.
946 * @subbuf_size: subbuffer size
947 * @num_subbuf: number of subbuffers
948 * @switch_timer_interval: Time interval (in us) to fill sub-buffers with
949 * padding to let readers get those sub-buffers.
950 * Used for live streaming.
951 * @read_timer_interval: Time interval (in us) to wake up pending readers.
952 * @stream_fds: array of stream file descriptors.
953 * @nr_stream_fds: number of file descriptors in array.
956 * Returns NULL on failure.
958 struct lttng_ust_shm_handle
*channel_create(const struct lttng_ust_lib_ring_buffer_config
*config
,
961 size_t priv_data_align
,
962 size_t priv_data_size
,
963 void *priv_data_init
,
964 void *buf_addr
, size_t subbuf_size
,
965 size_t num_subbuf
, unsigned int switch_timer_interval
,
966 unsigned int read_timer_interval
,
967 const int *stream_fds
, int nr_stream_fds
,
968 int64_t blocking_timeout
)
971 size_t shmsize
, chansize
;
972 struct lttng_ust_lib_ring_buffer_channel
*chan
;
973 struct lttng_ust_shm_handle
*handle
;
974 struct shm_object
*shmobj
;
975 unsigned int nr_streams
;
976 int64_t blocking_timeout_ms
;
978 if (config
->alloc
== RING_BUFFER_ALLOC_PER_CPU
)
979 nr_streams
= num_possible_cpus();
983 if (nr_stream_fds
!= nr_streams
)
986 if (blocking_timeout
< -1) {
990 if (blocking_timeout
== -1) {
991 blocking_timeout_ms
= -1;
993 blocking_timeout_ms
= blocking_timeout
/ 1000;
994 if (blocking_timeout_ms
!= (int32_t) blocking_timeout_ms
) {
999 if (lib_ring_buffer_check_config(config
, switch_timer_interval
,
1000 read_timer_interval
))
1003 handle
= zmalloc(sizeof(struct lttng_ust_shm_handle
));
1007 /* Allocate table for channel + per-cpu buffers */
1008 handle
->table
= shm_object_table_create(1 + num_possible_cpus());
1010 goto error_table_alloc
;
1012 /* Calculate the shm allocation layout */
1013 shmsize
= sizeof(struct lttng_ust_lib_ring_buffer_channel
);
1014 shmsize
+= lttng_ust_offset_align(shmsize
, __alignof__(struct lttng_ust_lib_ring_buffer_shmp
));
1015 shmsize
+= sizeof(struct lttng_ust_lib_ring_buffer_shmp
) * nr_streams
;
1017 if (priv_data_align
)
1018 shmsize
+= lttng_ust_offset_align(shmsize
, priv_data_align
);
1019 shmsize
+= priv_data_size
;
1021 /* Allocate normal memory for channel (not shared) */
1022 shmobj
= shm_object_table_alloc(handle
->table
, shmsize
, SHM_OBJECT_MEM
,
1026 /* struct lttng_ust_lib_ring_buffer_channel is at object 0, offset 0 (hardcoded) */
1027 set_shmp(handle
->chan
, zalloc_shm(shmobj
, chansize
));
1028 assert(handle
->chan
._ref
.index
== 0);
1029 assert(handle
->chan
._ref
.offset
== 0);
1030 chan
= shmp(handle
, handle
->chan
);
1033 chan
->nr_streams
= nr_streams
;
1035 /* space for private data */
1036 if (priv_data_size
) {
1037 DECLARE_SHMP(void, priv_data_alloc
);
1039 align_shm(shmobj
, priv_data_align
);
1040 chan
->priv_data_offset
= shmobj
->allocated_len
;
1041 set_shmp(priv_data_alloc
, zalloc_shm(shmobj
, priv_data_size
));
1042 if (!shmp(handle
, priv_data_alloc
))
1044 *priv_data
= channel_get_private(chan
);
1045 memcpy(*priv_data
, priv_data_init
, priv_data_size
);
1047 chan
->priv_data_offset
= -1;
1052 chan
->u
.s
.blocking_timeout_ms
= (int32_t) blocking_timeout_ms
;
1054 ret
= channel_backend_init(&chan
->backend
, name
, config
,
1055 subbuf_size
, num_subbuf
, handle
,
1058 goto error_backend_init
;
1060 chan
->handle
= handle
;
1061 chan
->commit_count_mask
= (~0UL >> chan
->backend
.num_subbuf_order
);
1063 chan
->switch_timer_interval
= switch_timer_interval
;
1064 chan
->read_timer_interval
= read_timer_interval
;
1065 lib_ring_buffer_channel_switch_timer_start(chan
);
1066 lib_ring_buffer_channel_read_timer_start(chan
);
1072 shm_object_table_destroy(handle
->table
, 1);
1078 struct lttng_ust_shm_handle
*channel_handle_create(void *data
,
1079 uint64_t memory_map_size
,
1082 struct lttng_ust_shm_handle
*handle
;
1083 struct shm_object
*object
;
1085 handle
= zmalloc(sizeof(struct lttng_ust_shm_handle
));
1089 /* Allocate table for channel + per-cpu buffers */
1090 handle
->table
= shm_object_table_create(1 + num_possible_cpus());
1092 goto error_table_alloc
;
1093 /* Add channel object */
1094 object
= shm_object_table_append_mem(handle
->table
, data
,
1095 memory_map_size
, wakeup_fd
);
1097 goto error_table_object
;
1098 /* struct lttng_ust_lib_ring_buffer_channel is at object 0, offset 0 (hardcoded) */
1099 handle
->chan
._ref
.index
= 0;
1100 handle
->chan
._ref
.offset
= 0;
1104 shm_object_table_destroy(handle
->table
, 0);
1110 int channel_handle_add_stream(struct lttng_ust_shm_handle
*handle
,
1111 int shm_fd
, int wakeup_fd
, uint32_t stream_nr
,
1112 uint64_t memory_map_size
)
1114 struct shm_object
*object
;
1116 /* Add stream object */
1117 object
= shm_object_table_append_shm(handle
->table
,
1118 shm_fd
, wakeup_fd
, stream_nr
,
1125 unsigned int channel_handle_get_nr_streams(struct lttng_ust_shm_handle
*handle
)
1127 assert(handle
->table
);
1128 return handle
->table
->allocated_len
- 1;
1132 void channel_release(struct lttng_ust_lib_ring_buffer_channel
*chan
, struct lttng_ust_shm_handle
*handle
,
1135 channel_free(chan
, handle
, consumer
);
1139 * channel_destroy - Finalize, wait for q.s. and destroy channel.
1140 * @chan: channel to destroy
1142 * Holds cpu hotplug.
1143 * Call "destroy" callback, finalize channels, decrement the channel
1144 * reference count. Note that when readers have completed data
1145 * consumption of finalized channels, get_subbuf() will return -ENODATA.
1146 * They should release their handle at that point.
1148 void channel_destroy(struct lttng_ust_lib_ring_buffer_channel
*chan
, struct lttng_ust_shm_handle
*handle
,
1153 * Note: the consumer takes care of finalizing and
1154 * switching the buffers.
1156 channel_unregister_notifiers(chan
, handle
);
1158 * The consumer prints errors.
1160 channel_print_errors(chan
, handle
);
1164 * sessiond/consumer are keeping a reference on the shm file
1165 * descriptor directly. No need to refcount.
1167 channel_release(chan
, handle
, consumer
);
1171 struct lttng_ust_lib_ring_buffer
*channel_get_ring_buffer(
1172 const struct lttng_ust_lib_ring_buffer_config
*config
,
1173 struct lttng_ust_lib_ring_buffer_channel
*chan
, int cpu
,
1174 struct lttng_ust_shm_handle
*handle
,
1175 int *shm_fd
, int *wait_fd
,
1177 uint64_t *memory_map_size
)
1179 struct shm_ref
*ref
;
1181 if (config
->alloc
== RING_BUFFER_ALLOC_GLOBAL
) {
1184 if (cpu
>= num_possible_cpus())
1187 ref
= &chan
->backend
.buf
[cpu
].shmp
._ref
;
1188 *shm_fd
= shm_get_shm_fd(handle
, ref
);
1189 *wait_fd
= shm_get_wait_fd(handle
, ref
);
1190 *wakeup_fd
= shm_get_wakeup_fd(handle
, ref
);
1191 if (shm_get_shm_size(handle
, ref
, memory_map_size
))
1193 return shmp(handle
, chan
->backend
.buf
[cpu
].shmp
);
1196 int ring_buffer_channel_close_wait_fd(const struct lttng_ust_lib_ring_buffer_config
*config
,
1197 struct lttng_ust_lib_ring_buffer_channel
*chan
,
1198 struct lttng_ust_shm_handle
*handle
)
1200 struct shm_ref
*ref
;
1202 ref
= &handle
->chan
._ref
;
1203 return shm_close_wait_fd(handle
, ref
);
1206 int ring_buffer_channel_close_wakeup_fd(const struct lttng_ust_lib_ring_buffer_config
*config
,
1207 struct lttng_ust_lib_ring_buffer_channel
*chan
,
1208 struct lttng_ust_shm_handle
*handle
)
1210 struct shm_ref
*ref
;
1212 ref
= &handle
->chan
._ref
;
1213 return shm_close_wakeup_fd(handle
, ref
);
1216 int ring_buffer_stream_close_wait_fd(const struct lttng_ust_lib_ring_buffer_config
*config
,
1217 struct lttng_ust_lib_ring_buffer_channel
*chan
,
1218 struct lttng_ust_shm_handle
*handle
,
1221 struct shm_ref
*ref
;
1223 if (config
->alloc
== RING_BUFFER_ALLOC_GLOBAL
) {
1226 if (cpu
>= num_possible_cpus())
1229 ref
= &chan
->backend
.buf
[cpu
].shmp
._ref
;
1230 return shm_close_wait_fd(handle
, ref
);
1233 int ring_buffer_stream_close_wakeup_fd(const struct lttng_ust_lib_ring_buffer_config
*config
,
1234 struct lttng_ust_lib_ring_buffer_channel
*chan
,
1235 struct lttng_ust_shm_handle
*handle
,
1238 struct shm_ref
*ref
;
1241 if (config
->alloc
== RING_BUFFER_ALLOC_GLOBAL
) {
1244 if (cpu
>= num_possible_cpus())
1247 ref
= &chan
->backend
.buf
[cpu
].shmp
._ref
;
1248 pthread_mutex_lock(&wakeup_fd_mutex
);
1249 ret
= shm_close_wakeup_fd(handle
, ref
);
1250 pthread_mutex_unlock(&wakeup_fd_mutex
);
1254 int lib_ring_buffer_open_read(struct lttng_ust_lib_ring_buffer
*buf
,
1255 struct lttng_ust_shm_handle
*handle
)
1257 if (uatomic_cmpxchg(&buf
->active_readers
, 0, 1) != 0)
1263 void lib_ring_buffer_release_read(struct lttng_ust_lib_ring_buffer
*buf
,
1264 struct lttng_ust_shm_handle
*handle
)
1266 struct lttng_ust_lib_ring_buffer_channel
*chan
= shmp(handle
, buf
->backend
.chan
);
1270 CHAN_WARN_ON(chan
, uatomic_read(&buf
->active_readers
) != 1);
1272 uatomic_dec(&buf
->active_readers
);
1276 * lib_ring_buffer_snapshot - save subbuffer position snapshot (for read)
1278 * @consumed: consumed count indicating the position where to read
1279 * @produced: produced count, indicates position when to stop reading
1281 * Returns -ENODATA if buffer is finalized, -EAGAIN if there is currently no
1282 * data to read at consumed position, or 0 if the get operation succeeds.
1285 int lib_ring_buffer_snapshot(struct lttng_ust_lib_ring_buffer
*buf
,
1286 unsigned long *consumed
, unsigned long *produced
,
1287 struct lttng_ust_shm_handle
*handle
)
1289 struct lttng_ust_lib_ring_buffer_channel
*chan
;
1290 const struct lttng_ust_lib_ring_buffer_config
*config
;
1291 unsigned long consumed_cur
, write_offset
;
1294 chan
= shmp(handle
, buf
->backend
.chan
);
1297 config
= &chan
->backend
.config
;
1298 finalized
= CMM_ACCESS_ONCE(buf
->finalized
);
1300 * Read finalized before counters.
1303 consumed_cur
= uatomic_read(&buf
->consumed
);
1305 * No need to issue a memory barrier between consumed count read and
1306 * write offset read, because consumed count can only change
1307 * concurrently in overwrite mode, and we keep a sequence counter
1308 * identifier derived from the write offset to check we are getting
1309 * the same sub-buffer we are expecting (the sub-buffers are atomically
1310 * "tagged" upon writes, tags are checked upon read).
1312 write_offset
= v_read(config
, &buf
->offset
);
1315 * Check that we are not about to read the same subbuffer in
1316 * which the writer head is.
1318 if (subbuf_trunc(write_offset
, chan
) - subbuf_trunc(consumed_cur
, chan
)
1322 *consumed
= consumed_cur
;
1323 *produced
= subbuf_trunc(write_offset
, chan
);
1329 * The memory barriers __wait_event()/wake_up_interruptible() take care
1330 * of "raw_spin_is_locked" memory ordering.
1339 * Performs the same function as lib_ring_buffer_snapshot(), but the positions
1340 * are saved regardless of whether the consumed and produced positions are
1341 * in the same subbuffer.
1343 * @consumed: consumed byte count indicating the last position read
1344 * @produced: produced byte count indicating the last position written
1346 * This function is meant to provide information on the exact producer and
1347 * consumer positions without regard for the "snapshot" feature.
1349 int lib_ring_buffer_snapshot_sample_positions(
1350 struct lttng_ust_lib_ring_buffer
*buf
,
1351 unsigned long *consumed
, unsigned long *produced
,
1352 struct lttng_ust_shm_handle
*handle
)
1354 struct lttng_ust_lib_ring_buffer_channel
*chan
;
1355 const struct lttng_ust_lib_ring_buffer_config
*config
;
1357 chan
= shmp(handle
, buf
->backend
.chan
);
1360 config
= &chan
->backend
.config
;
1362 *consumed
= uatomic_read(&buf
->consumed
);
1364 * No need to issue a memory barrier between consumed count read and
1365 * write offset read, because consumed count can only change
1366 * concurrently in overwrite mode, and we keep a sequence counter
1367 * identifier derived from the write offset to check we are getting
1368 * the same sub-buffer we are expecting (the sub-buffers are atomically
1369 * "tagged" upon writes, tags are checked upon read).
1371 *produced
= v_read(config
, &buf
->offset
);
1376 * lib_ring_buffer_move_consumer - move consumed counter forward
1378 * @consumed_new: new consumed count value
1380 void lib_ring_buffer_move_consumer(struct lttng_ust_lib_ring_buffer
*buf
,
1381 unsigned long consumed_new
,
1382 struct lttng_ust_shm_handle
*handle
)
1384 struct lttng_ust_lib_ring_buffer_backend
*bufb
= &buf
->backend
;
1385 struct lttng_ust_lib_ring_buffer_channel
*chan
;
1386 unsigned long consumed
;
1388 chan
= shmp(handle
, bufb
->chan
);
1391 CHAN_WARN_ON(chan
, uatomic_read(&buf
->active_readers
) != 1);
1394 * Only push the consumed value forward.
1395 * If the consumed cmpxchg fails, this is because we have been pushed by
1396 * the writer in flight recorder mode.
1398 consumed
= uatomic_read(&buf
->consumed
);
1399 while ((long) consumed
- (long) consumed_new
< 0)
1400 consumed
= uatomic_cmpxchg(&buf
->consumed
, consumed
,
1405 * lib_ring_buffer_get_subbuf - get exclusive access to subbuffer for reading
1407 * @consumed: consumed count indicating the position where to read
1409 * Returns -ENODATA if buffer is finalized, -EAGAIN if there is currently no
1410 * data to read at consumed position, or 0 if the get operation succeeds.
1412 int lib_ring_buffer_get_subbuf(struct lttng_ust_lib_ring_buffer
*buf
,
1413 unsigned long consumed
,
1414 struct lttng_ust_shm_handle
*handle
)
1416 struct lttng_ust_lib_ring_buffer_channel
*chan
;
1417 const struct lttng_ust_lib_ring_buffer_config
*config
;
1418 unsigned long consumed_cur
, consumed_idx
, commit_count
, write_offset
;
1419 int ret
, finalized
, nr_retry
= LTTNG_UST_RING_BUFFER_GET_RETRY
;
1420 struct commit_counters_cold
*cc_cold
;
1422 chan
= shmp(handle
, buf
->backend
.chan
);
1425 config
= &chan
->backend
.config
;
1427 finalized
= CMM_ACCESS_ONCE(buf
->finalized
);
1429 * Read finalized before counters.
1432 consumed_cur
= uatomic_read(&buf
->consumed
);
1433 consumed_idx
= subbuf_index(consumed
, chan
);
1434 cc_cold
= shmp_index(handle
, buf
->commit_cold
, consumed_idx
);
1437 commit_count
= v_read(config
, &cc_cold
->cc_sb
);
1439 * Make sure we read the commit count before reading the buffer
1440 * data and the write offset. Correct consumed offset ordering
1441 * wrt commit count is insured by the use of cmpxchg to update
1442 * the consumed offset.
1445 * Local rmb to match the remote wmb to read the commit count
1446 * before the buffer data and the write offset.
1450 write_offset
= v_read(config
, &buf
->offset
);
1453 * Check that the buffer we are getting is after or at consumed_cur
1456 if ((long) subbuf_trunc(consumed
, chan
)
1457 - (long) subbuf_trunc(consumed_cur
, chan
) < 0)
1461 * Check that the subbuffer we are trying to consume has been
1462 * already fully committed. There are a few causes that can make
1463 * this unavailability situation occur:
1465 * Temporary (short-term) situation:
1466 * - Application is running on a different CPU, between reserve
1467 * and commit ring buffer operations,
1468 * - Application is preempted between reserve and commit ring
1469 * buffer operations,
1471 * Long-term situation:
1472 * - Application is stopped (SIGSTOP) between reserve and commit
1473 * ring buffer operations. Could eventually be resumed by
1475 * - Application is killed (SIGTERM, SIGINT, SIGKILL) between
1476 * reserve and commit ring buffer operation.
1478 * From a consumer perspective, handling short-term
1479 * unavailability situations is performed by retrying a few
1480 * times after a delay. Handling long-term unavailability
1481 * situations is handled by failing to get the sub-buffer.
1483 * In all of those situations, if the application is taking a
1484 * long time to perform its commit after ring buffer space
1485 * reservation, we can end up in a situation where the producer
1486 * will fill the ring buffer and try to write into the same
1487 * sub-buffer again (which has a missing commit). This is
1488 * handled by the producer in the sub-buffer switch handling
1489 * code of the reserve routine by detecting unbalanced
1490 * reserve/commit counters and discarding all further events
1491 * until the situation is resolved in those situations. Two
1492 * scenarios can occur:
1494 * 1) The application causing the reserve/commit counters to be
1495 * unbalanced has been terminated. In this situation, all
1496 * further events will be discarded in the buffers, and no
1497 * further buffer data will be readable by the consumer
1498 * daemon. Tearing down the UST tracing session and starting
1499 * anew is a work-around for those situations. Note that this
1500 * only affects per-UID tracing. In per-PID tracing, the
1501 * application vanishes with the termination, and therefore
1502 * no more data needs to be written to the buffers.
1503 * 2) The application causing the unbalance has been delayed for
1504 * a long time, but will eventually try to increment the
1505 * commit counter after eventually writing to the sub-buffer.
1506 * This situation can cause events to be discarded until the
1507 * application resumes its operations.
1509 if (((commit_count
- chan
->backend
.subbuf_size
)
1510 & chan
->commit_count_mask
)
1511 - (buf_trunc(consumed
, chan
)
1512 >> chan
->backend
.num_subbuf_order
)
1514 if (nr_retry
-- > 0) {
1515 if (nr_retry
<= (LTTNG_UST_RING_BUFFER_GET_RETRY
>> 1))
1516 (void) poll(NULL
, 0, LTTNG_UST_RING_BUFFER_RETRY_DELAY_MS
);
1524 * Check that we are not about to read the same subbuffer in
1525 * which the writer head is.
1527 if (subbuf_trunc(write_offset
, chan
) - subbuf_trunc(consumed
, chan
)
1532 * Failure to get the subbuffer causes a busy-loop retry without going
1533 * to a wait queue. These are caused by short-lived race windows where
1534 * the writer is getting access to a subbuffer we were trying to get
1535 * access to. Also checks that the "consumed" buffer count we are
1536 * looking for matches the one contained in the subbuffer id.
1538 * The short-lived race window described here can be affected by
1539 * application signals and preemption, thus requiring to bound
1540 * the loop to a maximum number of retry.
1542 ret
= update_read_sb_index(config
, &buf
->backend
, &chan
->backend
,
1543 consumed_idx
, buf_trunc_val(consumed
, chan
),
1546 if (nr_retry
-- > 0) {
1547 if (nr_retry
<= (LTTNG_UST_RING_BUFFER_GET_RETRY
>> 1))
1548 (void) poll(NULL
, 0, LTTNG_UST_RING_BUFFER_RETRY_DELAY_MS
);
1554 subbuffer_id_clear_noref(config
, &buf
->backend
.buf_rsb
.id
);
1556 buf
->get_subbuf_consumed
= consumed
;
1557 buf
->get_subbuf
= 1;
1563 * The memory barriers __wait_event()/wake_up_interruptible() take care
1564 * of "raw_spin_is_locked" memory ordering.
1573 * lib_ring_buffer_put_subbuf - release exclusive subbuffer access
1576 void lib_ring_buffer_put_subbuf(struct lttng_ust_lib_ring_buffer
*buf
,
1577 struct lttng_ust_shm_handle
*handle
)
1579 struct lttng_ust_lib_ring_buffer_backend
*bufb
= &buf
->backend
;
1580 struct lttng_ust_lib_ring_buffer_channel
*chan
;
1581 const struct lttng_ust_lib_ring_buffer_config
*config
;
1582 unsigned long sb_bindex
, consumed_idx
, consumed
;
1583 struct lttng_ust_lib_ring_buffer_backend_pages_shmp
*rpages
;
1584 struct lttng_ust_lib_ring_buffer_backend_pages
*backend_pages
;
1586 chan
= shmp(handle
, bufb
->chan
);
1589 config
= &chan
->backend
.config
;
1590 CHAN_WARN_ON(chan
, uatomic_read(&buf
->active_readers
) != 1);
1592 if (!buf
->get_subbuf
) {
1594 * Reader puts a subbuffer it did not get.
1596 CHAN_WARN_ON(chan
, 1);
1599 consumed
= buf
->get_subbuf_consumed
;
1600 buf
->get_subbuf
= 0;
1603 * Clear the records_unread counter. (overruns counter)
1604 * Can still be non-zero if a file reader simply grabbed the data
1605 * without using iterators.
1606 * Can be below zero if an iterator is used on a snapshot more than
1609 sb_bindex
= subbuffer_id_get_index(config
, bufb
->buf_rsb
.id
);
1610 rpages
= shmp_index(handle
, bufb
->array
, sb_bindex
);
1613 backend_pages
= shmp(handle
, rpages
->shmp
);
1616 v_add(config
, v_read(config
, &backend_pages
->records_unread
),
1617 &bufb
->records_read
);
1618 v_set(config
, &backend_pages
->records_unread
, 0);
1619 CHAN_WARN_ON(chan
, config
->mode
== RING_BUFFER_OVERWRITE
1620 && subbuffer_id_is_noref(config
, bufb
->buf_rsb
.id
));
1621 subbuffer_id_set_noref(config
, &bufb
->buf_rsb
.id
);
1624 * Exchange the reader subbuffer with the one we put in its place in the
1625 * writer subbuffer table. Expect the original consumed count. If
1626 * update_read_sb_index fails, this is because the writer updated the
1627 * subbuffer concurrently. We should therefore keep the subbuffer we
1628 * currently have: it has become invalid to try reading this sub-buffer
1629 * consumed count value anyway.
1631 consumed_idx
= subbuf_index(consumed
, chan
);
1632 update_read_sb_index(config
, &buf
->backend
, &chan
->backend
,
1633 consumed_idx
, buf_trunc_val(consumed
, chan
),
1636 * update_read_sb_index return value ignored. Don't exchange sub-buffer
1637 * if the writer concurrently updated it.
1642 * cons_offset is an iterator on all subbuffer offsets between the reader
1643 * position and the writer position. (inclusive)
1646 void lib_ring_buffer_print_subbuffer_errors(struct lttng_ust_lib_ring_buffer
*buf
,
1647 struct lttng_ust_lib_ring_buffer_channel
*chan
,
1648 unsigned long cons_offset
,
1650 struct lttng_ust_shm_handle
*handle
)
1652 const struct lttng_ust_lib_ring_buffer_config
*config
= &chan
->backend
.config
;
1653 unsigned long cons_idx
, commit_count
, commit_count_sb
;
1654 struct commit_counters_hot
*cc_hot
;
1655 struct commit_counters_cold
*cc_cold
;
1657 cons_idx
= subbuf_index(cons_offset
, chan
);
1658 cc_hot
= shmp_index(handle
, buf
->commit_hot
, cons_idx
);
1661 cc_cold
= shmp_index(handle
, buf
->commit_cold
, cons_idx
);
1664 commit_count
= v_read(config
, &cc_hot
->cc
);
1665 commit_count_sb
= v_read(config
, &cc_cold
->cc_sb
);
1667 if (subbuf_offset(commit_count
, chan
) != 0)
1668 DBG("ring buffer %s, cpu %d: "
1669 "commit count in subbuffer %lu,\n"
1670 "expecting multiples of %lu bytes\n"
1671 " [ %lu bytes committed, %lu bytes reader-visible ]\n",
1672 chan
->backend
.name
, cpu
, cons_idx
,
1673 chan
->backend
.subbuf_size
,
1674 commit_count
, commit_count_sb
);
1676 DBG("ring buffer: %s, cpu %d: %lu bytes committed\n",
1677 chan
->backend
.name
, cpu
, commit_count
);
1681 void lib_ring_buffer_print_buffer_errors(struct lttng_ust_lib_ring_buffer
*buf
,
1682 struct lttng_ust_lib_ring_buffer_channel
*chan
,
1683 void *priv
, int cpu
,
1684 struct lttng_ust_shm_handle
*handle
)
1686 const struct lttng_ust_lib_ring_buffer_config
*config
= &chan
->backend
.config
;
1687 unsigned long write_offset
, cons_offset
;
1690 * No need to order commit_count, write_offset and cons_offset reads
1691 * because we execute at teardown when no more writer nor reader
1692 * references are left.
1694 write_offset
= v_read(config
, &buf
->offset
);
1695 cons_offset
= uatomic_read(&buf
->consumed
);
1696 if (write_offset
!= cons_offset
)
1697 DBG("ring buffer %s, cpu %d: "
1698 "non-consumed data\n"
1699 " [ %lu bytes written, %lu bytes read ]\n",
1700 chan
->backend
.name
, cpu
, write_offset
, cons_offset
);
1702 for (cons_offset
= uatomic_read(&buf
->consumed
);
1703 (long) (subbuf_trunc((unsigned long) v_read(config
, &buf
->offset
),
1706 cons_offset
= subbuf_align(cons_offset
, chan
))
1707 lib_ring_buffer_print_subbuffer_errors(buf
, chan
, cons_offset
,
1712 void lib_ring_buffer_print_errors(struct lttng_ust_lib_ring_buffer_channel
*chan
,
1713 struct lttng_ust_lib_ring_buffer
*buf
, int cpu
,
1714 struct lttng_ust_shm_handle
*handle
)
1716 const struct lttng_ust_lib_ring_buffer_config
*config
= &chan
->backend
.config
;
1717 void *priv
= channel_get_private(chan
);
1719 if (!strcmp(chan
->backend
.name
, "relay-metadata-mmap")) {
1720 DBG("ring buffer %s: %lu records written, "
1721 "%lu records overrun\n",
1723 v_read(config
, &buf
->records_count
),
1724 v_read(config
, &buf
->records_overrun
));
1726 DBG("ring buffer %s, cpu %d: %lu records written, "
1727 "%lu records overrun\n",
1728 chan
->backend
.name
, cpu
,
1729 v_read(config
, &buf
->records_count
),
1730 v_read(config
, &buf
->records_overrun
));
1732 if (v_read(config
, &buf
->records_lost_full
)
1733 || v_read(config
, &buf
->records_lost_wrap
)
1734 || v_read(config
, &buf
->records_lost_big
))
1735 DBG("ring buffer %s, cpu %d: records were lost. Caused by:\n"
1736 " [ %lu buffer full, %lu nest buffer wrap-around, "
1737 "%lu event too big ]\n",
1738 chan
->backend
.name
, cpu
,
1739 v_read(config
, &buf
->records_lost_full
),
1740 v_read(config
, &buf
->records_lost_wrap
),
1741 v_read(config
, &buf
->records_lost_big
));
1743 lib_ring_buffer_print_buffer_errors(buf
, chan
, priv
, cpu
, handle
);
1747 * lib_ring_buffer_switch_old_start: Populate old subbuffer header.
1749 * Only executed by SWITCH_FLUSH, which can be issued while tracing is
1750 * active or at buffer finalization (destroy).
1753 void lib_ring_buffer_switch_old_start(struct lttng_ust_lib_ring_buffer
*buf
,
1754 struct lttng_ust_lib_ring_buffer_channel
*chan
,
1755 struct switch_offsets
*offsets
,
1757 struct lttng_ust_shm_handle
*handle
)
1759 const struct lttng_ust_lib_ring_buffer_config
*config
= &chan
->backend
.config
;
1760 unsigned long oldidx
= subbuf_index(offsets
->old
, chan
);
1761 unsigned long commit_count
;
1762 struct commit_counters_hot
*cc_hot
;
1764 config
->cb
.buffer_begin(buf
, tsc
, oldidx
, handle
);
1767 * Order all writes to buffer before the commit count update that will
1768 * determine that the subbuffer is full.
1771 cc_hot
= shmp_index(handle
, buf
->commit_hot
, oldidx
);
1774 v_add(config
, config
->cb
.subbuffer_header_size(),
1776 commit_count
= v_read(config
, &cc_hot
->cc
);
1777 /* Check if the written buffer has to be delivered */
1778 lib_ring_buffer_check_deliver(config
, buf
, chan
, offsets
->old
,
1779 commit_count
, oldidx
, handle
, tsc
);
1780 lib_ring_buffer_write_commit_counter(config
, buf
, chan
,
1781 offsets
->old
+ config
->cb
.subbuffer_header_size(),
1782 commit_count
, handle
, cc_hot
);
1786 * lib_ring_buffer_switch_old_end: switch old subbuffer
1788 * Note : offset_old should never be 0 here. It is ok, because we never perform
1789 * buffer switch on an empty subbuffer in SWITCH_ACTIVE mode. The caller
1790 * increments the offset_old value when doing a SWITCH_FLUSH on an empty
1794 void lib_ring_buffer_switch_old_end(struct lttng_ust_lib_ring_buffer
*buf
,
1795 struct lttng_ust_lib_ring_buffer_channel
*chan
,
1796 struct switch_offsets
*offsets
,
1798 struct lttng_ust_shm_handle
*handle
)
1800 const struct lttng_ust_lib_ring_buffer_config
*config
= &chan
->backend
.config
;
1801 unsigned long oldidx
= subbuf_index(offsets
->old
- 1, chan
);
1802 unsigned long commit_count
, padding_size
, data_size
;
1803 struct commit_counters_hot
*cc_hot
;
1806 data_size
= subbuf_offset(offsets
->old
- 1, chan
) + 1;
1807 padding_size
= chan
->backend
.subbuf_size
- data_size
;
1808 subbuffer_set_data_size(config
, &buf
->backend
, oldidx
, data_size
,
1811 ts_end
= shmp_index(handle
, buf
->ts_end
, oldidx
);
1815 * This is the last space reservation in that sub-buffer before
1816 * it gets delivered. This provides exclusive access to write to
1817 * this sub-buffer's ts_end. There are also no concurrent
1818 * readers of that ts_end because delivery of that sub-buffer is
1819 * postponed until the commit counter is incremented for the
1820 * current space reservation.
1825 * Order all writes to buffer and store to ts_end before the commit
1826 * count update that will determine that the subbuffer is full.
1829 cc_hot
= shmp_index(handle
, buf
->commit_hot
, oldidx
);
1832 v_add(config
, padding_size
, &cc_hot
->cc
);
1833 commit_count
= v_read(config
, &cc_hot
->cc
);
1834 lib_ring_buffer_check_deliver(config
, buf
, chan
, offsets
->old
- 1,
1835 commit_count
, oldidx
, handle
, tsc
);
1836 lib_ring_buffer_write_commit_counter(config
, buf
, chan
,
1837 offsets
->old
+ padding_size
, commit_count
, handle
,
1842 * lib_ring_buffer_switch_new_start: Populate new subbuffer.
1844 * This code can be executed unordered : writers may already have written to the
1845 * sub-buffer before this code gets executed, caution. The commit makes sure
1846 * that this code is executed before the deliver of this sub-buffer.
1849 void lib_ring_buffer_switch_new_start(struct lttng_ust_lib_ring_buffer
*buf
,
1850 struct lttng_ust_lib_ring_buffer_channel
*chan
,
1851 struct switch_offsets
*offsets
,
1853 struct lttng_ust_shm_handle
*handle
)
1855 const struct lttng_ust_lib_ring_buffer_config
*config
= &chan
->backend
.config
;
1856 unsigned long beginidx
= subbuf_index(offsets
->begin
, chan
);
1857 unsigned long commit_count
;
1858 struct commit_counters_hot
*cc_hot
;
1860 config
->cb
.buffer_begin(buf
, tsc
, beginidx
, handle
);
1863 * Order all writes to buffer before the commit count update that will
1864 * determine that the subbuffer is full.
1867 cc_hot
= shmp_index(handle
, buf
->commit_hot
, beginidx
);
1870 v_add(config
, config
->cb
.subbuffer_header_size(), &cc_hot
->cc
);
1871 commit_count
= v_read(config
, &cc_hot
->cc
);
1872 /* Check if the written buffer has to be delivered */
1873 lib_ring_buffer_check_deliver(config
, buf
, chan
, offsets
->begin
,
1874 commit_count
, beginidx
, handle
, tsc
);
1875 lib_ring_buffer_write_commit_counter(config
, buf
, chan
,
1876 offsets
->begin
+ config
->cb
.subbuffer_header_size(),
1877 commit_count
, handle
, cc_hot
);
1881 * lib_ring_buffer_switch_new_end: finish switching current subbuffer
1883 * Calls subbuffer_set_data_size() to set the data size of the current
1884 * sub-buffer. We do not need to perform check_deliver nor commit here,
1885 * since this task will be done by the "commit" of the event for which
1886 * we are currently doing the space reservation.
1889 void lib_ring_buffer_switch_new_end(struct lttng_ust_lib_ring_buffer
*buf
,
1890 struct lttng_ust_lib_ring_buffer_channel
*chan
,
1891 struct switch_offsets
*offsets
,
1893 struct lttng_ust_shm_handle
*handle
)
1895 const struct lttng_ust_lib_ring_buffer_config
*config
= &chan
->backend
.config
;
1896 unsigned long endidx
, data_size
;
1899 endidx
= subbuf_index(offsets
->end
- 1, chan
);
1900 data_size
= subbuf_offset(offsets
->end
- 1, chan
) + 1;
1901 subbuffer_set_data_size(config
, &buf
->backend
, endidx
, data_size
,
1903 ts_end
= shmp_index(handle
, buf
->ts_end
, endidx
);
1907 * This is the last space reservation in that sub-buffer before
1908 * it gets delivered. This provides exclusive access to write to
1909 * this sub-buffer's ts_end. There are also no concurrent
1910 * readers of that ts_end because delivery of that sub-buffer is
1911 * postponed until the commit counter is incremented for the
1912 * current space reservation.
1920 * !0 if execution must be aborted.
1923 int lib_ring_buffer_try_switch_slow(enum switch_mode mode
,
1924 struct lttng_ust_lib_ring_buffer
*buf
,
1925 struct lttng_ust_lib_ring_buffer_channel
*chan
,
1926 struct switch_offsets
*offsets
,
1928 struct lttng_ust_shm_handle
*handle
)
1930 const struct lttng_ust_lib_ring_buffer_config
*config
= &chan
->backend
.config
;
1931 unsigned long off
, reserve_commit_diff
;
1933 offsets
->begin
= v_read(config
, &buf
->offset
);
1934 offsets
->old
= offsets
->begin
;
1935 offsets
->switch_old_start
= 0;
1936 off
= subbuf_offset(offsets
->begin
, chan
);
1938 *tsc
= config
->cb
.ring_buffer_clock_read(chan
);
1941 * Ensure we flush the header of an empty subbuffer when doing the
1942 * finalize (SWITCH_FLUSH). This ensures that we end up knowing the
1943 * total data gathering duration even if there were no records saved
1944 * after the last buffer switch.
1945 * In SWITCH_ACTIVE mode, switch the buffer when it contains events.
1946 * SWITCH_ACTIVE only flushes the current subbuffer, dealing with end of
1947 * subbuffer header as appropriate.
1948 * The next record that reserves space will be responsible for
1949 * populating the following subbuffer header. We choose not to populate
1950 * the next subbuffer header here because we want to be able to use
1951 * SWITCH_ACTIVE for periodical buffer flush, which must
1952 * guarantee that all the buffer content (records and header
1953 * timestamps) are visible to the reader. This is required for
1954 * quiescence guarantees for the fusion merge.
1956 if (mode
!= SWITCH_FLUSH
&& !off
)
1957 return -1; /* we do not have to switch : buffer is empty */
1959 if (caa_unlikely(off
== 0)) {
1960 unsigned long sb_index
, commit_count
;
1961 struct commit_counters_cold
*cc_cold
;
1964 * We are performing a SWITCH_FLUSH. There may be concurrent
1965 * writes into the buffer if e.g. invoked while performing a
1966 * snapshot on an active trace.
1968 * If the client does not save any header information
1969 * (sub-buffer header size == 0), don't switch empty subbuffer
1970 * on finalize, because it is invalid to deliver a completely
1973 if (!config
->cb
.subbuffer_header_size())
1976 /* Test new buffer integrity */
1977 sb_index
= subbuf_index(offsets
->begin
, chan
);
1978 cc_cold
= shmp_index(handle
, buf
->commit_cold
, sb_index
);
1981 commit_count
= v_read(config
, &cc_cold
->cc_sb
);
1982 reserve_commit_diff
=
1983 (buf_trunc(offsets
->begin
, chan
)
1984 >> chan
->backend
.num_subbuf_order
)
1985 - (commit_count
& chan
->commit_count_mask
);
1986 if (caa_likely(reserve_commit_diff
== 0)) {
1987 /* Next subbuffer not being written to. */
1988 if (caa_unlikely(config
->mode
!= RING_BUFFER_OVERWRITE
&&
1989 subbuf_trunc(offsets
->begin
, chan
)
1990 - subbuf_trunc((unsigned long)
1991 uatomic_read(&buf
->consumed
), chan
)
1992 >= chan
->backend
.buf_size
)) {
1994 * We do not overwrite non consumed buffers
1995 * and we are full : don't switch.
2000 * Next subbuffer not being written to, and we
2001 * are either in overwrite mode or the buffer is
2002 * not full. It's safe to write in this new
2008 * Next subbuffer reserve offset does not match the
2009 * commit offset. Don't perform switch in
2010 * producer-consumer and overwrite mode. Caused by
2011 * either a writer OOPS or too many nested writes over a
2012 * reserve/commit pair.
2018 * Need to write the subbuffer start header on finalize.
2020 offsets
->switch_old_start
= 1;
2022 offsets
->begin
= subbuf_align(offsets
->begin
, chan
);
2023 /* Note: old points to the next subbuf at offset 0 */
2024 offsets
->end
= offsets
->begin
;
2029 * Force a sub-buffer switch. This operation is completely reentrant : can be
2030 * called while tracing is active with absolutely no lock held.
2032 * For RING_BUFFER_SYNC_PER_CPU ring buffers, as a v_cmpxchg is used for
2033 * some atomic operations, this function must be called from the CPU
2034 * which owns the buffer for a ACTIVE flush. However, for
2035 * RING_BUFFER_SYNC_GLOBAL ring buffers, this function can be called
2038 void lib_ring_buffer_switch_slow(struct lttng_ust_lib_ring_buffer
*buf
, enum switch_mode mode
,
2039 struct lttng_ust_shm_handle
*handle
)
2041 struct lttng_ust_lib_ring_buffer_channel
*chan
;
2042 const struct lttng_ust_lib_ring_buffer_config
*config
;
2043 struct switch_offsets offsets
;
2044 unsigned long oldidx
;
2047 chan
= shmp(handle
, buf
->backend
.chan
);
2050 config
= &chan
->backend
.config
;
2055 * Perform retryable operations.
2058 if (lib_ring_buffer_try_switch_slow(mode
, buf
, chan
, &offsets
,
2060 return; /* Switch not needed */
2061 } while (v_cmpxchg(config
, &buf
->offset
, offsets
.old
, offsets
.end
)
2065 * Atomically update last_tsc. This update races against concurrent
2066 * atomic updates, but the race will always cause supplementary full TSC
2067 * records, never the opposite (missing a full TSC record when it would
2070 save_last_tsc(config
, buf
, tsc
);
2073 * Push the reader if necessary
2075 lib_ring_buffer_reserve_push_reader(buf
, chan
, offsets
.old
);
2077 oldidx
= subbuf_index(offsets
.old
, chan
);
2078 lib_ring_buffer_clear_noref(config
, &buf
->backend
, oldidx
, handle
);
2081 * May need to populate header start on SWITCH_FLUSH.
2083 if (offsets
.switch_old_start
) {
2084 lib_ring_buffer_switch_old_start(buf
, chan
, &offsets
, tsc
, handle
);
2085 offsets
.old
+= config
->cb
.subbuffer_header_size();
2089 * Switch old subbuffer.
2091 lib_ring_buffer_switch_old_end(buf
, chan
, &offsets
, tsc
, handle
);
2095 bool handle_blocking_retry(int *timeout_left_ms
)
2097 int timeout
= *timeout_left_ms
, delay
;
2099 if (caa_likely(!timeout
))
2100 return false; /* Do not retry, discard event. */
2101 if (timeout
< 0) /* Wait forever. */
2102 delay
= RETRY_DELAY_MS
;
2104 delay
= min_t(int, timeout
, RETRY_DELAY_MS
);
2105 (void) poll(NULL
, 0, delay
);
2107 *timeout_left_ms
-= delay
;
2108 return true; /* Retry. */
2114 * -ENOSPC if event size is too large for packet.
2115 * -ENOBUFS if there is currently not enough space in buffer for the event.
2116 * -EIO if data cannot be written into the buffer for any other reason.
2119 int lib_ring_buffer_try_reserve_slow(struct lttng_ust_lib_ring_buffer
*buf
,
2120 struct lttng_ust_lib_ring_buffer_channel
*chan
,
2121 struct switch_offsets
*offsets
,
2122 struct lttng_ust_lib_ring_buffer_ctx
*ctx
,
2125 const struct lttng_ust_lib_ring_buffer_config
*config
= &chan
->backend
.config
;
2126 struct lttng_ust_shm_handle
*handle
= ctx
->handle
;
2127 unsigned long reserve_commit_diff
, offset_cmp
;
2128 int timeout_left_ms
= lttng_ust_ringbuffer_get_timeout(chan
);
2131 offsets
->begin
= offset_cmp
= v_read(config
, &buf
->offset
);
2132 offsets
->old
= offsets
->begin
;
2133 offsets
->switch_new_start
= 0;
2134 offsets
->switch_new_end
= 0;
2135 offsets
->switch_old_end
= 0;
2136 offsets
->pre_header_padding
= 0;
2138 ctx
->tsc
= config
->cb
.ring_buffer_clock_read(chan
);
2139 if ((int64_t) ctx
->tsc
== -EIO
)
2142 if (last_tsc_overflow(config
, buf
, ctx
->tsc
))
2143 ctx
->rflags
|= RING_BUFFER_RFLAG_FULL_TSC
;
2145 if (caa_unlikely(subbuf_offset(offsets
->begin
, ctx
->chan
) == 0)) {
2146 offsets
->switch_new_start
= 1; /* For offsets->begin */
2148 offsets
->size
= config
->cb
.record_header_size(config
, chan
,
2150 &offsets
->pre_header_padding
,
2153 lib_ring_buffer_align(offsets
->begin
+ offsets
->size
,
2156 if (caa_unlikely(subbuf_offset(offsets
->begin
, chan
) +
2157 offsets
->size
> chan
->backend
.subbuf_size
)) {
2158 offsets
->switch_old_end
= 1; /* For offsets->old */
2159 offsets
->switch_new_start
= 1; /* For offsets->begin */
2162 if (caa_unlikely(offsets
->switch_new_start
)) {
2163 unsigned long sb_index
, commit_count
;
2164 struct commit_counters_cold
*cc_cold
;
2167 * We are typically not filling the previous buffer completely.
2169 if (caa_likely(offsets
->switch_old_end
))
2170 offsets
->begin
= subbuf_align(offsets
->begin
, chan
);
2171 offsets
->begin
= offsets
->begin
2172 + config
->cb
.subbuffer_header_size();
2173 /* Test new buffer integrity */
2174 sb_index
= subbuf_index(offsets
->begin
, chan
);
2176 * Read buf->offset before buf->commit_cold[sb_index].cc_sb.
2177 * lib_ring_buffer_check_deliver() has the matching
2178 * memory barriers required around commit_cold cc_sb
2179 * updates to ensure reserve and commit counter updates
2180 * are not seen reordered when updated by another CPU.
2183 cc_cold
= shmp_index(handle
, buf
->commit_cold
, sb_index
);
2186 commit_count
= v_read(config
, &cc_cold
->cc_sb
);
2187 /* Read buf->commit_cold[sb_index].cc_sb before buf->offset. */
2189 if (caa_unlikely(offset_cmp
!= v_read(config
, &buf
->offset
))) {
2191 * The reserve counter have been concurrently updated
2192 * while we read the commit counter. This means the
2193 * commit counter we read might not match buf->offset
2194 * due to concurrent update. We therefore need to retry.
2198 reserve_commit_diff
=
2199 (buf_trunc(offsets
->begin
, chan
)
2200 >> chan
->backend
.num_subbuf_order
)
2201 - (commit_count
& chan
->commit_count_mask
);
2202 if (caa_likely(reserve_commit_diff
== 0)) {
2203 /* Next subbuffer not being written to. */
2204 if (caa_unlikely(config
->mode
!= RING_BUFFER_OVERWRITE
&&
2205 subbuf_trunc(offsets
->begin
, chan
)
2206 - subbuf_trunc((unsigned long)
2207 uatomic_read(&buf
->consumed
), chan
)
2208 >= chan
->backend
.buf_size
)) {
2209 unsigned long nr_lost
;
2211 if (handle_blocking_retry(&timeout_left_ms
))
2215 * We do not overwrite non consumed buffers
2216 * and we are full : record is lost.
2218 nr_lost
= v_read(config
, &buf
->records_lost_full
);
2219 v_inc(config
, &buf
->records_lost_full
);
2220 if ((nr_lost
& (DBG_PRINT_NR_LOST
- 1)) == 0) {
2221 DBG("%lu or more records lost in (%s:%d) (buffer full)\n",
2222 nr_lost
+ 1, chan
->backend
.name
,
2228 * Next subbuffer not being written to, and we
2229 * are either in overwrite mode or the buffer is
2230 * not full. It's safe to write in this new
2235 unsigned long nr_lost
;
2238 * Next subbuffer reserve offset does not match the
2239 * commit offset, and this did not involve update to the
2240 * reserve counter. Drop record in producer-consumer and
2241 * overwrite mode. Caused by either a writer OOPS or too
2242 * many nested writes over a reserve/commit pair.
2244 nr_lost
= v_read(config
, &buf
->records_lost_wrap
);
2245 v_inc(config
, &buf
->records_lost_wrap
);
2246 if ((nr_lost
& (DBG_PRINT_NR_LOST
- 1)) == 0) {
2247 DBG("%lu or more records lost in (%s:%d) (wrap-around)\n",
2248 nr_lost
+ 1, chan
->backend
.name
,
2254 config
->cb
.record_header_size(config
, chan
,
2256 &offsets
->pre_header_padding
,
2259 lib_ring_buffer_align(offsets
->begin
+ offsets
->size
,
2262 if (caa_unlikely(subbuf_offset(offsets
->begin
, chan
)
2263 + offsets
->size
> chan
->backend
.subbuf_size
)) {
2264 unsigned long nr_lost
;
2267 * Record too big for subbuffers, report error, don't
2268 * complete the sub-buffer switch.
2270 nr_lost
= v_read(config
, &buf
->records_lost_big
);
2271 v_inc(config
, &buf
->records_lost_big
);
2272 if ((nr_lost
& (DBG_PRINT_NR_LOST
- 1)) == 0) {
2273 DBG("%lu or more records lost in (%s:%d) record size "
2274 " of %zu bytes is too large for buffer\n",
2275 nr_lost
+ 1, chan
->backend
.name
,
2276 buf
->backend
.cpu
, offsets
->size
);
2281 * We just made a successful buffer switch and the
2282 * record fits in the new subbuffer. Let's write.
2287 * Record fits in the current buffer and we are not on a switch
2288 * boundary. It's safe to write.
2291 offsets
->end
= offsets
->begin
+ offsets
->size
;
2293 if (caa_unlikely(subbuf_offset(offsets
->end
, chan
) == 0)) {
2295 * The offset_end will fall at the very beginning of the next
2298 offsets
->switch_new_end
= 1; /* For offsets->begin */
2304 * lib_ring_buffer_reserve_slow - Atomic slot reservation in a buffer.
2305 * @ctx: ring buffer context.
2307 * Return : -NOBUFS if not enough space, -ENOSPC if event size too large,
2308 * -EIO for other errors, else returns 0.
2309 * It will take care of sub-buffer switching.
2311 int lib_ring_buffer_reserve_slow(struct lttng_ust_lib_ring_buffer_ctx
*ctx
,
2314 struct lttng_ust_lib_ring_buffer_channel
*chan
= ctx
->chan
;
2315 struct lttng_ust_shm_handle
*handle
= ctx
->handle
;
2316 const struct lttng_ust_lib_ring_buffer_config
*config
= &chan
->backend
.config
;
2317 struct lttng_ust_lib_ring_buffer
*buf
;
2318 struct switch_offsets offsets
;
2321 if (config
->alloc
== RING_BUFFER_ALLOC_PER_CPU
)
2322 buf
= shmp(handle
, chan
->backend
.buf
[ctx
->cpu
].shmp
);
2324 buf
= shmp(handle
, chan
->backend
.buf
[0].shmp
);
2332 ret
= lib_ring_buffer_try_reserve_slow(buf
, chan
, &offsets
,
2334 if (caa_unlikely(ret
))
2336 } while (caa_unlikely(v_cmpxchg(config
, &buf
->offset
, offsets
.old
,
2341 * Atomically update last_tsc. This update races against concurrent
2342 * atomic updates, but the race will always cause supplementary full TSC
2343 * records, never the opposite (missing a full TSC record when it would
2346 save_last_tsc(config
, buf
, ctx
->tsc
);
2349 * Push the reader if necessary
2351 lib_ring_buffer_reserve_push_reader(buf
, chan
, offsets
.end
- 1);
2354 * Clear noref flag for this subbuffer.
2356 lib_ring_buffer_clear_noref(config
, &buf
->backend
,
2357 subbuf_index(offsets
.end
- 1, chan
),
2361 * Switch old subbuffer if needed.
2363 if (caa_unlikely(offsets
.switch_old_end
)) {
2364 lib_ring_buffer_clear_noref(config
, &buf
->backend
,
2365 subbuf_index(offsets
.old
- 1, chan
),
2367 lib_ring_buffer_switch_old_end(buf
, chan
, &offsets
, ctx
->tsc
, handle
);
2371 * Populate new subbuffer.
2373 if (caa_unlikely(offsets
.switch_new_start
))
2374 lib_ring_buffer_switch_new_start(buf
, chan
, &offsets
, ctx
->tsc
, handle
);
2376 if (caa_unlikely(offsets
.switch_new_end
))
2377 lib_ring_buffer_switch_new_end(buf
, chan
, &offsets
, ctx
->tsc
, handle
);
2379 ctx
->slot_size
= offsets
.size
;
2380 ctx
->pre_offset
= offsets
.begin
;
2381 ctx
->buf_offset
= offsets
.begin
+ offsets
.pre_header_padding
;
2386 void lib_ring_buffer_vmcore_check_deliver(const struct lttng_ust_lib_ring_buffer_config
*config
,
2387 struct lttng_ust_lib_ring_buffer
*buf
,
2388 unsigned long commit_count
,
2390 struct lttng_ust_shm_handle
*handle
)
2392 struct commit_counters_hot
*cc_hot
;
2394 if (config
->oops
!= RING_BUFFER_OOPS_CONSISTENCY
)
2396 cc_hot
= shmp_index(handle
, buf
->commit_hot
, idx
);
2399 v_set(config
, &cc_hot
->seq
, commit_count
);
2403 * The ring buffer can count events recorded and overwritten per buffer,
2404 * but it is disabled by default due to its performance overhead.
2406 #ifdef LTTNG_RING_BUFFER_COUNT_EVENTS
2408 void deliver_count_events(const struct lttng_ust_lib_ring_buffer_config
*config
,
2409 struct lttng_ust_lib_ring_buffer
*buf
,
2411 struct lttng_ust_shm_handle
*handle
)
2413 v_add(config
, subbuffer_get_records_count(config
,
2414 &buf
->backend
, idx
, handle
),
2415 &buf
->records_count
);
2416 v_add(config
, subbuffer_count_records_overrun(config
,
2417 &buf
->backend
, idx
, handle
),
2418 &buf
->records_overrun
);
2420 #else /* LTTNG_RING_BUFFER_COUNT_EVENTS */
2422 void deliver_count_events(const struct lttng_ust_lib_ring_buffer_config
*config
,
2423 struct lttng_ust_lib_ring_buffer
*buf
,
2425 struct lttng_ust_shm_handle
*handle
)
2428 #endif /* #else LTTNG_RING_BUFFER_COUNT_EVENTS */
2430 void lib_ring_buffer_check_deliver_slow(const struct lttng_ust_lib_ring_buffer_config
*config
,
2431 struct lttng_ust_lib_ring_buffer
*buf
,
2432 struct lttng_ust_lib_ring_buffer_channel
*chan
,
2433 unsigned long offset
,
2434 unsigned long commit_count
,
2436 struct lttng_ust_shm_handle
*handle
,
2439 unsigned long old_commit_count
= commit_count
2440 - chan
->backend
.subbuf_size
;
2441 struct commit_counters_cold
*cc_cold
;
2444 * If we succeeded at updating cc_sb below, we are the subbuffer
2445 * writer delivering the subbuffer. Deals with concurrent
2446 * updates of the "cc" value without adding a add_return atomic
2447 * operation to the fast path.
2449 * We are doing the delivery in two steps:
2450 * - First, we cmpxchg() cc_sb to the new value
2451 * old_commit_count + 1. This ensures that we are the only
2452 * subbuffer user successfully filling the subbuffer, but we
2453 * do _not_ set the cc_sb value to "commit_count" yet.
2454 * Therefore, other writers that would wrap around the ring
2455 * buffer and try to start writing to our subbuffer would
2456 * have to drop records, because it would appear as
2458 * We therefore have exclusive access to the subbuffer control
2459 * structures. This mutual exclusion with other writers is
2460 * crucially important to perform record overruns count in
2461 * flight recorder mode locklessly.
2462 * - When we are ready to release the subbuffer (either for
2463 * reading or for overrun by other writers), we simply set the
2464 * cc_sb value to "commit_count" and perform delivery.
2466 * The subbuffer size is least 2 bytes (minimum size: 1 page).
2467 * This guarantees that old_commit_count + 1 != commit_count.
2471 * Order prior updates to reserve count prior to the
2472 * commit_cold cc_sb update.
2475 cc_cold
= shmp_index(handle
, buf
->commit_cold
, idx
);
2478 if (caa_likely(v_cmpxchg(config
, &cc_cold
->cc_sb
,
2479 old_commit_count
, old_commit_count
+ 1)
2480 == old_commit_count
)) {
2484 * Start of exclusive subbuffer access. We are
2485 * guaranteed to be the last writer in this subbuffer
2486 * and any other writer trying to access this subbuffer
2487 * in this state is required to drop records.
2489 * We can read the ts_end for the current sub-buffer
2490 * which has been saved by the very last space
2491 * reservation for the current sub-buffer.
2493 * Order increment of commit counter before reading ts_end.
2496 ts_end
= shmp_index(handle
, buf
->ts_end
, idx
);
2499 deliver_count_events(config
, buf
, idx
, handle
);
2500 config
->cb
.buffer_end(buf
, *ts_end
, idx
,
2501 lib_ring_buffer_get_data_size(config
,
2508 * Increment the packet counter while we have exclusive
2511 subbuffer_inc_packet_count(config
, &buf
->backend
, idx
, handle
);
2514 * Set noref flag and offset for this subbuffer id.
2515 * Contains a memory barrier that ensures counter stores
2516 * are ordered before set noref and offset.
2518 lib_ring_buffer_set_noref_offset(config
, &buf
->backend
, idx
,
2519 buf_trunc_val(offset
, chan
), handle
);
2522 * Order set_noref and record counter updates before the
2523 * end of subbuffer exclusive access. Orders with
2524 * respect to writers coming into the subbuffer after
2525 * wrap around, and also order wrt concurrent readers.
2528 /* End of exclusive subbuffer access */
2529 v_set(config
, &cc_cold
->cc_sb
, commit_count
);
2531 * Order later updates to reserve count after
2532 * the commit cold cc_sb update.
2535 lib_ring_buffer_vmcore_check_deliver(config
, buf
,
2536 commit_count
, idx
, handle
);
2539 * RING_BUFFER_WAKEUP_BY_WRITER wakeup is not lock-free.
2541 if (config
->wakeup
== RING_BUFFER_WAKEUP_BY_WRITER
2542 && uatomic_read(&buf
->active_readers
)
2543 && lib_ring_buffer_poll_deliver(config
, buf
, chan
, handle
)) {
2544 lib_ring_buffer_wakeup(buf
, handle
);
2550 * Force a read (imply TLS fixup for dlopen) of TLS variables.
2552 void lttng_fixup_ringbuffer_tls(void)
2554 asm volatile ("" : : "m" (URCU_TLS(lib_ring_buffer_nesting
)));
2557 void lib_ringbuffer_signal_init(void)
2563 * Block signal for entire process, so only our thread processes
2567 ret
= pthread_sigmask(SIG_BLOCK
, &mask
, NULL
);
2570 PERROR("pthread_sigmask");