Use rseq for commit counter
[lttng-ust.git] / libringbuffer / ring_buffer_frontend.c
1 /*
2 * ring_buffer_frontend.c
3 *
4 * Copyright (C) 2005-2012 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
5 *
6 * This library is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU Lesser General Public
8 * License as published by the Free Software Foundation; only
9 * version 2.1 of the License.
10 *
11 * This library is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
14 * Lesser General Public License for more details.
15 *
16 * You should have received a copy of the GNU Lesser General Public
17 * License along with this library; if not, write to the Free Software
18 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
19 *
20 *
21 * Ring buffer wait-free buffer synchronization. Producer-consumer and flight
22 * recorder (overwrite) modes. See thesis:
23 *
24 * Desnoyers, Mathieu (2009), "Low-Impact Operating System Tracing", Ph.D.
25 * dissertation, Ecole Polytechnique de Montreal.
26 * http://www.lttng.org/pub/thesis/desnoyers-dissertation-2009-12.pdf
27 *
28 * - Algorithm presentation in Chapter 5:
29 * "Lockless Multi-Core High-Throughput Buffering".
30 * - Algorithm formal verification in Section 8.6:
31 * "Formal verification of LTTng"
32 *
33 * Author:
34 * Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
35 *
36 * Inspired from LTT and RelayFS:
37 * Karim Yaghmour <karim@opersys.com>
38 * Tom Zanussi <zanussi@us.ibm.com>
39 * Bob Wisniewski <bob@watson.ibm.com>
40 * And from K42 :
41 * Bob Wisniewski <bob@watson.ibm.com>
42 *
43 * Buffer reader semantic :
44 *
45 * - get_subbuf_size
46 * while buffer is not finalized and empty
47 * - get_subbuf
48 * - if return value != 0, continue
49 * - splice one subbuffer worth of data to a pipe
50 * - splice the data from pipe to disk/network
51 * - put_subbuf
52 */
53
54 #define _GNU_SOURCE
55 #define _LGPL_SOURCE
56 #include <sys/types.h>
57 #include <sys/mman.h>
58 #include <sys/stat.h>
59 #include <unistd.h>
60 #include <fcntl.h>
61 #include <signal.h>
62 #include <time.h>
63 #include <urcu/compiler.h>
64 #include <urcu/ref.h>
65 #include <urcu/tls-compat.h>
66 #include <poll.h>
67 #include <helper.h>
68
69 #include "smp.h"
70 #include <lttng/ringbuffer-config.h>
71 #include "vatomic.h"
72 #include "backend.h"
73 #include "frontend.h"
74 #include "shm.h"
75 #include "tlsfixup.h"
76 #include "../liblttng-ust/compat.h" /* For ENODATA */
77
78 /* Print DBG() messages about events lost only every 1048576 hits */
79 #define DBG_PRINT_NR_LOST (1UL << 20)
80
81 #define LTTNG_UST_RB_SIG_FLUSH SIGRTMIN
82 #define LTTNG_UST_RB_SIG_READ SIGRTMIN + 1
83 #define LTTNG_UST_RB_SIG_TEARDOWN SIGRTMIN + 2
84 #define CLOCKID CLOCK_MONOTONIC
85 #define LTTNG_UST_RING_BUFFER_GET_RETRY 10
86 #define LTTNG_UST_RING_BUFFER_RETRY_DELAY_MS 10
87
88 /*
89 * Non-static to ensure the compiler does not optimize away the xor.
90 */
91 uint8_t lttng_crash_magic_xor[] = RB_CRASH_DUMP_ABI_MAGIC_XOR;
92
93 /*
94 * Use POSIX SHM: shm_open(3) and shm_unlink(3).
95 * close(2) to close the fd returned by shm_open.
96 * shm_unlink releases the shared memory object name.
97 * ftruncate(2) sets the size of the memory object.
98 * mmap/munmap maps the shared memory obj to a virtual address in the
99 * calling proceess (should be done both in libust and consumer).
100 * See shm_overview(7) for details.
101 * Pass file descriptor returned by shm_open(3) to ltt-sessiond through
102 * a UNIX socket.
103 *
104 * Since we don't need to access the object using its name, we can
105 * immediately shm_unlink(3) it, and only keep the handle with its file
106 * descriptor.
107 */
108
109 /*
110 * Internal structure representing offsets to use at a sub-buffer switch.
111 */
112 struct switch_offsets {
113 unsigned long begin, end, old;
114 size_t pre_header_padding, size;
115 unsigned int switch_new_start:1, switch_new_end:1, switch_old_start:1,
116 switch_old_end:1;
117 };
118
119 DEFINE_URCU_TLS(unsigned int, lib_ring_buffer_nesting);
120
121 /*
122 * wakeup_fd_mutex protects wakeup fd use by timer from concurrent
123 * close.
124 */
125 static pthread_mutex_t wakeup_fd_mutex = PTHREAD_MUTEX_INITIALIZER;
126
127 static
128 void lib_ring_buffer_print_errors(struct channel *chan,
129 struct lttng_ust_lib_ring_buffer *buf, int cpu,
130 struct lttng_ust_shm_handle *handle);
131
132 /*
133 * Handle timer teardown race wrt memory free of private data by
134 * ring buffer signals are handled by a single thread, which permits
135 * a synchronization point between handling of each signal.
136 * Protected by the lock within the structure.
137 */
138 struct timer_signal_data {
139 pthread_t tid; /* thread id managing signals */
140 int setup_done;
141 int qs_done;
142 pthread_mutex_t lock;
143 };
144
145 static struct timer_signal_data timer_signal = {
146 .tid = 0,
147 .setup_done = 0,
148 .qs_done = 0,
149 .lock = PTHREAD_MUTEX_INITIALIZER,
150 };
151
152 /**
153 * lib_ring_buffer_reset - Reset ring buffer to initial values.
154 * @buf: Ring buffer.
155 *
156 * Effectively empty the ring buffer. Should be called when the buffer is not
157 * used for writing. The ring buffer can be opened for reading, but the reader
158 * should not be using the iterator concurrently with reset. The previous
159 * current iterator record is reset.
160 */
161 void lib_ring_buffer_reset(struct lttng_ust_lib_ring_buffer *buf,
162 struct lttng_ust_shm_handle *handle)
163 {
164 struct channel *chan;
165 const struct lttng_ust_lib_ring_buffer_config *config;
166 unsigned int i;
167
168 chan = shmp(handle, buf->backend.chan);
169 if (!chan)
170 return;
171 config = &chan->backend.config;
172 /*
173 * Reset iterator first. It will put the subbuffer if it currently holds
174 * it.
175 */
176 v_set(config, &buf->offset, 0);
177 for (i = 0; i < chan->backend.num_subbuf; i++) {
178 struct commit_counters_hot *cc_hot;
179 struct commit_counters_cold *cc_cold;
180
181 cc_hot = shmp_index(handle, buf->commit_hot, i);
182 if (!cc_hot)
183 return;
184 cc_cold = shmp_index(handle, buf->commit_cold, i);
185 if (!cc_cold)
186 return;
187 v_set(config, &cc_hot->cc, 0);
188 cc_hot->cc_rseq = 0;
189 v_set(config, &cc_hot->seq, 0);
190 v_set(config, &cc_cold->cc_sb, 0);
191 }
192 uatomic_set(&buf->consumed, 0);
193 uatomic_set(&buf->record_disabled, 0);
194 v_set(config, &buf->last_tsc, 0);
195 lib_ring_buffer_backend_reset(&buf->backend, handle);
196 /* Don't reset number of active readers */
197 v_set(config, &buf->records_lost_full, 0);
198 v_set(config, &buf->records_lost_wrap, 0);
199 v_set(config, &buf->records_lost_big, 0);
200 v_set(config, &buf->records_count, 0);
201 v_set(config, &buf->records_overrun, 0);
202 buf->finalized = 0;
203 }
204
205 /**
206 * channel_reset - Reset channel to initial values.
207 * @chan: Channel.
208 *
209 * Effectively empty the channel. Should be called when the channel is not used
210 * for writing. The channel can be opened for reading, but the reader should not
211 * be using the iterator concurrently with reset. The previous current iterator
212 * record is reset.
213 */
214 void channel_reset(struct channel *chan)
215 {
216 /*
217 * Reset iterators first. Will put the subbuffer if held for reading.
218 */
219 uatomic_set(&chan->record_disabled, 0);
220 /* Don't reset commit_count_mask, still valid */
221 channel_backend_reset(&chan->backend);
222 /* Don't reset switch/read timer interval */
223 /* Don't reset notifiers and notifier enable bits */
224 /* Don't reset reader reference count */
225 }
226
227 static
228 void init_crash_abi(const struct lttng_ust_lib_ring_buffer_config *config,
229 struct lttng_crash_abi *crash_abi,
230 struct lttng_ust_lib_ring_buffer *buf,
231 struct channel_backend *chanb,
232 struct shm_object *shmobj,
233 struct lttng_ust_shm_handle *handle)
234 {
235 int i;
236
237 for (i = 0; i < RB_CRASH_DUMP_ABI_MAGIC_LEN; i++)
238 crash_abi->magic[i] = lttng_crash_magic_xor[i] ^ 0xFF;
239 crash_abi->mmap_length = shmobj->memory_map_size;
240 crash_abi->endian = RB_CRASH_ENDIAN;
241 crash_abi->major = RB_CRASH_DUMP_ABI_MAJOR;
242 crash_abi->minor = RB_CRASH_DUMP_ABI_MINOR;
243 crash_abi->word_size = sizeof(unsigned long);
244 crash_abi->layout_type = LTTNG_CRASH_TYPE_UST;
245
246 /* Offset of fields */
247 crash_abi->offset.prod_offset =
248 (uint32_t) ((char *) &buf->offset - (char *) buf);
249 crash_abi->offset.consumed_offset =
250 (uint32_t) ((char *) &buf->consumed - (char *) buf);
251 crash_abi->offset.commit_hot_array =
252 (uint32_t) ((char *) shmp(handle, buf->commit_hot) - (char *) buf);
253 crash_abi->offset.commit_hot_seq =
254 offsetof(struct commit_counters_hot, seq);
255 crash_abi->offset.buf_wsb_array =
256 (uint32_t) ((char *) shmp(handle, buf->backend.buf_wsb) - (char *) buf);
257 crash_abi->offset.buf_wsb_id =
258 offsetof(struct lttng_ust_lib_ring_buffer_backend_subbuffer, id);
259 crash_abi->offset.sb_array =
260 (uint32_t) ((char *) shmp(handle, buf->backend.array) - (char *) buf);
261 crash_abi->offset.sb_array_shmp_offset =
262 offsetof(struct lttng_ust_lib_ring_buffer_backend_pages_shmp,
263 shmp._ref.offset);
264 crash_abi->offset.sb_backend_p_offset =
265 offsetof(struct lttng_ust_lib_ring_buffer_backend_pages,
266 p._ref.offset);
267
268 /* Field length */
269 crash_abi->length.prod_offset = sizeof(buf->offset);
270 crash_abi->length.consumed_offset = sizeof(buf->consumed);
271 crash_abi->length.commit_hot_seq =
272 sizeof(((struct commit_counters_hot *) NULL)->seq);
273 crash_abi->length.buf_wsb_id =
274 sizeof(((struct lttng_ust_lib_ring_buffer_backend_subbuffer *) NULL)->id);
275 crash_abi->length.sb_array_shmp_offset =
276 sizeof(((struct lttng_ust_lib_ring_buffer_backend_pages_shmp *) NULL)->shmp._ref.offset);
277 crash_abi->length.sb_backend_p_offset =
278 sizeof(((struct lttng_ust_lib_ring_buffer_backend_pages *) NULL)->p._ref.offset);
279
280 /* Array stride */
281 crash_abi->stride.commit_hot_array =
282 sizeof(struct commit_counters_hot);
283 crash_abi->stride.buf_wsb_array =
284 sizeof(struct lttng_ust_lib_ring_buffer_backend_subbuffer);
285 crash_abi->stride.sb_array =
286 sizeof(struct lttng_ust_lib_ring_buffer_backend_pages_shmp);
287
288 /* Buffer constants */
289 crash_abi->buf_size = chanb->buf_size;
290 crash_abi->subbuf_size = chanb->subbuf_size;
291 crash_abi->num_subbuf = chanb->num_subbuf;
292 crash_abi->mode = (uint32_t) chanb->config.mode;
293
294 if (config->cb.content_size_field) {
295 size_t offset, length;
296
297 config->cb.content_size_field(config, &offset, &length);
298 crash_abi->offset.content_size = offset;
299 crash_abi->length.content_size = length;
300 } else {
301 crash_abi->offset.content_size = 0;
302 crash_abi->length.content_size = 0;
303 }
304 if (config->cb.packet_size_field) {
305 size_t offset, length;
306
307 config->cb.packet_size_field(config, &offset, &length);
308 crash_abi->offset.packet_size = offset;
309 crash_abi->length.packet_size = length;
310 } else {
311 crash_abi->offset.packet_size = 0;
312 crash_abi->length.packet_size = 0;
313 }
314 }
315
316 /*
317 * Must be called under cpu hotplug protection.
318 */
319 int lib_ring_buffer_create(struct lttng_ust_lib_ring_buffer *buf,
320 struct channel_backend *chanb, int cpu,
321 struct lttng_ust_shm_handle *handle,
322 struct shm_object *shmobj)
323 {
324 const struct lttng_ust_lib_ring_buffer_config *config = &chanb->config;
325 struct channel *chan = caa_container_of(chanb, struct channel, backend);
326 struct lttng_ust_lib_ring_buffer_backend_subbuffer *wsb;
327 struct channel *shmp_chan;
328 struct commit_counters_hot *cc_hot;
329 void *priv = channel_get_private(chan);
330 size_t subbuf_header_size;
331 uint64_t tsc;
332 int ret;
333
334 /* Test for cpu hotplug */
335 if (buf->backend.allocated)
336 return 0;
337
338 align_shm(shmobj, __alignof__(struct commit_counters_hot));
339 set_shmp(buf->commit_hot,
340 zalloc_shm(shmobj,
341 sizeof(struct commit_counters_hot) * chan->backend.num_subbuf));
342 if (!shmp(handle, buf->commit_hot)) {
343 return -ENOMEM;
344 }
345
346 align_shm(shmobj, __alignof__(struct commit_counters_cold));
347 set_shmp(buf->commit_cold,
348 zalloc_shm(shmobj,
349 sizeof(struct commit_counters_cold) * chan->backend.num_subbuf));
350 if (!shmp(handle, buf->commit_cold)) {
351 ret = -ENOMEM;
352 goto free_commit;
353 }
354
355 ret = lib_ring_buffer_backend_create(&buf->backend, &chan->backend,
356 cpu, handle, shmobj);
357 if (ret) {
358 goto free_init;
359 }
360
361 /*
362 * Write the subbuffer header for first subbuffer so we know the total
363 * duration of data gathering.
364 */
365 subbuf_header_size = config->cb.subbuffer_header_size();
366 v_set(config, &buf->offset, subbuf_header_size);
367 wsb = shmp_index(handle, buf->backend.buf_wsb, 0);
368 if (!wsb) {
369 ret = -EPERM;
370 goto free_chanbuf;
371 }
372 subbuffer_id_clear_noref(config, &wsb->id);
373 shmp_chan = shmp(handle, buf->backend.chan);
374 if (!shmp_chan) {
375 ret = -EPERM;
376 goto free_chanbuf;
377 }
378 tsc = config->cb.ring_buffer_clock_read(shmp_chan);
379 config->cb.buffer_begin(buf, tsc, 0, handle);
380 cc_hot = shmp_index(handle, buf->commit_hot, 0);
381 if (!cc_hot) {
382 ret = -EPERM;
383 goto free_chanbuf;
384 }
385 v_add(config, subbuf_header_size, &cc_hot->cc);
386 v_add(config, subbuf_header_size, &cc_hot->seq);
387
388 if (config->cb.buffer_create) {
389 ret = config->cb.buffer_create(buf, priv, cpu, chanb->name, handle);
390 if (ret)
391 goto free_chanbuf;
392 }
393
394 init_crash_abi(config, &buf->crash_abi, buf, chanb, shmobj, handle);
395
396 buf->backend.allocated = 1;
397 return 0;
398
399 /* Error handling */
400 free_init:
401 /* commit_cold will be freed by shm teardown */
402 free_commit:
403 /* commit_hot will be freed by shm teardown */
404 free_chanbuf:
405 return ret;
406 }
407
408 static
409 void lib_ring_buffer_channel_switch_timer(int sig, siginfo_t *si, void *uc)
410 {
411 const struct lttng_ust_lib_ring_buffer_config *config;
412 struct lttng_ust_shm_handle *handle;
413 struct channel *chan;
414 int cpu;
415
416 assert(CMM_LOAD_SHARED(timer_signal.tid) == pthread_self());
417
418 chan = si->si_value.sival_ptr;
419 handle = chan->handle;
420 config = &chan->backend.config;
421
422 DBG("Switch timer for channel %p\n", chan);
423
424 /*
425 * Only flush buffers periodically if readers are active.
426 */
427 pthread_mutex_lock(&wakeup_fd_mutex);
428 if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
429 for_each_possible_cpu(cpu) {
430 struct lttng_ust_lib_ring_buffer *buf =
431 shmp(handle, chan->backend.buf[cpu].shmp);
432
433 if (!buf)
434 goto end;
435 if (uatomic_read(&buf->active_readers))
436 lib_ring_buffer_switch_slow(buf, SWITCH_ACTIVE,
437 chan->handle);
438 }
439 } else {
440 struct lttng_ust_lib_ring_buffer *buf =
441 shmp(handle, chan->backend.buf[0].shmp);
442
443 if (!buf)
444 goto end;
445 if (uatomic_read(&buf->active_readers))
446 lib_ring_buffer_switch_slow(buf, SWITCH_ACTIVE,
447 chan->handle);
448 }
449 end:
450 pthread_mutex_unlock(&wakeup_fd_mutex);
451 return;
452 }
453
454 static
455 int lib_ring_buffer_poll_deliver(const struct lttng_ust_lib_ring_buffer_config *config,
456 struct lttng_ust_lib_ring_buffer *buf,
457 struct channel *chan,
458 struct lttng_ust_shm_handle *handle)
459 {
460 unsigned long consumed_old, consumed_idx, commit_count, write_offset;
461 struct commit_counters_cold *cc_cold;
462
463 consumed_old = uatomic_read(&buf->consumed);
464 consumed_idx = subbuf_index(consumed_old, chan);
465 cc_cold = shmp_index(handle, buf->commit_cold, consumed_idx);
466 if (!cc_cold)
467 return 0;
468 commit_count = v_read(config, &cc_cold->cc_sb);
469 /*
470 * No memory barrier here, since we are only interested
471 * in a statistically correct polling result. The next poll will
472 * get the data is we are racing. The mb() that ensures correct
473 * memory order is in get_subbuf.
474 */
475 write_offset = v_read(config, &buf->offset);
476
477 /*
478 * Check that the subbuffer we are trying to consume has been
479 * already fully committed.
480 */
481
482 if (((commit_count - chan->backend.subbuf_size)
483 & chan->commit_count_mask)
484 - (buf_trunc(consumed_old, chan)
485 >> chan->backend.num_subbuf_order)
486 != 0)
487 return 0;
488
489 /*
490 * Check that we are not about to read the same subbuffer in
491 * which the writer head is.
492 */
493 if (subbuf_trunc(write_offset, chan) - subbuf_trunc(consumed_old, chan)
494 == 0)
495 return 0;
496
497 return 1;
498 }
499
500 static
501 void lib_ring_buffer_wakeup(struct lttng_ust_lib_ring_buffer *buf,
502 struct lttng_ust_shm_handle *handle)
503 {
504 int wakeup_fd = shm_get_wakeup_fd(handle, &buf->self._ref);
505 sigset_t sigpipe_set, pending_set, old_set;
506 int ret, sigpipe_was_pending = 0;
507
508 if (wakeup_fd < 0)
509 return;
510
511 /*
512 * Wake-up the other end by writing a null byte in the pipe
513 * (non-blocking). Important note: Because writing into the
514 * pipe is non-blocking (and therefore we allow dropping wakeup
515 * data, as long as there is wakeup data present in the pipe
516 * buffer to wake up the consumer), the consumer should perform
517 * the following sequence for waiting:
518 * 1) empty the pipe (reads).
519 * 2) check if there is data in the buffer.
520 * 3) wait on the pipe (poll).
521 *
522 * Discard the SIGPIPE from write(), not disturbing any SIGPIPE
523 * that might be already pending. If a bogus SIGPIPE is sent to
524 * the entire process concurrently by a malicious user, it may
525 * be simply discarded.
526 */
527 ret = sigemptyset(&pending_set);
528 assert(!ret);
529 /*
530 * sigpending returns the mask of signals that are _both_
531 * blocked for the thread _and_ pending for either the thread or
532 * the entire process.
533 */
534 ret = sigpending(&pending_set);
535 assert(!ret);
536 sigpipe_was_pending = sigismember(&pending_set, SIGPIPE);
537 /*
538 * If sigpipe was pending, it means it was already blocked, so
539 * no need to block it.
540 */
541 if (!sigpipe_was_pending) {
542 ret = sigemptyset(&sigpipe_set);
543 assert(!ret);
544 ret = sigaddset(&sigpipe_set, SIGPIPE);
545 assert(!ret);
546 ret = pthread_sigmask(SIG_BLOCK, &sigpipe_set, &old_set);
547 assert(!ret);
548 }
549 do {
550 ret = write(wakeup_fd, "", 1);
551 } while (ret == -1L && errno == EINTR);
552 if (ret == -1L && errno == EPIPE && !sigpipe_was_pending) {
553 struct timespec timeout = { 0, 0 };
554 do {
555 ret = sigtimedwait(&sigpipe_set, NULL,
556 &timeout);
557 } while (ret == -1L && errno == EINTR);
558 }
559 if (!sigpipe_was_pending) {
560 ret = pthread_sigmask(SIG_SETMASK, &old_set, NULL);
561 assert(!ret);
562 }
563 }
564
565 static
566 void lib_ring_buffer_channel_do_read(struct channel *chan)
567 {
568 const struct lttng_ust_lib_ring_buffer_config *config;
569 struct lttng_ust_shm_handle *handle;
570 int cpu;
571
572 handle = chan->handle;
573 config = &chan->backend.config;
574
575 /*
576 * Only flush buffers periodically if readers are active.
577 */
578 pthread_mutex_lock(&wakeup_fd_mutex);
579 if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
580 for_each_possible_cpu(cpu) {
581 struct lttng_ust_lib_ring_buffer *buf =
582 shmp(handle, chan->backend.buf[cpu].shmp);
583
584 if (!buf)
585 goto end;
586 if (uatomic_read(&buf->active_readers)
587 && lib_ring_buffer_poll_deliver(config, buf,
588 chan, handle)) {
589 lib_ring_buffer_wakeup(buf, handle);
590 }
591 }
592 } else {
593 struct lttng_ust_lib_ring_buffer *buf =
594 shmp(handle, chan->backend.buf[0].shmp);
595
596 if (!buf)
597 goto end;
598 if (uatomic_read(&buf->active_readers)
599 && lib_ring_buffer_poll_deliver(config, buf,
600 chan, handle)) {
601 lib_ring_buffer_wakeup(buf, handle);
602 }
603 }
604 end:
605 pthread_mutex_unlock(&wakeup_fd_mutex);
606 }
607
608 static
609 void lib_ring_buffer_channel_read_timer(int sig, siginfo_t *si, void *uc)
610 {
611 struct channel *chan;
612
613 assert(CMM_LOAD_SHARED(timer_signal.tid) == pthread_self());
614 chan = si->si_value.sival_ptr;
615 DBG("Read timer for channel %p\n", chan);
616 lib_ring_buffer_channel_do_read(chan);
617 return;
618 }
619
620 static
621 void rb_setmask(sigset_t *mask)
622 {
623 int ret;
624
625 ret = sigemptyset(mask);
626 if (ret) {
627 PERROR("sigemptyset");
628 }
629 ret = sigaddset(mask, LTTNG_UST_RB_SIG_FLUSH);
630 if (ret) {
631 PERROR("sigaddset");
632 }
633 ret = sigaddset(mask, LTTNG_UST_RB_SIG_READ);
634 if (ret) {
635 PERROR("sigaddset");
636 }
637 ret = sigaddset(mask, LTTNG_UST_RB_SIG_TEARDOWN);
638 if (ret) {
639 PERROR("sigaddset");
640 }
641 }
642
643 static
644 void *sig_thread(void *arg)
645 {
646 sigset_t mask;
647 siginfo_t info;
648 int signr;
649
650 /* Only self thread will receive signal mask. */
651 rb_setmask(&mask);
652 CMM_STORE_SHARED(timer_signal.tid, pthread_self());
653
654 for (;;) {
655 signr = sigwaitinfo(&mask, &info);
656 if (signr == -1) {
657 if (errno != EINTR)
658 PERROR("sigwaitinfo");
659 continue;
660 }
661 if (signr == LTTNG_UST_RB_SIG_FLUSH) {
662 lib_ring_buffer_channel_switch_timer(info.si_signo,
663 &info, NULL);
664 } else if (signr == LTTNG_UST_RB_SIG_READ) {
665 lib_ring_buffer_channel_read_timer(info.si_signo,
666 &info, NULL);
667 } else if (signr == LTTNG_UST_RB_SIG_TEARDOWN) {
668 cmm_smp_mb();
669 CMM_STORE_SHARED(timer_signal.qs_done, 1);
670 cmm_smp_mb();
671 } else {
672 ERR("Unexptected signal %d\n", info.si_signo);
673 }
674 }
675 return NULL;
676 }
677
678 /*
679 * Ensure only a single thread listens on the timer signal.
680 */
681 static
682 void lib_ring_buffer_setup_timer_thread(void)
683 {
684 pthread_t thread;
685 int ret;
686
687 pthread_mutex_lock(&timer_signal.lock);
688 if (timer_signal.setup_done)
689 goto end;
690
691 ret = pthread_create(&thread, NULL, &sig_thread, NULL);
692 if (ret) {
693 errno = ret;
694 PERROR("pthread_create");
695 }
696 ret = pthread_detach(thread);
697 if (ret) {
698 errno = ret;
699 PERROR("pthread_detach");
700 }
701 timer_signal.setup_done = 1;
702 end:
703 pthread_mutex_unlock(&timer_signal.lock);
704 }
705
706 /*
707 * Wait for signal-handling thread quiescent state.
708 */
709 static
710 void lib_ring_buffer_wait_signal_thread_qs(unsigned int signr)
711 {
712 sigset_t pending_set;
713 int ret;
714
715 /*
716 * We need to be the only thread interacting with the thread
717 * that manages signals for teardown synchronization.
718 */
719 pthread_mutex_lock(&timer_signal.lock);
720
721 /*
722 * Ensure we don't have any signal queued for this channel.
723 */
724 for (;;) {
725 ret = sigemptyset(&pending_set);
726 if (ret == -1) {
727 PERROR("sigemptyset");
728 }
729 ret = sigpending(&pending_set);
730 if (ret == -1) {
731 PERROR("sigpending");
732 }
733 if (!sigismember(&pending_set, signr))
734 break;
735 caa_cpu_relax();
736 }
737
738 /*
739 * From this point, no new signal handler will be fired that
740 * would try to access "chan". However, we still need to wait
741 * for any currently executing handler to complete.
742 */
743 cmm_smp_mb();
744 CMM_STORE_SHARED(timer_signal.qs_done, 0);
745 cmm_smp_mb();
746
747 /*
748 * Kill with LTTNG_UST_RB_SIG_TEARDOWN, so signal management
749 * thread wakes up.
750 */
751 kill(getpid(), LTTNG_UST_RB_SIG_TEARDOWN);
752
753 while (!CMM_LOAD_SHARED(timer_signal.qs_done))
754 caa_cpu_relax();
755 cmm_smp_mb();
756
757 pthread_mutex_unlock(&timer_signal.lock);
758 }
759
760 static
761 void lib_ring_buffer_channel_switch_timer_start(struct channel *chan)
762 {
763 struct sigevent sev;
764 struct itimerspec its;
765 int ret;
766
767 if (!chan->switch_timer_interval || chan->switch_timer_enabled)
768 return;
769
770 chan->switch_timer_enabled = 1;
771
772 lib_ring_buffer_setup_timer_thread();
773
774 sev.sigev_notify = SIGEV_SIGNAL;
775 sev.sigev_signo = LTTNG_UST_RB_SIG_FLUSH;
776 sev.sigev_value.sival_ptr = chan;
777 ret = timer_create(CLOCKID, &sev, &chan->switch_timer);
778 if (ret == -1) {
779 PERROR("timer_create");
780 }
781
782 its.it_value.tv_sec = chan->switch_timer_interval / 1000000;
783 its.it_value.tv_nsec = (chan->switch_timer_interval % 1000000) * 1000;
784 its.it_interval.tv_sec = its.it_value.tv_sec;
785 its.it_interval.tv_nsec = its.it_value.tv_nsec;
786
787 ret = timer_settime(chan->switch_timer, 0, &its, NULL);
788 if (ret == -1) {
789 PERROR("timer_settime");
790 }
791 }
792
793 static
794 void lib_ring_buffer_channel_switch_timer_stop(struct channel *chan)
795 {
796 int ret;
797
798 if (!chan->switch_timer_interval || !chan->switch_timer_enabled)
799 return;
800
801 ret = timer_delete(chan->switch_timer);
802 if (ret == -1) {
803 PERROR("timer_delete");
804 }
805
806 lib_ring_buffer_wait_signal_thread_qs(LTTNG_UST_RB_SIG_FLUSH);
807
808 chan->switch_timer = 0;
809 chan->switch_timer_enabled = 0;
810 }
811
812 static
813 void lib_ring_buffer_channel_read_timer_start(struct channel *chan)
814 {
815 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
816 struct sigevent sev;
817 struct itimerspec its;
818 int ret;
819
820 if (config->wakeup != RING_BUFFER_WAKEUP_BY_TIMER
821 || !chan->read_timer_interval || chan->read_timer_enabled)
822 return;
823
824 chan->read_timer_enabled = 1;
825
826 lib_ring_buffer_setup_timer_thread();
827
828 sev.sigev_notify = SIGEV_SIGNAL;
829 sev.sigev_signo = LTTNG_UST_RB_SIG_READ;
830 sev.sigev_value.sival_ptr = chan;
831 ret = timer_create(CLOCKID, &sev, &chan->read_timer);
832 if (ret == -1) {
833 PERROR("timer_create");
834 }
835
836 its.it_value.tv_sec = chan->read_timer_interval / 1000000;
837 its.it_value.tv_nsec = (chan->read_timer_interval % 1000000) * 1000;
838 its.it_interval.tv_sec = its.it_value.tv_sec;
839 its.it_interval.tv_nsec = its.it_value.tv_nsec;
840
841 ret = timer_settime(chan->read_timer, 0, &its, NULL);
842 if (ret == -1) {
843 PERROR("timer_settime");
844 }
845 }
846
847 static
848 void lib_ring_buffer_channel_read_timer_stop(struct channel *chan)
849 {
850 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
851 int ret;
852
853 if (config->wakeup != RING_BUFFER_WAKEUP_BY_TIMER
854 || !chan->read_timer_interval || !chan->read_timer_enabled)
855 return;
856
857 ret = timer_delete(chan->read_timer);
858 if (ret == -1) {
859 PERROR("timer_delete");
860 }
861
862 /*
863 * do one more check to catch data that has been written in the last
864 * timer period.
865 */
866 lib_ring_buffer_channel_do_read(chan);
867
868 lib_ring_buffer_wait_signal_thread_qs(LTTNG_UST_RB_SIG_READ);
869
870 chan->read_timer = 0;
871 chan->read_timer_enabled = 0;
872 }
873
874 static void channel_unregister_notifiers(struct channel *chan,
875 struct lttng_ust_shm_handle *handle)
876 {
877 lib_ring_buffer_channel_switch_timer_stop(chan);
878 lib_ring_buffer_channel_read_timer_stop(chan);
879 }
880
881 static void channel_print_errors(struct channel *chan,
882 struct lttng_ust_shm_handle *handle)
883 {
884 const struct lttng_ust_lib_ring_buffer_config *config =
885 &chan->backend.config;
886 int cpu;
887
888 if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
889 for_each_possible_cpu(cpu) {
890 struct lttng_ust_lib_ring_buffer *buf =
891 shmp(handle, chan->backend.buf[cpu].shmp);
892 if (buf)
893 lib_ring_buffer_print_errors(chan, buf, cpu, handle);
894 }
895 } else {
896 struct lttng_ust_lib_ring_buffer *buf =
897 shmp(handle, chan->backend.buf[0].shmp);
898
899 if (buf)
900 lib_ring_buffer_print_errors(chan, buf, -1, handle);
901 }
902 }
903
904 static void channel_free(struct channel *chan,
905 struct lttng_ust_shm_handle *handle,
906 int consumer)
907 {
908 channel_backend_free(&chan->backend, handle);
909 /* chan is freed by shm teardown */
910 shm_object_table_destroy(handle->table, consumer);
911 free(handle);
912 }
913
914 /**
915 * channel_create - Create channel.
916 * @config: ring buffer instance configuration
917 * @name: name of the channel
918 * @priv_data: ring buffer client private data area pointer (output)
919 * @priv_data_size: length, in bytes, of the private data area.
920 * @priv_data_init: initialization data for private data.
921 * @buf_addr: pointer the the beginning of the preallocated buffer contiguous
922 * address mapping. It is used only by RING_BUFFER_STATIC
923 * configuration. It can be set to NULL for other backends.
924 * @subbuf_size: subbuffer size
925 * @num_subbuf: number of subbuffers
926 * @switch_timer_interval: Time interval (in us) to fill sub-buffers with
927 * padding to let readers get those sub-buffers.
928 * Used for live streaming.
929 * @read_timer_interval: Time interval (in us) to wake up pending readers.
930 * @stream_fds: array of stream file descriptors.
931 * @nr_stream_fds: number of file descriptors in array.
932 *
933 * Holds cpu hotplug.
934 * Returns NULL on failure.
935 */
936 struct lttng_ust_shm_handle *channel_create(const struct lttng_ust_lib_ring_buffer_config *config,
937 const char *name,
938 void **priv_data,
939 size_t priv_data_align,
940 size_t priv_data_size,
941 void *priv_data_init,
942 void *buf_addr, size_t subbuf_size,
943 size_t num_subbuf, unsigned int switch_timer_interval,
944 unsigned int read_timer_interval,
945 const int *stream_fds, int nr_stream_fds)
946 {
947 int ret;
948 size_t shmsize, chansize;
949 struct channel *chan;
950 struct lttng_ust_shm_handle *handle;
951 struct shm_object *shmobj;
952 unsigned int nr_streams;
953
954 if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
955 nr_streams = num_possible_cpus();
956 else
957 nr_streams = 1;
958
959 if (nr_stream_fds != nr_streams)
960 return NULL;
961
962 if (lib_ring_buffer_check_config(config, switch_timer_interval,
963 read_timer_interval))
964 return NULL;
965
966 handle = zmalloc(sizeof(struct lttng_ust_shm_handle));
967 if (!handle)
968 return NULL;
969
970 /* Allocate table for channel + per-cpu buffers */
971 handle->table = shm_object_table_create(1 + num_possible_cpus());
972 if (!handle->table)
973 goto error_table_alloc;
974
975 /* Calculate the shm allocation layout */
976 shmsize = sizeof(struct channel);
977 shmsize += offset_align(shmsize, __alignof__(struct lttng_ust_lib_ring_buffer_shmp));
978 shmsize += sizeof(struct lttng_ust_lib_ring_buffer_shmp) * nr_streams;
979 chansize = shmsize;
980 if (priv_data_align)
981 shmsize += offset_align(shmsize, priv_data_align);
982 shmsize += priv_data_size;
983
984 /* Allocate normal memory for channel (not shared) */
985 shmobj = shm_object_table_alloc(handle->table, shmsize, SHM_OBJECT_MEM,
986 -1);
987 if (!shmobj)
988 goto error_append;
989 /* struct channel is at object 0, offset 0 (hardcoded) */
990 set_shmp(handle->chan, zalloc_shm(shmobj, chansize));
991 assert(handle->chan._ref.index == 0);
992 assert(handle->chan._ref.offset == 0);
993 chan = shmp(handle, handle->chan);
994 if (!chan)
995 goto error_append;
996 chan->nr_streams = nr_streams;
997
998 /* space for private data */
999 if (priv_data_size) {
1000 DECLARE_SHMP(void, priv_data_alloc);
1001
1002 align_shm(shmobj, priv_data_align);
1003 chan->priv_data_offset = shmobj->allocated_len;
1004 set_shmp(priv_data_alloc, zalloc_shm(shmobj, priv_data_size));
1005 if (!shmp(handle, priv_data_alloc))
1006 goto error_append;
1007 *priv_data = channel_get_private(chan);
1008 memcpy(*priv_data, priv_data_init, priv_data_size);
1009 } else {
1010 chan->priv_data_offset = -1;
1011 if (priv_data)
1012 *priv_data = NULL;
1013 }
1014
1015 ret = channel_backend_init(&chan->backend, name, config,
1016 subbuf_size, num_subbuf, handle,
1017 stream_fds);
1018 if (ret)
1019 goto error_backend_init;
1020
1021 chan->handle = handle;
1022 chan->commit_count_mask = (~0UL >> chan->backend.num_subbuf_order);
1023
1024 chan->switch_timer_interval = switch_timer_interval;
1025 chan->read_timer_interval = read_timer_interval;
1026 lib_ring_buffer_channel_switch_timer_start(chan);
1027 lib_ring_buffer_channel_read_timer_start(chan);
1028
1029 return handle;
1030
1031 error_backend_init:
1032 error_append:
1033 shm_object_table_destroy(handle->table, 1);
1034 error_table_alloc:
1035 free(handle);
1036 return NULL;
1037 }
1038
1039 struct lttng_ust_shm_handle *channel_handle_create(void *data,
1040 uint64_t memory_map_size,
1041 int wakeup_fd)
1042 {
1043 struct lttng_ust_shm_handle *handle;
1044 struct shm_object *object;
1045
1046 handle = zmalloc(sizeof(struct lttng_ust_shm_handle));
1047 if (!handle)
1048 return NULL;
1049
1050 /* Allocate table for channel + per-cpu buffers */
1051 handle->table = shm_object_table_create(1 + num_possible_cpus());
1052 if (!handle->table)
1053 goto error_table_alloc;
1054 /* Add channel object */
1055 object = shm_object_table_append_mem(handle->table, data,
1056 memory_map_size, wakeup_fd);
1057 if (!object)
1058 goto error_table_object;
1059 /* struct channel is at object 0, offset 0 (hardcoded) */
1060 handle->chan._ref.index = 0;
1061 handle->chan._ref.offset = 0;
1062 return handle;
1063
1064 error_table_object:
1065 shm_object_table_destroy(handle->table, 0);
1066 error_table_alloc:
1067 free(handle);
1068 return NULL;
1069 }
1070
1071 int channel_handle_add_stream(struct lttng_ust_shm_handle *handle,
1072 int shm_fd, int wakeup_fd, uint32_t stream_nr,
1073 uint64_t memory_map_size)
1074 {
1075 struct shm_object *object;
1076
1077 /* Add stream object */
1078 object = shm_object_table_append_shm(handle->table,
1079 shm_fd, wakeup_fd, stream_nr,
1080 memory_map_size);
1081 if (!object)
1082 return -EINVAL;
1083 return 0;
1084 }
1085
1086 unsigned int channel_handle_get_nr_streams(struct lttng_ust_shm_handle *handle)
1087 {
1088 assert(handle->table);
1089 return handle->table->allocated_len - 1;
1090 }
1091
1092 static
1093 void channel_release(struct channel *chan, struct lttng_ust_shm_handle *handle,
1094 int consumer)
1095 {
1096 channel_free(chan, handle, consumer);
1097 }
1098
1099 /**
1100 * channel_destroy - Finalize, wait for q.s. and destroy channel.
1101 * @chan: channel to destroy
1102 *
1103 * Holds cpu hotplug.
1104 * Call "destroy" callback, finalize channels, decrement the channel
1105 * reference count. Note that when readers have completed data
1106 * consumption of finalized channels, get_subbuf() will return -ENODATA.
1107 * They should release their handle at that point.
1108 */
1109 void channel_destroy(struct channel *chan, struct lttng_ust_shm_handle *handle,
1110 int consumer)
1111 {
1112 if (consumer) {
1113 /*
1114 * Note: the consumer takes care of finalizing and
1115 * switching the buffers.
1116 */
1117 channel_unregister_notifiers(chan, handle);
1118 /*
1119 * The consumer prints errors.
1120 */
1121 channel_print_errors(chan, handle);
1122 }
1123
1124 /*
1125 * sessiond/consumer are keeping a reference on the shm file
1126 * descriptor directly. No need to refcount.
1127 */
1128 channel_release(chan, handle, consumer);
1129 return;
1130 }
1131
1132 struct lttng_ust_lib_ring_buffer *channel_get_ring_buffer(
1133 const struct lttng_ust_lib_ring_buffer_config *config,
1134 struct channel *chan, int cpu,
1135 struct lttng_ust_shm_handle *handle,
1136 int *shm_fd, int *wait_fd,
1137 int *wakeup_fd,
1138 uint64_t *memory_map_size)
1139 {
1140 struct shm_ref *ref;
1141
1142 if (config->alloc == RING_BUFFER_ALLOC_GLOBAL) {
1143 cpu = 0;
1144 } else {
1145 if (cpu >= num_possible_cpus())
1146 return NULL;
1147 }
1148 ref = &chan->backend.buf[cpu].shmp._ref;
1149 *shm_fd = shm_get_shm_fd(handle, ref);
1150 *wait_fd = shm_get_wait_fd(handle, ref);
1151 *wakeup_fd = shm_get_wakeup_fd(handle, ref);
1152 if (shm_get_shm_size(handle, ref, memory_map_size))
1153 return NULL;
1154 return shmp(handle, chan->backend.buf[cpu].shmp);
1155 }
1156
1157 int ring_buffer_channel_close_wait_fd(const struct lttng_ust_lib_ring_buffer_config *config,
1158 struct channel *chan,
1159 struct lttng_ust_shm_handle *handle)
1160 {
1161 struct shm_ref *ref;
1162
1163 ref = &handle->chan._ref;
1164 return shm_close_wait_fd(handle, ref);
1165 }
1166
1167 int ring_buffer_channel_close_wakeup_fd(const struct lttng_ust_lib_ring_buffer_config *config,
1168 struct channel *chan,
1169 struct lttng_ust_shm_handle *handle)
1170 {
1171 struct shm_ref *ref;
1172
1173 ref = &handle->chan._ref;
1174 return shm_close_wakeup_fd(handle, ref);
1175 }
1176
1177 int ring_buffer_stream_close_wait_fd(const struct lttng_ust_lib_ring_buffer_config *config,
1178 struct channel *chan,
1179 struct lttng_ust_shm_handle *handle,
1180 int cpu)
1181 {
1182 struct shm_ref *ref;
1183
1184 if (config->alloc == RING_BUFFER_ALLOC_GLOBAL) {
1185 cpu = 0;
1186 } else {
1187 if (cpu >= num_possible_cpus())
1188 return -EINVAL;
1189 }
1190 ref = &chan->backend.buf[cpu].shmp._ref;
1191 return shm_close_wait_fd(handle, ref);
1192 }
1193
1194 int ring_buffer_stream_close_wakeup_fd(const struct lttng_ust_lib_ring_buffer_config *config,
1195 struct channel *chan,
1196 struct lttng_ust_shm_handle *handle,
1197 int cpu)
1198 {
1199 struct shm_ref *ref;
1200 int ret;
1201
1202 if (config->alloc == RING_BUFFER_ALLOC_GLOBAL) {
1203 cpu = 0;
1204 } else {
1205 if (cpu >= num_possible_cpus())
1206 return -EINVAL;
1207 }
1208 ref = &chan->backend.buf[cpu].shmp._ref;
1209 pthread_mutex_lock(&wakeup_fd_mutex);
1210 ret = shm_close_wakeup_fd(handle, ref);
1211 pthread_mutex_unlock(&wakeup_fd_mutex);
1212 return ret;
1213 }
1214
1215 int lib_ring_buffer_open_read(struct lttng_ust_lib_ring_buffer *buf,
1216 struct lttng_ust_shm_handle *handle)
1217 {
1218 if (uatomic_cmpxchg(&buf->active_readers, 0, 1) != 0)
1219 return -EBUSY;
1220 cmm_smp_mb();
1221 return 0;
1222 }
1223
1224 void lib_ring_buffer_release_read(struct lttng_ust_lib_ring_buffer *buf,
1225 struct lttng_ust_shm_handle *handle)
1226 {
1227 struct channel *chan = shmp(handle, buf->backend.chan);
1228
1229 if (!chan)
1230 return;
1231 CHAN_WARN_ON(chan, uatomic_read(&buf->active_readers) != 1);
1232 cmm_smp_mb();
1233 uatomic_dec(&buf->active_readers);
1234 }
1235
1236 /**
1237 * lib_ring_buffer_snapshot - save subbuffer position snapshot (for read)
1238 * @buf: ring buffer
1239 * @consumed: consumed count indicating the position where to read
1240 * @produced: produced count, indicates position when to stop reading
1241 *
1242 * Returns -ENODATA if buffer is finalized, -EAGAIN if there is currently no
1243 * data to read at consumed position, or 0 if the get operation succeeds.
1244 */
1245
1246 int lib_ring_buffer_snapshot(struct lttng_ust_lib_ring_buffer *buf,
1247 unsigned long *consumed, unsigned long *produced,
1248 struct lttng_ust_shm_handle *handle)
1249 {
1250 struct channel *chan;
1251 const struct lttng_ust_lib_ring_buffer_config *config;
1252 unsigned long consumed_cur, write_offset;
1253 int finalized;
1254
1255 chan = shmp(handle, buf->backend.chan);
1256 if (!chan)
1257 return -EPERM;
1258 config = &chan->backend.config;
1259 finalized = CMM_ACCESS_ONCE(buf->finalized);
1260 /*
1261 * Read finalized before counters.
1262 */
1263 cmm_smp_rmb();
1264 consumed_cur = uatomic_read(&buf->consumed);
1265 /*
1266 * No need to issue a memory barrier between consumed count read and
1267 * write offset read, because consumed count can only change
1268 * concurrently in overwrite mode, and we keep a sequence counter
1269 * identifier derived from the write offset to check we are getting
1270 * the same sub-buffer we are expecting (the sub-buffers are atomically
1271 * "tagged" upon writes, tags are checked upon read).
1272 */
1273 write_offset = v_read(config, &buf->offset);
1274
1275 /*
1276 * Check that we are not about to read the same subbuffer in
1277 * which the writer head is.
1278 */
1279 if (subbuf_trunc(write_offset, chan) - subbuf_trunc(consumed_cur, chan)
1280 == 0)
1281 goto nodata;
1282
1283 *consumed = consumed_cur;
1284 *produced = subbuf_trunc(write_offset, chan);
1285
1286 return 0;
1287
1288 nodata:
1289 /*
1290 * The memory barriers __wait_event()/wake_up_interruptible() take care
1291 * of "raw_spin_is_locked" memory ordering.
1292 */
1293 if (finalized)
1294 return -ENODATA;
1295 else
1296 return -EAGAIN;
1297 }
1298
1299 /**
1300 * lib_ring_buffer_move_consumer - move consumed counter forward
1301 * @buf: ring buffer
1302 * @consumed_new: new consumed count value
1303 */
1304 void lib_ring_buffer_move_consumer(struct lttng_ust_lib_ring_buffer *buf,
1305 unsigned long consumed_new,
1306 struct lttng_ust_shm_handle *handle)
1307 {
1308 struct lttng_ust_lib_ring_buffer_backend *bufb = &buf->backend;
1309 struct channel *chan;
1310 unsigned long consumed;
1311
1312 chan = shmp(handle, bufb->chan);
1313 if (!chan)
1314 return;
1315 CHAN_WARN_ON(chan, uatomic_read(&buf->active_readers) != 1);
1316
1317 /*
1318 * Only push the consumed value forward.
1319 * If the consumed cmpxchg fails, this is because we have been pushed by
1320 * the writer in flight recorder mode.
1321 */
1322 consumed = uatomic_read(&buf->consumed);
1323 while ((long) consumed - (long) consumed_new < 0)
1324 consumed = uatomic_cmpxchg(&buf->consumed, consumed,
1325 consumed_new);
1326 }
1327
1328 /**
1329 * lib_ring_buffer_get_subbuf - get exclusive access to subbuffer for reading
1330 * @buf: ring buffer
1331 * @consumed: consumed count indicating the position where to read
1332 *
1333 * Returns -ENODATA if buffer is finalized, -EAGAIN if there is currently no
1334 * data to read at consumed position, or 0 if the get operation succeeds.
1335 */
1336 int lib_ring_buffer_get_subbuf(struct lttng_ust_lib_ring_buffer *buf,
1337 unsigned long consumed,
1338 struct lttng_ust_shm_handle *handle)
1339 {
1340 struct channel *chan;
1341 const struct lttng_ust_lib_ring_buffer_config *config;
1342 unsigned long consumed_cur, consumed_idx, commit_count, write_offset;
1343 int ret, finalized, nr_retry = LTTNG_UST_RING_BUFFER_GET_RETRY;
1344 struct commit_counters_cold *cc_cold;
1345
1346 chan = shmp(handle, buf->backend.chan);
1347 if (!chan)
1348 return -EPERM;
1349 config = &chan->backend.config;
1350 retry:
1351 finalized = CMM_ACCESS_ONCE(buf->finalized);
1352 /*
1353 * Read finalized before counters.
1354 */
1355 cmm_smp_rmb();
1356 consumed_cur = uatomic_read(&buf->consumed);
1357 consumed_idx = subbuf_index(consumed, chan);
1358 cc_cold = shmp_index(handle, buf->commit_cold, consumed_idx);
1359 if (!cc_cold)
1360 return -EPERM;
1361 commit_count = v_read(config, &cc_cold->cc_sb);
1362 /*
1363 * Make sure we read the commit count before reading the buffer
1364 * data and the write offset. Correct consumed offset ordering
1365 * wrt commit count is insured by the use of cmpxchg to update
1366 * the consumed offset.
1367 */
1368 /*
1369 * Local rmb to match the remote wmb to read the commit count
1370 * before the buffer data and the write offset.
1371 */
1372 cmm_smp_rmb();
1373
1374 write_offset = v_read(config, &buf->offset);
1375
1376 /*
1377 * Check that the buffer we are getting is after or at consumed_cur
1378 * position.
1379 */
1380 if ((long) subbuf_trunc(consumed, chan)
1381 - (long) subbuf_trunc(consumed_cur, chan) < 0)
1382 goto nodata;
1383
1384 /*
1385 * Check that the subbuffer we are trying to consume has been
1386 * already fully committed. There are a few causes that can make
1387 * this unavailability situation occur:
1388 *
1389 * Temporary (short-term) situation:
1390 * - Application is running on a different CPU, between reserve
1391 * and commit ring buffer operations,
1392 * - Application is preempted between reserve and commit ring
1393 * buffer operations,
1394 *
1395 * Long-term situation:
1396 * - Application is stopped (SIGSTOP) between reserve and commit
1397 * ring buffer operations. Could eventually be resumed by
1398 * SIGCONT.
1399 * - Application is killed (SIGTERM, SIGINT, SIGKILL) between
1400 * reserve and commit ring buffer operation.
1401 *
1402 * From a consumer perspective, handling short-term
1403 * unavailability situations is performed by retrying a few
1404 * times after a delay. Handling long-term unavailability
1405 * situations is handled by failing to get the sub-buffer.
1406 *
1407 * In all of those situations, if the application is taking a
1408 * long time to perform its commit after ring buffer space
1409 * reservation, we can end up in a situation where the producer
1410 * will fill the ring buffer and try to write into the same
1411 * sub-buffer again (which has a missing commit). This is
1412 * handled by the producer in the sub-buffer switch handling
1413 * code of the reserve routine by detecting unbalanced
1414 * reserve/commit counters and discarding all further events
1415 * until the situation is resolved in those situations. Two
1416 * scenarios can occur:
1417 *
1418 * 1) The application causing the reserve/commit counters to be
1419 * unbalanced has been terminated. In this situation, all
1420 * further events will be discarded in the buffers, and no
1421 * further buffer data will be readable by the consumer
1422 * daemon. Tearing down the UST tracing session and starting
1423 * anew is a work-around for those situations. Note that this
1424 * only affects per-UID tracing. In per-PID tracing, the
1425 * application vanishes with the termination, and therefore
1426 * no more data needs to be written to the buffers.
1427 * 2) The application causing the unbalance has been delayed for
1428 * a long time, but will eventually try to increment the
1429 * commit counter after eventually writing to the sub-buffer.
1430 * This situation can cause events to be discarded until the
1431 * application resumes its operations.
1432 */
1433 if (((commit_count - chan->backend.subbuf_size)
1434 & chan->commit_count_mask)
1435 - (buf_trunc(consumed, chan)
1436 >> chan->backend.num_subbuf_order)
1437 != 0) {
1438 if (nr_retry-- > 0) {
1439 if (nr_retry <= (LTTNG_UST_RING_BUFFER_GET_RETRY >> 1))
1440 (void) poll(NULL, 0, LTTNG_UST_RING_BUFFER_RETRY_DELAY_MS);
1441 goto retry;
1442 } else {
1443 goto nodata;
1444 }
1445 }
1446
1447 /*
1448 * Check that we are not about to read the same subbuffer in
1449 * which the writer head is.
1450 */
1451 if (subbuf_trunc(write_offset, chan) - subbuf_trunc(consumed, chan)
1452 == 0)
1453 goto nodata;
1454
1455 /*
1456 * Failure to get the subbuffer causes a busy-loop retry without going
1457 * to a wait queue. These are caused by short-lived race windows where
1458 * the writer is getting access to a subbuffer we were trying to get
1459 * access to. Also checks that the "consumed" buffer count we are
1460 * looking for matches the one contained in the subbuffer id.
1461 *
1462 * The short-lived race window described here can be affected by
1463 * application signals and preemption, thus requiring to bound
1464 * the loop to a maximum number of retry.
1465 */
1466 ret = update_read_sb_index(config, &buf->backend, &chan->backend,
1467 consumed_idx, buf_trunc_val(consumed, chan),
1468 handle);
1469 if (ret) {
1470 if (nr_retry-- > 0) {
1471 if (nr_retry <= (LTTNG_UST_RING_BUFFER_GET_RETRY >> 1))
1472 (void) poll(NULL, 0, LTTNG_UST_RING_BUFFER_RETRY_DELAY_MS);
1473 goto retry;
1474 } else {
1475 goto nodata;
1476 }
1477 }
1478 subbuffer_id_clear_noref(config, &buf->backend.buf_rsb.id);
1479
1480 buf->get_subbuf_consumed = consumed;
1481 buf->get_subbuf = 1;
1482
1483 return 0;
1484
1485 nodata:
1486 /*
1487 * The memory barriers __wait_event()/wake_up_interruptible() take care
1488 * of "raw_spin_is_locked" memory ordering.
1489 */
1490 if (finalized)
1491 return -ENODATA;
1492 else
1493 return -EAGAIN;
1494 }
1495
1496 /**
1497 * lib_ring_buffer_put_subbuf - release exclusive subbuffer access
1498 * @buf: ring buffer
1499 */
1500 void lib_ring_buffer_put_subbuf(struct lttng_ust_lib_ring_buffer *buf,
1501 struct lttng_ust_shm_handle *handle)
1502 {
1503 struct lttng_ust_lib_ring_buffer_backend *bufb = &buf->backend;
1504 struct channel *chan;
1505 const struct lttng_ust_lib_ring_buffer_config *config;
1506 unsigned long sb_bindex, consumed_idx, consumed;
1507 struct lttng_ust_lib_ring_buffer_backend_pages_shmp *rpages;
1508 struct lttng_ust_lib_ring_buffer_backend_pages *backend_pages;
1509
1510 chan = shmp(handle, bufb->chan);
1511 if (!chan)
1512 return;
1513 config = &chan->backend.config;
1514 CHAN_WARN_ON(chan, uatomic_read(&buf->active_readers) != 1);
1515
1516 if (!buf->get_subbuf) {
1517 /*
1518 * Reader puts a subbuffer it did not get.
1519 */
1520 CHAN_WARN_ON(chan, 1);
1521 return;
1522 }
1523 consumed = buf->get_subbuf_consumed;
1524 buf->get_subbuf = 0;
1525
1526 /*
1527 * Clear the records_unread counter. (overruns counter)
1528 * Can still be non-zero if a file reader simply grabbed the data
1529 * without using iterators.
1530 * Can be below zero if an iterator is used on a snapshot more than
1531 * once.
1532 */
1533 sb_bindex = subbuffer_id_get_index(config, bufb->buf_rsb.id);
1534 rpages = shmp_index(handle, bufb->array, sb_bindex);
1535 if (!rpages)
1536 return;
1537 backend_pages = shmp(handle, rpages->shmp);
1538 if (!backend_pages)
1539 return;
1540 v_add(config, v_read(config, &backend_pages->records_unread),
1541 &bufb->records_read);
1542 v_set(config, &backend_pages->records_unread, 0);
1543 CHAN_WARN_ON(chan, config->mode == RING_BUFFER_OVERWRITE
1544 && subbuffer_id_is_noref(config, bufb->buf_rsb.id));
1545 subbuffer_id_set_noref(config, &bufb->buf_rsb.id);
1546
1547 /*
1548 * Exchange the reader subbuffer with the one we put in its place in the
1549 * writer subbuffer table. Expect the original consumed count. If
1550 * update_read_sb_index fails, this is because the writer updated the
1551 * subbuffer concurrently. We should therefore keep the subbuffer we
1552 * currently have: it has become invalid to try reading this sub-buffer
1553 * consumed count value anyway.
1554 */
1555 consumed_idx = subbuf_index(consumed, chan);
1556 update_read_sb_index(config, &buf->backend, &chan->backend,
1557 consumed_idx, buf_trunc_val(consumed, chan),
1558 handle);
1559 /*
1560 * update_read_sb_index return value ignored. Don't exchange sub-buffer
1561 * if the writer concurrently updated it.
1562 */
1563 }
1564
1565 /*
1566 * cons_offset is an iterator on all subbuffer offsets between the reader
1567 * position and the writer position. (inclusive)
1568 */
1569 static
1570 void lib_ring_buffer_print_subbuffer_errors(struct lttng_ust_lib_ring_buffer *buf,
1571 struct channel *chan,
1572 unsigned long cons_offset,
1573 int cpu,
1574 struct lttng_ust_shm_handle *handle)
1575 {
1576 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
1577 unsigned long cons_idx, commit_count, commit_count_sb;
1578 struct commit_counters_hot *cc_hot;
1579 struct commit_counters_cold *cc_cold;
1580
1581 cons_idx = subbuf_index(cons_offset, chan);
1582 cc_hot = shmp_index(handle, buf->commit_hot, cons_idx);
1583 if (!cc_hot)
1584 return;
1585 cc_cold = shmp_index(handle, buf->commit_cold, cons_idx);
1586 if (!cc_cold)
1587 return;
1588 commit_count = v_read(config, &cc_hot->cc);
1589 commit_count += cc_hot->cc_rseq;
1590 commit_count_sb = v_read(config, &cc_cold->cc_sb);
1591
1592 if (subbuf_offset(commit_count, chan) != 0)
1593 DBG("ring buffer %s, cpu %d: "
1594 "commit count in subbuffer %lu,\n"
1595 "expecting multiples of %lu bytes\n"
1596 " [ %lu bytes committed, %lu bytes reader-visible ]\n",
1597 chan->backend.name, cpu, cons_idx,
1598 chan->backend.subbuf_size,
1599 commit_count, commit_count_sb);
1600
1601 DBG("ring buffer: %s, cpu %d: %lu bytes committed\n",
1602 chan->backend.name, cpu, commit_count);
1603 }
1604
1605 static
1606 void lib_ring_buffer_print_buffer_errors(struct lttng_ust_lib_ring_buffer *buf,
1607 struct channel *chan,
1608 void *priv, int cpu,
1609 struct lttng_ust_shm_handle *handle)
1610 {
1611 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
1612 unsigned long write_offset, cons_offset;
1613
1614 /*
1615 * No need to order commit_count, write_offset and cons_offset reads
1616 * because we execute at teardown when no more writer nor reader
1617 * references are left.
1618 */
1619 write_offset = v_read(config, &buf->offset);
1620 cons_offset = uatomic_read(&buf->consumed);
1621 if (write_offset != cons_offset)
1622 DBG("ring buffer %s, cpu %d: "
1623 "non-consumed data\n"
1624 " [ %lu bytes written, %lu bytes read ]\n",
1625 chan->backend.name, cpu, write_offset, cons_offset);
1626
1627 for (cons_offset = uatomic_read(&buf->consumed);
1628 (long) (subbuf_trunc((unsigned long) v_read(config, &buf->offset),
1629 chan)
1630 - cons_offset) > 0;
1631 cons_offset = subbuf_align(cons_offset, chan))
1632 lib_ring_buffer_print_subbuffer_errors(buf, chan, cons_offset,
1633 cpu, handle);
1634 }
1635
1636 static
1637 void lib_ring_buffer_print_errors(struct channel *chan,
1638 struct lttng_ust_lib_ring_buffer *buf, int cpu,
1639 struct lttng_ust_shm_handle *handle)
1640 {
1641 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
1642 void *priv = channel_get_private(chan);
1643
1644 if (!strcmp(chan->backend.name, "relay-metadata-mmap")) {
1645 DBG("ring buffer %s: %lu records written, "
1646 "%lu records overrun\n",
1647 chan->backend.name,
1648 v_read(config, &buf->records_count),
1649 v_read(config, &buf->records_overrun));
1650 } else {
1651 DBG("ring buffer %s, cpu %d: %lu records written, "
1652 "%lu records overrun\n",
1653 chan->backend.name, cpu,
1654 v_read(config, &buf->records_count),
1655 v_read(config, &buf->records_overrun));
1656
1657 if (v_read(config, &buf->records_lost_full)
1658 || v_read(config, &buf->records_lost_wrap)
1659 || v_read(config, &buf->records_lost_big))
1660 DBG("ring buffer %s, cpu %d: records were lost. Caused by:\n"
1661 " [ %lu buffer full, %lu nest buffer wrap-around, "
1662 "%lu event too big ]\n",
1663 chan->backend.name, cpu,
1664 v_read(config, &buf->records_lost_full),
1665 v_read(config, &buf->records_lost_wrap),
1666 v_read(config, &buf->records_lost_big));
1667 }
1668 lib_ring_buffer_print_buffer_errors(buf, chan, priv, cpu, handle);
1669 }
1670
1671 /*
1672 * lib_ring_buffer_switch_old_start: Populate old subbuffer header.
1673 *
1674 * Only executed by SWITCH_FLUSH, which can be issued while tracing is
1675 * active or at buffer finalization (destroy).
1676 */
1677 static
1678 void lib_ring_buffer_switch_old_start(struct lttng_ust_lib_ring_buffer *buf,
1679 struct channel *chan,
1680 struct switch_offsets *offsets,
1681 uint64_t tsc,
1682 struct lttng_ust_shm_handle *handle)
1683 {
1684 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
1685 unsigned long oldidx = subbuf_index(offsets->old, chan);
1686 unsigned long commit_count;
1687 struct commit_counters_hot *cc_hot;
1688
1689 config->cb.buffer_begin(buf, tsc, oldidx, handle);
1690
1691 /*
1692 * Order all writes to buffer before the commit count update that will
1693 * determine that the subbuffer is full.
1694 */
1695 cmm_smp_wmb();
1696 cc_hot = shmp_index(handle, buf->commit_hot, oldidx);
1697 if (!cc_hot)
1698 return;
1699 v_add(config, config->cb.subbuffer_header_size(),
1700 &cc_hot->cc);
1701 commit_count = v_read(config, &cc_hot->cc);
1702 commit_count += cc_hot->cc_rseq;
1703 /* Check if the written buffer has to be delivered */
1704 lib_ring_buffer_check_deliver(config, buf, chan, offsets->old,
1705 commit_count, oldidx, handle, tsc);
1706 lib_ring_buffer_write_commit_counter(config, buf, chan,
1707 offsets->old + config->cb.subbuffer_header_size(),
1708 commit_count, handle, cc_hot);
1709 }
1710
1711 /*
1712 * lib_ring_buffer_switch_old_end: switch old subbuffer
1713 *
1714 * Note : offset_old should never be 0 here. It is ok, because we never perform
1715 * buffer switch on an empty subbuffer in SWITCH_ACTIVE mode. The caller
1716 * increments the offset_old value when doing a SWITCH_FLUSH on an empty
1717 * subbuffer.
1718 */
1719 static
1720 void lib_ring_buffer_switch_old_end(struct lttng_ust_lib_ring_buffer *buf,
1721 struct channel *chan,
1722 struct switch_offsets *offsets,
1723 uint64_t tsc,
1724 struct lttng_ust_shm_handle *handle)
1725 {
1726 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
1727 unsigned long oldidx = subbuf_index(offsets->old - 1, chan);
1728 unsigned long commit_count, padding_size, data_size;
1729 struct commit_counters_hot *cc_hot;
1730
1731 data_size = subbuf_offset(offsets->old - 1, chan) + 1;
1732 padding_size = chan->backend.subbuf_size - data_size;
1733 subbuffer_set_data_size(config, &buf->backend, oldidx, data_size,
1734 handle);
1735
1736 /*
1737 * Order all writes to buffer before the commit count update that will
1738 * determine that the subbuffer is full.
1739 */
1740 cmm_smp_wmb();
1741 cc_hot = shmp_index(handle, buf->commit_hot, oldidx);
1742 if (!cc_hot)
1743 return;
1744 v_add(config, padding_size, &cc_hot->cc);
1745 commit_count = v_read(config, &cc_hot->cc);
1746 commit_count += cc_hot->cc_rseq;
1747 lib_ring_buffer_check_deliver(config, buf, chan, offsets->old - 1,
1748 commit_count, oldidx, handle, tsc);
1749 lib_ring_buffer_write_commit_counter(config, buf, chan,
1750 offsets->old + padding_size, commit_count, handle,
1751 cc_hot);
1752 }
1753
1754 /*
1755 * lib_ring_buffer_switch_new_start: Populate new subbuffer.
1756 *
1757 * This code can be executed unordered : writers may already have written to the
1758 * sub-buffer before this code gets executed, caution. The commit makes sure
1759 * that this code is executed before the deliver of this sub-buffer.
1760 */
1761 static
1762 void lib_ring_buffer_switch_new_start(struct lttng_ust_lib_ring_buffer *buf,
1763 struct channel *chan,
1764 struct switch_offsets *offsets,
1765 uint64_t tsc,
1766 struct lttng_ust_shm_handle *handle)
1767 {
1768 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
1769 unsigned long beginidx = subbuf_index(offsets->begin, chan);
1770 unsigned long commit_count;
1771 struct commit_counters_hot *cc_hot;
1772
1773 config->cb.buffer_begin(buf, tsc, beginidx, handle);
1774
1775 /*
1776 * Order all writes to buffer before the commit count update that will
1777 * determine that the subbuffer is full.
1778 */
1779 cmm_smp_wmb();
1780 cc_hot = shmp_index(handle, buf->commit_hot, beginidx);
1781 if (!cc_hot)
1782 return;
1783 v_add(config, config->cb.subbuffer_header_size(), &cc_hot->cc);
1784 commit_count = v_read(config, &cc_hot->cc);
1785 commit_count += cc_hot->cc_rseq;
1786 /* Check if the written buffer has to be delivered */
1787 lib_ring_buffer_check_deliver(config, buf, chan, offsets->begin,
1788 commit_count, beginidx, handle, tsc);
1789 lib_ring_buffer_write_commit_counter(config, buf, chan,
1790 offsets->begin + config->cb.subbuffer_header_size(),
1791 commit_count, handle, cc_hot);
1792 }
1793
1794 /*
1795 * lib_ring_buffer_switch_new_end: finish switching current subbuffer
1796 *
1797 * Calls subbuffer_set_data_size() to set the data size of the current
1798 * sub-buffer. We do not need to perform check_deliver nor commit here,
1799 * since this task will be done by the "commit" of the event for which
1800 * we are currently doing the space reservation.
1801 */
1802 static
1803 void lib_ring_buffer_switch_new_end(struct lttng_ust_lib_ring_buffer *buf,
1804 struct channel *chan,
1805 struct switch_offsets *offsets,
1806 uint64_t tsc,
1807 struct lttng_ust_shm_handle *handle)
1808 {
1809 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
1810 unsigned long endidx, data_size;
1811
1812 endidx = subbuf_index(offsets->end - 1, chan);
1813 data_size = subbuf_offset(offsets->end - 1, chan) + 1;
1814 subbuffer_set_data_size(config, &buf->backend, endidx, data_size,
1815 handle);
1816 }
1817
1818 /*
1819 * Returns :
1820 * 0 if ok
1821 * !0 if execution must be aborted.
1822 */
1823 static
1824 int lib_ring_buffer_try_switch_slow(enum switch_mode mode,
1825 struct lttng_ust_lib_ring_buffer *buf,
1826 struct channel *chan,
1827 struct switch_offsets *offsets,
1828 uint64_t *tsc,
1829 struct lttng_ust_shm_handle *handle)
1830 {
1831 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
1832 unsigned long off, reserve_commit_diff;
1833
1834 offsets->begin = v_read(config, &buf->offset);
1835 offsets->old = offsets->begin;
1836 offsets->switch_old_start = 0;
1837 off = subbuf_offset(offsets->begin, chan);
1838
1839 *tsc = config->cb.ring_buffer_clock_read(chan);
1840
1841 /*
1842 * Ensure we flush the header of an empty subbuffer when doing the
1843 * finalize (SWITCH_FLUSH). This ensures that we end up knowing the
1844 * total data gathering duration even if there were no records saved
1845 * after the last buffer switch.
1846 * In SWITCH_ACTIVE mode, switch the buffer when it contains events.
1847 * SWITCH_ACTIVE only flushes the current subbuffer, dealing with end of
1848 * subbuffer header as appropriate.
1849 * The next record that reserves space will be responsible for
1850 * populating the following subbuffer header. We choose not to populate
1851 * the next subbuffer header here because we want to be able to use
1852 * SWITCH_ACTIVE for periodical buffer flush, which must
1853 * guarantee that all the buffer content (records and header
1854 * timestamps) are visible to the reader. This is required for
1855 * quiescence guarantees for the fusion merge.
1856 */
1857 if (mode != SWITCH_FLUSH && !off)
1858 return -1; /* we do not have to switch : buffer is empty */
1859
1860 if (caa_unlikely(off == 0)) {
1861 unsigned long sb_index, commit_count;
1862 struct commit_counters_cold *cc_cold;
1863
1864 /*
1865 * We are performing a SWITCH_FLUSH. There may be concurrent
1866 * writes into the buffer if e.g. invoked while performing a
1867 * snapshot on an active trace.
1868 *
1869 * If the client does not save any header information
1870 * (sub-buffer header size == 0), don't switch empty subbuffer
1871 * on finalize, because it is invalid to deliver a completely
1872 * empty subbuffer.
1873 */
1874 if (!config->cb.subbuffer_header_size())
1875 return -1;
1876
1877 /* Test new buffer integrity */
1878 sb_index = subbuf_index(offsets->begin, chan);
1879 cc_cold = shmp_index(handle, buf->commit_cold, sb_index);
1880 if (!cc_cold)
1881 return -1;
1882 commit_count = v_read(config, &cc_cold->cc_sb);
1883 reserve_commit_diff =
1884 (buf_trunc(offsets->begin, chan)
1885 >> chan->backend.num_subbuf_order)
1886 - (commit_count & chan->commit_count_mask);
1887 if (caa_likely(reserve_commit_diff == 0)) {
1888 /* Next subbuffer not being written to. */
1889 if (caa_unlikely(config->mode != RING_BUFFER_OVERWRITE &&
1890 subbuf_trunc(offsets->begin, chan)
1891 - subbuf_trunc((unsigned long)
1892 uatomic_read(&buf->consumed), chan)
1893 >= chan->backend.buf_size)) {
1894 /*
1895 * We do not overwrite non consumed buffers
1896 * and we are full : don't switch.
1897 */
1898 return -1;
1899 } else {
1900 /*
1901 * Next subbuffer not being written to, and we
1902 * are either in overwrite mode or the buffer is
1903 * not full. It's safe to write in this new
1904 * subbuffer.
1905 */
1906 }
1907 } else {
1908 /*
1909 * Next subbuffer reserve offset does not match the
1910 * commit offset. Don't perform switch in
1911 * producer-consumer and overwrite mode. Caused by
1912 * either a writer OOPS or too many nested writes over a
1913 * reserve/commit pair.
1914 */
1915 return -1;
1916 }
1917
1918 /*
1919 * Need to write the subbuffer start header on finalize.
1920 */
1921 offsets->switch_old_start = 1;
1922 }
1923 offsets->begin = subbuf_align(offsets->begin, chan);
1924 /* Note: old points to the next subbuf at offset 0 */
1925 offsets->end = offsets->begin;
1926 return 0;
1927 }
1928
1929 /*
1930 * Force a sub-buffer switch. This operation is completely reentrant : can be
1931 * called while tracing is active with absolutely no lock held.
1932 *
1933 * Note, however, that as a v_cmpxchg is used for some atomic
1934 * operations, this function must be called from the CPU which owns the buffer
1935 * for a ACTIVE flush.
1936 */
1937 void lib_ring_buffer_switch_slow(struct lttng_ust_lib_ring_buffer *buf, enum switch_mode mode,
1938 struct lttng_ust_shm_handle *handle)
1939 {
1940 struct channel *chan;
1941 const struct lttng_ust_lib_ring_buffer_config *config;
1942 struct switch_offsets offsets;
1943 unsigned long oldidx;
1944 uint64_t tsc;
1945
1946 chan = shmp(handle, buf->backend.chan);
1947 if (!chan)
1948 return;
1949 config = &chan->backend.config;
1950
1951 offsets.size = 0;
1952
1953 /*
1954 * Perform retryable operations.
1955 */
1956 do {
1957 if (lib_ring_buffer_try_switch_slow(mode, buf, chan, &offsets,
1958 &tsc, handle))
1959 return; /* Switch not needed */
1960 } while (v_cmpxchg(config, &buf->offset, offsets.old, offsets.end)
1961 != offsets.old);
1962
1963 /*
1964 * Atomically update last_tsc. This update races against concurrent
1965 * atomic updates, but the race will always cause supplementary full TSC
1966 * records, never the opposite (missing a full TSC record when it would
1967 * be needed).
1968 */
1969 save_last_tsc(config, buf, tsc);
1970
1971 /*
1972 * Push the reader if necessary
1973 */
1974 lib_ring_buffer_reserve_push_reader(buf, chan, offsets.old);
1975
1976 oldidx = subbuf_index(offsets.old, chan);
1977 lib_ring_buffer_clear_noref(config, &buf->backend, oldidx, handle);
1978
1979 /*
1980 * May need to populate header start on SWITCH_FLUSH.
1981 */
1982 if (offsets.switch_old_start) {
1983 lib_ring_buffer_switch_old_start(buf, chan, &offsets, tsc, handle);
1984 offsets.old += config->cb.subbuffer_header_size();
1985 }
1986
1987 /*
1988 * Switch old subbuffer.
1989 */
1990 lib_ring_buffer_switch_old_end(buf, chan, &offsets, tsc, handle);
1991 }
1992
1993 /*
1994 * Returns :
1995 * 0 if ok
1996 * -ENOSPC if event size is too large for packet.
1997 * -ENOBUFS if there is currently not enough space in buffer for the event.
1998 * -EIO if data cannot be written into the buffer for any other reason.
1999 */
2000 static
2001 int lib_ring_buffer_try_reserve_slow(struct lttng_ust_lib_ring_buffer *buf,
2002 struct channel *chan,
2003 struct switch_offsets *offsets,
2004 struct lttng_ust_lib_ring_buffer_ctx *ctx)
2005 {
2006 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
2007 struct lttng_ust_shm_handle *handle = ctx->handle;
2008 unsigned long reserve_commit_diff, offset_cmp;
2009
2010 retry:
2011 offsets->begin = offset_cmp = v_read(config, &buf->offset);
2012 offsets->old = offsets->begin;
2013 offsets->switch_new_start = 0;
2014 offsets->switch_new_end = 0;
2015 offsets->switch_old_end = 0;
2016 offsets->pre_header_padding = 0;
2017
2018 ctx->tsc = config->cb.ring_buffer_clock_read(chan);
2019 if ((int64_t) ctx->tsc == -EIO)
2020 return -EIO;
2021
2022 if (last_tsc_overflow(config, buf, ctx->tsc))
2023 ctx->rflags |= RING_BUFFER_RFLAG_FULL_TSC;
2024
2025 if (caa_unlikely(subbuf_offset(offsets->begin, ctx->chan) == 0)) {
2026 offsets->switch_new_start = 1; /* For offsets->begin */
2027 } else {
2028 offsets->size = config->cb.record_header_size(config, chan,
2029 offsets->begin,
2030 &offsets->pre_header_padding,
2031 ctx);
2032 offsets->size +=
2033 lib_ring_buffer_align(offsets->begin + offsets->size,
2034 ctx->largest_align)
2035 + ctx->data_size;
2036 if (caa_unlikely(subbuf_offset(offsets->begin, chan) +
2037 offsets->size > chan->backend.subbuf_size)) {
2038 offsets->switch_old_end = 1; /* For offsets->old */
2039 offsets->switch_new_start = 1; /* For offsets->begin */
2040 }
2041 }
2042 if (caa_unlikely(offsets->switch_new_start)) {
2043 unsigned long sb_index, commit_count;
2044 struct commit_counters_cold *cc_cold;
2045
2046 /*
2047 * We are typically not filling the previous buffer completely.
2048 */
2049 if (caa_likely(offsets->switch_old_end))
2050 offsets->begin = subbuf_align(offsets->begin, chan);
2051 offsets->begin = offsets->begin
2052 + config->cb.subbuffer_header_size();
2053 /* Test new buffer integrity */
2054 sb_index = subbuf_index(offsets->begin, chan);
2055 /*
2056 * Read buf->offset before buf->commit_cold[sb_index].cc_sb.
2057 * lib_ring_buffer_check_deliver() has the matching
2058 * memory barriers required around commit_cold cc_sb
2059 * updates to ensure reserve and commit counter updates
2060 * are not seen reordered when updated by another CPU.
2061 */
2062 cmm_smp_rmb();
2063 cc_cold = shmp_index(handle, buf->commit_cold, sb_index);
2064 if (!cc_cold)
2065 return -1;
2066 commit_count = v_read(config, &cc_cold->cc_sb);
2067 /* Read buf->commit_cold[sb_index].cc_sb before buf->offset. */
2068 cmm_smp_rmb();
2069 if (caa_unlikely(offset_cmp != v_read(config, &buf->offset))) {
2070 /*
2071 * The reserve counter have been concurrently updated
2072 * while we read the commit counter. This means the
2073 * commit counter we read might not match buf->offset
2074 * due to concurrent update. We therefore need to retry.
2075 */
2076 goto retry;
2077 }
2078 reserve_commit_diff =
2079 (buf_trunc(offsets->begin, chan)
2080 >> chan->backend.num_subbuf_order)
2081 - (commit_count & chan->commit_count_mask);
2082 if (caa_likely(reserve_commit_diff == 0)) {
2083 /* Next subbuffer not being written to. */
2084 if (caa_unlikely(config->mode != RING_BUFFER_OVERWRITE &&
2085 subbuf_trunc(offsets->begin, chan)
2086 - subbuf_trunc((unsigned long)
2087 uatomic_read(&buf->consumed), chan)
2088 >= chan->backend.buf_size)) {
2089 unsigned long nr_lost;
2090
2091 /*
2092 * We do not overwrite non consumed buffers
2093 * and we are full : record is lost.
2094 */
2095 nr_lost = v_read(config, &buf->records_lost_full);
2096 v_inc(config, &buf->records_lost_full);
2097 if ((nr_lost & (DBG_PRINT_NR_LOST - 1)) == 0) {
2098 DBG("%lu or more records lost in (%s:%d) (buffer full)\n",
2099 nr_lost + 1, chan->backend.name,
2100 buf->backend.cpu);
2101 }
2102 return -ENOBUFS;
2103 } else {
2104 /*
2105 * Next subbuffer not being written to, and we
2106 * are either in overwrite mode or the buffer is
2107 * not full. It's safe to write in this new
2108 * subbuffer.
2109 */
2110 }
2111 } else {
2112 unsigned long nr_lost;
2113
2114 /*
2115 * Next subbuffer reserve offset does not match the
2116 * commit offset, and this did not involve update to the
2117 * reserve counter. Drop record in producer-consumer and
2118 * overwrite mode. Caused by either a writer OOPS or too
2119 * many nested writes over a reserve/commit pair.
2120 */
2121 nr_lost = v_read(config, &buf->records_lost_wrap);
2122 v_inc(config, &buf->records_lost_wrap);
2123 if ((nr_lost & (DBG_PRINT_NR_LOST - 1)) == 0) {
2124 DBG("%lu or more records lost in (%s:%d) (wrap-around)\n",
2125 nr_lost + 1, chan->backend.name,
2126 buf->backend.cpu);
2127 }
2128 return -EIO;
2129 }
2130 offsets->size =
2131 config->cb.record_header_size(config, chan,
2132 offsets->begin,
2133 &offsets->pre_header_padding,
2134 ctx);
2135 offsets->size +=
2136 lib_ring_buffer_align(offsets->begin + offsets->size,
2137 ctx->largest_align)
2138 + ctx->data_size;
2139 if (caa_unlikely(subbuf_offset(offsets->begin, chan)
2140 + offsets->size > chan->backend.subbuf_size)) {
2141 unsigned long nr_lost;
2142
2143 /*
2144 * Record too big for subbuffers, report error, don't
2145 * complete the sub-buffer switch.
2146 */
2147 nr_lost = v_read(config, &buf->records_lost_big);
2148 v_inc(config, &buf->records_lost_big);
2149 if ((nr_lost & (DBG_PRINT_NR_LOST - 1)) == 0) {
2150 DBG("%lu or more records lost in (%s:%d) record size "
2151 " of %zu bytes is too large for buffer\n",
2152 nr_lost + 1, chan->backend.name,
2153 buf->backend.cpu, offsets->size);
2154 }
2155 return -ENOSPC;
2156 } else {
2157 /*
2158 * We just made a successful buffer switch and the
2159 * record fits in the new subbuffer. Let's write.
2160 */
2161 }
2162 } else {
2163 /*
2164 * Record fits in the current buffer and we are not on a switch
2165 * boundary. It's safe to write.
2166 */
2167 }
2168 offsets->end = offsets->begin + offsets->size;
2169
2170 if (caa_unlikely(subbuf_offset(offsets->end, chan) == 0)) {
2171 /*
2172 * The offset_end will fall at the very beginning of the next
2173 * subbuffer.
2174 */
2175 offsets->switch_new_end = 1; /* For offsets->begin */
2176 }
2177 return 0;
2178 }
2179
2180 /**
2181 * lib_ring_buffer_reserve_slow - Atomic slot reservation in a buffer.
2182 * @ctx: ring buffer context.
2183 *
2184 * Return : -NOBUFS if not enough space, -ENOSPC if event size too large,
2185 * -EIO for other errors, else returns 0.
2186 * It will take care of sub-buffer switching.
2187 */
2188 int lib_ring_buffer_reserve_slow(struct lttng_ust_lib_ring_buffer_ctx *ctx)
2189 {
2190 struct channel *chan = ctx->chan;
2191 struct lttng_ust_shm_handle *handle = ctx->handle;
2192 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
2193 struct lttng_ust_lib_ring_buffer *buf;
2194 struct switch_offsets offsets;
2195 int ret;
2196
2197 if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
2198 buf = shmp(handle, chan->backend.buf[ctx->cpu].shmp);
2199 else
2200 buf = shmp(handle, chan->backend.buf[0].shmp);
2201 if (!buf)
2202 return -EIO;
2203 ctx->buf = buf;
2204
2205 offsets.size = 0;
2206
2207 do {
2208 ret = lib_ring_buffer_try_reserve_slow(buf, chan, &offsets,
2209 ctx);
2210 if (caa_unlikely(ret))
2211 return ret;
2212 } while (caa_unlikely(v_cmpxchg(config, &buf->offset, offsets.old,
2213 offsets.end)
2214 != offsets.old));
2215
2216 /*
2217 * Atomically update last_tsc. This update races against concurrent
2218 * atomic updates, but the race will always cause supplementary full TSC
2219 * records, never the opposite (missing a full TSC record when it would
2220 * be needed).
2221 */
2222 save_last_tsc(config, buf, ctx->tsc);
2223
2224 /*
2225 * Push the reader if necessary
2226 */
2227 lib_ring_buffer_reserve_push_reader(buf, chan, offsets.end - 1);
2228
2229 /*
2230 * Clear noref flag for this subbuffer.
2231 */
2232 lib_ring_buffer_clear_noref(config, &buf->backend,
2233 subbuf_index(offsets.end - 1, chan),
2234 handle);
2235
2236 /*
2237 * Switch old subbuffer if needed.
2238 */
2239 if (caa_unlikely(offsets.switch_old_end)) {
2240 lib_ring_buffer_clear_noref(config, &buf->backend,
2241 subbuf_index(offsets.old - 1, chan),
2242 handle);
2243 lib_ring_buffer_switch_old_end(buf, chan, &offsets, ctx->tsc, handle);
2244 }
2245
2246 /*
2247 * Populate new subbuffer.
2248 */
2249 if (caa_unlikely(offsets.switch_new_start))
2250 lib_ring_buffer_switch_new_start(buf, chan, &offsets, ctx->tsc, handle);
2251
2252 if (caa_unlikely(offsets.switch_new_end))
2253 lib_ring_buffer_switch_new_end(buf, chan, &offsets, ctx->tsc, handle);
2254
2255 ctx->slot_size = offsets.size;
2256 ctx->pre_offset = offsets.begin;
2257 ctx->buf_offset = offsets.begin + offsets.pre_header_padding;
2258 return 0;
2259 }
2260
2261 static
2262 void lib_ring_buffer_vmcore_check_deliver(const struct lttng_ust_lib_ring_buffer_config *config,
2263 struct lttng_ust_lib_ring_buffer *buf,
2264 unsigned long commit_count,
2265 unsigned long idx,
2266 struct lttng_ust_shm_handle *handle)
2267 {
2268 struct commit_counters_hot *cc_hot;
2269
2270 if (config->oops != RING_BUFFER_OOPS_CONSISTENCY)
2271 return;
2272 cc_hot = shmp_index(handle, buf->commit_hot, idx);
2273 if (!cc_hot)
2274 return;
2275 v_set(config, &cc_hot->seq, commit_count);
2276 }
2277
2278 /*
2279 * The ring buffer can count events recorded and overwritten per buffer,
2280 * but it is disabled by default due to its performance overhead.
2281 */
2282 #ifdef LTTNG_RING_BUFFER_COUNT_EVENTS
2283 static
2284 void deliver_count_events(const struct lttng_ust_lib_ring_buffer_config *config,
2285 struct lttng_ust_lib_ring_buffer *buf,
2286 unsigned long idx,
2287 struct lttng_ust_shm_handle *handle)
2288 {
2289 v_add(config, subbuffer_get_records_count(config,
2290 &buf->backend, idx, handle),
2291 &buf->records_count);
2292 v_add(config, subbuffer_count_records_overrun(config,
2293 &buf->backend, idx, handle),
2294 &buf->records_overrun);
2295 }
2296 #else /* LTTNG_RING_BUFFER_COUNT_EVENTS */
2297 static
2298 void deliver_count_events(const struct lttng_ust_lib_ring_buffer_config *config,
2299 struct lttng_ust_lib_ring_buffer *buf,
2300 unsigned long idx,
2301 struct lttng_ust_shm_handle *handle)
2302 {
2303 }
2304 #endif /* #else LTTNG_RING_BUFFER_COUNT_EVENTS */
2305
2306 void lib_ring_buffer_check_deliver_slow(const struct lttng_ust_lib_ring_buffer_config *config,
2307 struct lttng_ust_lib_ring_buffer *buf,
2308 struct channel *chan,
2309 unsigned long offset,
2310 unsigned long commit_count,
2311 unsigned long idx,
2312 struct lttng_ust_shm_handle *handle,
2313 uint64_t tsc)
2314 {
2315 unsigned long old_commit_count = commit_count
2316 - chan->backend.subbuf_size;
2317 struct commit_counters_cold *cc_cold;
2318
2319 /*
2320 * If we succeeded at updating cc_sb below, we are the subbuffer
2321 * writer delivering the subbuffer. Deals with concurrent
2322 * updates of the "cc" value without adding a add_return atomic
2323 * operation to the fast path.
2324 *
2325 * We are doing the delivery in two steps:
2326 * - First, we cmpxchg() cc_sb to the new value
2327 * old_commit_count + 1. This ensures that we are the only
2328 * subbuffer user successfully filling the subbuffer, but we
2329 * do _not_ set the cc_sb value to "commit_count" yet.
2330 * Therefore, other writers that would wrap around the ring
2331 * buffer and try to start writing to our subbuffer would
2332 * have to drop records, because it would appear as
2333 * non-filled.
2334 * We therefore have exclusive access to the subbuffer control
2335 * structures. This mutual exclusion with other writers is
2336 * crucially important to perform record overruns count in
2337 * flight recorder mode locklessly.
2338 * - When we are ready to release the subbuffer (either for
2339 * reading or for overrun by other writers), we simply set the
2340 * cc_sb value to "commit_count" and perform delivery.
2341 *
2342 * The subbuffer size is least 2 bytes (minimum size: 1 page).
2343 * This guarantees that old_commit_count + 1 != commit_count.
2344 */
2345
2346 /*
2347 * Order prior updates to reserve count prior to the
2348 * commit_cold cc_sb update.
2349 */
2350 cmm_smp_wmb();
2351 cc_cold = shmp_index(handle, buf->commit_cold, idx);
2352 if (!cc_cold)
2353 return;
2354 if (caa_likely(v_cmpxchg(config, &cc_cold->cc_sb,
2355 old_commit_count, old_commit_count + 1)
2356 == old_commit_count)) {
2357 /*
2358 * Start of exclusive subbuffer access. We are
2359 * guaranteed to be the last writer in this subbuffer
2360 * and any other writer trying to access this subbuffer
2361 * in this state is required to drop records.
2362 */
2363 deliver_count_events(config, buf, idx, handle);
2364 config->cb.buffer_end(buf, tsc, idx,
2365 lib_ring_buffer_get_data_size(config,
2366 buf,
2367 idx,
2368 handle),
2369 handle);
2370
2371 /*
2372 * Increment the packet counter while we have exclusive
2373 * access.
2374 */
2375 subbuffer_inc_packet_count(config, &buf->backend, idx, handle);
2376
2377 /*
2378 * Set noref flag and offset for this subbuffer id.
2379 * Contains a memory barrier that ensures counter stores
2380 * are ordered before set noref and offset.
2381 */
2382 lib_ring_buffer_set_noref_offset(config, &buf->backend, idx,
2383 buf_trunc_val(offset, chan), handle);
2384
2385 /*
2386 * Order set_noref and record counter updates before the
2387 * end of subbuffer exclusive access. Orders with
2388 * respect to writers coming into the subbuffer after
2389 * wrap around, and also order wrt concurrent readers.
2390 */
2391 cmm_smp_mb();
2392 /* End of exclusive subbuffer access */
2393 v_set(config, &cc_cold->cc_sb, commit_count);
2394 /*
2395 * Order later updates to reserve count after
2396 * the commit cold cc_sb update.
2397 */
2398 cmm_smp_wmb();
2399 lib_ring_buffer_vmcore_check_deliver(config, buf,
2400 commit_count, idx, handle);
2401
2402 /*
2403 * RING_BUFFER_WAKEUP_BY_WRITER wakeup is not lock-free.
2404 */
2405 if (config->wakeup == RING_BUFFER_WAKEUP_BY_WRITER
2406 && uatomic_read(&buf->active_readers)
2407 && lib_ring_buffer_poll_deliver(config, buf, chan, handle)) {
2408 lib_ring_buffer_wakeup(buf, handle);
2409 }
2410 }
2411 }
2412
2413 /*
2414 * Force a read (imply TLS fixup for dlopen) of TLS variables.
2415 */
2416 void lttng_fixup_ringbuffer_tls(void)
2417 {
2418 asm volatile ("" : : "m" (URCU_TLS(lib_ring_buffer_nesting)));
2419 }
2420
2421 void lib_ringbuffer_signal_init(void)
2422 {
2423 sigset_t mask;
2424 int ret;
2425
2426 /*
2427 * Block signal for entire process, so only our thread processes
2428 * it.
2429 */
2430 rb_setmask(&mask);
2431 ret = pthread_sigmask(SIG_BLOCK, &mask, NULL);
2432 if (ret) {
2433 errno = ret;
2434 PERROR("pthread_sigmask");
2435 }
2436 }
This page took 0.082978 seconds and 5 git commands to generate.