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/align.h>
15 #include <lttng/ringbuffer-config.h>
23 * lib_ring_buffer_backend_allocate - allocate a channel buffer
24 * @config: ring buffer instance configuration
25 * @buf: the buffer struct
26 * @size: total size of the buffer
27 * @num_subbuf: number of subbuffers
28 * @extra_reader_sb: need extra subbuffer for reader
31 int lib_ring_buffer_backend_allocate(const struct lttng_ust_lib_ring_buffer_config
*config
,
32 struct lttng_ust_lib_ring_buffer_backend
*bufb
,
33 size_t size
, size_t num_subbuf
,
35 struct lttng_ust_shm_handle
*handle
,
36 struct shm_object
*shmobj
)
38 struct channel_backend
*chanb
;
39 unsigned long subbuf_size
, mmap_offset
= 0;
40 unsigned long num_subbuf_alloc
;
44 chanb
= &shmp(handle
, bufb
->chan
)->backend
;
48 subbuf_size
= chanb
->subbuf_size
;
49 num_subbuf_alloc
= num_subbuf
;
54 page_size
= LTTNG_UST_PAGE_SIZE
;
59 align_shm(shmobj
, __alignof__(struct lttng_ust_lib_ring_buffer_backend_pages_shmp
));
60 set_shmp(bufb
->array
, zalloc_shm(shmobj
,
61 sizeof(struct lttng_ust_lib_ring_buffer_backend_pages_shmp
) * num_subbuf_alloc
));
62 if (caa_unlikely(!shmp(handle
, bufb
->array
)))
66 * This is the largest element (the buffer pages) which needs to
67 * be aligned on page size.
69 align_shm(shmobj
, page_size
);
70 set_shmp(bufb
->memory_map
, zalloc_shm(shmobj
,
71 subbuf_size
* num_subbuf_alloc
));
72 if (caa_unlikely(!shmp(handle
, bufb
->memory_map
)))
73 goto memory_map_error
;
75 /* Allocate backend pages array elements */
76 for (i
= 0; i
< num_subbuf_alloc
; i
++) {
77 align_shm(shmobj
, __alignof__(struct lttng_ust_lib_ring_buffer_backend_pages
));
78 set_shmp(shmp_index(handle
, bufb
->array
, i
)->shmp
,
80 sizeof(struct lttng_ust_lib_ring_buffer_backend_pages
)));
81 if (!shmp(handle
, shmp_index(handle
, bufb
->array
, i
)->shmp
))
85 /* Allocate write-side subbuffer table */
86 align_shm(shmobj
, __alignof__(struct lttng_ust_lib_ring_buffer_backend_subbuffer
));
87 set_shmp(bufb
->buf_wsb
, zalloc_shm(shmobj
,
88 sizeof(struct lttng_ust_lib_ring_buffer_backend_subbuffer
)
90 if (caa_unlikely(!shmp(handle
, bufb
->buf_wsb
)))
93 for (i
= 0; i
< num_subbuf
; i
++) {
94 struct lttng_ust_lib_ring_buffer_backend_subbuffer
*sb
;
96 sb
= shmp_index(handle
, bufb
->buf_wsb
, i
);
99 sb
->id
= subbuffer_id(config
, 0, 1, i
);
102 /* Assign read-side subbuffer table */
104 bufb
->buf_rsb
.id
= subbuffer_id(config
, 0, 1,
105 num_subbuf_alloc
- 1);
107 bufb
->buf_rsb
.id
= subbuffer_id(config
, 0, 1, 0);
109 /* Allocate subbuffer packet counter table */
110 align_shm(shmobj
, __alignof__(struct lttng_ust_lib_ring_buffer_backend_counts
));
111 set_shmp(bufb
->buf_cnt
, zalloc_shm(shmobj
,
112 sizeof(struct lttng_ust_lib_ring_buffer_backend_counts
)
114 if (caa_unlikely(!shmp(handle
, bufb
->buf_cnt
)))
117 /* Assign pages to page index */
118 for (i
= 0; i
< num_subbuf_alloc
; i
++) {
119 struct lttng_ust_lib_ring_buffer_backend_pages_shmp
*sbp
;
120 struct lttng_ust_lib_ring_buffer_backend_pages
*pages
;
123 ref
.index
= bufb
->memory_map
._ref
.index
;
124 ref
.offset
= bufb
->memory_map
._ref
.offset
;
125 ref
.offset
+= i
* subbuf_size
;
127 sbp
= shmp_index(handle
, bufb
->array
, i
);
130 pages
= shmp(handle
, sbp
->shmp
);
133 set_shmp(pages
->p
, ref
);
134 if (config
->output
== RING_BUFFER_MMAP
) {
135 pages
->mmap_offset
= mmap_offset
;
136 mmap_offset
+= subbuf_size
;
142 /* bufb->buf_wsb will be freed by shm teardown */
144 /* bufb->array[i] will be freed by shm teardown */
146 /* bufb->array will be freed by shm teardown */
152 int lib_ring_buffer_backend_create(struct lttng_ust_lib_ring_buffer_backend
*bufb
,
153 struct channel_backend
*chanb
, int cpu
,
154 struct lttng_ust_shm_handle
*handle
,
155 struct shm_object
*shmobj
)
157 const struct lttng_ust_lib_ring_buffer_config
*config
= &chanb
->config
;
159 set_shmp(bufb
->chan
, handle
->chan
._ref
);
162 return lib_ring_buffer_backend_allocate(config
, bufb
, chanb
->buf_size
,
164 chanb
->extra_reader_sb
,
168 void lib_ring_buffer_backend_reset(struct lttng_ust_lib_ring_buffer_backend
*bufb
,
169 struct lttng_ust_shm_handle
*handle
)
171 struct channel_backend
*chanb
;
172 const struct lttng_ust_lib_ring_buffer_config
*config
;
173 unsigned long num_subbuf_alloc
;
176 chanb
= &shmp(handle
, bufb
->chan
)->backend
;
179 config
= &chanb
->config
;
181 num_subbuf_alloc
= chanb
->num_subbuf
;
182 if (chanb
->extra_reader_sb
)
185 for (i
= 0; i
< chanb
->num_subbuf
; i
++) {
186 struct lttng_ust_lib_ring_buffer_backend_subbuffer
*sb
;
188 sb
= shmp_index(handle
, bufb
->buf_wsb
, i
);
191 sb
->id
= subbuffer_id(config
, 0, 1, i
);
193 if (chanb
->extra_reader_sb
)
194 bufb
->buf_rsb
.id
= subbuffer_id(config
, 0, 1,
195 num_subbuf_alloc
- 1);
197 bufb
->buf_rsb
.id
= subbuffer_id(config
, 0, 1, 0);
199 for (i
= 0; i
< num_subbuf_alloc
; i
++) {
200 struct lttng_ust_lib_ring_buffer_backend_pages_shmp
*sbp
;
201 struct lttng_ust_lib_ring_buffer_backend_pages
*pages
;
203 sbp
= shmp_index(handle
, bufb
->array
, i
);
206 pages
= shmp(handle
, sbp
->shmp
);
209 /* Don't reset mmap_offset */
210 v_set(config
, &pages
->records_commit
, 0);
211 v_set(config
, &pages
->records_unread
, 0);
212 pages
->data_size
= 0;
213 /* Don't reset backend page and virt addresses */
215 /* Don't reset num_pages_per_subbuf, cpu, allocated */
216 v_set(config
, &bufb
->records_read
, 0);
220 * The frontend is responsible for also calling ring_buffer_backend_reset for
221 * each buffer when calling channel_backend_reset.
223 void channel_backend_reset(struct channel_backend
*chanb
)
225 struct channel
*chan
= caa_container_of(chanb
, struct channel
, backend
);
226 const struct lttng_ust_lib_ring_buffer_config
*config
= &chanb
->config
;
229 * Don't reset buf_size, subbuf_size, subbuf_size_order,
230 * num_subbuf_order, buf_size_order, extra_reader_sb, num_subbuf,
231 * priv, notifiers, config, cpumask and name.
233 chanb
->start_tsc
= config
->cb
.ring_buffer_clock_read(chan
);
237 * channel_backend_init - initialize a channel backend
238 * @chanb: channel backend
239 * @name: channel name
240 * @config: client ring buffer configuration
241 * @parent: dentry of parent directory, %NULL for root directory
242 * @subbuf_size: size of sub-buffers (> page size, power of 2)
243 * @num_subbuf: number of sub-buffers (power of 2)
244 * @lttng_ust_shm_handle: shared memory handle
245 * @stream_fds: stream file descriptors.
247 * Returns channel pointer if successful, %NULL otherwise.
249 * Creates per-cpu channel buffers using the sizes and attributes
250 * specified. The created channel buffer files will be named
251 * name_0...name_N-1. File permissions will be %S_IRUSR.
253 * Called with CPU hotplug disabled.
255 int channel_backend_init(struct channel_backend
*chanb
,
257 const struct lttng_ust_lib_ring_buffer_config
*config
,
258 size_t subbuf_size
, size_t num_subbuf
,
259 struct lttng_ust_shm_handle
*handle
,
260 const int *stream_fds
)
262 struct channel
*chan
= caa_container_of(chanb
, struct channel
, backend
);
265 size_t shmsize
= 0, num_subbuf_alloc
;
271 page_size
= LTTNG_UST_PAGE_SIZE
;
272 if (page_size
<= 0) {
275 /* Check that the subbuffer size is larger than a page. */
276 if (subbuf_size
< page_size
)
280 * Make sure the number of subbuffers and subbuffer size are
281 * power of 2, and nonzero.
283 if (!subbuf_size
|| (subbuf_size
& (subbuf_size
- 1)))
285 if (!num_subbuf
|| (num_subbuf
& (num_subbuf
- 1)))
288 * Overwrite mode buffers require at least 2 subbuffers per
291 if (config
->mode
== RING_BUFFER_OVERWRITE
&& num_subbuf
< 2)
294 ret
= subbuffer_id_check_index(config
, num_subbuf
);
298 chanb
->buf_size
= num_subbuf
* subbuf_size
;
299 chanb
->subbuf_size
= subbuf_size
;
300 chanb
->buf_size_order
= get_count_order(chanb
->buf_size
);
301 chanb
->subbuf_size_order
= get_count_order(subbuf_size
);
302 chanb
->num_subbuf_order
= get_count_order(num_subbuf
);
303 chanb
->extra_reader_sb
=
304 (config
->mode
== RING_BUFFER_OVERWRITE
) ? 1 : 0;
305 chanb
->num_subbuf
= num_subbuf
;
306 strncpy(chanb
->name
, name
, NAME_MAX
);
307 chanb
->name
[NAME_MAX
- 1] = '\0';
308 memcpy(&chanb
->config
, config
, sizeof(*config
));
310 /* Per-cpu buffer size: control (prior to backend) */
311 shmsize
= lttng_ust_offset_align(shmsize
, __alignof__(struct lttng_ust_lib_ring_buffer
));
312 shmsize
+= sizeof(struct lttng_ust_lib_ring_buffer
);
313 shmsize
+= lttng_ust_offset_align(shmsize
, __alignof__(struct commit_counters_hot
));
314 shmsize
+= sizeof(struct commit_counters_hot
) * num_subbuf
;
315 shmsize
+= lttng_ust_offset_align(shmsize
, __alignof__(struct commit_counters_cold
));
316 shmsize
+= sizeof(struct commit_counters_cold
) * num_subbuf
;
317 /* Sampled timestamp end */
318 shmsize
+= lttng_ust_offset_align(shmsize
, __alignof__(uint64_t));
319 shmsize
+= sizeof(uint64_t) * num_subbuf
;
321 /* Per-cpu buffer size: backend */
322 /* num_subbuf + 1 is the worse case */
323 num_subbuf_alloc
= num_subbuf
+ 1;
324 shmsize
+= lttng_ust_offset_align(shmsize
, __alignof__(struct lttng_ust_lib_ring_buffer_backend_pages_shmp
));
325 shmsize
+= sizeof(struct lttng_ust_lib_ring_buffer_backend_pages_shmp
) * num_subbuf_alloc
;
326 shmsize
+= lttng_ust_offset_align(shmsize
, page_size
);
327 shmsize
+= subbuf_size
* num_subbuf_alloc
;
328 shmsize
+= lttng_ust_offset_align(shmsize
, __alignof__(struct lttng_ust_lib_ring_buffer_backend_pages
));
329 shmsize
+= sizeof(struct lttng_ust_lib_ring_buffer_backend_pages
) * num_subbuf_alloc
;
330 shmsize
+= lttng_ust_offset_align(shmsize
, __alignof__(struct lttng_ust_lib_ring_buffer_backend_subbuffer
));
331 shmsize
+= sizeof(struct lttng_ust_lib_ring_buffer_backend_subbuffer
) * num_subbuf
;
332 shmsize
+= lttng_ust_offset_align(shmsize
, __alignof__(struct lttng_ust_lib_ring_buffer_backend_counts
));
333 shmsize
+= sizeof(struct lttng_ust_lib_ring_buffer_backend_counts
) * num_subbuf
;
335 if (config
->alloc
== RING_BUFFER_ALLOC_PER_CPU
) {
336 struct lttng_ust_lib_ring_buffer
*buf
;
338 * We need to allocate for all possible cpus.
340 for_each_possible_cpu(i
) {
341 struct shm_object
*shmobj
;
343 shmobj
= shm_object_table_alloc(handle
->table
, shmsize
,
344 SHM_OBJECT_SHM
, stream_fds
[i
], i
);
347 align_shm(shmobj
, __alignof__(struct lttng_ust_lib_ring_buffer
));
348 set_shmp(chanb
->buf
[i
].shmp
, zalloc_shm(shmobj
, sizeof(struct lttng_ust_lib_ring_buffer
)));
349 buf
= shmp(handle
, chanb
->buf
[i
].shmp
);
352 set_shmp(buf
->self
, chanb
->buf
[i
].shmp
._ref
);
353 ret
= lib_ring_buffer_create(buf
, chanb
, i
,
356 goto free_bufs
; /* cpu hotplug locked */
359 struct shm_object
*shmobj
;
360 struct lttng_ust_lib_ring_buffer
*buf
;
362 shmobj
= shm_object_table_alloc(handle
->table
, shmsize
,
363 SHM_OBJECT_SHM
, stream_fds
[0], -1);
366 align_shm(shmobj
, __alignof__(struct lttng_ust_lib_ring_buffer
));
367 set_shmp(chanb
->buf
[0].shmp
, zalloc_shm(shmobj
, sizeof(struct lttng_ust_lib_ring_buffer
)));
368 buf
= shmp(handle
, chanb
->buf
[0].shmp
);
371 set_shmp(buf
->self
, chanb
->buf
[0].shmp
._ref
);
372 ret
= lib_ring_buffer_create(buf
, chanb
, -1,
377 chanb
->start_tsc
= config
->cb
.ring_buffer_clock_read(chan
);
382 /* We only free the buffer data upon shm teardown */
388 * channel_backend_free - destroy the channel
391 * Destroy all channel buffers and frees the channel.
393 void channel_backend_free(struct channel_backend
*chanb
,
394 struct lttng_ust_shm_handle
*handle
)
396 /* SHM teardown takes care of everything */
400 * lib_ring_buffer_read - read data from ring_buffer_buffer.
401 * @bufb : buffer backend
402 * @offset : offset within the buffer
403 * @dest : destination address
404 * @len : length to copy to destination
406 * Should be protected by get_subbuf/put_subbuf.
407 * Returns the length copied.
409 size_t lib_ring_buffer_read(struct lttng_ust_lib_ring_buffer_backend
*bufb
, size_t offset
,
410 void *dest
, size_t len
, struct lttng_ust_shm_handle
*handle
)
412 struct channel_backend
*chanb
;
413 const struct lttng_ust_lib_ring_buffer_config
*config
;
415 struct lttng_ust_lib_ring_buffer_backend_pages_shmp
*rpages
;
416 struct lttng_ust_lib_ring_buffer_backend_pages
*backend_pages
;
417 unsigned long sb_bindex
, id
;
420 chanb
= &shmp(handle
, bufb
->chan
)->backend
;
423 config
= &chanb
->config
;
425 offset
&= chanb
->buf_size
- 1;
427 if (caa_unlikely(!len
))
429 id
= bufb
->buf_rsb
.id
;
430 sb_bindex
= subbuffer_id_get_index(config
, id
);
431 rpages
= shmp_index(handle
, bufb
->array
, sb_bindex
);
435 * Underlying layer should never ask for reads across
438 CHAN_WARN_ON(chanb
, offset
>= chanb
->buf_size
);
439 CHAN_WARN_ON(chanb
, config
->mode
== RING_BUFFER_OVERWRITE
440 && subbuffer_id_is_noref(config
, id
));
441 backend_pages
= shmp(handle
, rpages
->shmp
);
444 src
= shmp_index(handle
, backend_pages
->p
, offset
& (chanb
->subbuf_size
- 1));
445 if (caa_unlikely(!src
))
447 memcpy(dest
, src
, len
);
452 * lib_ring_buffer_read_cstr - read a C-style string from ring_buffer.
453 * @bufb : buffer backend
454 * @offset : offset within the buffer
455 * @dest : destination address
456 * @len : destination's length
458 * Return string's length, or -EINVAL on error.
459 * Should be protected by get_subbuf/put_subbuf.
460 * Destination length should be at least 1 to hold '\0'.
462 int lib_ring_buffer_read_cstr(struct lttng_ust_lib_ring_buffer_backend
*bufb
, size_t offset
,
463 void *dest
, size_t len
, struct lttng_ust_shm_handle
*handle
)
465 struct channel_backend
*chanb
;
466 const struct lttng_ust_lib_ring_buffer_config
*config
;
467 ssize_t string_len
, orig_offset
;
469 struct lttng_ust_lib_ring_buffer_backend_pages_shmp
*rpages
;
470 struct lttng_ust_lib_ring_buffer_backend_pages
*backend_pages
;
471 unsigned long sb_bindex
, id
;
473 chanb
= &shmp(handle
, bufb
->chan
)->backend
;
476 config
= &chanb
->config
;
477 if (caa_unlikely(!len
))
479 offset
&= chanb
->buf_size
- 1;
480 orig_offset
= offset
;
481 id
= bufb
->buf_rsb
.id
;
482 sb_bindex
= subbuffer_id_get_index(config
, id
);
483 rpages
= shmp_index(handle
, bufb
->array
, sb_bindex
);
487 * Underlying layer should never ask for reads across
490 CHAN_WARN_ON(chanb
, offset
>= chanb
->buf_size
);
491 CHAN_WARN_ON(chanb
, config
->mode
== RING_BUFFER_OVERWRITE
492 && subbuffer_id_is_noref(config
, id
));
493 backend_pages
= shmp(handle
, rpages
->shmp
);
496 str
= shmp_index(handle
, backend_pages
->p
, offset
& (chanb
->subbuf_size
- 1));
497 if (caa_unlikely(!str
))
499 string_len
= strnlen(str
, len
);
501 memcpy(dest
, str
, string_len
);
502 ((char *)dest
)[0] = 0;
504 return offset
- orig_offset
;
508 * lib_ring_buffer_read_offset_address - get address of a buffer location
509 * @bufb : buffer backend
510 * @offset : offset within the buffer.
512 * Return the address where a given offset is located (for read).
513 * Should be used to get the current subbuffer header pointer. Given we know
514 * it's never on a page boundary, it's safe to read/write directly
515 * from/to this address, as long as the read/write is never bigger than
518 void *lib_ring_buffer_read_offset_address(struct lttng_ust_lib_ring_buffer_backend
*bufb
,
520 struct lttng_ust_shm_handle
*handle
)
522 struct lttng_ust_lib_ring_buffer_backend_pages_shmp
*rpages
;
523 struct lttng_ust_lib_ring_buffer_backend_pages
*backend_pages
;
524 struct channel_backend
*chanb
;
525 const struct lttng_ust_lib_ring_buffer_config
*config
;
526 unsigned long sb_bindex
, id
;
528 chanb
= &shmp(handle
, bufb
->chan
)->backend
;
531 config
= &chanb
->config
;
532 offset
&= chanb
->buf_size
- 1;
533 id
= bufb
->buf_rsb
.id
;
534 sb_bindex
= subbuffer_id_get_index(config
, id
);
535 rpages
= shmp_index(handle
, bufb
->array
, sb_bindex
);
538 CHAN_WARN_ON(chanb
, config
->mode
== RING_BUFFER_OVERWRITE
539 && subbuffer_id_is_noref(config
, id
));
540 backend_pages
= shmp(handle
, rpages
->shmp
);
543 return shmp_index(handle
, backend_pages
->p
, offset
& (chanb
->subbuf_size
- 1));
547 * lib_ring_buffer_offset_address - get address of a location within the buffer
548 * @bufb : buffer backend
549 * @offset : offset within the buffer.
551 * Return the address where a given offset is located.
552 * Should be used to get the current subbuffer header pointer. Given we know
553 * it's always at the beginning of a page, it's safe to write directly to this
554 * address, as long as the write is never bigger than a page size.
556 void *lib_ring_buffer_offset_address(struct lttng_ust_lib_ring_buffer_backend
*bufb
,
558 struct lttng_ust_shm_handle
*handle
)
561 struct lttng_ust_lib_ring_buffer_backend_pages_shmp
*rpages
;
562 struct lttng_ust_lib_ring_buffer_backend_pages
*backend_pages
;
563 struct channel_backend
*chanb
;
564 const struct lttng_ust_lib_ring_buffer_config
*config
;
565 unsigned long sb_bindex
, id
;
566 struct lttng_ust_lib_ring_buffer_backend_subbuffer
*sb
;
568 chanb
= &shmp(handle
, bufb
->chan
)->backend
;
571 config
= &chanb
->config
;
572 offset
&= chanb
->buf_size
- 1;
573 sbidx
= offset
>> chanb
->subbuf_size_order
;
574 sb
= shmp_index(handle
, bufb
->buf_wsb
, sbidx
);
578 sb_bindex
= subbuffer_id_get_index(config
, id
);
579 rpages
= shmp_index(handle
, bufb
->array
, sb_bindex
);
582 CHAN_WARN_ON(chanb
, config
->mode
== RING_BUFFER_OVERWRITE
583 && subbuffer_id_is_noref(config
, id
));
584 backend_pages
= shmp(handle
, rpages
->shmp
);
587 return shmp_index(handle
, backend_pages
->p
, offset
& (chanb
->subbuf_size
- 1));