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