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
27 #include <ust/kernelcompat.h>
30 #include "tracerconst.h"
31 #include "tracercore.h"
32 #include "header-inline.h"
34 /***** FIXME: SHOULD BE REMOVED ***** */
37 * BUFFER_TRUNC zeroes the subbuffer offset and the subbuffer number parts of
38 * the offset, which leaves only the buffer number.
40 #define BUFFER_TRUNC(offset, chan) \
41 ((offset) & (~((chan)->alloc_size-1)))
42 #define BUFFER_OFFSET(offset, chan) ((offset) & ((chan)->alloc_size - 1))
43 #define SUBBUF_OFFSET(offset, chan) ((offset) & ((chan)->subbuf_size - 1))
44 #define SUBBUF_ALIGN(offset, chan) \
45 (((offset) + (chan)->subbuf_size) & (~((chan)->subbuf_size - 1)))
46 #define SUBBUF_TRUNC(offset, chan) \
47 ((offset) & (~((chan)->subbuf_size - 1)))
48 #define SUBBUF_INDEX(offset, chan) \
49 (BUFFER_OFFSET((offset), chan) >> (chan)->subbuf_size_order)
52 * Tracks changes to rchan/rchan_buf structs
54 #define UST_CHANNEL_VERSION 8
56 /**************************************/
58 struct commit_counters
{
60 long cc_sb
; /* ATOMIC - Incremented _once_ at sb switch */
64 /* First 32 bytes cache-hot cacheline */
65 long offset
; /* Current offset in the buffer *atomic* */
66 struct commit_counters
*commit_count
; /* Commit count per sub-buffer */
67 long consumed
; /* Current offset in the buffer *atomic* access (shared) */
68 unsigned long last_tsc
; /*
69 * Last timestamp written in the buffer.
71 /* End of first 32 bytes cacheline */
72 long active_readers
; /* ATOMIC - Active readers count standard atomic access (shared) */
73 long events_lost
; /* ATOMIC */
74 long corrupted_subbuffers
; /* *ATOMIC* */
75 /* one byte is written to this pipe when data is available, in order
76 to wake the consumer */
77 /* portability: Single byte writes must be as quick as possible. The kernel-side
78 buffer must be large enough so the writer doesn't block. From the pipe(7)
79 man page: Since linux 2.6.11, the pipe capacity is 65536 bytes. */
80 int data_ready_fd_write
;
81 /* the reading end of the pipe */
82 int data_ready_fd_read
;
84 unsigned int finalized
;
85 //ust// struct timer_list switch_timer; /* timer for periodical switch */
86 unsigned long switch_timer_interval
; /* 0 = unset */
88 struct ust_channel
*chan
;
96 /* commit count per subbuffer; must be at end of struct */
97 long commit_seq
[0] ____cacheline_aligned
; /* ATOMIC */
98 } ____cacheline_aligned
;
101 * A switch is done during tracing or as a final flush after tracing (so it
102 * won't write in the new sub-buffer).
103 * FIXME: make this message clearer
105 enum force_switch_mode
{ FORCE_ACTIVE
, FORCE_FLUSH
};
107 extern int ltt_reserve_slot_lockless_slow(struct ust_trace
*trace
,
108 struct ust_channel
*ltt_channel
, void **transport_data
,
109 size_t data_size
, size_t *slot_size
, long *buf_offset
, u64
*tsc
,
110 unsigned int *rflags
, int largest_align
, int cpu
);
112 extern void ltt_force_switch_lockless_slow(struct ust_buffer
*buf
,
113 enum force_switch_mode mode
);
116 static __inline__
void ust_buffers_do_copy(void *dest
, const void *src
, size_t len
)
124 } u
= { .src
= src
};
128 case 1: *(u8
*)dest
= *u
.src8
;
130 case 2: *(u16
*)dest
= *u
.src16
;
132 case 4: *(u32
*)dest
= *u
.src32
;
134 case 8: *(u64
*)dest
= *u
.src64
;
137 memcpy(dest
, src
, len
);
141 static __inline__
void *ust_buffers_offset_address(struct ust_buffer
*buf
, size_t offset
)
143 return ((char *)buf
->buf_data
)+offset
;
147 * Last TSC comparison functions. Check if the current TSC overflows
148 * LTT_TSC_BITS bits from the last TSC read. Reads and writes last_tsc
152 /* FIXME: does this test work properly? */
153 #if (BITS_PER_LONG == 32)
154 static __inline__
void save_last_tsc(struct ust_buffer
*ltt_buf
,
157 ltt_buf
->last_tsc
= (unsigned long)(tsc
>> LTT_TSC_BITS
);
160 static __inline__
int last_tsc_overflow(struct ust_buffer
*ltt_buf
,
163 unsigned long tsc_shifted
= (unsigned long)(tsc
>> LTT_TSC_BITS
);
165 if (unlikely((tsc_shifted
- ltt_buf
->last_tsc
)))
171 static __inline__
void save_last_tsc(struct ust_buffer
*ltt_buf
,
174 ltt_buf
->last_tsc
= (unsigned long)tsc
;
177 static __inline__
int last_tsc_overflow(struct ust_buffer
*ltt_buf
,
180 if (unlikely((tsc
- ltt_buf
->last_tsc
) >> LTT_TSC_BITS
))
187 static __inline__
void ltt_reserve_push_reader(
188 struct ust_channel
*rchan
,
189 struct ust_buffer
*buf
,
192 long consumed_old
, consumed_new
;
195 consumed_old
= uatomic_read(&buf
->consumed
);
197 * If buffer is in overwrite mode, push the reader consumed
198 * count if the write position has reached it and we are not
199 * at the first iteration (don't push the reader farther than
200 * the writer). This operation can be done concurrently by many
201 * writers in the same buffer, the writer being at the farthest
202 * write position sub-buffer index in the buffer being the one
203 * which will win this loop.
204 * If the buffer is not in overwrite mode, pushing the reader
205 * only happens if a sub-buffer is corrupted.
207 if (unlikely((SUBBUF_TRUNC(offset
, buf
->chan
)
208 - SUBBUF_TRUNC(consumed_old
, buf
->chan
))
209 >= rchan
->alloc_size
))
210 consumed_new
= SUBBUF_ALIGN(consumed_old
, buf
->chan
);
213 } while (unlikely(uatomic_cmpxchg(&buf
->consumed
, consumed_old
,
214 consumed_new
) != consumed_old
));
217 static __inline__
void ltt_vmcore_check_deliver(
218 struct ust_buffer
*buf
,
219 long commit_count
, long idx
)
221 uatomic_set(&buf
->commit_seq
[idx
], commit_count
);
224 static __inline__
void ltt_check_deliver(struct ust_channel
*chan
,
225 struct ust_buffer
*buf
,
226 long offset
, long commit_count
, long idx
)
228 long old_commit_count
= commit_count
- chan
->subbuf_size
;
230 /* Check if all commits have been done */
231 if (unlikely((BUFFER_TRUNC(offset
, chan
)
232 >> chan
->n_subbufs_order
)
234 & chan
->commit_count_mask
) == 0)) {
236 * If we succeeded in updating the cc_sb, we are delivering
237 * the subbuffer. Deals with concurrent updates of the "cc"
238 * value without adding a add_return atomic operation to the
241 if (likely(uatomic_cmpxchg(&buf
->commit_count
[idx
].cc_sb
,
242 old_commit_count
, commit_count
)
243 == old_commit_count
)) {
247 * Set noref flag for this subbuffer.
249 //ust// ltt_set_noref_flag(rchan, buf, idx);
250 ltt_vmcore_check_deliver(buf
, commit_count
, idx
);
252 /* wakeup consumer */
253 result
= write(buf
->data_ready_fd_write
, "1", 1);
255 PERROR("write (in ltt_relay_buffer_flush)");
256 ERR("this should never happen!");
262 static __inline__
int ltt_poll_deliver(struct ust_channel
*chan
, struct ust_buffer
*buf
)
264 long consumed_old
, consumed_idx
, commit_count
, write_offset
;
266 consumed_old
= uatomic_read(&buf
->consumed
);
267 consumed_idx
= SUBBUF_INDEX(consumed_old
, buf
->chan
);
268 commit_count
= uatomic_read(&buf
->commit_count
[consumed_idx
].cc_sb
);
270 * No memory barrier here, since we are only interested
271 * in a statistically correct polling result. The next poll will
272 * get the data is we are racing. The mb() that ensures correct
273 * memory order is in get_subbuf.
275 write_offset
= uatomic_read(&buf
->offset
);
278 * Check that the subbuffer we are trying to consume has been
279 * already fully committed.
282 if (((commit_count
- chan
->subbuf_size
)
283 & chan
->commit_count_mask
)
284 - (BUFFER_TRUNC(consumed_old
, buf
->chan
)
285 >> chan
->n_subbufs_order
)
290 * Check that we are not about to read the same subbuffer in
291 * which the writer head is.
293 if ((SUBBUF_TRUNC(write_offset
, buf
->chan
)
294 - SUBBUF_TRUNC(consumed_old
, buf
->chan
))
303 * returns 0 if reserve ok, or 1 if the slow path must be taken.
305 static __inline__
int ltt_relay_try_reserve(
306 struct ust_channel
*chan
,
307 struct ust_buffer
*buf
,
309 u64
*tsc
, unsigned int *rflags
, int largest_align
,
310 long *o_begin
, long *o_end
, long *o_old
,
311 size_t *before_hdr_pad
, size_t *size
)
313 *o_begin
= uatomic_read(&buf
->offset
);
316 *tsc
= trace_clock_read64();
318 //ust// #ifdef CONFIG_LTT_VMCORE
319 //ust// prefetch(&buf->commit_count[SUBBUF_INDEX(*o_begin, rchan)]);
320 //ust// prefetch(&buf->commit_seq[SUBBUF_INDEX(*o_begin, rchan)]);
322 //ust// prefetchw(&buf->commit_count[SUBBUF_INDEX(*o_begin, rchan)]);
324 if (last_tsc_overflow(buf
, *tsc
))
325 *rflags
= LTT_RFLAG_ID_SIZE_TSC
;
327 if (unlikely(SUBBUF_OFFSET(*o_begin
, buf
->chan
) == 0))
330 *size
= ust_get_header_size(chan
,
332 before_hdr_pad
, *rflags
);
333 *size
+= ltt_align(*o_begin
+ *size
, largest_align
) + data_size
;
334 if (unlikely((SUBBUF_OFFSET(*o_begin
, buf
->chan
) + *size
)
335 > buf
->chan
->subbuf_size
))
339 * Event fits in the current buffer and we are not on a switch
340 * boundary. It's safe to write.
342 *o_end
= *o_begin
+ *size
;
344 if (unlikely((SUBBUF_OFFSET(*o_end
, buf
->chan
)) == 0))
346 * The offset_end will fall at the very beginning of the next
354 static __inline__
int ltt_reserve_slot(struct ust_trace
*trace
,
355 struct ust_channel
*chan
, void **transport_data
,
356 size_t data_size
, size_t *slot_size
, long *buf_offset
, u64
*tsc
,
357 unsigned int *rflags
, int largest_align
, int cpu
)
359 struct ust_buffer
*buf
= chan
->buf
[cpu
];
360 long o_begin
, o_end
, o_old
;
361 size_t before_hdr_pad
;
364 * Perform retryable operations.
366 /* FIXME: make this rellay per cpu? */
367 if (unlikely(LOAD_SHARED(ltt_nesting
) > 4)) {
368 DBG("Dropping event because nesting is too deep.");
369 uatomic_inc(&buf
->events_lost
);
373 if (unlikely(ltt_relay_try_reserve(chan
, buf
,
374 data_size
, tsc
, rflags
,
375 largest_align
, &o_begin
, &o_end
, &o_old
,
376 &before_hdr_pad
, slot_size
)))
379 if (unlikely(uatomic_cmpxchg(&buf
->offset
, o_old
, o_end
) != o_old
))
383 * Atomically update last_tsc. This update races against concurrent
384 * atomic updates, but the race will always cause supplementary full TSC
385 * events, never the opposite (missing a full TSC event when it would be
388 save_last_tsc(buf
, *tsc
);
391 * Push the reader if necessary
393 ltt_reserve_push_reader(chan
, buf
, o_end
- 1);
396 * Clear noref flag for this subbuffer.
398 //ust// ltt_clear_noref_flag(chan, buf, SUBBUF_INDEX(o_end - 1, chan));
400 *buf_offset
= o_begin
+ before_hdr_pad
;
403 return ltt_reserve_slot_lockless_slow(trace
, chan
,
404 transport_data
, data_size
, slot_size
, buf_offset
, tsc
,
405 rflags
, largest_align
, cpu
);
409 * Force a sub-buffer switch for a per-cpu buffer. This operation is
410 * completely reentrant : can be called while tracing is active with
411 * absolutely no lock held.
413 static __inline__
void ltt_force_switch(struct ust_buffer
*buf
,
414 enum force_switch_mode mode
)
416 return ltt_force_switch_lockless_slow(buf
, mode
);
420 * for flight recording. must be called after relay_commit.
421 * This function increments the subbuffers's commit_seq counter each time the
422 * commit count reaches back the reserve offset (module subbuffer size). It is
423 * useful for crash dump.
425 //ust// #ifdef CONFIG_LTT_VMCORE
426 static __inline__
void ltt_write_commit_counter(struct ust_channel
*chan
,
427 struct ust_buffer
*buf
, long idx
, long buf_offset
,
428 long commit_count
, size_t data_size
)
433 offset
= buf_offset
+ data_size
;
436 * SUBBUF_OFFSET includes commit_count_mask. We can simply
437 * compare the offsets within the subbuffer without caring about
438 * buffer full/empty mismatch because offset is never zero here
439 * (subbuffer header and event headers have non-zero length).
441 if (unlikely(SUBBUF_OFFSET(offset
- commit_count
, buf
->chan
)))
444 commit_seq_old
= uatomic_read(&buf
->commit_seq
[idx
]);
445 while (commit_seq_old
< commit_count
)
446 commit_seq_old
= uatomic_cmpxchg(&buf
->commit_seq
[idx
],
447 commit_seq_old
, commit_count
);
449 DBG("commit_seq for channel %s_%d, subbuf %ld is now %ld", buf
->chan
->channel_name
, buf
->cpu
, idx
, commit_count
);
452 //ust// static __inline__ void ltt_write_commit_counter(struct ust_buffer *buf,
453 //ust// long idx, long buf_offset, long commit_count, size_t data_size)
459 * Atomic unordered slot commit. Increments the commit count in the
460 * specified sub-buffer, and delivers it if necessary.
464 * @ltt_channel : channel structure
465 * @transport_data: transport-specific data
466 * @buf_offset : offset following the event header.
467 * @data_size : size of the event data.
468 * @slot_size : size of the reserved slot.
470 static __inline__
void ltt_commit_slot(
471 struct ust_channel
*chan
,
472 struct ust_buffer
*buf
, long buf_offset
,
473 size_t data_size
, size_t slot_size
)
475 long offset_end
= buf_offset
;
476 long endidx
= SUBBUF_INDEX(offset_end
- 1, chan
);
479 #ifdef LTT_NO_IPI_BARRIER
483 * Must write slot data before incrementing commit count.
484 * This compiler barrier is upgraded into a smp_mb() by the IPI
485 * sent by get_subbuf().
489 uatomic_add(&buf
->commit_count
[endidx
].cc
, slot_size
);
491 * commit count read can race with concurrent OOO commit count updates.
492 * This is only needed for ltt_check_deliver (for non-polling delivery
493 * only) and for ltt_write_commit_counter. The race can only cause the
494 * counter to be read with the same value more than once, which could
496 * - Multiple delivery for the same sub-buffer (which is handled
497 * gracefully by the reader code) if the value is for a full
498 * sub-buffer. It's important that we can never miss a sub-buffer
499 * delivery. Re-reading the value after the uatomic_add ensures this.
500 * - Reading a commit_count with a higher value that what was actually
501 * added to it for the ltt_write_commit_counter call (again caused by
502 * a concurrent committer). It does not matter, because this function
503 * is interested in the fact that the commit count reaches back the
504 * reserve offset for a specific sub-buffer, which is completely
505 * independent of the order.
507 commit_count
= uatomic_read(&buf
->commit_count
[endidx
].cc
);
509 ltt_check_deliver(chan
, buf
, offset_end
- 1, commit_count
, endidx
);
511 * Update data_size for each commit. It's needed only for extracting
512 * ltt buffers from vmcore, after crash.
514 ltt_write_commit_counter(chan
, buf
, endidx
, buf_offset
, commit_count
, data_size
);
517 void _ust_buffers_write(struct ust_buffer
*buf
, size_t offset
,
518 const void *src
, size_t len
, ssize_t cpy
);
520 static __inline__
int ust_buffers_write(struct ust_buffer
*buf
, size_t offset
,
521 const void *src
, size_t len
)
524 size_t buf_offset
= BUFFER_OFFSET(offset
, buf
->chan
);
526 assert(buf_offset
< buf
->chan
->subbuf_size
*buf
->chan
->subbuf_cnt
);
528 cpy
= min_t(size_t, len
, buf
->buf_size
- buf_offset
);
529 ust_buffers_do_copy(buf
->buf_data
+ buf_offset
, src
, cpy
);
531 if (unlikely(len
!= cpy
))
532 _ust_buffers_write(buf
, buf_offset
, src
, len
, cpy
);
536 extern int ust_buffers_get_subbuf(struct ust_buffer
*buf
, long *consumed
);
537 extern int ust_buffers_put_subbuf(struct ust_buffer
*buf
, unsigned long uconsumed_old
);
539 extern void init_ustrelay_transport(void);
541 #endif /* _UST_BUFFERS_H */
This page took 0.121927 seconds and 4 git commands to generate.