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
22 #include <urcu/arch.h>
25 #include <lttng/ringbuffer-config.h>
33 * lib_ring_buffer_backend_allocate - allocate a channel buffer
34 * @config: ring buffer instance configuration
35 * @buf: the buffer struct
36 * @size: total size of the buffer
37 * @num_subbuf: number of subbuffers
38 * @extra_reader_sb: need extra subbuffer for reader
41 int lib_ring_buffer_backend_allocate(const struct lttng_ust_lib_ring_buffer_config
*config
,
42 struct lttng_ust_lib_ring_buffer_backend
*bufb
,
43 size_t size
, size_t num_subbuf
,
45 struct lttng_ust_shm_handle
*handle
,
46 struct shm_object
*shmobj
)
48 struct channel_backend
*chanb
= &shmp(handle
, bufb
->chan
)->backend
;
49 unsigned long subbuf_size
, mmap_offset
= 0;
50 unsigned long num_subbuf_alloc
;
53 subbuf_size
= chanb
->subbuf_size
;
54 num_subbuf_alloc
= num_subbuf
;
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 shmp_index(handle
, bufb
->buf_wsb
, i
)->id
= subbuffer_id(config
, 0, 1, i
);
96 /* Assign read-side subbuffer table */
98 bufb
->buf_rsb
.id
= subbuffer_id(config
, 0, 1,
99 num_subbuf_alloc
- 1);
101 bufb
->buf_rsb
.id
= subbuffer_id(config
, 0, 1, 0);
103 /* Assign pages to page index */
104 for (i
= 0; i
< num_subbuf_alloc
; i
++) {
107 ref
.index
= bufb
->memory_map
._ref
.index
;
108 ref
.offset
= bufb
->memory_map
._ref
.offset
;
109 ref
.offset
+= i
* subbuf_size
;
111 set_shmp(shmp(handle
, shmp_index(handle
, bufb
->array
, i
)->shmp
)->p
,
113 if (config
->output
== RING_BUFFER_MMAP
) {
114 shmp(handle
, shmp_index(handle
, bufb
->array
, i
)->shmp
)->mmap_offset
= mmap_offset
;
115 mmap_offset
+= subbuf_size
;
122 /* bufb->array[i] will be freed by shm teardown */
124 /* bufb->array will be freed by shm teardown */
129 int lib_ring_buffer_backend_create(struct lttng_ust_lib_ring_buffer_backend
*bufb
,
130 struct channel_backend
*chanb
, int cpu
,
131 struct lttng_ust_shm_handle
*handle
,
132 struct shm_object
*shmobj
)
134 const struct lttng_ust_lib_ring_buffer_config
*config
= &chanb
->config
;
136 set_shmp(bufb
->chan
, handle
->chan
._ref
);
139 return lib_ring_buffer_backend_allocate(config
, bufb
, chanb
->buf_size
,
141 chanb
->extra_reader_sb
,
145 void lib_ring_buffer_backend_reset(struct lttng_ust_lib_ring_buffer_backend
*bufb
,
146 struct lttng_ust_shm_handle
*handle
)
148 struct channel_backend
*chanb
= &shmp(handle
, bufb
->chan
)->backend
;
149 const struct lttng_ust_lib_ring_buffer_config
*config
= &chanb
->config
;
150 unsigned long num_subbuf_alloc
;
153 num_subbuf_alloc
= chanb
->num_subbuf
;
154 if (chanb
->extra_reader_sb
)
157 for (i
= 0; i
< chanb
->num_subbuf
; i
++)
158 shmp_index(handle
, bufb
->buf_wsb
, i
)->id
= subbuffer_id(config
, 0, 1, i
);
159 if (chanb
->extra_reader_sb
)
160 bufb
->buf_rsb
.id
= subbuffer_id(config
, 0, 1,
161 num_subbuf_alloc
- 1);
163 bufb
->buf_rsb
.id
= subbuffer_id(config
, 0, 1, 0);
165 for (i
= 0; i
< num_subbuf_alloc
; i
++) {
166 /* Don't reset mmap_offset */
167 v_set(config
, &shmp(handle
, shmp_index(handle
, bufb
->array
, i
)->shmp
)->records_commit
, 0);
168 v_set(config
, &shmp(handle
, shmp_index(handle
, bufb
->array
, i
)->shmp
)->records_unread
, 0);
169 shmp(handle
, shmp_index(handle
, bufb
->array
, i
)->shmp
)->data_size
= 0;
170 /* Don't reset backend page and virt addresses */
172 /* Don't reset num_pages_per_subbuf, cpu, allocated */
173 v_set(config
, &bufb
->records_read
, 0);
177 * The frontend is responsible for also calling ring_buffer_backend_reset for
178 * each buffer when calling channel_backend_reset.
180 void channel_backend_reset(struct channel_backend
*chanb
)
182 struct channel
*chan
= caa_container_of(chanb
, struct channel
, backend
);
183 const struct lttng_ust_lib_ring_buffer_config
*config
= &chanb
->config
;
186 * Don't reset buf_size, subbuf_size, subbuf_size_order,
187 * num_subbuf_order, buf_size_order, extra_reader_sb, num_subbuf,
188 * priv, notifiers, config, cpumask and name.
190 chanb
->start_tsc
= config
->cb
.ring_buffer_clock_read(chan
);
194 * channel_backend_init - initialize a channel backend
195 * @chanb: channel backend
196 * @name: channel name
197 * @config: client ring buffer configuration
198 * @parent: dentry of parent directory, %NULL for root directory
199 * @subbuf_size: size of sub-buffers (> PAGE_SIZE, power of 2)
200 * @num_subbuf: number of sub-buffers (power of 2)
201 * @lttng_ust_shm_handle: shared memory handle
203 * Returns channel pointer if successful, %NULL otherwise.
205 * Creates per-cpu channel buffers using the sizes and attributes
206 * specified. The created channel buffer files will be named
207 * name_0...name_N-1. File permissions will be %S_IRUSR.
209 * Called with CPU hotplug disabled.
211 int channel_backend_init(struct channel_backend
*chanb
,
213 const struct lttng_ust_lib_ring_buffer_config
*config
,
214 size_t subbuf_size
, size_t num_subbuf
,
215 struct lttng_ust_shm_handle
*handle
)
217 struct channel
*chan
= caa_container_of(chanb
, struct channel
, backend
);
220 size_t shmsize
= 0, num_subbuf_alloc
;
225 /* Check that the subbuffer size is larger than a page. */
226 if (subbuf_size
< PAGE_SIZE
)
230 * Make sure the number of subbuffers and subbuffer size are
231 * power of 2, and nonzero.
233 if (!subbuf_size
|| (subbuf_size
& (subbuf_size
- 1)))
235 if (!num_subbuf
|| (num_subbuf
& (num_subbuf
- 1)))
238 ret
= subbuffer_id_check_index(config
, num_subbuf
);
242 chanb
->buf_size
= num_subbuf
* subbuf_size
;
243 chanb
->subbuf_size
= subbuf_size
;
244 chanb
->buf_size_order
= get_count_order(chanb
->buf_size
);
245 chanb
->subbuf_size_order
= get_count_order(subbuf_size
);
246 chanb
->num_subbuf_order
= get_count_order(num_subbuf
);
247 chanb
->extra_reader_sb
=
248 (config
->mode
== RING_BUFFER_OVERWRITE
) ? 1 : 0;
249 chanb
->num_subbuf
= num_subbuf
;
250 strncpy(chanb
->name
, name
, NAME_MAX
);
251 chanb
->name
[NAME_MAX
- 1] = '\0';
252 memcpy(&chanb
->config
, config
, sizeof(*config
));
254 /* Per-cpu buffer size: control (prior to backend) */
255 shmsize
= offset_align(shmsize
, __alignof__(struct lttng_ust_lib_ring_buffer
));
256 shmsize
+= sizeof(struct lttng_ust_lib_ring_buffer
);
258 /* Per-cpu buffer size: backend */
259 /* num_subbuf + 1 is the worse case */
260 num_subbuf_alloc
= num_subbuf
+ 1;
261 shmsize
+= offset_align(shmsize
, __alignof__(struct lttng_ust_lib_ring_buffer_backend_pages_shmp
));
262 shmsize
+= sizeof(struct lttng_ust_lib_ring_buffer_backend_pages_shmp
) * num_subbuf_alloc
;
263 shmsize
+= offset_align(shmsize
, PAGE_SIZE
);
264 shmsize
+= subbuf_size
* num_subbuf_alloc
;
265 shmsize
+= offset_align(shmsize
, __alignof__(struct lttng_ust_lib_ring_buffer_backend_pages
));
266 shmsize
+= sizeof(struct lttng_ust_lib_ring_buffer_backend_pages
) * num_subbuf_alloc
;
267 shmsize
+= offset_align(shmsize
, __alignof__(struct lttng_ust_lib_ring_buffer_backend_subbuffer
));
268 shmsize
+= sizeof(struct lttng_ust_lib_ring_buffer_backend_subbuffer
) * num_subbuf
;
269 /* Per-cpu buffer size: control (after backend) */
270 shmsize
+= offset_align(shmsize
, __alignof__(struct commit_counters_hot
));
271 shmsize
+= sizeof(struct commit_counters_hot
) * num_subbuf
;
272 shmsize
+= offset_align(shmsize
, __alignof__(struct commit_counters_cold
));
273 shmsize
+= sizeof(struct commit_counters_cold
) * num_subbuf
;
275 if (config
->alloc
== RING_BUFFER_ALLOC_PER_CPU
) {
276 struct lttng_ust_lib_ring_buffer
*buf
;
278 * We need to allocate for all possible cpus.
280 for_each_possible_cpu(i
) {
281 struct shm_object
*shmobj
;
283 shmobj
= shm_object_table_append(handle
->table
, shmsize
);
286 align_shm(shmobj
, __alignof__(struct lttng_ust_lib_ring_buffer
));
287 set_shmp(chanb
->buf
[i
].shmp
, zalloc_shm(shmobj
, sizeof(struct lttng_ust_lib_ring_buffer
)));
288 buf
= shmp(handle
, chanb
->buf
[i
].shmp
);
291 set_shmp(buf
->self
, chanb
->buf
[i
].shmp
._ref
);
292 ret
= lib_ring_buffer_create(buf
, chanb
, i
,
295 goto free_bufs
; /* cpu hotplug locked */
298 struct shm_object
*shmobj
;
299 struct lttng_ust_lib_ring_buffer
*buf
;
301 shmobj
= shm_object_table_append(handle
->table
, shmsize
);
304 align_shm(shmobj
, __alignof__(struct lttng_ust_lib_ring_buffer
));
305 set_shmp(chanb
->buf
[0].shmp
, zalloc_shm(shmobj
, sizeof(struct lttng_ust_lib_ring_buffer
)));
306 buf
= shmp(handle
, chanb
->buf
[0].shmp
);
309 set_shmp(buf
->self
, chanb
->buf
[0].shmp
._ref
);
310 ret
= lib_ring_buffer_create(buf
, chanb
, -1,
315 chanb
->start_tsc
= config
->cb
.ring_buffer_clock_read(chan
);
320 /* We only free the buffer data upon shm teardown */
326 * channel_backend_free - destroy the channel
329 * Destroy all channel buffers and frees the channel.
331 void channel_backend_free(struct channel_backend
*chanb
,
332 struct lttng_ust_shm_handle
*handle
)
334 /* SHM teardown takes care of everything */
338 * lib_ring_buffer_read - read data from ring_buffer_buffer.
339 * @bufb : buffer backend
340 * @offset : offset within the buffer
341 * @dest : destination address
342 * @len : length to copy to destination
344 * Should be protected by get_subbuf/put_subbuf.
345 * Returns the length copied.
347 size_t lib_ring_buffer_read(struct lttng_ust_lib_ring_buffer_backend
*bufb
, size_t offset
,
348 void *dest
, size_t len
, struct lttng_ust_shm_handle
*handle
)
350 struct channel_backend
*chanb
= &shmp(handle
, bufb
->chan
)->backend
;
351 const struct lttng_ust_lib_ring_buffer_config
*config
= &chanb
->config
;
353 struct lttng_ust_lib_ring_buffer_backend_pages_shmp
*rpages
;
354 unsigned long sb_bindex
, id
;
357 offset
&= chanb
->buf_size
- 1;
359 if (caa_unlikely(!len
))
361 id
= bufb
->buf_rsb
.id
;
362 sb_bindex
= subbuffer_id_get_index(config
, id
);
363 rpages
= shmp_index(handle
, bufb
->array
, sb_bindex
);
365 * Underlying layer should never ask for reads across
368 CHAN_WARN_ON(chanb
, offset
>= chanb
->buf_size
);
369 CHAN_WARN_ON(chanb
, config
->mode
== RING_BUFFER_OVERWRITE
370 && subbuffer_id_is_noref(config
, id
));
371 memcpy(dest
, shmp_index(handle
, shmp(handle
, rpages
->shmp
)->p
, offset
& (chanb
->subbuf_size
- 1)), len
);
376 * lib_ring_buffer_read_cstr - read a C-style string from ring_buffer.
377 * @bufb : buffer backend
378 * @offset : offset within the buffer
379 * @dest : destination address
380 * @len : destination's length
382 * return string's length
383 * Should be protected by get_subbuf/put_subbuf.
385 int lib_ring_buffer_read_cstr(struct lttng_ust_lib_ring_buffer_backend
*bufb
, size_t offset
,
386 void *dest
, size_t len
, struct lttng_ust_shm_handle
*handle
)
388 struct channel_backend
*chanb
= &shmp(handle
, bufb
->chan
)->backend
;
389 const struct lttng_ust_lib_ring_buffer_config
*config
= &chanb
->config
;
390 ssize_t string_len
, orig_offset
;
392 struct lttng_ust_lib_ring_buffer_backend_pages_shmp
*rpages
;
393 unsigned long sb_bindex
, id
;
395 offset
&= chanb
->buf_size
- 1;
396 orig_offset
= offset
;
397 id
= bufb
->buf_rsb
.id
;
398 sb_bindex
= subbuffer_id_get_index(config
, id
);
399 rpages
= shmp_index(handle
, bufb
->array
, sb_bindex
);
401 * Underlying layer should never ask for reads across
404 CHAN_WARN_ON(chanb
, offset
>= chanb
->buf_size
);
405 CHAN_WARN_ON(chanb
, config
->mode
== RING_BUFFER_OVERWRITE
406 && subbuffer_id_is_noref(config
, id
));
407 str
= shmp_index(handle
, shmp(handle
, rpages
->shmp
)->p
, offset
& (chanb
->subbuf_size
- 1));
408 string_len
= strnlen(str
, len
);
410 memcpy(dest
, str
, string_len
);
411 ((char *)dest
)[0] = 0;
413 return offset
- orig_offset
;
417 * lib_ring_buffer_read_offset_address - get address of a buffer location
418 * @bufb : buffer backend
419 * @offset : offset within the buffer.
421 * Return the address where a given offset is located (for read).
422 * Should be used to get the current subbuffer header pointer. Given we know
423 * it's never on a page boundary, it's safe to write directly to this address,
424 * as long as the write is never bigger than a page size.
426 void *lib_ring_buffer_read_offset_address(struct lttng_ust_lib_ring_buffer_backend
*bufb
,
428 struct lttng_ust_shm_handle
*handle
)
430 struct lttng_ust_lib_ring_buffer_backend_pages_shmp
*rpages
;
431 struct channel_backend
*chanb
= &shmp(handle
, bufb
->chan
)->backend
;
432 const struct lttng_ust_lib_ring_buffer_config
*config
= &chanb
->config
;
433 unsigned long sb_bindex
, id
;
435 offset
&= chanb
->buf_size
- 1;
436 id
= bufb
->buf_rsb
.id
;
437 sb_bindex
= subbuffer_id_get_index(config
, id
);
438 rpages
= shmp_index(handle
, bufb
->array
, sb_bindex
);
439 CHAN_WARN_ON(chanb
, config
->mode
== RING_BUFFER_OVERWRITE
440 && subbuffer_id_is_noref(config
, id
));
441 return shmp_index(handle
, shmp(handle
, rpages
->shmp
)->p
, offset
& (chanb
->subbuf_size
- 1));
445 * lib_ring_buffer_offset_address - get address of a location within the buffer
446 * @bufb : buffer backend
447 * @offset : offset within the buffer.
449 * Return the address where a given offset is located.
450 * Should be used to get the current subbuffer header pointer. Given we know
451 * it's always at the beginning of a page, it's safe to write directly to this
452 * address, as long as the write is never bigger than a page size.
454 void *lib_ring_buffer_offset_address(struct lttng_ust_lib_ring_buffer_backend
*bufb
,
456 struct lttng_ust_shm_handle
*handle
)
459 struct lttng_ust_lib_ring_buffer_backend_pages_shmp
*rpages
;
460 struct channel_backend
*chanb
= &shmp(handle
, bufb
->chan
)->backend
;
461 const struct lttng_ust_lib_ring_buffer_config
*config
= &chanb
->config
;
462 unsigned long sb_bindex
, id
;
464 offset
&= chanb
->buf_size
- 1;
465 sbidx
= offset
>> chanb
->subbuf_size_order
;
466 id
= shmp_index(handle
, bufb
->buf_wsb
, sbidx
)->id
;
467 sb_bindex
= subbuffer_id_get_index(config
, id
);
468 rpages
= shmp_index(handle
, bufb
->array
, sb_bindex
);
469 CHAN_WARN_ON(chanb
, config
->mode
== RING_BUFFER_OVERWRITE
470 && subbuffer_id_is_noref(config
, id
));
471 return shmp_index(handle
, shmp(handle
, rpages
->shmp
)->p
, offset
& (chanb
->subbuf_size
- 1));