2 * ring_buffer_backend.c
4 * Copyright (C) 2005-2012 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
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.
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.
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
23 #include <urcu/arch.h>
26 #include <lttng/ringbuffer-config.h>
34 * lib_ring_buffer_backend_allocate - allocate a channel buffer
35 * @config: ring buffer instance configuration
36 * @buf: the buffer struct
37 * @size: total size of the buffer
38 * @num_subbuf: number of subbuffers
39 * @extra_reader_sb: need extra subbuffer for reader
42 int lib_ring_buffer_backend_allocate(const struct lttng_ust_lib_ring_buffer_config
*config
,
43 struct lttng_ust_lib_ring_buffer_backend
*bufb
,
44 size_t size
, size_t num_subbuf
,
46 struct lttng_ust_shm_handle
*handle
,
47 struct shm_object
*shmobj
)
49 struct channel_backend
*chanb
;
50 unsigned long subbuf_size
, mmap_offset
= 0;
51 unsigned long num_subbuf_alloc
;
55 chanb
= &shmp(handle
, bufb
->chan
)->backend
;
59 subbuf_size
= chanb
->subbuf_size
;
60 num_subbuf_alloc
= num_subbuf
;
65 page_size
= sysconf(_SC_PAGE_SIZE
);
70 align_shm(shmobj
, __alignof__(struct lttng_ust_lib_ring_buffer_backend_pages_shmp
));
71 set_shmp(bufb
->array
, zalloc_shm(shmobj
,
72 sizeof(struct lttng_ust_lib_ring_buffer_backend_pages_shmp
) * num_subbuf_alloc
));
73 if (caa_unlikely(!shmp(handle
, bufb
->array
)))
77 * This is the largest element (the buffer pages) which needs to
78 * be aligned on page size.
80 align_shm(shmobj
, page_size
);
81 set_shmp(bufb
->memory_map
, zalloc_shm(shmobj
,
82 subbuf_size
* num_subbuf_alloc
));
83 if (caa_unlikely(!shmp(handle
, bufb
->memory_map
)))
84 goto memory_map_error
;
86 /* Allocate backend pages array elements */
87 for (i
= 0; i
< num_subbuf_alloc
; i
++) {
88 align_shm(shmobj
, __alignof__(struct lttng_ust_lib_ring_buffer_backend_pages
));
89 set_shmp(shmp_index(handle
, bufb
->array
, i
)->shmp
,
91 sizeof(struct lttng_ust_lib_ring_buffer_backend_pages
)));
92 if (!shmp(handle
, shmp_index(handle
, bufb
->array
, i
)->shmp
))
96 /* Allocate write-side subbuffer table */
97 align_shm(shmobj
, __alignof__(struct lttng_ust_lib_ring_buffer_backend_subbuffer
));
98 set_shmp(bufb
->buf_wsb
, zalloc_shm(shmobj
,
99 sizeof(struct lttng_ust_lib_ring_buffer_backend_subbuffer
)
101 if (caa_unlikely(!shmp(handle
, bufb
->buf_wsb
)))
104 for (i
= 0; i
< num_subbuf
; i
++) {
105 struct lttng_ust_lib_ring_buffer_backend_subbuffer
*sb
;
107 sb
= shmp_index(handle
, bufb
->buf_wsb
, i
);
110 sb
->id
= subbuffer_id(config
, 0, 1, i
);
113 /* Assign read-side subbuffer table */
115 bufb
->buf_rsb
.id
= subbuffer_id(config
, 0, 1,
116 num_subbuf_alloc
- 1);
118 bufb
->buf_rsb
.id
= subbuffer_id(config
, 0, 1, 0);
120 /* Assign pages to page index */
121 for (i
= 0; i
< num_subbuf_alloc
; i
++) {
122 struct lttng_ust_lib_ring_buffer_backend_pages_shmp
*sbp
;
123 struct lttng_ust_lib_ring_buffer_backend_pages
*pages
;
126 ref
.index
= bufb
->memory_map
._ref
.index
;
127 ref
.offset
= bufb
->memory_map
._ref
.offset
;
128 ref
.offset
+= i
* subbuf_size
;
130 sbp
= shmp_index(handle
, bufb
->array
, i
);
133 pages
= shmp(handle
, sbp
->shmp
);
136 set_shmp(pages
->p
, ref
);
137 if (config
->output
== RING_BUFFER_MMAP
) {
138 pages
->mmap_offset
= mmap_offset
;
139 mmap_offset
+= subbuf_size
;
146 /* bufb->array[i] will be freed by shm teardown */
148 /* bufb->array will be freed by shm teardown */
154 int lib_ring_buffer_backend_create(struct lttng_ust_lib_ring_buffer_backend
*bufb
,
155 struct channel_backend
*chanb
, int cpu
,
156 struct lttng_ust_shm_handle
*handle
,
157 struct shm_object
*shmobj
)
159 const struct lttng_ust_lib_ring_buffer_config
*config
= &chanb
->config
;
161 set_shmp(bufb
->chan
, handle
->chan
._ref
);
164 return lib_ring_buffer_backend_allocate(config
, bufb
, chanb
->buf_size
,
166 chanb
->extra_reader_sb
,
170 void lib_ring_buffer_backend_reset(struct lttng_ust_lib_ring_buffer_backend
*bufb
,
171 struct lttng_ust_shm_handle
*handle
)
173 struct channel_backend
*chanb
;
174 const struct lttng_ust_lib_ring_buffer_config
*config
;
175 unsigned long num_subbuf_alloc
;
178 chanb
= &shmp(handle
, bufb
->chan
)->backend
;
181 config
= &chanb
->config
;
183 num_subbuf_alloc
= chanb
->num_subbuf
;
184 if (chanb
->extra_reader_sb
)
187 for (i
= 0; i
< chanb
->num_subbuf
; i
++) {
188 struct lttng_ust_lib_ring_buffer_backend_subbuffer
*sb
;
190 sb
= shmp_index(handle
, bufb
->buf_wsb
, i
);
193 sb
->id
= subbuffer_id(config
, 0, 1, i
);
195 if (chanb
->extra_reader_sb
)
196 bufb
->buf_rsb
.id
= subbuffer_id(config
, 0, 1,
197 num_subbuf_alloc
- 1);
199 bufb
->buf_rsb
.id
= subbuffer_id(config
, 0, 1, 0);
201 for (i
= 0; i
< num_subbuf_alloc
; i
++) {
202 struct lttng_ust_lib_ring_buffer_backend_pages_shmp
*sbp
;
203 struct lttng_ust_lib_ring_buffer_backend_pages
*pages
;
205 sbp
= shmp_index(handle
, bufb
->array
, i
);
208 pages
= shmp(handle
, sbp
->shmp
);
211 /* Don't reset mmap_offset */
212 v_set(config
, &pages
->records_commit
, 0);
213 v_set(config
, &pages
->records_unread
, 0);
214 pages
->data_size
= 0;
215 /* Don't reset backend page and virt addresses */
217 /* Don't reset num_pages_per_subbuf, cpu, allocated */
218 v_set(config
, &bufb
->records_read
, 0);
222 * The frontend is responsible for also calling ring_buffer_backend_reset for
223 * each buffer when calling channel_backend_reset.
225 void channel_backend_reset(struct channel_backend
*chanb
)
227 struct channel
*chan
= caa_container_of(chanb
, struct channel
, backend
);
228 const struct lttng_ust_lib_ring_buffer_config
*config
= &chanb
->config
;
231 * Don't reset buf_size, subbuf_size, subbuf_size_order,
232 * num_subbuf_order, buf_size_order, extra_reader_sb, num_subbuf,
233 * priv, notifiers, config, cpumask and name.
235 chanb
->start_tsc
= config
->cb
.ring_buffer_clock_read(chan
);
239 * channel_backend_init - initialize a channel backend
240 * @chanb: channel backend
241 * @name: channel name
242 * @config: client ring buffer configuration
243 * @parent: dentry of parent directory, %NULL for root directory
244 * @subbuf_size: size of sub-buffers (> page size, power of 2)
245 * @num_subbuf: number of sub-buffers (power of 2)
246 * @lttng_ust_shm_handle: shared memory handle
248 * Returns channel pointer if successful, %NULL otherwise.
250 * Creates per-cpu channel buffers using the sizes and attributes
251 * specified. The created channel buffer files will be named
252 * name_0...name_N-1. File permissions will be %S_IRUSR.
254 * Called with CPU hotplug disabled.
256 int channel_backend_init(struct channel_backend
*chanb
,
258 const struct lttng_ust_lib_ring_buffer_config
*config
,
259 size_t subbuf_size
, size_t num_subbuf
,
260 struct lttng_ust_shm_handle
*handle
)
262 struct channel
*chan
= caa_container_of(chanb
, struct channel
, backend
);
265 size_t shmsize
= 0, num_subbuf_alloc
;
271 page_size
= sysconf(_SC_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
= offset_align(shmsize
, __alignof__(struct lttng_ust_lib_ring_buffer
));
312 shmsize
+= sizeof(struct lttng_ust_lib_ring_buffer
);
314 /* Per-cpu buffer size: backend */
315 /* num_subbuf + 1 is the worse case */
316 num_subbuf_alloc
= num_subbuf
+ 1;
317 shmsize
+= offset_align(shmsize
, __alignof__(struct lttng_ust_lib_ring_buffer_backend_pages_shmp
));
318 shmsize
+= sizeof(struct lttng_ust_lib_ring_buffer_backend_pages_shmp
) * num_subbuf_alloc
;
319 shmsize
+= offset_align(shmsize
, page_size
);
320 shmsize
+= subbuf_size
* num_subbuf_alloc
;
321 shmsize
+= offset_align(shmsize
, __alignof__(struct lttng_ust_lib_ring_buffer_backend_pages
));
322 shmsize
+= sizeof(struct lttng_ust_lib_ring_buffer_backend_pages
) * num_subbuf_alloc
;
323 shmsize
+= offset_align(shmsize
, __alignof__(struct lttng_ust_lib_ring_buffer_backend_subbuffer
));
324 shmsize
+= sizeof(struct lttng_ust_lib_ring_buffer_backend_subbuffer
) * num_subbuf
;
325 /* Per-cpu buffer size: control (after backend) */
326 shmsize
+= offset_align(shmsize
, __alignof__(struct commit_counters_hot
));
327 shmsize
+= sizeof(struct commit_counters_hot
) * num_subbuf
;
328 shmsize
+= offset_align(shmsize
, __alignof__(struct commit_counters_cold
));
329 shmsize
+= sizeof(struct commit_counters_cold
) * num_subbuf
;
331 if (config
->alloc
== RING_BUFFER_ALLOC_PER_CPU
) {
332 struct lttng_ust_lib_ring_buffer
*buf
;
334 * We need to allocate for all possible cpus.
336 for_each_possible_cpu(i
) {
337 struct shm_object
*shmobj
;
339 shmobj
= shm_object_table_alloc(handle
->table
, shmsize
,
343 align_shm(shmobj
, __alignof__(struct lttng_ust_lib_ring_buffer
));
344 set_shmp(chanb
->buf
[i
].shmp
, zalloc_shm(shmobj
, sizeof(struct lttng_ust_lib_ring_buffer
)));
345 buf
= shmp(handle
, chanb
->buf
[i
].shmp
);
348 set_shmp(buf
->self
, chanb
->buf
[i
].shmp
._ref
);
349 ret
= lib_ring_buffer_create(buf
, chanb
, i
,
352 goto free_bufs
; /* cpu hotplug locked */
355 struct shm_object
*shmobj
;
356 struct lttng_ust_lib_ring_buffer
*buf
;
358 shmobj
= shm_object_table_alloc(handle
->table
, shmsize
,
362 align_shm(shmobj
, __alignof__(struct lttng_ust_lib_ring_buffer
));
363 set_shmp(chanb
->buf
[0].shmp
, zalloc_shm(shmobj
, sizeof(struct lttng_ust_lib_ring_buffer
)));
364 buf
= shmp(handle
, chanb
->buf
[0].shmp
);
367 set_shmp(buf
->self
, chanb
->buf
[0].shmp
._ref
);
368 ret
= lib_ring_buffer_create(buf
, chanb
, -1,
373 chanb
->start_tsc
= config
->cb
.ring_buffer_clock_read(chan
);
378 /* We only free the buffer data upon shm teardown */
384 * channel_backend_free - destroy the channel
387 * Destroy all channel buffers and frees the channel.
389 void channel_backend_free(struct channel_backend
*chanb
,
390 struct lttng_ust_shm_handle
*handle
)
392 /* SHM teardown takes care of everything */
396 * lib_ring_buffer_read - read data from ring_buffer_buffer.
397 * @bufb : buffer backend
398 * @offset : offset within the buffer
399 * @dest : destination address
400 * @len : length to copy to destination
402 * Should be protected by get_subbuf/put_subbuf.
403 * Returns the length copied.
405 size_t lib_ring_buffer_read(struct lttng_ust_lib_ring_buffer_backend
*bufb
, size_t offset
,
406 void *dest
, size_t len
, struct lttng_ust_shm_handle
*handle
)
408 struct channel_backend
*chanb
;
409 const struct lttng_ust_lib_ring_buffer_config
*config
;
411 struct lttng_ust_lib_ring_buffer_backend_pages_shmp
*rpages
;
412 unsigned long sb_bindex
, id
;
415 chanb
= &shmp(handle
, bufb
->chan
)->backend
;
418 config
= &chanb
->config
;
420 offset
&= chanb
->buf_size
- 1;
422 if (caa_unlikely(!len
))
424 id
= bufb
->buf_rsb
.id
;
425 sb_bindex
= subbuffer_id_get_index(config
, id
);
426 rpages
= shmp_index(handle
, bufb
->array
, sb_bindex
);
428 * Underlying layer should never ask for reads across
431 CHAN_WARN_ON(chanb
, offset
>= chanb
->buf_size
);
432 CHAN_WARN_ON(chanb
, config
->mode
== RING_BUFFER_OVERWRITE
433 && subbuffer_id_is_noref(config
, id
));
434 src
= shmp_index(handle
, shmp(handle
, rpages
->shmp
)->p
,
435 offset
& (chanb
->subbuf_size
- 1));
436 if (caa_unlikely(!src
))
438 memcpy(dest
, src
, len
);
443 * lib_ring_buffer_read_cstr - read a C-style string from ring_buffer.
444 * @bufb : buffer backend
445 * @offset : offset within the buffer
446 * @dest : destination address
447 * @len : destination's length
449 * Return string's length, or -EINVAL on error.
450 * Should be protected by get_subbuf/put_subbuf.
451 * Destination length should be at least 1 to hold '\0'.
453 int lib_ring_buffer_read_cstr(struct lttng_ust_lib_ring_buffer_backend
*bufb
, size_t offset
,
454 void *dest
, size_t len
, struct lttng_ust_shm_handle
*handle
)
456 struct channel_backend
*chanb
;
457 const struct lttng_ust_lib_ring_buffer_config
*config
;
458 ssize_t string_len
, orig_offset
;
460 struct lttng_ust_lib_ring_buffer_backend_pages_shmp
*rpages
;
461 unsigned long sb_bindex
, id
;
463 chanb
= &shmp(handle
, bufb
->chan
)->backend
;
466 config
= &chanb
->config
;
467 if (caa_unlikely(!len
))
469 offset
&= chanb
->buf_size
- 1;
470 orig_offset
= offset
;
471 id
= bufb
->buf_rsb
.id
;
472 sb_bindex
= subbuffer_id_get_index(config
, id
);
473 rpages
= shmp_index(handle
, bufb
->array
, sb_bindex
);
475 * Underlying layer should never ask for reads across
478 CHAN_WARN_ON(chanb
, offset
>= chanb
->buf_size
);
479 CHAN_WARN_ON(chanb
, config
->mode
== RING_BUFFER_OVERWRITE
480 && subbuffer_id_is_noref(config
, id
));
481 str
= shmp_index(handle
, shmp(handle
, rpages
->shmp
)->p
, offset
& (chanb
->subbuf_size
- 1));
482 if (caa_unlikely(!str
))
484 string_len
= strnlen(str
, len
);
486 memcpy(dest
, str
, string_len
);
487 ((char *)dest
)[0] = 0;
489 return offset
- orig_offset
;
493 * lib_ring_buffer_read_offset_address - get address of a buffer location
494 * @bufb : buffer backend
495 * @offset : offset within the buffer.
497 * Return the address where a given offset is located (for read).
498 * Should be used to get the current subbuffer header pointer. Given we know
499 * it's never on a page boundary, it's safe to read/write directly
500 * from/to this address, as long as the read/write is never bigger than
503 void *lib_ring_buffer_read_offset_address(struct lttng_ust_lib_ring_buffer_backend
*bufb
,
505 struct lttng_ust_shm_handle
*handle
)
507 struct lttng_ust_lib_ring_buffer_backend_pages_shmp
*rpages
;
508 struct channel_backend
*chanb
;
509 const struct lttng_ust_lib_ring_buffer_config
*config
;
510 unsigned long sb_bindex
, id
;
512 chanb
= &shmp(handle
, bufb
->chan
)->backend
;
515 config
= &chanb
->config
;
516 offset
&= chanb
->buf_size
- 1;
517 id
= bufb
->buf_rsb
.id
;
518 sb_bindex
= subbuffer_id_get_index(config
, id
);
519 rpages
= shmp_index(handle
, bufb
->array
, sb_bindex
);
520 CHAN_WARN_ON(chanb
, config
->mode
== RING_BUFFER_OVERWRITE
521 && subbuffer_id_is_noref(config
, id
));
522 return shmp_index(handle
, shmp(handle
, rpages
->shmp
)->p
, offset
& (chanb
->subbuf_size
- 1));
526 * lib_ring_buffer_offset_address - get address of a location within the buffer
527 * @bufb : buffer backend
528 * @offset : offset within the buffer.
530 * Return the address where a given offset is located.
531 * Should be used to get the current subbuffer header pointer. Given we know
532 * it's always at the beginning of a page, it's safe to write directly to this
533 * address, as long as the write is never bigger than a page size.
535 void *lib_ring_buffer_offset_address(struct lttng_ust_lib_ring_buffer_backend
*bufb
,
537 struct lttng_ust_shm_handle
*handle
)
540 struct lttng_ust_lib_ring_buffer_backend_pages_shmp
*rpages
;
541 struct channel_backend
*chanb
;
542 const struct lttng_ust_lib_ring_buffer_config
*config
;
543 unsigned long sb_bindex
, id
;
544 struct lttng_ust_lib_ring_buffer_backend_subbuffer
*sb
;
546 chanb
= &shmp(handle
, bufb
->chan
)->backend
;
549 config
= &chanb
->config
;
550 offset
&= chanb
->buf_size
- 1;
551 sbidx
= offset
>> chanb
->subbuf_size_order
;
552 sb
= shmp_index(handle
, bufb
->buf_wsb
, sbidx
);
556 sb_bindex
= subbuffer_id_get_index(config
, id
);
557 rpages
= shmp_index(handle
, bufb
->array
, sb_bindex
);
558 CHAN_WARN_ON(chanb
, config
->mode
== RING_BUFFER_OVERWRITE
559 && subbuffer_id_is_noref(config
, id
));
560 return shmp_index(handle
, shmp(handle
, rpages
->shmp
)->p
, offset
& (chanb
->subbuf_size
- 1));