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
23 #ifndef _UST_BUFFERS_H
24 #define _UST_BUFFERS_H
29 #include <ust/clock.h>
31 #include "usterr_signal_safe.h"
33 #include "tracerconst.h"
34 #include "tracercore.h"
36 /***** FIXME: SHOULD BE REMOVED ***** */
39 * BUFFER_TRUNC zeroes the subbuffer offset and the subbuffer number parts of
40 * the offset, which leaves only the buffer number.
42 #define BUFFER_TRUNC(offset, chan) \
43 ((offset) & (~((chan)->alloc_size-1)))
44 #define BUFFER_OFFSET(offset, chan) ((offset) & ((chan)->alloc_size - 1))
45 #define SUBBUF_OFFSET(offset, chan) ((offset) & ((chan)->subbuf_size - 1))
46 #define SUBBUF_ALIGN(offset, chan) \
47 (((offset) + (chan)->subbuf_size) & (~((chan)->subbuf_size - 1)))
48 #define SUBBUF_TRUNC(offset, chan) \
49 ((offset) & (~((chan)->subbuf_size - 1)))
50 #define SUBBUF_INDEX(offset, chan) \
51 (BUFFER_OFFSET((offset), chan) >> (chan)->subbuf_size_order)
54 * Tracks changes to rchan/rchan_buf structs
56 #define UST_CHANNEL_VERSION 8
58 /**************************************/
60 struct commit_counters
{
62 long cc_sb
; /* ATOMIC - Incremented _once_ at sb switch */
66 /* First 32 bytes cache-hot cacheline */
67 long offset
; /* Current offset in the buffer *atomic* */
68 struct commit_counters
*commit_count
; /* Commit count per sub-buffer */
69 long consumed
; /* Current offset in the buffer *atomic* access (shared) */
70 unsigned long last_tsc
; /*
71 * Last timestamp written in the buffer.
73 /* End of first 32 bytes cacheline */
74 long active_readers
; /* ATOMIC - Active readers count standard atomic access (shared) */
75 long events_lost
; /* ATOMIC */
76 long corrupted_subbuffers
; /* *ATOMIC* */
77 /* one byte is written to this pipe when data is available, in order
78 to wake the consumer */
79 /* portability: Single byte writes must be as quick as possible. The kernel-side
80 buffer must be large enough so the writer doesn't block. From the pipe(7)
81 man page: Since linux 2.6.11, the pipe capacity is 65536 bytes. */
82 int data_ready_fd_write
;
83 /* the reading end of the pipe */
84 int data_ready_fd_read
;
86 * List of buffers with an open pipe, used for fork and forced subbuffer
89 struct cds_list_head open_buffers_list
;
91 unsigned int finalized
;
92 //ust// struct timer_list switch_timer; /* timer for periodical switch */
93 unsigned long switch_timer_interval
; /* 0 = unset */
95 struct ust_channel
*chan
;
97 struct urcu_ref urcu_ref
;
103 /* commit count per subbuffer; must be at end of struct */
104 long commit_seq
[0]; /* ATOMIC */
105 } ____cacheline_aligned
;
108 * A switch is done during tracing or as a final flush after tracing (so it
109 * won't write in the new sub-buffer).
110 * FIXME: make this message clearer
112 enum force_switch_mode
{ FORCE_ACTIVE
, FORCE_FLUSH
};
114 extern int ltt_reserve_slot_lockless_slow(struct ust_channel
*chan
,
115 struct ust_trace
*trace
, size_t data_size
,
116 int largest_align
, int cpu
,
117 struct ust_buffer
**ret_buf
,
118 size_t *slot_size
, long *buf_offset
,
119 u64
*tsc
, unsigned int *rflags
);
121 extern void ltt_force_switch_lockless_slow(struct ust_buffer
*buf
,
122 enum force_switch_mode mode
);
124 #ifndef HAVE_EFFICIENT_UNALIGNED_ACCESS
127 * Calculate the offset needed to align the type.
128 * size_of_type must be non-zero.
130 static inline unsigned int ltt_align(size_t align_drift
, size_t size_of_type
)
132 size_t alignment
= min(sizeof(void *), size_of_type
);
133 return (alignment
- align_drift
) & (alignment
- 1);
135 /* Default arch alignment */
138 static inline int ltt_get_alignment(void)
140 return sizeof(void *);
143 #else /* HAVE_EFFICIENT_UNALIGNED_ACCESS */
145 static inline unsigned int ltt_align(size_t align_drift
,
151 #define LTT_ALIGN __attribute__((packed))
153 static inline int ltt_get_alignment(void)
157 #endif /* HAVE_EFFICIENT_UNALIGNED_ACCESS */
159 static __inline__
void ust_buffers_do_copy(void *dest
, const void *src
, size_t len
)
167 } u
= { .src
= src
};
171 case 1: *(u8
*)dest
= *u
.src8
;
173 case 2: *(u16
*)dest
= *u
.src16
;
175 case 4: *(u32
*)dest
= *u
.src32
;
177 case 8: *(u64
*)dest
= *u
.src64
;
180 memcpy(dest
, src
, len
);
184 static __inline__
void *ust_buffers_offset_address(struct ust_buffer
*buf
, size_t offset
)
186 return ((char *)buf
->buf_data
)+offset
;
190 * Last TSC comparison functions. Check if the current TSC overflows
191 * LTT_TSC_BITS bits from the last TSC read. Reads and writes last_tsc
195 /* FIXME: does this test work properly? */
196 #if (BITS_PER_LONG == 32)
197 static __inline__
void save_last_tsc(struct ust_buffer
*ltt_buf
,
200 ltt_buf
->last_tsc
= (unsigned long)(tsc
>> LTT_TSC_BITS
);
203 static __inline__
int last_tsc_overflow(struct ust_buffer
*ltt_buf
,
206 unsigned long tsc_shifted
= (unsigned long)(tsc
>> LTT_TSC_BITS
);
208 if (unlikely((tsc_shifted
- ltt_buf
->last_tsc
)))
214 static __inline__
void save_last_tsc(struct ust_buffer
*ltt_buf
,
217 ltt_buf
->last_tsc
= (unsigned long)tsc
;
220 static __inline__
int last_tsc_overflow(struct ust_buffer
*ltt_buf
,
223 if (unlikely((tsc
- ltt_buf
->last_tsc
) >> LTT_TSC_BITS
))
231 * ust_get_header_size
233 * Calculate alignment offset to 32-bits. This is the alignment offset of the
237 * The event header must be 32-bits. The total offset calculated here :
239 * Alignment of header struct on 32 bits (min arch size, header size)
240 * + sizeof(header struct) (32-bits)
241 * + (opt) u16 (ext. event id)
242 * + (opt) u16 (event_size) (if event_size == 0xFFFFUL, has ext. event size)
243 * + (opt) u32 (ext. event size)
244 * + (opt) u64 full TSC (aligned on min(64-bits, arch size))
246 * The payload must itself determine its own alignment from the biggest type it
249 static __inline__
unsigned char ust_get_header_size(
250 struct ust_channel
*channel
,
253 size_t *before_hdr_pad
,
256 size_t orig_offset
= offset
;
259 padding
= ltt_align(offset
, sizeof(struct ltt_event_header
));
261 offset
+= sizeof(struct ltt_event_header
);
263 if(unlikely(rflags
)) {
265 case LTT_RFLAG_ID_SIZE_TSC
:
266 offset
+= sizeof(u16
) + sizeof(u16
);
267 if (data_size
>= 0xFFFFU
)
268 offset
+= sizeof(u32
);
269 offset
+= ltt_align(offset
, sizeof(u64
));
270 offset
+= sizeof(u64
);
272 case LTT_RFLAG_ID_SIZE
:
273 offset
+= sizeof(u16
) + sizeof(u16
);
274 if (data_size
>= 0xFFFFU
)
275 offset
+= sizeof(u32
);
278 offset
+= sizeof(u16
);
283 *before_hdr_pad
= padding
;
284 return offset
- orig_offset
;
287 static __inline__
void ltt_reserve_push_reader(
288 struct ust_channel
*rchan
,
289 struct ust_buffer
*buf
,
292 long consumed_old
, consumed_new
;
295 consumed_old
= uatomic_read(&buf
->consumed
);
297 * If buffer is in overwrite mode, push the reader consumed
298 * count if the write position has reached it and we are not
299 * at the first iteration (don't push the reader farther than
300 * the writer). This operation can be done concurrently by many
301 * writers in the same buffer, the writer being at the farthest
302 * write position sub-buffer index in the buffer being the one
303 * which will win this loop.
304 * If the buffer is not in overwrite mode, pushing the reader
305 * only happens if a sub-buffer is corrupted.
307 if (unlikely((SUBBUF_TRUNC(offset
, buf
->chan
)
308 - SUBBUF_TRUNC(consumed_old
, buf
->chan
))
309 >= rchan
->alloc_size
))
310 consumed_new
= SUBBUF_ALIGN(consumed_old
, buf
->chan
);
313 } while (unlikely(uatomic_cmpxchg(&buf
->consumed
, consumed_old
,
314 consumed_new
) != consumed_old
));
317 static __inline__
void ltt_vmcore_check_deliver(
318 struct ust_buffer
*buf
,
319 long commit_count
, long idx
)
321 uatomic_set(&buf
->commit_seq
[idx
], commit_count
);
324 static __inline__
void ltt_check_deliver(struct ust_channel
*chan
,
325 struct ust_buffer
*buf
,
326 long offset
, long commit_count
, long idx
)
328 long old_commit_count
= commit_count
- chan
->subbuf_size
;
330 /* Check if all commits have been done */
331 if (unlikely((BUFFER_TRUNC(offset
, chan
)
332 >> chan
->n_subbufs_order
)
334 & chan
->commit_count_mask
) == 0)) {
336 * If we succeeded in updating the cc_sb, we are delivering
337 * the subbuffer. Deals with concurrent updates of the "cc"
338 * value without adding a add_return atomic operation to the
341 if (likely(uatomic_cmpxchg(&buf
->commit_count
[idx
].cc_sb
,
342 old_commit_count
, commit_count
)
343 == old_commit_count
)) {
347 * Set noref flag for this subbuffer.
349 //ust// ltt_set_noref_flag(rchan, buf, idx);
350 ltt_vmcore_check_deliver(buf
, commit_count
, idx
);
352 /* wakeup consumer */
353 result
= write(buf
->data_ready_fd_write
, "1", 1);
355 PERROR("write (in ltt_relay_buffer_flush)");
356 ERR("this should never happen!");
362 static __inline__
int ltt_poll_deliver(struct ust_channel
*chan
, struct ust_buffer
*buf
)
364 long consumed_old
, consumed_idx
, commit_count
, write_offset
;
366 consumed_old
= uatomic_read(&buf
->consumed
);
367 consumed_idx
= SUBBUF_INDEX(consumed_old
, buf
->chan
);
368 commit_count
= uatomic_read(&buf
->commit_count
[consumed_idx
].cc_sb
);
370 * No memory cmm_barrier here, since we are only interested
371 * in a statistically correct polling result. The next poll will
372 * get the data is we are racing. The mb() that ensures correct
373 * memory order is in get_subbuf.
375 write_offset
= uatomic_read(&buf
->offset
);
378 * Check that the subbuffer we are trying to consume has been
379 * already fully committed.
382 if (((commit_count
- chan
->subbuf_size
)
383 & chan
->commit_count_mask
)
384 - (BUFFER_TRUNC(consumed_old
, buf
->chan
)
385 >> chan
->n_subbufs_order
)
390 * Check that we are not about to read the same subbuffer in
391 * which the writer head is.
393 if ((SUBBUF_TRUNC(write_offset
, buf
->chan
)
394 - SUBBUF_TRUNC(consumed_old
, buf
->chan
))
403 * returns 0 if reserve ok, or 1 if the slow path must be taken.
405 static __inline__
int ltt_relay_try_reserve(
406 struct ust_channel
*chan
,
407 struct ust_buffer
*buf
,
409 u64
*tsc
, unsigned int *rflags
, int largest_align
,
410 long *o_begin
, long *o_end
, long *o_old
,
411 size_t *before_hdr_pad
, size_t *size
)
413 *o_begin
= uatomic_read(&buf
->offset
);
416 *tsc
= trace_clock_read64();
418 //ust// #ifdef CONFIG_LTT_VMCORE
419 //ust// prefetch(&buf->commit_count[SUBBUF_INDEX(*o_begin, rchan)]);
420 //ust// prefetch(&buf->commit_seq[SUBBUF_INDEX(*o_begin, rchan)]);
422 //ust// prefetchw(&buf->commit_count[SUBBUF_INDEX(*o_begin, rchan)]);
424 if (last_tsc_overflow(buf
, *tsc
))
425 *rflags
= LTT_RFLAG_ID_SIZE_TSC
;
427 if (unlikely(SUBBUF_OFFSET(*o_begin
, buf
->chan
) == 0))
430 *size
= ust_get_header_size(chan
,
432 before_hdr_pad
, *rflags
);
433 *size
+= ltt_align(*o_begin
+ *size
, largest_align
) + data_size
;
434 if (unlikely((SUBBUF_OFFSET(*o_begin
, buf
->chan
) + *size
)
435 > buf
->chan
->subbuf_size
))
439 * Event fits in the current buffer and we are not on a switch
440 * boundary. It's safe to write.
442 *o_end
= *o_begin
+ *size
;
444 if (unlikely((SUBBUF_OFFSET(*o_end
, buf
->chan
)) == 0))
446 * The offset_end will fall at the very beginning of the next
454 static __inline__
int ltt_reserve_slot(struct ust_channel
*chan
,
455 struct ust_trace
*trace
, size_t data_size
,
456 int largest_align
, int cpu
,
457 struct ust_buffer
**ret_buf
,
458 size_t *slot_size
, long *buf_offset
, u64
*tsc
,
459 unsigned int *rflags
)
461 struct ust_buffer
*buf
= *ret_buf
= chan
->buf
[cpu
];
462 long o_begin
, o_end
, o_old
;
463 size_t before_hdr_pad
;
466 * Perform retryable operations.
468 /* FIXME: make this really per cpu? */
469 if (unlikely(CMM_LOAD_SHARED(ltt_nesting
) > 4)) {
470 DBG("Dropping event because nesting is too deep.");
471 uatomic_inc(&buf
->events_lost
);
475 if (unlikely(ltt_relay_try_reserve(chan
, buf
,
476 data_size
, tsc
, rflags
,
477 largest_align
, &o_begin
, &o_end
, &o_old
,
478 &before_hdr_pad
, slot_size
)))
481 if (unlikely(uatomic_cmpxchg(&buf
->offset
, o_old
, o_end
) != o_old
))
485 * Atomically update last_tsc. This update races against concurrent
486 * atomic updates, but the race will always cause supplementary full TSC
487 * events, never the opposite (missing a full TSC event when it would be
490 save_last_tsc(buf
, *tsc
);
493 * Push the reader if necessary
495 ltt_reserve_push_reader(chan
, buf
, o_end
- 1);
498 * Clear noref flag for this subbuffer.
500 //ust// ltt_clear_noref_flag(chan, buf, SUBBUF_INDEX(o_end - 1, chan));
502 *buf_offset
= o_begin
+ before_hdr_pad
;
505 return ltt_reserve_slot_lockless_slow(chan
, trace
, data_size
,
506 largest_align
, cpu
, ret_buf
,
507 slot_size
, buf_offset
, tsc
,
512 * Force a sub-buffer switch for a per-cpu buffer. This operation is
513 * completely reentrant : can be called while tracing is active with
514 * absolutely no lock held.
516 static __inline__
void ltt_force_switch(struct ust_buffer
*buf
,
517 enum force_switch_mode mode
)
519 return ltt_force_switch_lockless_slow(buf
, mode
);
523 * for flight recording. must be called after relay_commit.
524 * This function increments the subbuffers's commit_seq counter each time the
525 * commit count reaches back the reserve offset (module subbuffer size). It is
526 * useful for crash dump.
528 //ust// #ifdef CONFIG_LTT_VMCORE
529 static __inline__
void ltt_write_commit_counter(struct ust_channel
*chan
,
530 struct ust_buffer
*buf
, long idx
, long buf_offset
,
531 long commit_count
, size_t data_size
)
536 offset
= buf_offset
+ data_size
;
539 * SUBBUF_OFFSET includes commit_count_mask. We can simply
540 * compare the offsets within the subbuffer without caring about
541 * buffer full/empty mismatch because offset is never zero here
542 * (subbuffer header and event headers have non-zero length).
544 if (unlikely(SUBBUF_OFFSET(offset
- commit_count
, buf
->chan
)))
547 commit_seq_old
= uatomic_read(&buf
->commit_seq
[idx
]);
548 while (commit_seq_old
< commit_count
)
549 commit_seq_old
= uatomic_cmpxchg(&buf
->commit_seq
[idx
],
550 commit_seq_old
, commit_count
);
552 DBG("commit_seq for channel %s_%d, subbuf %ld is now %ld", buf
->chan
->channel_name
, buf
->cpu
, idx
, commit_count
);
555 //ust// static __inline__ void ltt_write_commit_counter(struct ust_buffer *buf,
556 //ust// long idx, long buf_offset, long commit_count, size_t data_size)
562 * Atomic unordered slot commit. Increments the commit count in the
563 * specified sub-buffer, and delivers it if necessary.
567 * @ltt_channel : channel structure
568 * @transport_data: transport-specific data
569 * @buf_offset : offset following the event header.
570 * @data_size : size of the event data.
571 * @slot_size : size of the reserved slot.
573 static __inline__
void ltt_commit_slot(
574 struct ust_channel
*chan
,
575 struct ust_buffer
*buf
, long buf_offset
,
576 size_t data_size
, size_t slot_size
)
578 long offset_end
= buf_offset
;
579 long endidx
= SUBBUF_INDEX(offset_end
- 1, chan
);
584 uatomic_add(&buf
->commit_count
[endidx
].cc
, slot_size
);
586 * commit count read can race with concurrent OOO commit count updates.
587 * This is only needed for ltt_check_deliver (for non-polling delivery
588 * only) and for ltt_write_commit_counter. The race can only cause the
589 * counter to be read with the same value more than once, which could
591 * - Multiple delivery for the same sub-buffer (which is handled
592 * gracefully by the reader code) if the value is for a full
593 * sub-buffer. It's important that we can never miss a sub-buffer
594 * delivery. Re-reading the value after the uatomic_add ensures this.
595 * - Reading a commit_count with a higher value that what was actually
596 * added to it for the ltt_write_commit_counter call (again caused by
597 * a concurrent committer). It does not matter, because this function
598 * is interested in the fact that the commit count reaches back the
599 * reserve offset for a specific sub-buffer, which is completely
600 * independent of the order.
602 commit_count
= uatomic_read(&buf
->commit_count
[endidx
].cc
);
604 ltt_check_deliver(chan
, buf
, offset_end
- 1, commit_count
, endidx
);
606 * Update data_size for each commit. It's needed only for extracting
607 * ltt buffers from vmcore, after crash.
609 ltt_write_commit_counter(chan
, buf
, endidx
, buf_offset
, commit_count
, data_size
);
612 void _ust_buffers_strncpy_fixup(struct ust_buffer
*buf
, size_t offset
,
613 size_t len
, size_t copied
, int terminated
);
615 static __inline__
int ust_buffers_write(struct ust_buffer
*buf
, size_t offset
,
616 const void *src
, size_t len
)
618 size_t buf_offset
= BUFFER_OFFSET(offset
, buf
->chan
);
620 assert(buf_offset
< buf
->chan
->subbuf_size
*buf
->chan
->subbuf_cnt
);
621 assert(buf_offset
+ len
622 <= buf
->chan
->subbuf_size
*buf
->chan
->subbuf_cnt
);
624 ust_buffers_do_copy(buf
->buf_data
+ buf_offset
, src
, len
);
630 * ust_buffers_do_memset - write character into dest.
632 * @src: source character
633 * @len: length to write
636 void ust_buffers_do_memset(void *dest
, char src
, size_t len
)
639 * What we really want here is an __inline__ memset, but we
640 * don't have constants, so gcc generally uses a function call.
642 for (; len
> 0; len
--)
647 * ust_buffers_do_strncpy - copy a string up to a certain number of bytes
650 * @len: max. length to copy
651 * @terminated: output string ends with \0 (output)
653 * returns the number of bytes copied. Does not finalize with \0 if len is
657 size_t ust_buffers_do_strncpy(void *dest
, const void *src
, size_t len
,
660 size_t orig_len
= len
;
664 * What we really want here is an __inline__ strncpy, but we
665 * don't have constants, so gcc generally uses a function call.
667 for (; len
> 0; len
--) {
668 *(u8
*)dest
= CMM_LOAD_SHARED(*(const u8
*)src
);
669 /* Check with dest, because src may be modified concurrently */
670 if (*(const u8
*)dest
== '\0') {
678 return orig_len
- len
;
682 int ust_buffers_strncpy(struct ust_buffer
*buf
, size_t offset
, const void *src
,
685 size_t buf_offset
= BUFFER_OFFSET(offset
, buf
->chan
);
689 assert(buf_offset
< buf
->chan
->subbuf_size
*buf
->chan
->subbuf_cnt
);
690 assert(buf_offset
+ len
691 <= buf
->chan
->subbuf_size
*buf
->chan
->subbuf_cnt
);
693 copied
= ust_buffers_do_strncpy(buf
->buf_data
+ buf_offset
,
694 src
, len
, &terminated
);
695 if (unlikely(copied
< len
|| !terminated
))
696 _ust_buffers_strncpy_fixup(buf
, offset
, len
, copied
,
701 extern int ust_buffers_get_subbuf(struct ust_buffer
*buf
, long *consumed
);
702 extern int ust_buffers_put_subbuf(struct ust_buffer
*buf
, unsigned long uconsumed_old
);
704 extern void init_ustrelay_transport(void);
706 #endif /* _UST_BUFFERS_H */