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