2 * SPDX-License-Identifier: LGPL-2.1-only
4 * Copyright (C) 2005-2012 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
11 #include <urcu/arch.h>
14 #include <lttng/ust-utils.h>
15 #include <lttng/ust-ringbuffer-context.h>
17 #include "ringbuffer-config.h"
21 #include "common/smp.h"
23 #include "common/align.h"
24 #include "common/populate.h"
27 * lib_ring_buffer_backend_allocate - allocate a channel buffer
28 * @config: ring buffer instance configuration
29 * @buf: the buffer struct
30 * @size: total size of the buffer
31 * @num_subbuf: number of subbuffers
32 * @extra_reader_sb: need extra subbuffer for reader
35 int lib_ring_buffer_backend_allocate(const struct lttng_ust_ring_buffer_config
*config
,
36 struct lttng_ust_ring_buffer_backend
*bufb
,
37 size_t size
__attribute__((unused
)), size_t num_subbuf
,
39 struct lttng_ust_shm_handle
*handle
,
40 struct shm_object
*shmobj
)
42 struct channel_backend
*chanb
;
43 unsigned long subbuf_size
, mmap_offset
= 0;
44 unsigned long num_subbuf_alloc
;
48 chanb
= &shmp(handle
, bufb
->chan
)->backend
;
52 subbuf_size
= chanb
->subbuf_size
;
53 num_subbuf_alloc
= num_subbuf
;
58 page_size
= LTTNG_UST_PAGE_SIZE
;
63 align_shm(shmobj
, __alignof__(struct lttng_ust_ring_buffer_backend_pages_shmp
));
64 set_shmp(bufb
->array
, zalloc_shm(shmobj
,
65 sizeof(struct lttng_ust_ring_buffer_backend_pages_shmp
) * num_subbuf_alloc
));
66 if (caa_unlikely(!shmp(handle
, bufb
->array
)))
70 * This is the largest element (the buffer pages) which needs to
71 * be aligned on page size.
73 align_shm(shmobj
, page_size
);
74 set_shmp(bufb
->memory_map
, zalloc_shm(shmobj
,
75 subbuf_size
* num_subbuf_alloc
));
76 if (caa_unlikely(!shmp(handle
, bufb
->memory_map
)))
77 goto memory_map_error
;
79 /* Allocate backend pages array elements */
80 for (i
= 0; i
< num_subbuf_alloc
; i
++) {
81 align_shm(shmobj
, __alignof__(struct lttng_ust_ring_buffer_backend_pages
));
82 set_shmp(shmp_index(handle
, bufb
->array
, i
)->shmp
,
84 sizeof(struct lttng_ust_ring_buffer_backend_pages
)));
85 if (!shmp(handle
, shmp_index(handle
, bufb
->array
, i
)->shmp
))
89 /* Allocate write-side subbuffer table */
90 align_shm(shmobj
, __alignof__(struct lttng_ust_ring_buffer_backend_subbuffer
));
91 set_shmp(bufb
->buf_wsb
, zalloc_shm(shmobj
,
92 sizeof(struct lttng_ust_ring_buffer_backend_subbuffer
)
94 if (caa_unlikely(!shmp(handle
, bufb
->buf_wsb
)))
97 for (i
= 0; i
< num_subbuf
; i
++) {
98 struct lttng_ust_ring_buffer_backend_subbuffer
*sb
;
100 sb
= shmp_index(handle
, bufb
->buf_wsb
, i
);
103 sb
->id
= subbuffer_id(config
, 0, 1, i
);
106 /* Assign read-side subbuffer table */
108 bufb
->buf_rsb
.id
= subbuffer_id(config
, 0, 1,
109 num_subbuf_alloc
- 1);
111 bufb
->buf_rsb
.id
= subbuffer_id(config
, 0, 1, 0);
113 /* Allocate subbuffer packet counter table */
114 align_shm(shmobj
, __alignof__(struct lttng_ust_ring_buffer_backend_counts
));
115 set_shmp(bufb
->buf_cnt
, zalloc_shm(shmobj
,
116 sizeof(struct lttng_ust_ring_buffer_backend_counts
)
118 if (caa_unlikely(!shmp(handle
, bufb
->buf_cnt
)))
121 /* Assign pages to page index */
122 for (i
= 0; i
< num_subbuf_alloc
; i
++) {
123 struct lttng_ust_ring_buffer_backend_pages_shmp
*sbp
;
124 struct lttng_ust_ring_buffer_backend_pages
*pages
;
127 ref
.index
= bufb
->memory_map
._ref
.index
;
128 ref
.offset
= bufb
->memory_map
._ref
.offset
;
129 ref
.offset
+= i
* subbuf_size
;
131 sbp
= shmp_index(handle
, bufb
->array
, i
);
134 pages
= shmp(handle
, sbp
->shmp
);
137 set_shmp(pages
->p
, ref
);
138 if (config
->output
== RING_BUFFER_MMAP
) {
139 pages
->mmap_offset
= mmap_offset
;
140 mmap_offset
+= subbuf_size
;
146 /* bufb->buf_wsb will be freed by shm teardown */
148 /* bufb->array[i] will be freed by shm teardown */
150 /* bufb->array will be freed by shm teardown */
156 int lib_ring_buffer_backend_create(struct lttng_ust_ring_buffer_backend
*bufb
,
157 struct channel_backend
*chanb
, int cpu
,
158 struct lttng_ust_shm_handle
*handle
,
159 struct shm_object
*shmobj
)
161 const struct lttng_ust_ring_buffer_config
*config
= &chanb
->config
;
163 set_shmp(bufb
->chan
, handle
->chan
._ref
);
166 return lib_ring_buffer_backend_allocate(config
, bufb
, chanb
->buf_size
,
168 chanb
->extra_reader_sb
,
172 void lib_ring_buffer_backend_reset(struct lttng_ust_ring_buffer_backend
*bufb
,
173 struct lttng_ust_shm_handle
*handle
)
175 struct channel_backend
*chanb
;
176 const struct lttng_ust_ring_buffer_config
*config
;
177 unsigned long num_subbuf_alloc
;
180 chanb
= &shmp(handle
, bufb
->chan
)->backend
;
183 config
= &chanb
->config
;
185 num_subbuf_alloc
= chanb
->num_subbuf
;
186 if (chanb
->extra_reader_sb
)
189 for (i
= 0; i
< chanb
->num_subbuf
; i
++) {
190 struct lttng_ust_ring_buffer_backend_subbuffer
*sb
;
192 sb
= shmp_index(handle
, bufb
->buf_wsb
, i
);
195 sb
->id
= subbuffer_id(config
, 0, 1, i
);
197 if (chanb
->extra_reader_sb
)
198 bufb
->buf_rsb
.id
= subbuffer_id(config
, 0, 1,
199 num_subbuf_alloc
- 1);
201 bufb
->buf_rsb
.id
= subbuffer_id(config
, 0, 1, 0);
203 for (i
= 0; i
< num_subbuf_alloc
; i
++) {
204 struct lttng_ust_ring_buffer_backend_pages_shmp
*sbp
;
205 struct lttng_ust_ring_buffer_backend_pages
*pages
;
207 sbp
= shmp_index(handle
, bufb
->array
, i
);
210 pages
= shmp(handle
, sbp
->shmp
);
213 /* Don't reset mmap_offset */
214 v_set(config
, &pages
->records_commit
, 0);
215 v_set(config
, &pages
->records_unread
, 0);
216 pages
->data_size
= 0;
217 /* Don't reset backend page and virt addresses */
219 /* Don't reset num_pages_per_subbuf, cpu, allocated */
220 v_set(config
, &bufb
->records_read
, 0);
224 * The frontend is responsible for also calling ring_buffer_backend_reset for
225 * each buffer when calling channel_backend_reset.
227 void channel_backend_reset(struct channel_backend
*chanb
)
229 struct lttng_ust_ring_buffer_channel
*chan
= caa_container_of(chanb
,
230 struct lttng_ust_ring_buffer_channel
, backend
);
231 const struct lttng_ust_ring_buffer_config
*config
= &chanb
->config
;
234 * Don't reset buf_size, subbuf_size, subbuf_size_order,
235 * num_subbuf_order, buf_size_order, extra_reader_sb, num_subbuf,
236 * priv, notifiers, config, cpumask and name.
238 chanb
->start_timestamp
= config
->cb
.ring_buffer_clock_read(chan
);
242 * channel_backend_init - initialize a channel backend
243 * @chanb: channel backend
244 * @name: channel name
245 * @config: client ring buffer configuration
246 * @parent: dentry of parent directory, %NULL for root directory
247 * @subbuf_size: size of sub-buffers (> page size, power of 2)
248 * @num_subbuf: number of sub-buffers (power of 2)
249 * @lttng_ust_shm_handle: shared memory handle
250 * @stream_fds: stream file descriptors.
252 * Returns channel pointer if successful, %NULL otherwise.
254 * Creates per-cpu channel buffers using the sizes and attributes
255 * specified. The created channel buffer files will be named
256 * name_0...name_N-1. File permissions will be %S_IRUSR.
258 * Called with CPU hotplug disabled.
260 int channel_backend_init(struct channel_backend
*chanb
,
262 const struct lttng_ust_ring_buffer_config
*config
,
263 size_t subbuf_size
, size_t num_subbuf
,
264 struct lttng_ust_shm_handle
*handle
,
265 const int *stream_fds
)
267 struct lttng_ust_ring_buffer_channel
*chan
= caa_container_of(chanb
,
268 struct lttng_ust_ring_buffer_channel
, backend
);
271 size_t shmsize
= 0, num_subbuf_alloc
;
277 page_size
= LTTNG_UST_PAGE_SIZE
;
278 if (page_size
<= 0) {
281 /* Check that the subbuffer size is larger than a page. */
282 if (subbuf_size
< page_size
)
286 * Make sure the number of subbuffers and subbuffer size are
287 * power of 2, and nonzero.
289 if (!subbuf_size
|| (subbuf_size
& (subbuf_size
- 1)))
291 if (!num_subbuf
|| (num_subbuf
& (num_subbuf
- 1)))
294 * Overwrite mode buffers require at least 2 subbuffers per
297 if (config
->mode
== RING_BUFFER_OVERWRITE
&& num_subbuf
< 2)
300 ret
= subbuffer_id_check_index(config
, num_subbuf
);
304 chanb
->buf_size
= num_subbuf
* subbuf_size
;
305 chanb
->subbuf_size
= subbuf_size
;
306 chanb
->buf_size_order
= get_count_order(chanb
->buf_size
);
307 chanb
->subbuf_size_order
= get_count_order(subbuf_size
);
308 chanb
->num_subbuf_order
= get_count_order(num_subbuf
);
309 chanb
->extra_reader_sb
=
310 (config
->mode
== RING_BUFFER_OVERWRITE
) ? 1 : 0;
311 chanb
->num_subbuf
= num_subbuf
;
312 strncpy(chanb
->name
, name
, NAME_MAX
);
313 chanb
->name
[NAME_MAX
- 1] = '\0';
314 memcpy(&chanb
->config
, config
, sizeof(*config
));
316 /* Per-cpu buffer size: control (prior to backend) */
317 shmsize
= lttng_ust_offset_align(shmsize
, __alignof__(struct lttng_ust_ring_buffer
));
318 shmsize
+= sizeof(struct lttng_ust_ring_buffer
);
319 shmsize
+= lttng_ust_offset_align(shmsize
, __alignof__(struct commit_counters_hot
));
320 shmsize
+= sizeof(struct commit_counters_hot
) * num_subbuf
;
321 shmsize
+= lttng_ust_offset_align(shmsize
, __alignof__(struct commit_counters_cold
));
322 shmsize
+= sizeof(struct commit_counters_cold
) * num_subbuf
;
323 /* Sampled timestamp end */
324 shmsize
+= lttng_ust_offset_align(shmsize
, __alignof__(uint64_t));
325 shmsize
+= sizeof(uint64_t) * num_subbuf
;
327 /* Per-cpu buffer size: backend */
328 /* num_subbuf + 1 is the worse case */
329 num_subbuf_alloc
= num_subbuf
+ 1;
330 shmsize
+= lttng_ust_offset_align(shmsize
, __alignof__(struct lttng_ust_ring_buffer_backend_pages_shmp
));
331 shmsize
+= sizeof(struct lttng_ust_ring_buffer_backend_pages_shmp
) * num_subbuf_alloc
;
332 shmsize
+= lttng_ust_offset_align(shmsize
, page_size
);
333 shmsize
+= subbuf_size
* num_subbuf_alloc
;
334 shmsize
+= lttng_ust_offset_align(shmsize
, __alignof__(struct lttng_ust_ring_buffer_backend_pages
));
335 shmsize
+= sizeof(struct lttng_ust_ring_buffer_backend_pages
) * num_subbuf_alloc
;
336 shmsize
+= lttng_ust_offset_align(shmsize
, __alignof__(struct lttng_ust_ring_buffer_backend_subbuffer
));
337 shmsize
+= sizeof(struct lttng_ust_ring_buffer_backend_subbuffer
) * num_subbuf
;
338 shmsize
+= lttng_ust_offset_align(shmsize
, __alignof__(struct lttng_ust_ring_buffer_backend_counts
));
339 shmsize
+= sizeof(struct lttng_ust_ring_buffer_backend_counts
) * num_subbuf
;
341 if (config
->alloc
== RING_BUFFER_ALLOC_PER_CPU
) {
342 struct lttng_ust_ring_buffer
*buf
;
344 * We need to allocate for all possible cpus.
346 for_each_possible_cpu(i
) {
347 struct shm_object
*shmobj
;
349 shmobj
= shm_object_table_alloc(handle
->table
, shmsize
,
350 SHM_OBJECT_SHM
, stream_fds
[i
], i
,
351 lttng_ust_map_populate_cpu_is_enabled(i
));
354 align_shm(shmobj
, __alignof__(struct lttng_ust_ring_buffer
));
355 set_shmp(chanb
->buf
[i
].shmp
, zalloc_shm(shmobj
, sizeof(struct lttng_ust_ring_buffer
)));
356 buf
= shmp(handle
, chanb
->buf
[i
].shmp
);
359 set_shmp(buf
->self
, chanb
->buf
[i
].shmp
._ref
);
360 ret
= lib_ring_buffer_create(buf
, chanb
, i
,
363 goto free_bufs
; /* cpu hotplug locked */
366 struct shm_object
*shmobj
;
367 struct lttng_ust_ring_buffer
*buf
;
369 shmobj
= shm_object_table_alloc(handle
->table
, shmsize
,
370 SHM_OBJECT_SHM
, stream_fds
[0], -1,
371 lttng_ust_map_populate_is_enabled());
374 align_shm(shmobj
, __alignof__(struct lttng_ust_ring_buffer
));
375 set_shmp(chanb
->buf
[0].shmp
, zalloc_shm(shmobj
, sizeof(struct lttng_ust_ring_buffer
)));
376 buf
= shmp(handle
, chanb
->buf
[0].shmp
);
379 set_shmp(buf
->self
, chanb
->buf
[0].shmp
._ref
);
380 ret
= lib_ring_buffer_create(buf
, chanb
, -1,
385 chanb
->start_timestamp
= config
->cb
.ring_buffer_clock_read(chan
);
390 /* We only free the buffer data upon shm teardown */
396 * channel_backend_free - destroy the channel
399 * Destroy all channel buffers and frees the channel.
401 void channel_backend_free(struct channel_backend
*chanb
__attribute__((unused
)),
402 struct lttng_ust_shm_handle
*handle
__attribute__((unused
)))
404 /* SHM teardown takes care of everything */
408 * lib_ring_buffer_read - read data from ring_buffer_buffer.
409 * @bufb : buffer backend
410 * @offset : offset within the buffer
411 * @dest : destination address
412 * @len : length to copy to destination
414 * Should be protected by get_subbuf/put_subbuf.
415 * Returns the length copied.
417 size_t lib_ring_buffer_read(struct lttng_ust_ring_buffer_backend
*bufb
, size_t offset
,
418 void *dest
, size_t len
, struct lttng_ust_shm_handle
*handle
)
420 struct channel_backend
*chanb
;
421 const struct lttng_ust_ring_buffer_config
*config
;
423 struct lttng_ust_ring_buffer_backend_pages_shmp
*rpages
;
424 struct lttng_ust_ring_buffer_backend_pages
*backend_pages
;
425 unsigned long sb_bindex
, id
;
428 chanb
= &shmp(handle
, bufb
->chan
)->backend
;
431 config
= &chanb
->config
;
433 offset
&= chanb
->buf_size
- 1;
435 if (caa_unlikely(!len
))
437 id
= bufb
->buf_rsb
.id
;
438 sb_bindex
= subbuffer_id_get_index(config
, id
);
439 rpages
= shmp_index(handle
, bufb
->array
, sb_bindex
);
443 * Underlying layer should never ask for reads across
446 CHAN_WARN_ON(chanb
, offset
>= chanb
->buf_size
);
447 CHAN_WARN_ON(chanb
, config
->mode
== RING_BUFFER_OVERWRITE
448 && subbuffer_id_is_noref(config
, id
));
449 backend_pages
= shmp(handle
, rpages
->shmp
);
452 src
= shmp_index(handle
, backend_pages
->p
, offset
& (chanb
->subbuf_size
- 1));
453 if (caa_unlikely(!src
))
455 memcpy(dest
, src
, len
);
460 * lib_ring_buffer_read_cstr - read a C-style string from ring_buffer.
461 * @bufb : buffer backend
462 * @offset : offset within the buffer
463 * @dest : destination address
464 * @len : destination's length
466 * Return string's length, or -EINVAL on error.
467 * Should be protected by get_subbuf/put_subbuf.
468 * Destination length should be at least 1 to hold '\0'.
470 int lib_ring_buffer_read_cstr(struct lttng_ust_ring_buffer_backend
*bufb
, size_t offset
,
471 void *dest
, size_t len
, struct lttng_ust_shm_handle
*handle
)
473 struct channel_backend
*chanb
;
474 const struct lttng_ust_ring_buffer_config
*config
;
475 ssize_t string_len
, orig_offset
;
477 struct lttng_ust_ring_buffer_backend_pages_shmp
*rpages
;
478 struct lttng_ust_ring_buffer_backend_pages
*backend_pages
;
479 unsigned long sb_bindex
, id
;
481 chanb
= &shmp(handle
, bufb
->chan
)->backend
;
484 config
= &chanb
->config
;
485 if (caa_unlikely(!len
))
487 offset
&= chanb
->buf_size
- 1;
488 orig_offset
= offset
;
489 id
= bufb
->buf_rsb
.id
;
490 sb_bindex
= subbuffer_id_get_index(config
, id
);
491 rpages
= shmp_index(handle
, bufb
->array
, sb_bindex
);
495 * Underlying layer should never ask for reads across
498 CHAN_WARN_ON(chanb
, offset
>= chanb
->buf_size
);
499 CHAN_WARN_ON(chanb
, config
->mode
== RING_BUFFER_OVERWRITE
500 && subbuffer_id_is_noref(config
, id
));
501 backend_pages
= shmp(handle
, rpages
->shmp
);
504 str
= shmp_index(handle
, backend_pages
->p
, offset
& (chanb
->subbuf_size
- 1));
505 if (caa_unlikely(!str
))
507 string_len
= strnlen(str
, len
);
509 memcpy(dest
, str
, string_len
);
510 ((char *)dest
)[0] = 0;
512 return offset
- orig_offset
;
516 * lib_ring_buffer_read_offset_address - get address of a buffer location
517 * @bufb : buffer backend
518 * @offset : offset within the buffer.
520 * Return the address where a given offset is located (for read).
521 * Should be used to get the current subbuffer header pointer. Given we know
522 * it's never on a page boundary, it's safe to read/write directly
523 * from/to this address, as long as the read/write is never bigger than
526 void *lib_ring_buffer_read_offset_address(struct lttng_ust_ring_buffer_backend
*bufb
,
528 struct lttng_ust_shm_handle
*handle
)
530 struct lttng_ust_ring_buffer_backend_pages_shmp
*rpages
;
531 struct lttng_ust_ring_buffer_backend_pages
*backend_pages
;
532 struct channel_backend
*chanb
;
533 const struct lttng_ust_ring_buffer_config
*config
;
534 unsigned long sb_bindex
, id
;
536 chanb
= &shmp(handle
, bufb
->chan
)->backend
;
539 config
= &chanb
->config
;
540 offset
&= chanb
->buf_size
- 1;
541 id
= bufb
->buf_rsb
.id
;
542 sb_bindex
= subbuffer_id_get_index(config
, id
);
543 rpages
= shmp_index(handle
, bufb
->array
, sb_bindex
);
546 CHAN_WARN_ON(chanb
, config
->mode
== RING_BUFFER_OVERWRITE
547 && subbuffer_id_is_noref(config
, id
));
548 backend_pages
= shmp(handle
, rpages
->shmp
);
551 return shmp_index(handle
, backend_pages
->p
, offset
& (chanb
->subbuf_size
- 1));
555 * lib_ring_buffer_offset_address - get address of a location within the buffer
556 * @bufb : buffer backend
557 * @offset : offset within the buffer.
559 * Return the address where a given offset is located.
560 * Should be used to get the current subbuffer header pointer. Given we know
561 * it's always at the beginning of a page, it's safe to write directly to this
562 * address, as long as the write is never bigger than a page size.
564 void *lib_ring_buffer_offset_address(struct lttng_ust_ring_buffer_backend
*bufb
,
566 struct lttng_ust_shm_handle
*handle
)
569 struct lttng_ust_ring_buffer_backend_pages_shmp
*rpages
;
570 struct lttng_ust_ring_buffer_backend_pages
*backend_pages
;
571 struct channel_backend
*chanb
;
572 const struct lttng_ust_ring_buffer_config
*config
;
573 unsigned long sb_bindex
, id
;
574 struct lttng_ust_ring_buffer_backend_subbuffer
*sb
;
576 chanb
= &shmp(handle
, bufb
->chan
)->backend
;
579 config
= &chanb
->config
;
580 offset
&= chanb
->buf_size
- 1;
581 sbidx
= offset
>> chanb
->subbuf_size_order
;
582 sb
= shmp_index(handle
, bufb
->buf_wsb
, sbidx
);
586 sb_bindex
= subbuffer_id_get_index(config
, id
);
587 rpages
= shmp_index(handle
, bufb
->array
, sb_bindex
);
590 CHAN_WARN_ON(chanb
, config
->mode
== RING_BUFFER_OVERWRITE
591 && subbuffer_id_is_noref(config
, id
));
592 backend_pages
= shmp(handle
, rpages
->shmp
);
595 return shmp_index(handle
, backend_pages
->p
, offset
& (chanb
->subbuf_size
- 1));