Allow consumer to flush buffers, fix alignment bug
[lttng-ust.git] / libringbuffer / ring_buffer_frontend.c
1 /*
2 * ring_buffer_frontend.c
3 *
4 * (C) Copyright 2005-2010 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
5 *
6 * Ring buffer wait-free buffer synchronization. Producer-consumer and flight
7 * recorder (overwrite) modes. See thesis:
8 *
9 * Desnoyers, Mathieu (2009), "Low-Impact Operating System Tracing", Ph.D.
10 * dissertation, Ecole Polytechnique de Montreal.
11 * http://www.lttng.org/pub/thesis/desnoyers-dissertation-2009-12.pdf
12 *
13 * - Algorithm presentation in Chapter 5:
14 * "Lockless Multi-Core High-Throughput Buffering".
15 * - Algorithm formal verification in Section 8.6:
16 * "Formal verification of LTTng"
17 *
18 * Author:
19 * Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
20 *
21 * Inspired from LTT and RelayFS:
22 * Karim Yaghmour <karim@opersys.com>
23 * Tom Zanussi <zanussi@us.ibm.com>
24 * Bob Wisniewski <bob@watson.ibm.com>
25 * And from K42 :
26 * Bob Wisniewski <bob@watson.ibm.com>
27 *
28 * Buffer reader semantic :
29 *
30 * - get_subbuf_size
31 * while buffer is not finalized and empty
32 * - get_subbuf
33 * - if return value != 0, continue
34 * - splice one subbuffer worth of data to a pipe
35 * - splice the data from pipe to disk/network
36 * - put_subbuf
37 *
38 * Dual LGPL v2.1/GPL v2 license.
39 */
40
41 #include <sys/types.h>
42 #include <sys/mman.h>
43 #include <sys/stat.h>
44 #include <fcntl.h>
45 #include <urcu/compiler.h>
46 #include <urcu/ref.h>
47
48 #include "smp.h"
49 #include <lttng/ringbuffer-config.h>
50 #include "backend.h"
51 #include "frontend.h"
52 #include "shm.h"
53
54 #ifndef max
55 #define max(a, b) ((a) > (b) ? (a) : (b))
56 #endif
57
58 /*
59 * Use POSIX SHM: shm_open(3) and shm_unlink(3).
60 * close(2) to close the fd returned by shm_open.
61 * shm_unlink releases the shared memory object name.
62 * ftruncate(2) sets the size of the memory object.
63 * mmap/munmap maps the shared memory obj to a virtual address in the
64 * calling proceess (should be done both in libust and consumer).
65 * See shm_overview(7) for details.
66 * Pass file descriptor returned by shm_open(3) to ltt-sessiond through
67 * a UNIX socket.
68 *
69 * Since we don't need to access the object using its name, we can
70 * immediately shm_unlink(3) it, and only keep the handle with its file
71 * descriptor.
72 */
73
74 /*
75 * Internal structure representing offsets to use at a sub-buffer switch.
76 */
77 struct switch_offsets {
78 unsigned long begin, end, old;
79 size_t pre_header_padding, size;
80 unsigned int switch_new_start:1, switch_new_end:1, switch_old_start:1,
81 switch_old_end:1;
82 };
83
84 __thread unsigned int lib_ring_buffer_nesting;
85
86 static
87 void lib_ring_buffer_print_errors(struct channel *chan,
88 struct lttng_ust_lib_ring_buffer *buf, int cpu,
89 struct lttng_ust_shm_handle *handle);
90
91 /*
92 * Must be called under cpu hotplug protection.
93 */
94 void lib_ring_buffer_free(struct lttng_ust_lib_ring_buffer *buf,
95 struct lttng_ust_shm_handle *handle)
96 {
97 struct channel *chan = shmp(handle, buf->backend.chan);
98
99 lib_ring_buffer_print_errors(chan, buf, buf->backend.cpu, handle);
100 /* buf->commit_hot will be freed by shm teardown */
101 /* buf->commit_cold will be freed by shm teardown */
102
103 lib_ring_buffer_backend_free(&buf->backend);
104 }
105
106 /**
107 * lib_ring_buffer_reset - Reset ring buffer to initial values.
108 * @buf: Ring buffer.
109 *
110 * Effectively empty the ring buffer. Should be called when the buffer is not
111 * used for writing. The ring buffer can be opened for reading, but the reader
112 * should not be using the iterator concurrently with reset. The previous
113 * current iterator record is reset.
114 */
115 void lib_ring_buffer_reset(struct lttng_ust_lib_ring_buffer *buf,
116 struct lttng_ust_shm_handle *handle)
117 {
118 struct channel *chan = shmp(handle, buf->backend.chan);
119 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
120 unsigned int i;
121
122 /*
123 * Reset iterator first. It will put the subbuffer if it currently holds
124 * it.
125 */
126 v_set(config, &buf->offset, 0);
127 for (i = 0; i < chan->backend.num_subbuf; i++) {
128 v_set(config, &shmp_index(handle, buf->commit_hot, i)->cc, 0);
129 v_set(config, &shmp_index(handle, buf->commit_hot, i)->seq, 0);
130 v_set(config, &shmp_index(handle, buf->commit_cold, i)->cc_sb, 0);
131 }
132 uatomic_set(&buf->consumed, 0);
133 uatomic_set(&buf->record_disabled, 0);
134 v_set(config, &buf->last_tsc, 0);
135 lib_ring_buffer_backend_reset(&buf->backend, handle);
136 /* Don't reset number of active readers */
137 v_set(config, &buf->records_lost_full, 0);
138 v_set(config, &buf->records_lost_wrap, 0);
139 v_set(config, &buf->records_lost_big, 0);
140 v_set(config, &buf->records_count, 0);
141 v_set(config, &buf->records_overrun, 0);
142 buf->finalized = 0;
143 }
144
145 /**
146 * channel_reset - Reset channel to initial values.
147 * @chan: Channel.
148 *
149 * Effectively empty the channel. Should be called when the channel is not used
150 * for writing. The channel can be opened for reading, but the reader should not
151 * be using the iterator concurrently with reset. The previous current iterator
152 * record is reset.
153 */
154 void channel_reset(struct channel *chan)
155 {
156 /*
157 * Reset iterators first. Will put the subbuffer if held for reading.
158 */
159 uatomic_set(&chan->record_disabled, 0);
160 /* Don't reset commit_count_mask, still valid */
161 channel_backend_reset(&chan->backend);
162 /* Don't reset switch/read timer interval */
163 /* Don't reset notifiers and notifier enable bits */
164 /* Don't reset reader reference count */
165 }
166
167 /*
168 * Must be called under cpu hotplug protection.
169 */
170 int lib_ring_buffer_create(struct lttng_ust_lib_ring_buffer *buf,
171 struct channel_backend *chanb, int cpu,
172 struct lttng_ust_shm_handle *handle,
173 struct shm_object *shmobj)
174 {
175 const struct lttng_ust_lib_ring_buffer_config *config = &chanb->config;
176 struct channel *chan = caa_container_of(chanb, struct channel, backend);
177 void *priv = chanb->priv;
178 unsigned int num_subbuf;
179 size_t subbuf_header_size;
180 u64 tsc;
181 int ret;
182
183 /* Test for cpu hotplug */
184 if (buf->backend.allocated)
185 return 0;
186
187 ret = lib_ring_buffer_backend_create(&buf->backend, &chan->backend,
188 cpu, handle, shmobj);
189 if (ret)
190 return ret;
191
192 align_shm(shmobj, __alignof__(struct commit_counters_hot));
193 set_shmp(buf->commit_hot,
194 zalloc_shm(shmobj,
195 sizeof(struct commit_counters_hot) * chan->backend.num_subbuf));
196 if (!shmp(handle, buf->commit_hot)) {
197 ret = -ENOMEM;
198 goto free_chanbuf;
199 }
200
201 align_shm(shmobj, __alignof__(struct commit_counters_cold));
202 set_shmp(buf->commit_cold,
203 zalloc_shm(shmobj,
204 sizeof(struct commit_counters_cold) * chan->backend.num_subbuf));
205 if (!shmp(handle, buf->commit_cold)) {
206 ret = -ENOMEM;
207 goto free_commit;
208 }
209
210 num_subbuf = chan->backend.num_subbuf;
211 //init_waitqueue_head(&buf->read_wait);
212
213 /*
214 * Write the subbuffer header for first subbuffer so we know the total
215 * duration of data gathering.
216 */
217 subbuf_header_size = config->cb.subbuffer_header_size();
218 v_set(config, &buf->offset, subbuf_header_size);
219 subbuffer_id_clear_noref(config, &shmp_index(handle, buf->backend.buf_wsb, 0)->id);
220 tsc = config->cb.ring_buffer_clock_read(shmp(handle, buf->backend.chan));
221 config->cb.buffer_begin(buf, tsc, 0, handle);
222 v_add(config, subbuf_header_size, &shmp_index(handle, buf->commit_hot, 0)->cc);
223
224 if (config->cb.buffer_create) {
225 ret = config->cb.buffer_create(buf, priv, cpu, chanb->name, handle);
226 if (ret)
227 goto free_init;
228 }
229 buf->backend.allocated = 1;
230 return 0;
231
232 /* Error handling */
233 free_init:
234 /* commit_cold will be freed by shm teardown */
235 free_commit:
236 /* commit_hot will be freed by shm teardown */
237 free_chanbuf:
238 lib_ring_buffer_backend_free(&buf->backend);
239 return ret;
240 }
241
242 #if 0
243 static void switch_buffer_timer(unsigned long data)
244 {
245 struct lttng_ust_lib_ring_buffer *buf = (struct lttng_ust_lib_ring_buffer *)data;
246 struct channel *chan = shmp(handle, buf->backend.chan);
247 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
248
249 /*
250 * Only flush buffers periodically if readers are active.
251 */
252 if (uatomic_read(&buf->active_readers) || uatomic_read(&buf->active_shadow_readers))
253 lib_ring_buffer_switch_slow(buf, SWITCH_ACTIVE, handle);
254
255 //TODO timers
256 //if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
257 // mod_timer_pinned(&buf->switch_timer,
258 // jiffies + chan->switch_timer_interval);
259 //else
260 // mod_timer(&buf->switch_timer,
261 // jiffies + chan->switch_timer_interval);
262 }
263 #endif //0
264
265 static void lib_ring_buffer_start_switch_timer(struct lttng_ust_lib_ring_buffer *buf,
266 struct lttng_ust_shm_handle *handle)
267 {
268 struct channel *chan = shmp(handle, buf->backend.chan);
269 //const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
270
271 if (!chan->switch_timer_interval || buf->switch_timer_enabled)
272 return;
273 //TODO
274 //init_timer(&buf->switch_timer);
275 //buf->switch_timer.function = switch_buffer_timer;
276 //buf->switch_timer.expires = jiffies + chan->switch_timer_interval;
277 //buf->switch_timer.data = (unsigned long)buf;
278 //if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
279 // add_timer_on(&buf->switch_timer, buf->backend.cpu);
280 //else
281 // add_timer(&buf->switch_timer);
282 buf->switch_timer_enabled = 1;
283 }
284
285 static void lib_ring_buffer_stop_switch_timer(struct lttng_ust_lib_ring_buffer *buf,
286 struct lttng_ust_shm_handle *handle)
287 {
288 struct channel *chan = shmp(handle, buf->backend.chan);
289
290 if (!chan->switch_timer_interval || !buf->switch_timer_enabled)
291 return;
292
293 //TODO
294 //del_timer_sync(&buf->switch_timer);
295 buf->switch_timer_enabled = 0;
296 }
297
298 #if 0
299 /*
300 * Polling timer to check the channels for data.
301 */
302 static void read_buffer_timer(unsigned long data)
303 {
304 struct lttng_ust_lib_ring_buffer *buf = (struct lttng_ust_lib_ring_buffer *)data;
305 struct channel *chan = shmp(handle, buf->backend.chan);
306 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
307
308 CHAN_WARN_ON(chan, !buf->backend.allocated);
309
310 if (uatomic_read(&buf->active_readers) || uatomic_read(&buf->active_shadow_readers))
311 && lib_ring_buffer_poll_deliver(config, buf, chan)) {
312 //TODO
313 //wake_up_interruptible(&buf->read_wait);
314 //wake_up_interruptible(&chan->read_wait);
315 }
316
317 //TODO
318 //if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
319 // mod_timer_pinned(&buf->read_timer,
320 // jiffies + chan->read_timer_interval);
321 //else
322 // mod_timer(&buf->read_timer,
323 // jiffies + chan->read_timer_interval);
324 }
325 #endif //0
326
327 static void lib_ring_buffer_start_read_timer(struct lttng_ust_lib_ring_buffer *buf,
328 struct lttng_ust_shm_handle *handle)
329 {
330 struct channel *chan = shmp(handle, buf->backend.chan);
331 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
332
333 if (config->wakeup != RING_BUFFER_WAKEUP_BY_TIMER
334 || !chan->read_timer_interval
335 || buf->read_timer_enabled)
336 return;
337
338 //TODO
339 //init_timer(&buf->read_timer);
340 //buf->read_timer.function = read_buffer_timer;
341 //buf->read_timer.expires = jiffies + chan->read_timer_interval;
342 //buf->read_timer.data = (unsigned long)buf;
343
344 //if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
345 // add_timer_on(&buf->read_timer, buf->backend.cpu);
346 //else
347 // add_timer(&buf->read_timer);
348 buf->read_timer_enabled = 1;
349 }
350
351 static void lib_ring_buffer_stop_read_timer(struct lttng_ust_lib_ring_buffer *buf,
352 struct lttng_ust_shm_handle *handle)
353 {
354 struct channel *chan = shmp(handle, buf->backend.chan);
355 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
356
357 if (config->wakeup != RING_BUFFER_WAKEUP_BY_TIMER
358 || !chan->read_timer_interval
359 || !buf->read_timer_enabled)
360 return;
361
362 //TODO
363 //del_timer_sync(&buf->read_timer);
364 /*
365 * do one more check to catch data that has been written in the last
366 * timer period.
367 */
368 if (lib_ring_buffer_poll_deliver(config, buf, chan, handle)) {
369 //TODO
370 //wake_up_interruptible(&buf->read_wait);
371 //wake_up_interruptible(&chan->read_wait);
372 }
373 buf->read_timer_enabled = 0;
374 }
375
376 static void channel_unregister_notifiers(struct channel *chan,
377 struct lttng_ust_shm_handle *handle)
378 {
379 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
380 int cpu;
381
382 if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
383 for_each_possible_cpu(cpu) {
384 struct lttng_ust_lib_ring_buffer *buf = shmp(handle, chan->backend.buf[cpu].shmp);
385
386 lib_ring_buffer_stop_switch_timer(buf, handle);
387 lib_ring_buffer_stop_read_timer(buf, handle);
388 }
389 } else {
390 struct lttng_ust_lib_ring_buffer *buf = shmp(handle, chan->backend.buf[0].shmp);
391
392 lib_ring_buffer_stop_switch_timer(buf, handle);
393 lib_ring_buffer_stop_read_timer(buf, handle);
394 }
395 //channel_backend_unregister_notifiers(&chan->backend);
396 }
397
398 static void channel_free(struct channel *chan, struct lttng_ust_shm_handle *handle,
399 int shadow)
400 {
401 if (!shadow)
402 channel_backend_free(&chan->backend, handle);
403 /* chan is freed by shm teardown */
404 shm_object_table_destroy(handle->table);
405 free(handle);
406 }
407
408 /**
409 * channel_create - Create channel.
410 * @config: ring buffer instance configuration
411 * @name: name of the channel
412 * @priv: ring buffer client private data
413 * @buf_addr: pointer the the beginning of the preallocated buffer contiguous
414 * address mapping. It is used only by RING_BUFFER_STATIC
415 * configuration. It can be set to NULL for other backends.
416 * @subbuf_size: subbuffer size
417 * @num_subbuf: number of subbuffers
418 * @switch_timer_interval: Time interval (in us) to fill sub-buffers with
419 * padding to let readers get those sub-buffers.
420 * Used for live streaming.
421 * @read_timer_interval: Time interval (in us) to wake up pending readers.
422 *
423 * Holds cpu hotplug.
424 * Returns NULL on failure.
425 */
426 struct lttng_ust_shm_handle *channel_create(const struct lttng_ust_lib_ring_buffer_config *config,
427 const char *name, void *priv, void *buf_addr,
428 size_t subbuf_size,
429 size_t num_subbuf, unsigned int switch_timer_interval,
430 unsigned int read_timer_interval,
431 int *shm_fd, int *wait_fd, uint64_t *memory_map_size)
432 {
433 int ret, cpu;
434 size_t shmsize;
435 struct channel *chan;
436 struct lttng_ust_shm_handle *handle;
437 struct shm_object *shmobj;
438 struct shm_ref *ref;
439
440 if (lib_ring_buffer_check_config(config, switch_timer_interval,
441 read_timer_interval))
442 return NULL;
443
444 handle = zmalloc(sizeof(struct lttng_ust_shm_handle));
445 if (!handle)
446 return NULL;
447
448 /* Allocate table for channel + per-cpu buffers */
449 handle->table = shm_object_table_create(1 + num_possible_cpus());
450 if (!handle->table)
451 goto error_table_alloc;
452
453 /* Calculate the shm allocation layout */
454 shmsize = sizeof(struct channel);
455 shmsize += offset_align(shmsize, __alignof__(struct lttng_ust_lib_ring_buffer_shmp));
456 if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
457 shmsize += sizeof(struct lttng_ust_lib_ring_buffer_shmp) * num_possible_cpus();
458 else
459 shmsize += sizeof(struct lttng_ust_lib_ring_buffer_shmp);
460
461 shmobj = shm_object_table_append(handle->table, shmsize);
462 if (!shmobj)
463 goto error_append;
464 /* struct channel is at object 0, offset 0 (hardcoded) */
465 set_shmp(handle->chan, zalloc_shm(shmobj, shmsize));
466 assert(handle->chan._ref.index == 0);
467 assert(handle->chan._ref.offset == 0);
468 chan = shmp(handle, handle->chan);
469 if (!chan)
470 goto error_append;
471
472 ret = channel_backend_init(&chan->backend, name, config, priv,
473 subbuf_size, num_subbuf, handle);
474 if (ret)
475 goto error_backend_init;
476
477 chan->commit_count_mask = (~0UL >> chan->backend.num_subbuf_order);
478 //TODO
479 //chan->switch_timer_interval = usecs_to_jiffies(switch_timer_interval);
480 //chan->read_timer_interval = usecs_to_jiffies(read_timer_interval);
481 //TODO
482 //init_waitqueue_head(&chan->read_wait);
483 //init_waitqueue_head(&chan->hp_wait);
484
485 if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
486 /*
487 * In case of non-hotplug cpu, if the ring-buffer is allocated
488 * in early initcall, it will not be notified of secondary cpus.
489 * In that off case, we need to allocate for all possible cpus.
490 */
491 for_each_possible_cpu(cpu) {
492 struct lttng_ust_lib_ring_buffer *buf = shmp(handle, chan->backend.buf[cpu].shmp);
493 lib_ring_buffer_start_switch_timer(buf, handle);
494 lib_ring_buffer_start_read_timer(buf, handle);
495 }
496 } else {
497 struct lttng_ust_lib_ring_buffer *buf = shmp(handle, chan->backend.buf[0].shmp);
498
499 lib_ring_buffer_start_switch_timer(buf, handle);
500 lib_ring_buffer_start_read_timer(buf, handle);
501 }
502 ref = &handle->chan._ref;
503 shm_get_object_data(handle, ref, shm_fd, wait_fd, memory_map_size);
504 return handle;
505
506 error_backend_init:
507 error_append:
508 shm_object_table_destroy(handle->table);
509 error_table_alloc:
510 free(handle);
511 return NULL;
512 }
513
514 struct lttng_ust_shm_handle *channel_handle_create(int shm_fd, int wait_fd,
515 uint64_t memory_map_size)
516 {
517 struct lttng_ust_shm_handle *handle;
518 struct shm_object *object;
519
520 handle = zmalloc(sizeof(struct lttng_ust_shm_handle));
521 if (!handle)
522 return NULL;
523
524 /* Allocate table for channel + per-cpu buffers */
525 handle->table = shm_object_table_create(1 + num_possible_cpus());
526 if (!handle->table)
527 goto error_table_alloc;
528 /* Add channel object */
529 object = shm_object_table_append_shadow(handle->table,
530 shm_fd, wait_fd, memory_map_size);
531 if (!object)
532 goto error_table_object;
533 /* struct channel is at object 0, offset 0 (hardcoded) */
534 handle->chan._ref.index = 0;
535 handle->chan._ref.offset = 0;
536 return handle;
537
538 error_table_object:
539 shm_object_table_destroy(handle->table);
540 error_table_alloc:
541 free(handle);
542 return NULL;
543 }
544
545 int channel_handle_add_stream(struct lttng_ust_shm_handle *handle,
546 int shm_fd, int wait_fd, uint64_t memory_map_size)
547 {
548 struct shm_object *object;
549
550 /* Add stream object */
551 object = shm_object_table_append_shadow(handle->table,
552 shm_fd, wait_fd, memory_map_size);
553 if (!object)
554 return -1;
555 return 0;
556 }
557
558 static
559 void channel_release(struct channel *chan, struct lttng_ust_shm_handle *handle,
560 int shadow)
561 {
562 channel_free(chan, handle, shadow);
563 }
564
565 /**
566 * channel_destroy - Finalize, wait for q.s. and destroy channel.
567 * @chan: channel to destroy
568 *
569 * Holds cpu hotplug.
570 * Call "destroy" callback, finalize channels, decrement the channel
571 * reference count. Note that when readers have completed data
572 * consumption of finalized channels, get_subbuf() will return -ENODATA.
573 * They should release their handle at that point. Returns the private
574 * data pointer.
575 */
576 void *channel_destroy(struct channel *chan, struct lttng_ust_shm_handle *handle,
577 int shadow)
578 {
579 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
580 void *priv;
581 int cpu;
582
583 if (shadow) {
584 channel_release(chan, handle, shadow);
585 return NULL;
586 }
587
588 channel_unregister_notifiers(chan, handle);
589
590 if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
591 for_each_channel_cpu(cpu, chan) {
592 struct lttng_ust_lib_ring_buffer *buf = shmp(handle, chan->backend.buf[cpu].shmp);
593
594 if (config->cb.buffer_finalize)
595 config->cb.buffer_finalize(buf,
596 chan->backend.priv,
597 cpu, handle);
598 if (buf->backend.allocated)
599 lib_ring_buffer_switch_slow(buf, SWITCH_FLUSH,
600 handle);
601 /*
602 * Perform flush before writing to finalized.
603 */
604 cmm_smp_wmb();
605 CMM_ACCESS_ONCE(buf->finalized) = 1;
606 //wake_up_interruptible(&buf->read_wait);
607 }
608 } else {
609 struct lttng_ust_lib_ring_buffer *buf = shmp(handle, chan->backend.buf[0].shmp);
610
611 if (config->cb.buffer_finalize)
612 config->cb.buffer_finalize(buf, chan->backend.priv, -1, handle);
613 if (buf->backend.allocated)
614 lib_ring_buffer_switch_slow(buf, SWITCH_FLUSH,
615 handle);
616 /*
617 * Perform flush before writing to finalized.
618 */
619 cmm_smp_wmb();
620 CMM_ACCESS_ONCE(buf->finalized) = 1;
621 //wake_up_interruptible(&buf->read_wait);
622 }
623 CMM_ACCESS_ONCE(chan->finalized) = 1;
624 //wake_up_interruptible(&chan->hp_wait);
625 //wake_up_interruptible(&chan->read_wait);
626 /*
627 * sessiond/consumer are keeping a reference on the shm file
628 * descriptor directly. No need to refcount.
629 */
630 priv = chan->backend.priv;
631 channel_release(chan, handle, shadow);
632 return priv;
633 }
634
635 struct lttng_ust_lib_ring_buffer *channel_get_ring_buffer(
636 const struct lttng_ust_lib_ring_buffer_config *config,
637 struct channel *chan, int cpu,
638 struct lttng_ust_shm_handle *handle,
639 int *shm_fd, int *wait_fd,
640 uint64_t *memory_map_size)
641 {
642 struct shm_ref *ref;
643
644 if (config->alloc == RING_BUFFER_ALLOC_GLOBAL) {
645 ref = &chan->backend.buf[0].shmp._ref;
646 shm_get_object_data(handle, ref, shm_fd, wait_fd,
647 memory_map_size);
648 return shmp(handle, chan->backend.buf[0].shmp);
649 } else {
650 if (cpu >= num_possible_cpus())
651 return NULL;
652 ref = &chan->backend.buf[cpu].shmp._ref;
653 shm_get_object_data(handle, ref, shm_fd, wait_fd,
654 memory_map_size);
655 return shmp(handle, chan->backend.buf[cpu].shmp);
656 }
657 }
658
659 int lib_ring_buffer_open_read(struct lttng_ust_lib_ring_buffer *buf,
660 struct lttng_ust_shm_handle *handle,
661 int shadow)
662 {
663 if (shadow) {
664 if (uatomic_cmpxchg(&buf->active_shadow_readers, 0, 1) != 0)
665 return -EBUSY;
666 cmm_smp_mb();
667 return 0;
668 }
669 if (uatomic_cmpxchg(&buf->active_readers, 0, 1) != 0)
670 return -EBUSY;
671 cmm_smp_mb();
672 return 0;
673 }
674
675 void lib_ring_buffer_release_read(struct lttng_ust_lib_ring_buffer *buf,
676 struct lttng_ust_shm_handle *handle,
677 int shadow)
678 {
679 struct channel *chan = shmp(handle, buf->backend.chan);
680
681 if (shadow) {
682 CHAN_WARN_ON(chan, uatomic_read(&buf->active_shadow_readers) != 1);
683 cmm_smp_mb();
684 uatomic_dec(&buf->active_shadow_readers);
685 return;
686 }
687 CHAN_WARN_ON(chan, uatomic_read(&buf->active_readers) != 1);
688 cmm_smp_mb();
689 uatomic_dec(&buf->active_readers);
690 }
691
692 /**
693 * lib_ring_buffer_snapshot - save subbuffer position snapshot (for read)
694 * @buf: ring buffer
695 * @consumed: consumed count indicating the position where to read
696 * @produced: produced count, indicates position when to stop reading
697 *
698 * Returns -ENODATA if buffer is finalized, -EAGAIN if there is currently no
699 * data to read at consumed position, or 0 if the get operation succeeds.
700 */
701
702 int lib_ring_buffer_snapshot(struct lttng_ust_lib_ring_buffer *buf,
703 unsigned long *consumed, unsigned long *produced,
704 struct lttng_ust_shm_handle *handle)
705 {
706 struct channel *chan = shmp(handle, buf->backend.chan);
707 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
708 unsigned long consumed_cur, write_offset;
709 int finalized;
710
711 finalized = CMM_ACCESS_ONCE(buf->finalized);
712 /*
713 * Read finalized before counters.
714 */
715 cmm_smp_rmb();
716 consumed_cur = uatomic_read(&buf->consumed);
717 /*
718 * No need to issue a memory barrier between consumed count read and
719 * write offset read, because consumed count can only change
720 * concurrently in overwrite mode, and we keep a sequence counter
721 * identifier derived from the write offset to check we are getting
722 * the same sub-buffer we are expecting (the sub-buffers are atomically
723 * "tagged" upon writes, tags are checked upon read).
724 */
725 write_offset = v_read(config, &buf->offset);
726
727 /*
728 * Check that we are not about to read the same subbuffer in
729 * which the writer head is.
730 */
731 if (subbuf_trunc(write_offset, chan) - subbuf_trunc(consumed_cur, chan)
732 == 0)
733 goto nodata;
734
735 *consumed = consumed_cur;
736 *produced = subbuf_trunc(write_offset, chan);
737
738 return 0;
739
740 nodata:
741 /*
742 * The memory barriers __wait_event()/wake_up_interruptible() take care
743 * of "raw_spin_is_locked" memory ordering.
744 */
745 if (finalized)
746 return -ENODATA;
747 else
748 return -EAGAIN;
749 }
750
751 /**
752 * lib_ring_buffer_put_snapshot - move consumed counter forward
753 * @buf: ring buffer
754 * @consumed_new: new consumed count value
755 */
756 void lib_ring_buffer_move_consumer(struct lttng_ust_lib_ring_buffer *buf,
757 unsigned long consumed_new,
758 struct lttng_ust_shm_handle *handle)
759 {
760 struct lttng_ust_lib_ring_buffer_backend *bufb = &buf->backend;
761 struct channel *chan = shmp(handle, bufb->chan);
762 unsigned long consumed;
763
764 CHAN_WARN_ON(chan, uatomic_read(&buf->active_readers) != 1
765 && uatomic_read(&buf->active_shadow_readers) != 1);
766
767 /*
768 * Only push the consumed value forward.
769 * If the consumed cmpxchg fails, this is because we have been pushed by
770 * the writer in flight recorder mode.
771 */
772 consumed = uatomic_read(&buf->consumed);
773 while ((long) consumed - (long) consumed_new < 0)
774 consumed = uatomic_cmpxchg(&buf->consumed, consumed,
775 consumed_new);
776 }
777
778 /**
779 * lib_ring_buffer_get_subbuf - get exclusive access to subbuffer for reading
780 * @buf: ring buffer
781 * @consumed: consumed count indicating the position where to read
782 *
783 * Returns -ENODATA if buffer is finalized, -EAGAIN if there is currently no
784 * data to read at consumed position, or 0 if the get operation succeeds.
785 */
786 int lib_ring_buffer_get_subbuf(struct lttng_ust_lib_ring_buffer *buf,
787 unsigned long consumed,
788 struct lttng_ust_shm_handle *handle)
789 {
790 struct channel *chan = shmp(handle, buf->backend.chan);
791 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
792 unsigned long consumed_cur, consumed_idx, commit_count, write_offset;
793 int ret;
794 int finalized;
795
796 retry:
797 finalized = CMM_ACCESS_ONCE(buf->finalized);
798 /*
799 * Read finalized before counters.
800 */
801 cmm_smp_rmb();
802 consumed_cur = uatomic_read(&buf->consumed);
803 consumed_idx = subbuf_index(consumed, chan);
804 commit_count = v_read(config, &shmp_index(handle, buf->commit_cold, consumed_idx)->cc_sb);
805 /*
806 * Make sure we read the commit count before reading the buffer
807 * data and the write offset. Correct consumed offset ordering
808 * wrt commit count is insured by the use of cmpxchg to update
809 * the consumed offset.
810 */
811 /*
812 * Local rmb to match the remote wmb to read the commit count
813 * before the buffer data and the write offset.
814 */
815 cmm_smp_rmb();
816
817 write_offset = v_read(config, &buf->offset);
818
819 /*
820 * Check that the buffer we are getting is after or at consumed_cur
821 * position.
822 */
823 if ((long) subbuf_trunc(consumed, chan)
824 - (long) subbuf_trunc(consumed_cur, chan) < 0)
825 goto nodata;
826
827 /*
828 * Check that the subbuffer we are trying to consume has been
829 * already fully committed.
830 */
831 if (((commit_count - chan->backend.subbuf_size)
832 & chan->commit_count_mask)
833 - (buf_trunc(consumed_cur, chan)
834 >> chan->backend.num_subbuf_order)
835 != 0)
836 goto nodata;
837
838 /*
839 * Check that we are not about to read the same subbuffer in
840 * which the writer head is.
841 */
842 if (subbuf_trunc(write_offset, chan) - subbuf_trunc(consumed_cur, chan)
843 == 0)
844 goto nodata;
845
846 /*
847 * Failure to get the subbuffer causes a busy-loop retry without going
848 * to a wait queue. These are caused by short-lived race windows where
849 * the writer is getting access to a subbuffer we were trying to get
850 * access to. Also checks that the "consumed" buffer count we are
851 * looking for matches the one contained in the subbuffer id.
852 */
853 ret = update_read_sb_index(config, &buf->backend, &chan->backend,
854 consumed_idx, buf_trunc_val(consumed, chan),
855 handle);
856 if (ret)
857 goto retry;
858 subbuffer_id_clear_noref(config, &buf->backend.buf_rsb.id);
859
860 buf->get_subbuf_consumed = consumed;
861 buf->get_subbuf = 1;
862
863 return 0;
864
865 nodata:
866 /*
867 * The memory barriers __wait_event()/wake_up_interruptible() take care
868 * of "raw_spin_is_locked" memory ordering.
869 */
870 if (finalized)
871 return -ENODATA;
872 else
873 return -EAGAIN;
874 }
875
876 /**
877 * lib_ring_buffer_put_subbuf - release exclusive subbuffer access
878 * @buf: ring buffer
879 */
880 void lib_ring_buffer_put_subbuf(struct lttng_ust_lib_ring_buffer *buf,
881 struct lttng_ust_shm_handle *handle)
882 {
883 struct lttng_ust_lib_ring_buffer_backend *bufb = &buf->backend;
884 struct channel *chan = shmp(handle, bufb->chan);
885 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
886 unsigned long read_sb_bindex, consumed_idx, consumed;
887
888 CHAN_WARN_ON(chan, uatomic_read(&buf->active_readers) != 1
889 && uatomic_read(&buf->active_shadow_readers) != 1);
890
891 if (!buf->get_subbuf) {
892 /*
893 * Reader puts a subbuffer it did not get.
894 */
895 CHAN_WARN_ON(chan, 1);
896 return;
897 }
898 consumed = buf->get_subbuf_consumed;
899 buf->get_subbuf = 0;
900
901 /*
902 * Clear the records_unread counter. (overruns counter)
903 * Can still be non-zero if a file reader simply grabbed the data
904 * without using iterators.
905 * Can be below zero if an iterator is used on a snapshot more than
906 * once.
907 */
908 read_sb_bindex = subbuffer_id_get_index(config, bufb->buf_rsb.id);
909 v_add(config, v_read(config,
910 &shmp(handle, shmp_index(handle, bufb->array, read_sb_bindex)->shmp)->records_unread),
911 &bufb->records_read);
912 v_set(config, &shmp(handle, shmp_index(handle, bufb->array, read_sb_bindex)->shmp)->records_unread, 0);
913 CHAN_WARN_ON(chan, config->mode == RING_BUFFER_OVERWRITE
914 && subbuffer_id_is_noref(config, bufb->buf_rsb.id));
915 subbuffer_id_set_noref(config, &bufb->buf_rsb.id);
916
917 /*
918 * Exchange the reader subbuffer with the one we put in its place in the
919 * writer subbuffer table. Expect the original consumed count. If
920 * update_read_sb_index fails, this is because the writer updated the
921 * subbuffer concurrently. We should therefore keep the subbuffer we
922 * currently have: it has become invalid to try reading this sub-buffer
923 * consumed count value anyway.
924 */
925 consumed_idx = subbuf_index(consumed, chan);
926 update_read_sb_index(config, &buf->backend, &chan->backend,
927 consumed_idx, buf_trunc_val(consumed, chan),
928 handle);
929 /*
930 * update_read_sb_index return value ignored. Don't exchange sub-buffer
931 * if the writer concurrently updated it.
932 */
933 }
934
935 /*
936 * cons_offset is an iterator on all subbuffer offsets between the reader
937 * position and the writer position. (inclusive)
938 */
939 static
940 void lib_ring_buffer_print_subbuffer_errors(struct lttng_ust_lib_ring_buffer *buf,
941 struct channel *chan,
942 unsigned long cons_offset,
943 int cpu,
944 struct lttng_ust_shm_handle *handle)
945 {
946 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
947 unsigned long cons_idx, commit_count, commit_count_sb;
948
949 cons_idx = subbuf_index(cons_offset, chan);
950 commit_count = v_read(config, &shmp_index(handle, buf->commit_hot, cons_idx)->cc);
951 commit_count_sb = v_read(config, &shmp_index(handle, buf->commit_cold, cons_idx)->cc_sb);
952
953 if (subbuf_offset(commit_count, chan) != 0)
954 ERRMSG("ring buffer %s, cpu %d: "
955 "commit count in subbuffer %lu,\n"
956 "expecting multiples of %lu bytes\n"
957 " [ %lu bytes committed, %lu bytes reader-visible ]\n",
958 chan->backend.name, cpu, cons_idx,
959 chan->backend.subbuf_size,
960 commit_count, commit_count_sb);
961
962 ERRMSG("ring buffer: %s, cpu %d: %lu bytes committed\n",
963 chan->backend.name, cpu, commit_count);
964 }
965
966 static
967 void lib_ring_buffer_print_buffer_errors(struct lttng_ust_lib_ring_buffer *buf,
968 struct channel *chan,
969 void *priv, int cpu,
970 struct lttng_ust_shm_handle *handle)
971 {
972 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
973 unsigned long write_offset, cons_offset;
974
975 /*
976 * Can be called in the error path of allocation when
977 * trans_channel_data is not yet set.
978 */
979 if (!chan)
980 return;
981 /*
982 * No need to order commit_count, write_offset and cons_offset reads
983 * because we execute at teardown when no more writer nor reader
984 * references are left.
985 */
986 write_offset = v_read(config, &buf->offset);
987 cons_offset = uatomic_read(&buf->consumed);
988 if (write_offset != cons_offset)
989 ERRMSG("ring buffer %s, cpu %d: "
990 "non-consumed data\n"
991 " [ %lu bytes written, %lu bytes read ]\n",
992 chan->backend.name, cpu, write_offset, cons_offset);
993
994 for (cons_offset = uatomic_read(&buf->consumed);
995 (long) (subbuf_trunc((unsigned long) v_read(config, &buf->offset),
996 chan)
997 - cons_offset) > 0;
998 cons_offset = subbuf_align(cons_offset, chan))
999 lib_ring_buffer_print_subbuffer_errors(buf, chan, cons_offset,
1000 cpu, handle);
1001 }
1002
1003 static
1004 void lib_ring_buffer_print_errors(struct channel *chan,
1005 struct lttng_ust_lib_ring_buffer *buf, int cpu,
1006 struct lttng_ust_shm_handle *handle)
1007 {
1008 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
1009 void *priv = chan->backend.priv;
1010
1011 ERRMSG("ring buffer %s, cpu %d: %lu records written, "
1012 "%lu records overrun\n",
1013 chan->backend.name, cpu,
1014 v_read(config, &buf->records_count),
1015 v_read(config, &buf->records_overrun));
1016
1017 if (v_read(config, &buf->records_lost_full)
1018 || v_read(config, &buf->records_lost_wrap)
1019 || v_read(config, &buf->records_lost_big))
1020 ERRMSG("ring buffer %s, cpu %d: records were lost. Caused by:\n"
1021 " [ %lu buffer full, %lu nest buffer wrap-around, "
1022 "%lu event too big ]\n",
1023 chan->backend.name, cpu,
1024 v_read(config, &buf->records_lost_full),
1025 v_read(config, &buf->records_lost_wrap),
1026 v_read(config, &buf->records_lost_big));
1027
1028 lib_ring_buffer_print_buffer_errors(buf, chan, priv, cpu, handle);
1029 }
1030
1031 /*
1032 * lib_ring_buffer_switch_old_start: Populate old subbuffer header.
1033 *
1034 * Only executed when the buffer is finalized, in SWITCH_FLUSH.
1035 */
1036 static
1037 void lib_ring_buffer_switch_old_start(struct lttng_ust_lib_ring_buffer *buf,
1038 struct channel *chan,
1039 struct switch_offsets *offsets,
1040 u64 tsc,
1041 struct lttng_ust_shm_handle *handle)
1042 {
1043 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
1044 unsigned long oldidx = subbuf_index(offsets->old, chan);
1045 unsigned long commit_count;
1046
1047 config->cb.buffer_begin(buf, tsc, oldidx, handle);
1048
1049 /*
1050 * Order all writes to buffer before the commit count update that will
1051 * determine that the subbuffer is full.
1052 */
1053 cmm_smp_wmb();
1054 v_add(config, config->cb.subbuffer_header_size(),
1055 &shmp_index(handle, buf->commit_hot, oldidx)->cc);
1056 commit_count = v_read(config, &shmp_index(handle, buf->commit_hot, oldidx)->cc);
1057 /* Check if the written buffer has to be delivered */
1058 lib_ring_buffer_check_deliver(config, buf, chan, offsets->old,
1059 commit_count, oldidx, handle);
1060 lib_ring_buffer_write_commit_counter(config, buf, chan, oldidx,
1061 offsets->old, commit_count,
1062 config->cb.subbuffer_header_size(),
1063 handle);
1064 }
1065
1066 /*
1067 * lib_ring_buffer_switch_old_end: switch old subbuffer
1068 *
1069 * Note : offset_old should never be 0 here. It is ok, because we never perform
1070 * buffer switch on an empty subbuffer in SWITCH_ACTIVE mode. The caller
1071 * increments the offset_old value when doing a SWITCH_FLUSH on an empty
1072 * subbuffer.
1073 */
1074 static
1075 void lib_ring_buffer_switch_old_end(struct lttng_ust_lib_ring_buffer *buf,
1076 struct channel *chan,
1077 struct switch_offsets *offsets,
1078 u64 tsc,
1079 struct lttng_ust_shm_handle *handle)
1080 {
1081 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
1082 unsigned long oldidx = subbuf_index(offsets->old - 1, chan);
1083 unsigned long commit_count, padding_size, data_size;
1084
1085 data_size = subbuf_offset(offsets->old - 1, chan) + 1;
1086 padding_size = chan->backend.subbuf_size - data_size;
1087 subbuffer_set_data_size(config, &buf->backend, oldidx, data_size,
1088 handle);
1089
1090 /*
1091 * Order all writes to buffer before the commit count update that will
1092 * determine that the subbuffer is full.
1093 */
1094 cmm_smp_wmb();
1095 v_add(config, padding_size, &shmp_index(handle, buf->commit_hot, oldidx)->cc);
1096 commit_count = v_read(config, &shmp_index(handle, buf->commit_hot, oldidx)->cc);
1097 lib_ring_buffer_check_deliver(config, buf, chan, offsets->old - 1,
1098 commit_count, oldidx, handle);
1099 lib_ring_buffer_write_commit_counter(config, buf, chan, oldidx,
1100 offsets->old, commit_count,
1101 padding_size, handle);
1102 }
1103
1104 /*
1105 * lib_ring_buffer_switch_new_start: Populate new subbuffer.
1106 *
1107 * This code can be executed unordered : writers may already have written to the
1108 * sub-buffer before this code gets executed, caution. The commit makes sure
1109 * that this code is executed before the deliver of this sub-buffer.
1110 */
1111 static
1112 void lib_ring_buffer_switch_new_start(struct lttng_ust_lib_ring_buffer *buf,
1113 struct channel *chan,
1114 struct switch_offsets *offsets,
1115 u64 tsc,
1116 struct lttng_ust_shm_handle *handle)
1117 {
1118 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
1119 unsigned long beginidx = subbuf_index(offsets->begin, chan);
1120 unsigned long commit_count;
1121
1122 config->cb.buffer_begin(buf, tsc, beginidx, handle);
1123
1124 /*
1125 * Order all writes to buffer before the commit count update that will
1126 * determine that the subbuffer is full.
1127 */
1128 cmm_smp_wmb();
1129 v_add(config, config->cb.subbuffer_header_size(),
1130 &shmp_index(handle, buf->commit_hot, beginidx)->cc);
1131 commit_count = v_read(config, &shmp_index(handle, buf->commit_hot, beginidx)->cc);
1132 /* Check if the written buffer has to be delivered */
1133 lib_ring_buffer_check_deliver(config, buf, chan, offsets->begin,
1134 commit_count, beginidx, handle);
1135 lib_ring_buffer_write_commit_counter(config, buf, chan, beginidx,
1136 offsets->begin, commit_count,
1137 config->cb.subbuffer_header_size(),
1138 handle);
1139 }
1140
1141 /*
1142 * lib_ring_buffer_switch_new_end: finish switching current subbuffer
1143 *
1144 * The only remaining threads could be the ones with pending commits. They will
1145 * have to do the deliver themselves.
1146 */
1147 static
1148 void lib_ring_buffer_switch_new_end(struct lttng_ust_lib_ring_buffer *buf,
1149 struct channel *chan,
1150 struct switch_offsets *offsets,
1151 u64 tsc,
1152 struct lttng_ust_shm_handle *handle)
1153 {
1154 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
1155 unsigned long endidx = subbuf_index(offsets->end - 1, chan);
1156 unsigned long commit_count, padding_size, data_size;
1157
1158 data_size = subbuf_offset(offsets->end - 1, chan) + 1;
1159 padding_size = chan->backend.subbuf_size - data_size;
1160 subbuffer_set_data_size(config, &buf->backend, endidx, data_size,
1161 handle);
1162
1163 /*
1164 * Order all writes to buffer before the commit count update that will
1165 * determine that the subbuffer is full.
1166 */
1167 cmm_smp_wmb();
1168 v_add(config, padding_size, &shmp_index(handle, buf->commit_hot, endidx)->cc);
1169 commit_count = v_read(config, &shmp_index(handle, buf->commit_hot, endidx)->cc);
1170 lib_ring_buffer_check_deliver(config, buf, chan, offsets->end - 1,
1171 commit_count, endidx, handle);
1172 lib_ring_buffer_write_commit_counter(config, buf, chan, endidx,
1173 offsets->end, commit_count,
1174 padding_size, handle);
1175 }
1176
1177 /*
1178 * Returns :
1179 * 0 if ok
1180 * !0 if execution must be aborted.
1181 */
1182 static
1183 int lib_ring_buffer_try_switch_slow(enum switch_mode mode,
1184 struct lttng_ust_lib_ring_buffer *buf,
1185 struct channel *chan,
1186 struct switch_offsets *offsets,
1187 u64 *tsc)
1188 {
1189 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
1190 unsigned long off;
1191
1192 offsets->begin = v_read(config, &buf->offset);
1193 offsets->old = offsets->begin;
1194 offsets->switch_old_start = 0;
1195 off = subbuf_offset(offsets->begin, chan);
1196
1197 *tsc = config->cb.ring_buffer_clock_read(chan);
1198
1199 /*
1200 * Ensure we flush the header of an empty subbuffer when doing the
1201 * finalize (SWITCH_FLUSH). This ensures that we end up knowing the
1202 * total data gathering duration even if there were no records saved
1203 * after the last buffer switch.
1204 * In SWITCH_ACTIVE mode, switch the buffer when it contains events.
1205 * SWITCH_ACTIVE only flushes the current subbuffer, dealing with end of
1206 * subbuffer header as appropriate.
1207 * The next record that reserves space will be responsible for
1208 * populating the following subbuffer header. We choose not to populate
1209 * the next subbuffer header here because we want to be able to use
1210 * SWITCH_ACTIVE for periodical buffer flush, which must
1211 * guarantee that all the buffer content (records and header
1212 * timestamps) are visible to the reader. This is required for
1213 * quiescence guarantees for the fusion merge.
1214 */
1215 if (mode == SWITCH_FLUSH || off > 0) {
1216 if (caa_unlikely(off == 0)) {
1217 /*
1218 * The client does not save any header information.
1219 * Don't switch empty subbuffer on finalize, because it
1220 * is invalid to deliver a completely empty subbuffer.
1221 */
1222 if (!config->cb.subbuffer_header_size())
1223 return -1;
1224 /*
1225 * Need to write the subbuffer start header on finalize.
1226 */
1227 offsets->switch_old_start = 1;
1228 }
1229 offsets->begin = subbuf_align(offsets->begin, chan);
1230 } else
1231 return -1; /* we do not have to switch : buffer is empty */
1232 /* Note: old points to the next subbuf at offset 0 */
1233 offsets->end = offsets->begin;
1234 return 0;
1235 }
1236
1237 /*
1238 * Force a sub-buffer switch. This operation is completely reentrant : can be
1239 * called while tracing is active with absolutely no lock held.
1240 *
1241 * Note, however, that as a v_cmpxchg is used for some atomic
1242 * operations, this function must be called from the CPU which owns the buffer
1243 * for a ACTIVE flush.
1244 */
1245 void lib_ring_buffer_switch_slow(struct lttng_ust_lib_ring_buffer *buf, enum switch_mode mode,
1246 struct lttng_ust_shm_handle *handle)
1247 {
1248 struct channel *chan = shmp(handle, buf->backend.chan);
1249 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
1250 struct switch_offsets offsets;
1251 unsigned long oldidx;
1252 u64 tsc;
1253
1254 offsets.size = 0;
1255
1256 /*
1257 * Perform retryable operations.
1258 */
1259 do {
1260 if (lib_ring_buffer_try_switch_slow(mode, buf, chan, &offsets,
1261 &tsc))
1262 return; /* Switch not needed */
1263 } while (v_cmpxchg(config, &buf->offset, offsets.old, offsets.end)
1264 != offsets.old);
1265
1266 /*
1267 * Atomically update last_tsc. This update races against concurrent
1268 * atomic updates, but the race will always cause supplementary full TSC
1269 * records, never the opposite (missing a full TSC record when it would
1270 * be needed).
1271 */
1272 save_last_tsc(config, buf, tsc);
1273
1274 /*
1275 * Push the reader if necessary
1276 */
1277 lib_ring_buffer_reserve_push_reader(buf, chan, offsets.old);
1278
1279 oldidx = subbuf_index(offsets.old, chan);
1280 lib_ring_buffer_clear_noref(config, &buf->backend, oldidx, handle);
1281
1282 /*
1283 * May need to populate header start on SWITCH_FLUSH.
1284 */
1285 if (offsets.switch_old_start) {
1286 lib_ring_buffer_switch_old_start(buf, chan, &offsets, tsc, handle);
1287 offsets.old += config->cb.subbuffer_header_size();
1288 }
1289
1290 /*
1291 * Switch old subbuffer.
1292 */
1293 lib_ring_buffer_switch_old_end(buf, chan, &offsets, tsc, handle);
1294 }
1295
1296 /*
1297 * Returns :
1298 * 0 if ok
1299 * -ENOSPC if event size is too large for packet.
1300 * -ENOBUFS if there is currently not enough space in buffer for the event.
1301 * -EIO if data cannot be written into the buffer for any other reason.
1302 */
1303 static
1304 int lib_ring_buffer_try_reserve_slow(struct lttng_ust_lib_ring_buffer *buf,
1305 struct channel *chan,
1306 struct switch_offsets *offsets,
1307 struct lttng_ust_lib_ring_buffer_ctx *ctx)
1308 {
1309 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
1310 struct lttng_ust_shm_handle *handle = ctx->handle;
1311 unsigned long reserve_commit_diff;
1312
1313 offsets->begin = v_read(config, &buf->offset);
1314 offsets->old = offsets->begin;
1315 offsets->switch_new_start = 0;
1316 offsets->switch_new_end = 0;
1317 offsets->switch_old_end = 0;
1318 offsets->pre_header_padding = 0;
1319
1320 ctx->tsc = config->cb.ring_buffer_clock_read(chan);
1321 if ((int64_t) ctx->tsc == -EIO)
1322 return -EIO;
1323
1324 if (last_tsc_overflow(config, buf, ctx->tsc))
1325 ctx->rflags |= RING_BUFFER_RFLAG_FULL_TSC;
1326
1327 if (caa_unlikely(subbuf_offset(offsets->begin, ctx->chan) == 0)) {
1328 offsets->switch_new_start = 1; /* For offsets->begin */
1329 } else {
1330 offsets->size = config->cb.record_header_size(config, chan,
1331 offsets->begin,
1332 &offsets->pre_header_padding,
1333 ctx);
1334 offsets->size +=
1335 lib_ring_buffer_align(offsets->begin + offsets->size,
1336 ctx->largest_align)
1337 + ctx->data_size;
1338 if (caa_unlikely(subbuf_offset(offsets->begin, chan) +
1339 offsets->size > chan->backend.subbuf_size)) {
1340 offsets->switch_old_end = 1; /* For offsets->old */
1341 offsets->switch_new_start = 1; /* For offsets->begin */
1342 }
1343 }
1344 if (caa_unlikely(offsets->switch_new_start)) {
1345 unsigned long sb_index;
1346
1347 /*
1348 * We are typically not filling the previous buffer completely.
1349 */
1350 if (caa_likely(offsets->switch_old_end))
1351 offsets->begin = subbuf_align(offsets->begin, chan);
1352 offsets->begin = offsets->begin
1353 + config->cb.subbuffer_header_size();
1354 /* Test new buffer integrity */
1355 sb_index = subbuf_index(offsets->begin, chan);
1356 reserve_commit_diff =
1357 (buf_trunc(offsets->begin, chan)
1358 >> chan->backend.num_subbuf_order)
1359 - ((unsigned long) v_read(config,
1360 &shmp_index(handle, buf->commit_cold, sb_index)->cc_sb)
1361 & chan->commit_count_mask);
1362 if (caa_likely(reserve_commit_diff == 0)) {
1363 /* Next subbuffer not being written to. */
1364 if (caa_unlikely(config->mode != RING_BUFFER_OVERWRITE &&
1365 subbuf_trunc(offsets->begin, chan)
1366 - subbuf_trunc((unsigned long)
1367 uatomic_read(&buf->consumed), chan)
1368 >= chan->backend.buf_size)) {
1369 /*
1370 * We do not overwrite non consumed buffers
1371 * and we are full : record is lost.
1372 */
1373 v_inc(config, &buf->records_lost_full);
1374 return -ENOBUFS;
1375 } else {
1376 /*
1377 * Next subbuffer not being written to, and we
1378 * are either in overwrite mode or the buffer is
1379 * not full. It's safe to write in this new
1380 * subbuffer.
1381 */
1382 }
1383 } else {
1384 /*
1385 * Next subbuffer reserve offset does not match the
1386 * commit offset. Drop record in producer-consumer and
1387 * overwrite mode. Caused by either a writer OOPS or too
1388 * many nested writes over a reserve/commit pair.
1389 */
1390 v_inc(config, &buf->records_lost_wrap);
1391 return -EIO;
1392 }
1393 offsets->size =
1394 config->cb.record_header_size(config, chan,
1395 offsets->begin,
1396 &offsets->pre_header_padding,
1397 ctx);
1398 offsets->size +=
1399 lib_ring_buffer_align(offsets->begin + offsets->size,
1400 ctx->largest_align)
1401 + ctx->data_size;
1402 if (caa_unlikely(subbuf_offset(offsets->begin, chan)
1403 + offsets->size > chan->backend.subbuf_size)) {
1404 /*
1405 * Record too big for subbuffers, report error, don't
1406 * complete the sub-buffer switch.
1407 */
1408 v_inc(config, &buf->records_lost_big);
1409 return -ENOSPC;
1410 } else {
1411 /*
1412 * We just made a successful buffer switch and the
1413 * record fits in the new subbuffer. Let's write.
1414 */
1415 }
1416 } else {
1417 /*
1418 * Record fits in the current buffer and we are not on a switch
1419 * boundary. It's safe to write.
1420 */
1421 }
1422 offsets->end = offsets->begin + offsets->size;
1423
1424 if (caa_unlikely(subbuf_offset(offsets->end, chan) == 0)) {
1425 /*
1426 * The offset_end will fall at the very beginning of the next
1427 * subbuffer.
1428 */
1429 offsets->switch_new_end = 1; /* For offsets->begin */
1430 }
1431 return 0;
1432 }
1433
1434 /**
1435 * lib_ring_buffer_reserve_slow - Atomic slot reservation in a buffer.
1436 * @ctx: ring buffer context.
1437 *
1438 * Return : -NOBUFS if not enough space, -ENOSPC if event size too large,
1439 * -EIO for other errors, else returns 0.
1440 * It will take care of sub-buffer switching.
1441 */
1442 int lib_ring_buffer_reserve_slow(struct lttng_ust_lib_ring_buffer_ctx *ctx)
1443 {
1444 struct channel *chan = ctx->chan;
1445 struct lttng_ust_shm_handle *handle = ctx->handle;
1446 const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
1447 struct lttng_ust_lib_ring_buffer *buf;
1448 struct switch_offsets offsets;
1449 int ret;
1450
1451 if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
1452 buf = shmp(handle, chan->backend.buf[ctx->cpu].shmp);
1453 else
1454 buf = shmp(handle, chan->backend.buf[0].shmp);
1455 ctx->buf = buf;
1456
1457 offsets.size = 0;
1458
1459 do {
1460 ret = lib_ring_buffer_try_reserve_slow(buf, chan, &offsets,
1461 ctx);
1462 if (caa_unlikely(ret))
1463 return ret;
1464 } while (caa_unlikely(v_cmpxchg(config, &buf->offset, offsets.old,
1465 offsets.end)
1466 != offsets.old));
1467
1468 /*
1469 * Atomically update last_tsc. This update races against concurrent
1470 * atomic updates, but the race will always cause supplementary full TSC
1471 * records, never the opposite (missing a full TSC record when it would
1472 * be needed).
1473 */
1474 save_last_tsc(config, buf, ctx->tsc);
1475
1476 /*
1477 * Push the reader if necessary
1478 */
1479 lib_ring_buffer_reserve_push_reader(buf, chan, offsets.end - 1);
1480
1481 /*
1482 * Clear noref flag for this subbuffer.
1483 */
1484 lib_ring_buffer_clear_noref(config, &buf->backend,
1485 subbuf_index(offsets.end - 1, chan),
1486 handle);
1487
1488 /*
1489 * Switch old subbuffer if needed.
1490 */
1491 if (caa_unlikely(offsets.switch_old_end)) {
1492 lib_ring_buffer_clear_noref(config, &buf->backend,
1493 subbuf_index(offsets.old - 1, chan),
1494 handle);
1495 lib_ring_buffer_switch_old_end(buf, chan, &offsets, ctx->tsc, handle);
1496 }
1497
1498 /*
1499 * Populate new subbuffer.
1500 */
1501 if (caa_unlikely(offsets.switch_new_start))
1502 lib_ring_buffer_switch_new_start(buf, chan, &offsets, ctx->tsc, handle);
1503
1504 if (caa_unlikely(offsets.switch_new_end))
1505 lib_ring_buffer_switch_new_end(buf, chan, &offsets, ctx->tsc, handle);
1506
1507 ctx->slot_size = offsets.size;
1508 ctx->pre_offset = offsets.begin;
1509 ctx->buf_offset = offsets.begin + offsets.pre_header_padding;
1510 return 0;
1511 }
This page took 0.090056 seconds and 5 git commands to generate.