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