3 * LTTng userspace tracer buffering system
5 * Copyright (C) 2009 - Pierre-Marc Fournier (pierre-marc dot fournier at polymtl dot ca)
6 * Copyright (C) 2008 - Mathieu Desnoyers (mathieu.desnoyers@polymtl.ca)
8 * This library is free software; you can redistribute it and/or
9 * modify it under the terms of the GNU Lesser General Public
10 * License as published by the Free Software Foundation; either
11 * version 2.1 of the License, or (at your option) any later version.
13 * This library is distributed in the hope that it will be useful,
14 * but WITHOUT ANY WARRANTY; without even the implied warranty of
15 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
16 * Lesser General Public License for more details.
18 * You should have received a copy of the GNU Lesser General Public
19 * License along with this library; if not, write to the Free Software
20 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
30 #include <ust/clock.h>
35 #include "tracercore.h"
38 struct ltt_reserve_switch_offsets
{
40 long begin_switch
, end_switch_current
, end_switch_old
;
41 size_t before_hdr_pad
, size
;
45 static DEFINE_MUTEX(ust_buffers_channels_mutex
);
46 static CDS_LIST_HEAD(ust_buffers_channels
);
48 static int get_n_cpus(void)
51 static int n_cpus
= 0;
54 /* On Linux, when some processors are offline
55 * _SC_NPROCESSORS_CONF counts the offline
56 * processors, whereas _SC_NPROCESSORS_ONLN
57 * does not. If we used _SC_NPROCESSORS_ONLN,
58 * getcpu() could return a value greater than
59 * this sysconf, in which case the arrays
60 * indexed by processor would overflow.
62 result
= sysconf(_SC_NPROCESSORS_CONF
);
74 * _ust_buffers_strncpy_fixup - Fix an incomplete string in a ltt_relay buffer.
76 * @offset : offset within the buffer
77 * @len : length to write
78 * @copied: string actually copied
79 * @terminated: does string end with \0
81 * Fills string with "X" if incomplete.
83 void _ust_buffers_strncpy_fixup(struct ust_buffer
*buf
, size_t offset
,
84 size_t len
, size_t copied
, int terminated
)
86 size_t buf_offset
, cpy
;
90 * Deal with non-terminated string.
94 buf_offset
= BUFFER_OFFSET(offset
, buf
->chan
);
96 * Underlying layer should never ask for writes across
100 < buf
->chan
->subbuf_size
*buf
->chan
->subbuf_cnt
);
101 ust_buffers_do_memset(buf
->buf_data
+ buf_offset
, '\0', 1);
106 * Deal with incomplete string.
107 * Overwrite string's \0 with X too.
113 buf_offset
= BUFFER_OFFSET(offset
, buf
->chan
);
116 * Underlying layer should never ask for writes across subbuffers.
119 < buf
->chan
->subbuf_size
*buf
->chan
->subbuf_cnt
);
121 ust_buffers_do_memset(buf
->buf_data
+ buf_offset
,
125 * Overwrite last 'X' with '\0'.
128 buf_offset
= BUFFER_OFFSET(offset
, buf
->chan
);
130 * Underlying layer should never ask for writes across subbuffers.
133 < buf
->chan
->subbuf_size
*buf
->chan
->subbuf_cnt
);
134 ust_buffers_do_memset(buf
->buf_data
+ buf_offset
, '\0', 1);
137 static void ltt_buffer_begin(struct ust_buffer
*buf
,
138 u64 tsc
, unsigned int subbuf_idx
)
140 struct ust_channel
*channel
= buf
->chan
;
141 struct ltt_subbuffer_header
*header
=
142 (struct ltt_subbuffer_header
*)
143 ust_buffers_offset_address(buf
,
144 subbuf_idx
* buf
->chan
->subbuf_size
);
146 header
->cycle_count_begin
= tsc
;
147 header
->data_size
= 0xFFFFFFFF; /* for recognizing crashed buffers */
148 header
->sb_size
= 0xFFFFFFFF; /* for recognizing crashed buffers */
149 /* FIXME: add memory barrier? */
150 ltt_write_trace_header(channel
->trace
, header
);
153 static int map_buf_data(struct ust_buffer
*buf
, size_t *size
)
158 *size
= PAGE_ALIGN(*size
);
160 result
= buf
->shmid
= shmget(getpid(), *size
, IPC_CREAT
| IPC_EXCL
| 0700);
161 if (result
< 0 && errno
== EINVAL
) {
162 ERR("shmget() returned EINVAL; maybe /proc/sys/kernel/shmmax should be increased.");
164 } else if (result
< 0) {
169 ptr
= shmat(buf
->shmid
, NULL
, 0);
170 if (ptr
== (void *) -1) {
175 /* Already mark the shared memory for destruction. This will occur only
176 * when all users have detached.
178 result
= shmctl(buf
->shmid
, IPC_RMID
, NULL
);
185 buf
->buf_size
= *size
;
190 result
= shmctl(buf
->shmid
, IPC_RMID
, NULL
);
198 static int open_buf(struct ust_channel
*chan
, int cpu
)
202 struct ust_trace
*trace
= chan
->trace
;
203 struct ust_buffer
*buf
= chan
->buf
[cpu
];
204 unsigned int n_subbufs
= chan
->subbuf_cnt
;
207 result
= map_buf_data(buf
, &chan
->alloc_size
);
212 zmalloc(sizeof(*buf
->commit_count
) * n_subbufs
);
213 if (!buf
->commit_count
)
219 goto free_commit_count
;
221 buf
->data_ready_fd_read
= fds
[0];
222 buf
->data_ready_fd_write
= fds
[1];
227 uatomic_set(&buf
->offset
, ltt_subbuffer_header_size());
228 uatomic_set(&buf
->consumed
, 0);
229 uatomic_set(&buf
->active_readers
, 0);
230 for (j
= 0; j
< n_subbufs
; j
++) {
231 uatomic_set(&buf
->commit_count
[j
].cc
, 0);
232 uatomic_set(&buf
->commit_count
[j
].cc_sb
, 0);
235 ltt_buffer_begin(buf
, trace
->start_tsc
, 0);
237 uatomic_add(&buf
->commit_count
[0].cc
, ltt_subbuffer_header_size());
239 uatomic_set(&buf
->events_lost
, 0);
240 uatomic_set(&buf
->corrupted_subbuffers
, 0);
242 memset(buf
->commit_seq
, 0, sizeof(buf
->commit_seq
[0]) * n_subbufs
);
247 free(buf
->commit_count
);
250 if (shmdt(buf
->buf_data
) < 0) {
251 PERROR("shmdt failed");
257 static void ltt_relay_print_buffer_errors(struct ust_channel
*chan
, int cpu
);
259 static void close_buf(struct ust_buffer
*buf
)
261 struct ust_channel
*chan
= buf
->chan
;
265 result
= shmdt(buf
->buf_data
);
270 free(buf
->commit_count
);
272 result
= close(buf
->data_ready_fd_read
);
277 result
= close(buf
->data_ready_fd_write
);
278 if (result
< 0 && errno
!= EBADF
) {
282 /* FIXME: This spews out errors, are they real?:
283 * ltt_relay_print_buffer_errors(chan, cpu); */
287 static int open_channel(struct ust_channel
*chan
, size_t subbuf_size
,
293 if(subbuf_size
== 0 || subbuf_cnt
== 0)
296 /* Check that the subbuffer size is larger than a page. */
297 WARN_ON_ONCE(subbuf_size
< PAGE_SIZE
);
300 * Make sure the number of subbuffers and subbuffer size are power of 2.
302 WARN_ON_ONCE(hweight32(subbuf_size
) != 1);
303 WARN_ON(hweight32(subbuf_cnt
) != 1);
305 chan
->version
= UST_CHANNEL_VERSION
;
306 chan
->subbuf_cnt
= subbuf_cnt
;
307 chan
->subbuf_size
= subbuf_size
;
308 chan
->subbuf_size_order
= get_count_order(subbuf_size
);
309 chan
->alloc_size
= subbuf_size
* subbuf_cnt
;
311 pthread_mutex_lock(&ust_buffers_channels_mutex
);
312 for (i
=0; i
< chan
->n_cpus
; i
++) {
313 result
= open_buf(chan
, i
);
317 cds_list_add(&chan
->list
, &ust_buffers_channels
);
318 pthread_mutex_unlock(&ust_buffers_channels_mutex
);
322 /* Jump directly inside the loop to close the buffers that were already
325 close_buf(chan
->buf
[i
]);
330 pthread_mutex_unlock(&ust_buffers_channels_mutex
);
334 static void close_channel(struct ust_channel
*chan
)
340 pthread_mutex_lock(&ust_buffers_channels_mutex
);
341 for(i
=0; i
<chan
->n_cpus
; i
++) {
342 /* FIXME: if we make it here, then all buffers were necessarily allocated. Moreover, we don't
343 * initialize to NULL so we cannot use this check. Should we? */
344 //ust// if (chan->buf[i])
345 close_buf(chan
->buf
[i
]);
348 cds_list_del(&chan
->list
);
350 pthread_mutex_unlock(&ust_buffers_channels_mutex
);
353 static void ltt_force_switch(struct ust_buffer
*buf
,
354 enum force_switch_mode mode
);
359 * offset is assumed to never be 0 here : never deliver a completely empty
360 * subbuffer. The lost size is between 0 and subbuf_size-1.
362 static notrace
void ltt_buffer_end(struct ust_buffer
*buf
,
363 u64 tsc
, unsigned int offset
, unsigned int subbuf_idx
)
365 struct ltt_subbuffer_header
*header
=
366 (struct ltt_subbuffer_header
*)
367 ust_buffers_offset_address(buf
,
368 subbuf_idx
* buf
->chan
->subbuf_size
);
369 u32 data_size
= SUBBUF_OFFSET(offset
- 1, buf
->chan
) + 1;
371 header
->data_size
= data_size
;
372 header
->sb_size
= PAGE_ALIGN(data_size
);
373 header
->cycle_count_end
= tsc
;
374 header
->events_lost
= uatomic_read(&buf
->events_lost
);
375 header
->subbuf_corrupt
= uatomic_read(&buf
->corrupted_subbuffers
);
376 if(unlikely(header
->events_lost
> 0)) {
377 DBG("Some events (%d) were lost in %s_%d", header
->events_lost
, buf
->chan
->channel_name
, buf
->cpu
);
382 * This function should not be called from NMI interrupt context
384 static notrace
void ltt_buf_unfull(struct ust_buffer
*buf
,
385 unsigned int subbuf_idx
,
391 * Promote compiler cmm_barrier to a smp_mb().
392 * For the specific LTTng case, this IPI call should be removed if the
393 * architecture does not reorder writes. This should eventually be provided by
394 * a separate architecture-specific infrastructure.
396 //ust// static void remote_mb(void *info)
401 int ust_buffers_get_subbuf(struct ust_buffer
*buf
, long *consumed
)
403 struct ust_channel
*channel
= buf
->chan
;
404 long consumed_old
, consumed_idx
, commit_count
, write_offset
;
407 consumed_old
= uatomic_read(&buf
->consumed
);
408 consumed_idx
= SUBBUF_INDEX(consumed_old
, buf
->chan
);
409 commit_count
= uatomic_read(&buf
->commit_count
[consumed_idx
].cc_sb
);
411 * Make sure we read the commit count before reading the buffer
412 * data and the write offset. Correct consumed offset ordering
413 * wrt commit count is insured by the use of cmpxchg to update
414 * the consumed offset.
415 * smp_call_function_single can fail if the remote CPU is offline,
416 * this is OK because then there is no wmb to execute there.
417 * If our thread is executing on the same CPU as the on the buffers
418 * belongs to, we don't have to synchronize it at all. If we are
419 * migrated, the scheduler will take care of the memory cmm_barriers.
420 * Normally, smp_call_function_single() should ensure program order when
421 * executing the remote function, which implies that it surrounds the
422 * function execution with :
433 * However, smp_call_function_single() does not seem to clearly execute
434 * such barriers. It depends on spinlock semantic to provide the barrier
435 * before executing the IPI and, when busy-looping, csd_lock_wait only
436 * executes smp_mb() when it has to wait for the other CPU.
438 * I don't trust this code. Therefore, let's add the smp_mb() sequence
439 * required ourself, even if duplicated. It has no performance impact
442 * smp_mb() is needed because cmm_smp_rmb() and cmm_smp_wmb() only order read vs
443 * read and write vs write. They do not ensure core synchronization. We
444 * really have to ensure total order between the 3 cmm_barriers running on
447 //ust// #ifdef LTT_NO_IPI_BARRIER
449 * Local rmb to match the remote wmb to read the commit count before the
450 * buffer data and the write offset.
454 //ust// if (raw_smp_processor_id() != buf->cpu) {
455 //ust// smp_mb(); /* Total order with IPI handler smp_mb() */
456 //ust// smp_call_function_single(buf->cpu, remote_mb, NULL, 1);
457 //ust// smp_mb(); /* Total order with IPI handler smp_mb() */
461 write_offset
= uatomic_read(&buf
->offset
);
463 * Check that the subbuffer we are trying to consume has been
464 * already fully committed.
466 if (((commit_count
- buf
->chan
->subbuf_size
)
467 & channel
->commit_count_mask
)
468 - (BUFFER_TRUNC(consumed_old
, buf
->chan
)
469 >> channel
->n_subbufs_order
)
474 * Check that we are not about to read the same subbuffer in
475 * which the writer head is.
477 if ((SUBBUF_TRUNC(write_offset
, buf
->chan
)
478 - SUBBUF_TRUNC(consumed_old
, buf
->chan
))
483 /* FIXME: is this ok to disable the reading feature? */
484 //ust// retval = update_read_sb_index(buf, consumed_idx);
486 //ust// return retval;
488 *consumed
= consumed_old
;
493 int ust_buffers_put_subbuf(struct ust_buffer
*buf
, unsigned long uconsumed_old
)
495 long consumed_new
, consumed_old
;
497 consumed_old
= uatomic_read(&buf
->consumed
);
498 consumed_old
= consumed_old
& (~0xFFFFFFFFL
);
499 consumed_old
= consumed_old
| uconsumed_old
;
500 consumed_new
= SUBBUF_ALIGN(consumed_old
, buf
->chan
);
502 //ust// spin_lock(<t_buf->full_lock);
503 if (uatomic_cmpxchg(&buf
->consumed
, consumed_old
,
506 /* We have been pushed by the writer : the last
507 * buffer read _is_ corrupted! It can also
508 * happen if this is a buffer we never got. */
509 //ust// spin_unlock(<t_buf->full_lock);
512 /* tell the client that buffer is now unfull */
515 index
= SUBBUF_INDEX(consumed_old
, buf
->chan
);
516 data
= BUFFER_OFFSET(consumed_old
, buf
->chan
);
517 ltt_buf_unfull(buf
, index
, data
);
518 //ust// spin_unlock(<t_buf->full_lock);
523 static void ltt_relay_print_subbuffer_errors(
524 struct ust_channel
*channel
,
525 long cons_off
, int cpu
)
527 struct ust_buffer
*ltt_buf
= channel
->buf
[cpu
];
528 long cons_idx
, commit_count
, commit_count_sb
, write_offset
;
530 cons_idx
= SUBBUF_INDEX(cons_off
, channel
);
531 commit_count
= uatomic_read(<t_buf
->commit_count
[cons_idx
].cc
);
532 commit_count_sb
= uatomic_read(<t_buf
->commit_count
[cons_idx
].cc_sb
);
535 * No need to order commit_count and write_offset reads because we
536 * execute after trace is stopped when there are no readers left.
538 write_offset
= uatomic_read(<t_buf
->offset
);
539 WARN( "LTT : unread channel %s offset is %ld "
540 "and cons_off : %ld (cpu %d)\n",
541 channel
->channel_name
, write_offset
, cons_off
, cpu
);
542 /* Check each sub-buffer for non filled commit count */
543 if (((commit_count
- channel
->subbuf_size
) & channel
->commit_count_mask
)
544 - (BUFFER_TRUNC(cons_off
, channel
) >> channel
->n_subbufs_order
) != 0) {
545 ERR("LTT : %s : subbuffer %lu has non filled "
546 "commit count [cc, cc_sb] [%lu,%lu].\n",
547 channel
->channel_name
, cons_idx
, commit_count
, commit_count_sb
);
549 ERR("LTT : %s : commit count : %lu, subbuf size %zd\n",
550 channel
->channel_name
, commit_count
,
551 channel
->subbuf_size
);
554 static void ltt_relay_print_errors(struct ust_trace
*trace
,
555 struct ust_channel
*channel
, int cpu
)
557 struct ust_buffer
*ltt_buf
= channel
->buf
[cpu
];
561 * Can be called in the error path of allocation when
562 * trans_channel_data is not yet set.
567 //ust// for (cons_off = 0; cons_off < rchan->alloc_size;
568 //ust// cons_off = SUBBUF_ALIGN(cons_off, rchan))
569 //ust// ust_buffers_print_written(ltt_chan, cons_off, cpu);
570 for (cons_off
= uatomic_read(<t_buf
->consumed
);
571 (SUBBUF_TRUNC(uatomic_read(<t_buf
->offset
),
574 cons_off
= SUBBUF_ALIGN(cons_off
, channel
))
575 ltt_relay_print_subbuffer_errors(channel
, cons_off
, cpu
);
578 static void ltt_relay_print_buffer_errors(struct ust_channel
*channel
, int cpu
)
580 struct ust_trace
*trace
= channel
->trace
;
581 struct ust_buffer
*ltt_buf
= channel
->buf
[cpu
];
583 if (uatomic_read(<t_buf
->events_lost
))
584 ERR("channel %s: %ld events lost (cpu %d)",
585 channel
->channel_name
,
586 uatomic_read(<t_buf
->events_lost
), cpu
);
587 if (uatomic_read(<t_buf
->corrupted_subbuffers
))
588 ERR("channel %s : %ld corrupted subbuffers (cpu %d)",
589 channel
->channel_name
,
590 uatomic_read(<t_buf
->corrupted_subbuffers
), cpu
);
592 ltt_relay_print_errors(trace
, channel
, cpu
);
595 static int map_buf_structs(struct ust_channel
*chan
)
602 size
= PAGE_ALIGN(1);
604 for(i
=0; i
<chan
->n_cpus
; i
++) {
606 result
= chan
->buf_struct_shmids
[i
] = shmget(getpid(), size
, IPC_CREAT
| IPC_EXCL
| 0700);
609 goto destroy_previous
;
612 ptr
= shmat(chan
->buf_struct_shmids
[i
], NULL
, 0);
613 if(ptr
== (void *) -1) {
618 /* Already mark the shared memory for destruction. This will occur only
619 * when all users have detached.
621 result
= shmctl(chan
->buf_struct_shmids
[i
], IPC_RMID
, NULL
);
624 goto destroy_previous
;
632 /* Jumping inside this loop occurs from within the other loop above with i as
633 * counter, so it unallocates the structures for the cpu = current_i down to
637 result
= shmctl(chan
->buf_struct_shmids
[i
], IPC_RMID
, NULL
);
649 static int unmap_buf_structs(struct ust_channel
*chan
)
653 for (i
=0; i
< chan
->n_cpus
; i
++) {
654 if (shmdt(chan
->buf
[i
]) < 0) {
664 static int create_channel(const char *trace_name
, struct ust_trace
*trace
,
665 const char *channel_name
, struct ust_channel
*chan
,
666 unsigned int subbuf_size
, unsigned int n_subbufs
, int overwrite
)
671 chan
->overwrite
= overwrite
;
672 chan
->n_subbufs_order
= get_count_order(n_subbufs
);
673 chan
->commit_count_mask
= (~0UL >> chan
->n_subbufs_order
);
674 chan
->n_cpus
= get_n_cpus();
676 /* These mappings should ideall be per-cpu, if somebody can do that
677 * from userspace, that would be cool!
679 chan
->buf
= (void *) zmalloc(chan
->n_cpus
* sizeof(void *));
680 if(chan
->buf
== NULL
) {
683 chan
->buf_struct_shmids
= (int *) zmalloc(chan
->n_cpus
* sizeof(int));
684 if(chan
->buf_struct_shmids
== NULL
)
687 result
= map_buf_structs(chan
);
689 goto free_buf_struct_shmids
;
692 result
= open_channel(chan
, subbuf_size
, n_subbufs
);
694 ERR("Cannot open channel for trace %s", trace_name
);
695 goto unmap_buf_structs
;
701 for (i
=0; i
< chan
->n_cpus
; i
++) {
702 if (shmdt(chan
->buf
[i
]) < 0) {
703 PERROR("shmdt bufstruct");
707 free_buf_struct_shmids
:
708 free(chan
->buf_struct_shmids
);
718 static void remove_channel(struct ust_channel
*chan
)
722 unmap_buf_structs(chan
);
724 free(chan
->buf_struct_shmids
);
730 static void ltt_relay_async_wakeup_chan(struct ust_channel
*ltt_channel
)
732 //ust// unsigned int i;
733 //ust// struct rchan *rchan = ltt_channel->trans_channel_data;
735 //ust// for_each_possible_cpu(i) {
736 //ust// struct ltt_channel_buf_struct *ltt_buf =
737 //ust// percpu_ptr(ltt_channel->buf, i);
739 //ust// if (uatomic_read(<t_buf->wakeup_readers) == 1) {
740 //ust// uatomic_set(<t_buf->wakeup_readers, 0);
741 //ust// wake_up_interruptible(&rchan->buf[i]->read_wait);
746 static void ltt_relay_finish_buffer(struct ust_channel
*channel
, unsigned int cpu
)
750 if (channel
->buf
[cpu
]) {
751 struct ust_buffer
*buf
= channel
->buf
[cpu
];
752 ltt_force_switch(buf
, FORCE_FLUSH
);
754 /* closing the pipe tells the consumer the buffer is finished */
755 close(buf
->data_ready_fd_write
);
760 static void finish_channel(struct ust_channel
*channel
)
764 for(i
=0; i
<channel
->n_cpus
; i
++) {
765 ltt_relay_finish_buffer(channel
, i
);
771 * ltt_reserve_switch_old_subbuf: switch old subbuffer
773 * Concurrency safe because we are the last and only thread to alter this
774 * sub-buffer. As long as it is not delivered and read, no other thread can
775 * alter the offset, alter the reserve_count or call the
776 * client_buffer_end_callback on this sub-buffer.
778 * The only remaining threads could be the ones with pending commits. They will
779 * have to do the deliver themselves. Not concurrency safe in overwrite mode.
780 * We detect corrupted subbuffers with commit and reserve counts. We keep a
781 * corrupted sub-buffers count and push the readers across these sub-buffers.
783 * Not concurrency safe if a writer is stalled in a subbuffer and another writer
784 * switches in, finding out it's corrupted. The result will be than the old
785 * (uncommited) subbuffer will be declared corrupted, and that the new subbuffer
786 * will be declared corrupted too because of the commit count adjustment.
788 * Note : offset_old should never be 0 here.
790 static void ltt_reserve_switch_old_subbuf(
791 struct ust_channel
*chan
, struct ust_buffer
*buf
,
792 struct ltt_reserve_switch_offsets
*offsets
, u64
*tsc
)
794 long oldidx
= SUBBUF_INDEX(offsets
->old
- 1, chan
);
795 long commit_count
, padding_size
;
797 padding_size
= chan
->subbuf_size
798 - (SUBBUF_OFFSET(offsets
->old
- 1, chan
) + 1);
799 ltt_buffer_end(buf
, *tsc
, offsets
->old
, oldidx
);
802 * Must write slot data before incrementing commit count.
803 * This compiler barrier is upgraded into a cmm_smp_wmb() by the IPI
804 * sent by get_subbuf() when it does its cmm_smp_rmb().
807 uatomic_add(&buf
->commit_count
[oldidx
].cc
, padding_size
);
808 commit_count
= uatomic_read(&buf
->commit_count
[oldidx
].cc
);
809 ltt_check_deliver(chan
, buf
, offsets
->old
- 1, commit_count
, oldidx
);
810 ltt_write_commit_counter(chan
, buf
, oldidx
,
811 offsets
->old
, commit_count
, padding_size
);
815 * ltt_reserve_switch_new_subbuf: Populate new subbuffer.
817 * This code can be executed unordered : writers may already have written to the
818 * sub-buffer before this code gets executed, caution. The commit makes sure
819 * that this code is executed before the deliver of this sub-buffer.
821 static void ltt_reserve_switch_new_subbuf(
822 struct ust_channel
*chan
, struct ust_buffer
*buf
,
823 struct ltt_reserve_switch_offsets
*offsets
, u64
*tsc
)
825 long beginidx
= SUBBUF_INDEX(offsets
->begin
, chan
);
828 ltt_buffer_begin(buf
, *tsc
, beginidx
);
831 * Must write slot data before incrementing commit count.
832 * This compiler barrier is upgraded into a cmm_smp_wmb() by the IPI
833 * sent by get_subbuf() when it does its cmm_smp_rmb().
836 uatomic_add(&buf
->commit_count
[beginidx
].cc
, ltt_subbuffer_header_size());
837 commit_count
= uatomic_read(&buf
->commit_count
[beginidx
].cc
);
838 /* Check if the written buffer has to be delivered */
839 ltt_check_deliver(chan
, buf
, offsets
->begin
, commit_count
, beginidx
);
840 ltt_write_commit_counter(chan
, buf
, beginidx
,
841 offsets
->begin
, commit_count
, ltt_subbuffer_header_size());
845 * ltt_reserve_end_switch_current: finish switching current subbuffer
847 * Concurrency safe because we are the last and only thread to alter this
848 * sub-buffer. As long as it is not delivered and read, no other thread can
849 * alter the offset, alter the reserve_count or call the
850 * client_buffer_end_callback on this sub-buffer.
852 * The only remaining threads could be the ones with pending commits. They will
853 * have to do the deliver themselves. Not concurrency safe in overwrite mode.
854 * We detect corrupted subbuffers with commit and reserve counts. We keep a
855 * corrupted sub-buffers count and push the readers across these sub-buffers.
857 * Not concurrency safe if a writer is stalled in a subbuffer and another writer
858 * switches in, finding out it's corrupted. The result will be than the old
859 * (uncommited) subbuffer will be declared corrupted, and that the new subbuffer
860 * will be declared corrupted too because of the commit count adjustment.
862 static void ltt_reserve_end_switch_current(
863 struct ust_channel
*chan
,
864 struct ust_buffer
*buf
,
865 struct ltt_reserve_switch_offsets
*offsets
, u64
*tsc
)
867 long endidx
= SUBBUF_INDEX(offsets
->end
- 1, chan
);
868 long commit_count
, padding_size
;
870 padding_size
= chan
->subbuf_size
871 - (SUBBUF_OFFSET(offsets
->end
- 1, chan
) + 1);
873 ltt_buffer_end(buf
, *tsc
, offsets
->end
, endidx
);
876 * Must write slot data before incrementing commit count.
877 * This compiler barrier is upgraded into a cmm_smp_wmb() by the IPI
878 * sent by get_subbuf() when it does its cmm_smp_rmb().
881 uatomic_add(&buf
->commit_count
[endidx
].cc
, padding_size
);
882 commit_count
= uatomic_read(&buf
->commit_count
[endidx
].cc
);
883 ltt_check_deliver(chan
, buf
,
884 offsets
->end
- 1, commit_count
, endidx
);
885 ltt_write_commit_counter(chan
, buf
, endidx
,
886 offsets
->end
, commit_count
, padding_size
);
892 * !0 if execution must be aborted.
894 static int ltt_relay_try_switch_slow(
895 enum force_switch_mode mode
,
896 struct ust_channel
*chan
,
897 struct ust_buffer
*buf
,
898 struct ltt_reserve_switch_offsets
*offsets
,
902 long reserve_commit_diff
;
904 offsets
->begin
= uatomic_read(&buf
->offset
);
905 offsets
->old
= offsets
->begin
;
906 offsets
->begin_switch
= 0;
907 offsets
->end_switch_old
= 0;
909 *tsc
= trace_clock_read64();
911 if (SUBBUF_OFFSET(offsets
->begin
, buf
->chan
) != 0) {
912 offsets
->begin
= SUBBUF_ALIGN(offsets
->begin
, buf
->chan
);
913 offsets
->end_switch_old
= 1;
915 /* we do not have to switch : buffer is empty */
918 if (mode
== FORCE_ACTIVE
)
919 offsets
->begin
+= ltt_subbuffer_header_size();
921 * Always begin_switch in FORCE_ACTIVE mode.
922 * Test new buffer integrity
924 subbuf_index
= SUBBUF_INDEX(offsets
->begin
, buf
->chan
);
925 reserve_commit_diff
=
926 (BUFFER_TRUNC(offsets
->begin
, buf
->chan
)
927 >> chan
->n_subbufs_order
)
928 - (uatomic_read(&buf
->commit_count
[subbuf_index
].cc_sb
)
929 & chan
->commit_count_mask
);
930 if (reserve_commit_diff
== 0) {
931 /* Next buffer not corrupted. */
932 if (mode
== FORCE_ACTIVE
934 && offsets
->begin
- uatomic_read(&buf
->consumed
)
935 >= chan
->alloc_size
) {
937 * We do not overwrite non consumed buffers and we are
938 * full : ignore switch while tracing is active.
944 * Next subbuffer corrupted. Force pushing reader even in normal
948 offsets
->end
= offsets
->begin
;
953 * Force a sub-buffer switch for a per-cpu buffer. This operation is
954 * completely reentrant : can be called while tracing is active with
955 * absolutely no lock held.
957 void ltt_force_switch_lockless_slow(struct ust_buffer
*buf
,
958 enum force_switch_mode mode
)
960 struct ust_channel
*chan
= buf
->chan
;
961 struct ltt_reserve_switch_offsets offsets
;
966 DBG("Switching (forced) %s_%d", chan
->channel_name
, buf
->cpu
);
968 * Perform retryable operations.
971 if (ltt_relay_try_switch_slow(mode
, chan
, buf
,
974 } while (uatomic_cmpxchg(&buf
->offset
, offsets
.old
,
975 offsets
.end
) != offsets
.old
);
978 * Atomically update last_tsc. This update races against concurrent
979 * atomic updates, but the race will always cause supplementary full TSC
980 * events, never the opposite (missing a full TSC event when it would be
983 save_last_tsc(buf
, tsc
);
986 * Push the reader if necessary
988 if (mode
== FORCE_ACTIVE
) {
989 ltt_reserve_push_reader(chan
, buf
, offsets
.end
- 1);
990 //ust// ltt_clear_noref_flag(chan, buf, SUBBUF_INDEX(offsets.end - 1, chan));
994 * Switch old subbuffer if needed.
996 if (offsets
.end_switch_old
) {
997 //ust// ltt_clear_noref_flag(rchan, buf, SUBBUF_INDEX(offsets.old - 1, rchan));
998 ltt_reserve_switch_old_subbuf(chan
, buf
, &offsets
, &tsc
);
1002 * Populate new subbuffer.
1004 if (mode
== FORCE_ACTIVE
)
1005 ltt_reserve_switch_new_subbuf(chan
, buf
, &offsets
, &tsc
);
1011 * !0 if execution must be aborted.
1013 static int ltt_relay_try_reserve_slow(struct ust_channel
*chan
, struct ust_buffer
*buf
,
1014 struct ltt_reserve_switch_offsets
*offsets
, size_t data_size
,
1015 u64
*tsc
, unsigned int *rflags
, int largest_align
)
1017 long reserve_commit_diff
;
1019 offsets
->begin
= uatomic_read(&buf
->offset
);
1020 offsets
->old
= offsets
->begin
;
1021 offsets
->begin_switch
= 0;
1022 offsets
->end_switch_current
= 0;
1023 offsets
->end_switch_old
= 0;
1025 *tsc
= trace_clock_read64();
1026 if (last_tsc_overflow(buf
, *tsc
))
1027 *rflags
= LTT_RFLAG_ID_SIZE_TSC
;
1029 if (unlikely(SUBBUF_OFFSET(offsets
->begin
, buf
->chan
) == 0)) {
1030 offsets
->begin_switch
= 1; /* For offsets->begin */
1032 offsets
->size
= ust_get_header_size(chan
,
1033 offsets
->begin
, data_size
,
1034 &offsets
->before_hdr_pad
, *rflags
);
1035 offsets
->size
+= ltt_align(offsets
->begin
+ offsets
->size
,
1038 if (unlikely((SUBBUF_OFFSET(offsets
->begin
, buf
->chan
) +
1039 offsets
->size
) > buf
->chan
->subbuf_size
)) {
1040 offsets
->end_switch_old
= 1; /* For offsets->old */
1041 offsets
->begin_switch
= 1; /* For offsets->begin */
1044 if (unlikely(offsets
->begin_switch
)) {
1048 * We are typically not filling the previous buffer completely.
1050 if (likely(offsets
->end_switch_old
))
1051 offsets
->begin
= SUBBUF_ALIGN(offsets
->begin
,
1053 offsets
->begin
= offsets
->begin
+ ltt_subbuffer_header_size();
1054 /* Test new buffer integrity */
1055 subbuf_index
= SUBBUF_INDEX(offsets
->begin
, buf
->chan
);
1056 reserve_commit_diff
=
1057 (BUFFER_TRUNC(offsets
->begin
, buf
->chan
)
1058 >> chan
->n_subbufs_order
)
1059 - (uatomic_read(&buf
->commit_count
[subbuf_index
].cc_sb
)
1060 & chan
->commit_count_mask
);
1061 if (likely(reserve_commit_diff
== 0)) {
1062 /* Next buffer not corrupted. */
1063 if (unlikely(!chan
->overwrite
&&
1064 (SUBBUF_TRUNC(offsets
->begin
, buf
->chan
)
1065 - SUBBUF_TRUNC(uatomic_read(
1068 >= chan
->alloc_size
)) {
1070 * We do not overwrite non consumed buffers
1071 * and we are full : event is lost.
1073 uatomic_inc(&buf
->events_lost
);
1077 * next buffer not corrupted, we are either in
1078 * overwrite mode or the buffer is not full.
1079 * It's safe to write in this new subbuffer.
1084 * Next subbuffer corrupted. Drop event in normal and
1085 * overwrite mode. Caused by either a writer OOPS or
1086 * too many nested writes over a reserve/commit pair.
1088 uatomic_inc(&buf
->events_lost
);
1091 offsets
->size
= ust_get_header_size(chan
,
1092 offsets
->begin
, data_size
,
1093 &offsets
->before_hdr_pad
, *rflags
);
1094 offsets
->size
+= ltt_align(offsets
->begin
+ offsets
->size
,
1097 if (unlikely((SUBBUF_OFFSET(offsets
->begin
, buf
->chan
)
1098 + offsets
->size
) > buf
->chan
->subbuf_size
)) {
1100 * Event too big for subbuffers, report error, don't
1101 * complete the sub-buffer switch.
1103 uatomic_inc(&buf
->events_lost
);
1107 * We just made a successful buffer switch and the event
1108 * fits in the new subbuffer. Let's write.
1113 * Event fits in the current buffer and we are not on a switch
1114 * boundary. It's safe to write.
1117 offsets
->end
= offsets
->begin
+ offsets
->size
;
1119 if (unlikely((SUBBUF_OFFSET(offsets
->end
, buf
->chan
)) == 0)) {
1121 * The offset_end will fall at the very beginning of the next
1124 offsets
->end_switch_current
= 1; /* For offsets->begin */
1130 * ltt_relay_reserve_slot_lockless_slow - Atomic slot reservation in a buffer.
1131 * @trace: the trace structure to log to.
1132 * @ltt_channel: channel structure
1133 * @transport_data: data structure specific to ltt relay
1134 * @data_size: size of the variable length data to log.
1135 * @slot_size: pointer to total size of the slot (out)
1136 * @buf_offset : pointer to reserved buffer offset (out)
1137 * @tsc: pointer to the tsc at the slot reservation (out)
1140 * Return : -ENOSPC if not enough space, else returns 0.
1141 * It will take care of sub-buffer switching.
1143 int ltt_reserve_slot_lockless_slow(struct ust_channel
*chan
,
1144 struct ust_trace
*trace
, size_t data_size
,
1145 int largest_align
, int cpu
,
1146 struct ust_buffer
**ret_buf
,
1147 size_t *slot_size
, long *buf_offset
,
1148 u64
*tsc
, unsigned int *rflags
)
1150 struct ust_buffer
*buf
= *ret_buf
= chan
->buf
[cpu
];
1151 struct ltt_reserve_switch_offsets offsets
;
1156 if (unlikely(ltt_relay_try_reserve_slow(chan
, buf
, &offsets
,
1157 data_size
, tsc
, rflags
, largest_align
)))
1159 } while (unlikely(uatomic_cmpxchg(&buf
->offset
, offsets
.old
,
1160 offsets
.end
) != offsets
.old
));
1163 * Atomically update last_tsc. This update races against concurrent
1164 * atomic updates, but the race will always cause supplementary full TSC
1165 * events, never the opposite (missing a full TSC event when it would be
1168 save_last_tsc(buf
, *tsc
);
1171 * Push the reader if necessary
1173 ltt_reserve_push_reader(chan
, buf
, offsets
.end
- 1);
1176 * Clear noref flag for this subbuffer.
1178 //ust// ltt_clear_noref_flag(chan, buf, SUBBUF_INDEX(offsets.end - 1, chan));
1181 * Switch old subbuffer if needed.
1183 if (unlikely(offsets
.end_switch_old
)) {
1184 //ust// ltt_clear_noref_flag(chan, buf, SUBBUF_INDEX(offsets.old - 1, chan));
1185 ltt_reserve_switch_old_subbuf(chan
, buf
, &offsets
, tsc
);
1186 DBG("Switching %s_%d", chan
->channel_name
, cpu
);
1190 * Populate new subbuffer.
1192 if (unlikely(offsets
.begin_switch
))
1193 ltt_reserve_switch_new_subbuf(chan
, buf
, &offsets
, tsc
);
1195 if (unlikely(offsets
.end_switch_current
))
1196 ltt_reserve_end_switch_current(chan
, buf
, &offsets
, tsc
);
1198 *slot_size
= offsets
.size
;
1199 *buf_offset
= offsets
.begin
+ offsets
.before_hdr_pad
;
1203 static struct ltt_transport ust_relay_transport
= {
1206 .create_channel
= create_channel
,
1207 .finish_channel
= finish_channel
,
1208 .remove_channel
= remove_channel
,
1209 .wakeup_channel
= ltt_relay_async_wakeup_chan
,
1213 static char initialized
= 0;
1215 void __attribute__((constructor
)) init_ustrelay_transport(void)
1218 ltt_transport_register(&ust_relay_transport
);
1223 static void __attribute__((destructor
)) ust_buffers_exit(void)
1225 ltt_transport_unregister(&ust_relay_transport
);
1228 size_t ltt_write_event_header_slow(struct ust_channel
*channel
,
1229 struct ust_buffer
*buf
, long buf_offset
,
1230 u16 eID
, u32 event_size
,
1231 u64 tsc
, unsigned int rflags
)
1233 struct ltt_event_header header
;
1237 case LTT_RFLAG_ID_SIZE_TSC
:
1238 header
.id_time
= 29 << LTT_TSC_BITS
;
1240 case LTT_RFLAG_ID_SIZE
:
1241 header
.id_time
= 30 << LTT_TSC_BITS
;
1244 header
.id_time
= 31 << LTT_TSC_BITS
;
1252 header
.id_time
|= (u32
)tsc
& LTT_TSC_MASK
;
1253 ust_buffers_write(buf
, buf_offset
, &header
, sizeof(header
));
1254 buf_offset
+= sizeof(header
);
1257 case LTT_RFLAG_ID_SIZE_TSC
:
1258 small_size
= (u16
)min_t(u32
, event_size
, LTT_MAX_SMALL_SIZE
);
1259 ust_buffers_write(buf
, buf_offset
,
1261 buf_offset
+= sizeof(u16
);
1262 ust_buffers_write(buf
, buf_offset
,
1263 &small_size
, sizeof(u16
));
1264 buf_offset
+= sizeof(u16
);
1265 if (small_size
== LTT_MAX_SMALL_SIZE
) {
1266 ust_buffers_write(buf
, buf_offset
,
1267 &event_size
, sizeof(u32
));
1268 buf_offset
+= sizeof(u32
);
1270 buf_offset
+= ltt_align(buf_offset
, sizeof(u64
));
1271 ust_buffers_write(buf
, buf_offset
,
1273 buf_offset
+= sizeof(u64
);
1275 case LTT_RFLAG_ID_SIZE
:
1276 small_size
= (u16
)min_t(u32
, event_size
, LTT_MAX_SMALL_SIZE
);
1277 ust_buffers_write(buf
, buf_offset
,
1279 buf_offset
+= sizeof(u16
);
1280 ust_buffers_write(buf
, buf_offset
,
1281 &small_size
, sizeof(u16
));
1282 buf_offset
+= sizeof(u16
);
1283 if (small_size
== LTT_MAX_SMALL_SIZE
) {
1284 ust_buffers_write(buf
, buf_offset
,
1285 &event_size
, sizeof(u32
));
1286 buf_offset
+= sizeof(u32
);
1290 ust_buffers_write(buf
, buf_offset
,
1292 buf_offset
+= sizeof(u16
);