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