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
28 #include <ust/kernelcompat.h>
29 #include <kcompat/kref.h>
33 #include "tracercore.h"
36 struct ltt_reserve_switch_offsets
{
38 long begin_switch
, end_switch_current
, end_switch_old
;
39 size_t before_hdr_pad
, size
;
43 static DEFINE_MUTEX(ust_buffers_channels_mutex
);
44 static LIST_HEAD(ust_buffers_channels
);
46 static int get_n_cpus(void)
49 static int n_cpus
= 0;
55 /* On Linux, when some processors are offline
56 * _SC_NPROCESSORS_CONF counts the offline
57 * processors, whereas _SC_NPROCESSORS_ONLN
58 * does not. If we used _SC_NPROCESSORS_ONLN,
59 * getcpu() could return a value greater than
60 * this sysconf, in which case the arrays
61 * indexed by processor would overflow.
63 result
= sysconf(_SC_NPROCESSORS_CONF
);
73 /* _ust_buffers_write()
75 * @buf: destination buffer
76 * @offset: offset in destination
78 * @len: length of source
79 * @cpy: already copied
82 void _ust_buffers_write(struct ust_buffer
*buf
, size_t offset
,
83 const void *src
, size_t len
, ssize_t cpy
)
90 WARN_ON(offset
>= buf
->buf_size
);
92 cpy
= min_t(size_t, len
, buf
->buf_size
- offset
);
93 ust_buffers_do_copy(buf
->buf_data
+ offset
, src
, cpy
);
94 } while (unlikely(len
!= cpy
));
97 static int ust_buffers_init_buffer(struct ust_trace
*trace
,
98 struct ust_channel
*ltt_chan
,
99 struct ust_buffer
*buf
,
100 unsigned int n_subbufs
);
102 static int ust_buffers_alloc_buf(struct ust_buffer
*buf
, size_t *size
)
107 *size
= PAGE_ALIGN(*size
);
109 result
= buf
->shmid
= shmget(getpid(), *size
, IPC_CREAT
| IPC_EXCL
| 0700);
110 if(result
== -1 && errno
== EINVAL
) {
111 ERR("shmget() returned EINVAL; maybe /proc/sys/kernel/shmmax should be increased.");
114 else if(result
== -1) {
119 /* FIXME: should have matching call to shmdt */
120 ptr
= shmat(buf
->shmid
, NULL
, 0);
121 if(ptr
== (void *) -1) {
126 /* Already mark the shared memory for destruction. This will occur only
127 * when all users have detached.
129 result
= shmctl(buf
->shmid
, IPC_RMID
, NULL
);
136 buf
->buf_size
= *size
;
141 result
= shmctl(buf
->shmid
, IPC_RMID
, NULL
);
149 int ust_buffers_create_buf(struct ust_channel
*channel
, int cpu
)
152 struct ust_buffer
*buf
= channel
->buf
[cpu
];
155 result
= ust_buffers_alloc_buf(buf
, &channel
->alloc_size
);
160 kref_get(&channel
->kref
);
164 static void ust_buffers_destroy_channel(struct kref
*kref
)
166 struct ust_channel
*chan
= container_of(kref
, struct ust_channel
, kref
);
170 static void ust_buffers_destroy_buf(struct ust_buffer
*buf
)
172 struct ust_channel
*chan
= buf
->chan
;
175 result
= munmap(buf
->buf_data
, buf
->buf_size
);
180 //ust// chan->buf[buf->cpu] = NULL;
182 kref_put(&chan
->kref
, ust_buffers_destroy_channel
);
185 /* called from kref_put */
186 static void ust_buffers_remove_buf(struct kref
*kref
)
188 struct ust_buffer
*buf
= container_of(kref
, struct ust_buffer
, kref
);
189 ust_buffers_destroy_buf(buf
);
192 int ust_buffers_open_buf(struct ust_channel
*chan
, int cpu
)
196 result
= ust_buffers_create_buf(chan
, cpu
);
200 kref_init(&chan
->buf
[cpu
]->kref
);
202 result
= ust_buffers_init_buffer(chan
->trace
, chan
, chan
->buf
[cpu
], chan
->subbuf_cnt
);
208 /* FIXME: decrementally destroy on error? */
212 * ust_buffers_close_buf - close a channel buffer
215 static void ust_buffers_close_buf(struct ust_buffer
*buf
)
217 kref_put(&buf
->kref
, ust_buffers_remove_buf
);
220 int ust_buffers_channel_open(struct ust_channel
*chan
, size_t subbuf_size
, size_t subbuf_cnt
)
225 if(subbuf_size
== 0 || subbuf_cnt
== 0)
228 /* Check that the subbuffer size is larger than a page. */
229 WARN_ON_ONCE(subbuf_size
< PAGE_SIZE
);
232 * Make sure the number of subbuffers and subbuffer size are power of 2.
234 WARN_ON_ONCE(hweight32(subbuf_size
) != 1);
235 WARN_ON(hweight32(subbuf_cnt
) != 1);
237 chan
->version
= UST_CHANNEL_VERSION
;
238 chan
->subbuf_cnt
= subbuf_cnt
;
239 chan
->subbuf_size
= subbuf_size
;
240 chan
->subbuf_size_order
= get_count_order(subbuf_size
);
241 chan
->alloc_size
= subbuf_size
* subbuf_cnt
;
243 kref_init(&chan
->kref
);
245 mutex_lock(&ust_buffers_channels_mutex
);
246 for(i
=0; i
<chan
->n_cpus
; i
++) {
247 result
= ust_buffers_open_buf(chan
, i
);
251 list_add(&chan
->list
, &ust_buffers_channels
);
252 mutex_unlock(&ust_buffers_channels_mutex
);
256 /* Jump directly inside the loop to close the buffers that were already
259 ust_buffers_close_buf(chan
->buf
[i
]);
264 kref_put(&chan
->kref
, ust_buffers_destroy_channel
);
265 mutex_unlock(&ust_buffers_channels_mutex
);
269 void ust_buffers_channel_close(struct ust_channel
*chan
)
275 mutex_lock(&ust_buffers_channels_mutex
);
276 for(i
=0; i
<chan
->n_cpus
; i
++) {
277 /* FIXME: if we make it here, then all buffers were necessarily allocated. Moreover, we don't
278 * initialize to NULL so we cannot use this check. Should we? */
279 //ust// if (chan->buf[i])
280 ust_buffers_close_buf(chan
->buf
[i
]);
283 list_del(&chan
->list
);
284 kref_put(&chan
->kref
, ust_buffers_destroy_channel
);
285 mutex_unlock(&ust_buffers_channels_mutex
);
292 static void ust_buffers_destroy_buffer(struct ust_channel
*ltt_chan
, int cpu
);
294 static void ltt_force_switch(struct ust_buffer
*buf
,
295 enum force_switch_mode mode
);
300 static void ltt_buffer_begin(struct ust_buffer
*buf
,
301 u64 tsc
, unsigned int subbuf_idx
)
303 struct ust_channel
*channel
= buf
->chan
;
304 struct ltt_subbuffer_header
*header
=
305 (struct ltt_subbuffer_header
*)
306 ust_buffers_offset_address(buf
,
307 subbuf_idx
* buf
->chan
->subbuf_size
);
309 header
->cycle_count_begin
= tsc
;
310 header
->data_size
= 0xFFFFFFFF; /* for recognizing crashed buffers */
311 header
->sb_size
= 0xFFFFFFFF; /* for recognizing crashed buffers */
312 /* FIXME: add memory barrier? */
313 ltt_write_trace_header(channel
->trace
, header
);
317 * offset is assumed to never be 0 here : never deliver a completely empty
318 * subbuffer. The lost size is between 0 and subbuf_size-1.
320 static notrace
void ltt_buffer_end(struct ust_buffer
*buf
,
321 u64 tsc
, unsigned int offset
, unsigned int subbuf_idx
)
323 struct ltt_subbuffer_header
*header
=
324 (struct ltt_subbuffer_header
*)
325 ust_buffers_offset_address(buf
,
326 subbuf_idx
* buf
->chan
->subbuf_size
);
327 u32 data_size
= SUBBUF_OFFSET(offset
- 1, buf
->chan
) + 1;
329 header
->data_size
= data_size
;
330 header
->sb_size
= PAGE_ALIGN(data_size
);
331 header
->cycle_count_end
= tsc
;
332 header
->events_lost
= local_read(&buf
->events_lost
);
333 header
->subbuf_corrupt
= local_read(&buf
->corrupted_subbuffers
);
334 if(unlikely(header
->events_lost
> 0)) {
335 DBG("Some events (%d) were lost in %s_%d", header
->events_lost
, buf
->chan
->channel_name
, buf
->cpu
);
340 * This function should not be called from NMI interrupt context
342 static notrace
void ltt_buf_unfull(struct ust_buffer
*buf
,
343 unsigned int subbuf_idx
,
349 * Promote compiler barrier to a smp_mb().
350 * For the specific LTTng case, this IPI call should be removed if the
351 * architecture does not reorder writes. This should eventually be provided by
352 * a separate architecture-specific infrastructure.
354 //ust// static void remote_mb(void *info)
359 int ust_buffers_get_subbuf(struct ust_buffer
*buf
, long *consumed
)
361 struct ust_channel
*channel
= buf
->chan
;
362 long consumed_old
, consumed_idx
, commit_count
, write_offset
;
365 consumed_old
= atomic_long_read(&buf
->consumed
);
366 consumed_idx
= SUBBUF_INDEX(consumed_old
, buf
->chan
);
367 commit_count
= local_read(&buf
->commit_count
[consumed_idx
].cc_sb
);
369 * Make sure we read the commit count before reading the buffer
370 * data and the write offset. Correct consumed offset ordering
371 * wrt commit count is insured by the use of cmpxchg to update
372 * the consumed offset.
373 * smp_call_function_single can fail if the remote CPU is offline,
374 * this is OK because then there is no wmb to execute there.
375 * If our thread is executing on the same CPU as the on the buffers
376 * belongs to, we don't have to synchronize it at all. If we are
377 * migrated, the scheduler will take care of the memory barriers.
378 * Normally, smp_call_function_single() should ensure program order when
379 * executing the remote function, which implies that it surrounds the
380 * function execution with :
391 * However, smp_call_function_single() does not seem to clearly execute
392 * such barriers. It depends on spinlock semantic to provide the barrier
393 * before executing the IPI and, when busy-looping, csd_lock_wait only
394 * executes smp_mb() when it has to wait for the other CPU.
396 * I don't trust this code. Therefore, let's add the smp_mb() sequence
397 * required ourself, even if duplicated. It has no performance impact
400 * smp_mb() is needed because smp_rmb() and smp_wmb() only order read vs
401 * read and write vs write. They do not ensure core synchronization. We
402 * really have to ensure total order between the 3 barriers running on
405 //ust// #ifdef LTT_NO_IPI_BARRIER
407 * Local rmb to match the remote wmb to read the commit count before the
408 * buffer data and the write offset.
412 //ust// if (raw_smp_processor_id() != buf->cpu) {
413 //ust// smp_mb(); /* Total order with IPI handler smp_mb() */
414 //ust// smp_call_function_single(buf->cpu, remote_mb, NULL, 1);
415 //ust// smp_mb(); /* Total order with IPI handler smp_mb() */
419 write_offset
= local_read(&buf
->offset
);
421 * Check that the subbuffer we are trying to consume has been
422 * already fully committed.
424 if (((commit_count
- buf
->chan
->subbuf_size
)
425 & channel
->commit_count_mask
)
426 - (BUFFER_TRUNC(consumed_old
, buf
->chan
)
427 >> channel
->n_subbufs_order
)
432 * Check that we are not about to read the same subbuffer in
433 * which the writer head is.
435 if ((SUBBUF_TRUNC(write_offset
, buf
->chan
)
436 - SUBBUF_TRUNC(consumed_old
, buf
->chan
))
441 /* FIXME: is this ok to disable the reading feature? */
442 //ust// retval = update_read_sb_index(buf, consumed_idx);
444 //ust// return retval;
446 *consumed
= consumed_old
;
451 int ust_buffers_put_subbuf(struct ust_buffer
*buf
, unsigned long uconsumed_old
)
453 long consumed_new
, consumed_old
;
455 consumed_old
= atomic_long_read(&buf
->consumed
);
456 consumed_old
= consumed_old
& (~0xFFFFFFFFL
);
457 consumed_old
= consumed_old
| uconsumed_old
;
458 consumed_new
= SUBBUF_ALIGN(consumed_old
, buf
->chan
);
460 //ust// spin_lock(<t_buf->full_lock);
461 if (atomic_long_cmpxchg(&buf
->consumed
, consumed_old
,
464 /* We have been pushed by the writer : the last
465 * buffer read _is_ corrupted! It can also
466 * happen if this is a buffer we never got. */
467 //ust// spin_unlock(<t_buf->full_lock);
470 /* tell the client that buffer is now unfull */
473 index
= SUBBUF_INDEX(consumed_old
, buf
->chan
);
474 data
= BUFFER_OFFSET(consumed_old
, buf
->chan
);
475 ltt_buf_unfull(buf
, index
, data
);
476 //ust// spin_unlock(<t_buf->full_lock);
481 //ust// static void switch_buffer(unsigned long data)
483 //ust// struct ltt_channel_buf_struct *ltt_buf =
484 //ust// (struct ltt_channel_buf_struct *)data;
485 //ust// struct rchan_buf *buf = ltt_buf->rbuf;
488 //ust// ltt_force_switch(buf, FORCE_ACTIVE);
490 //ust// ltt_buf->switch_timer.expires += ltt_buf->switch_timer_interval;
491 //ust// add_timer_on(<t_buf->switch_timer, smp_processor_id());
494 //ust// static void start_switch_timer(struct ltt_channel_struct *ltt_channel)
496 //ust// struct rchan *rchan = ltt_channel->trans_channel_data;
499 //ust// if (!ltt_channel->switch_timer_interval)
502 //ust// // TODO : hotplug
503 //ust// for_each_online_cpu(cpu) {
504 //ust// struct ltt_channel_buf_struct *ltt_buf;
505 //ust// struct rchan_buf *buf;
507 //ust// buf = rchan->buf[cpu];
508 //ust// ltt_buf = buf->chan_private;
509 //ust// buf->random_access = 1;
510 //ust// ltt_buf->switch_timer_interval =
511 //ust// ltt_channel->switch_timer_interval;
512 //ust// init_timer(<t_buf->switch_timer);
513 //ust// ltt_buf->switch_timer.function = switch_buffer;
514 //ust// ltt_buf->switch_timer.expires = jiffies +
515 //ust// ltt_buf->switch_timer_interval;
516 //ust// ltt_buf->switch_timer.data = (unsigned long)ltt_buf;
517 //ust// add_timer_on(<t_buf->switch_timer, cpu);
522 //ust// * Cannot use del_timer_sync with add_timer_on, so use an IPI to locally
523 //ust// * delete the timer.
525 //ust// static void stop_switch_timer_ipi(void *info)
527 //ust// struct ltt_channel_buf_struct *ltt_buf =
528 //ust// (struct ltt_channel_buf_struct *)info;
530 //ust// del_timer(<t_buf->switch_timer);
533 //ust// static void stop_switch_timer(struct ltt_channel_struct *ltt_channel)
535 //ust// struct rchan *rchan = ltt_channel->trans_channel_data;
538 //ust// if (!ltt_channel->switch_timer_interval)
541 //ust// // TODO : hotplug
542 //ust// for_each_online_cpu(cpu) {
543 //ust// struct ltt_channel_buf_struct *ltt_buf;
544 //ust// struct rchan_buf *buf;
546 //ust// buf = rchan->buf[cpu];
547 //ust// ltt_buf = buf->chan_private;
548 //ust// smp_call_function(stop_switch_timer_ipi, ltt_buf, 1);
549 //ust// buf->random_access = 0;
553 //ust// static void ust_buffers_print_written(struct ust_channel *chan,
554 //ust// long cons_off, unsigned int cpu)
556 //ust// struct ust_buffer *buf = chan->buf[cpu];
557 //ust// long cons_idx, events_count;
559 //ust// cons_idx = SUBBUF_INDEX(cons_off, chan);
560 //ust// events_count = local_read(&buf->commit_count[cons_idx].events);
562 //ust// if (events_count)
563 //ust// printk(KERN_INFO
564 //ust// "channel %s: %lu events written (cpu %u, index %lu)\n",
565 //ust// chan->channel_name, events_count, cpu, cons_idx);
568 static void ltt_relay_print_subbuffer_errors(
569 struct ust_channel
*channel
,
570 long cons_off
, int cpu
)
572 struct ust_buffer
*ltt_buf
= channel
->buf
[cpu
];
573 long cons_idx
, commit_count
, commit_count_sb
, write_offset
;
575 cons_idx
= SUBBUF_INDEX(cons_off
, channel
);
576 commit_count
= local_read(<t_buf
->commit_count
[cons_idx
].cc
);
577 commit_count_sb
= local_read(<t_buf
->commit_count
[cons_idx
].cc_sb
);
580 * No need to order commit_count and write_offset reads because we
581 * execute after trace is stopped when there are no readers left.
583 write_offset
= local_read(<t_buf
->offset
);
584 WARN( "LTT : unread channel %s offset is %ld "
585 "and cons_off : %ld (cpu %d)\n",
586 channel
->channel_name
, write_offset
, cons_off
, cpu
);
587 /* Check each sub-buffer for non filled commit count */
588 if (((commit_count
- channel
->subbuf_size
) & channel
->commit_count_mask
)
589 - (BUFFER_TRUNC(cons_off
, channel
) >> channel
->n_subbufs_order
) != 0) {
590 ERR("LTT : %s : subbuffer %lu has non filled "
591 "commit count [cc, cc_sb] [%lu,%lu].\n",
592 channel
->channel_name
, cons_idx
, commit_count
, commit_count_sb
);
594 ERR("LTT : %s : commit count : %lu, subbuf size %zd\n",
595 channel
->channel_name
, commit_count
,
596 channel
->subbuf_size
);
599 static void ltt_relay_print_errors(struct ust_trace
*trace
,
600 struct ust_channel
*channel
, int cpu
)
602 struct ust_buffer
*ltt_buf
= channel
->buf
[cpu
];
606 * Can be called in the error path of allocation when
607 * trans_channel_data is not yet set.
612 //ust// for (cons_off = 0; cons_off < rchan->alloc_size;
613 //ust// cons_off = SUBBUF_ALIGN(cons_off, rchan))
614 //ust// ust_buffers_print_written(ltt_chan, cons_off, cpu);
615 for (cons_off
= atomic_long_read(<t_buf
->consumed
);
616 (SUBBUF_TRUNC(local_read(<t_buf
->offset
),
619 cons_off
= SUBBUF_ALIGN(cons_off
, channel
))
620 ltt_relay_print_subbuffer_errors(channel
, cons_off
, cpu
);
623 static void ltt_relay_print_buffer_errors(struct ust_channel
*channel
, int cpu
)
625 struct ust_trace
*trace
= channel
->trace
;
626 struct ust_buffer
*ltt_buf
= channel
->buf
[cpu
];
628 if (local_read(<t_buf
->events_lost
))
629 ERR("channel %s: %ld events lost (cpu %d)",
630 channel
->channel_name
,
631 local_read(<t_buf
->events_lost
), cpu
);
632 if (local_read(<t_buf
->corrupted_subbuffers
))
633 ERR("channel %s : %ld corrupted subbuffers (cpu %d)",
634 channel
->channel_name
,
635 local_read(<t_buf
->corrupted_subbuffers
), cpu
);
637 ltt_relay_print_errors(trace
, channel
, cpu
);
640 static void ltt_relay_release_channel(struct kref
*kref
)
642 struct ust_channel
*ltt_chan
= container_of(kref
,
643 struct ust_channel
, kref
);
650 //ust// static int ltt_relay_create_buffer(struct ust_trace *trace,
651 //ust// struct ltt_channel_struct *ltt_chan, struct rchan_buf *buf,
652 //ust// unsigned int cpu, unsigned int n_subbufs)
654 //ust// struct ltt_channel_buf_struct *ltt_buf =
655 //ust// percpu_ptr(ltt_chan->buf, cpu);
656 //ust// unsigned int j;
658 //ust// ltt_buf->commit_count =
659 //ust// kzalloc_node(sizeof(ltt_buf->commit_count) * n_subbufs,
660 //ust// GFP_KERNEL, cpu_to_node(cpu));
661 //ust// if (!ltt_buf->commit_count)
662 //ust// return -ENOMEM;
663 //ust// kref_get(&trace->kref);
664 //ust// kref_get(&trace->ltt_transport_kref);
665 //ust// kref_get(<t_chan->kref);
666 //ust// local_set(<t_buf->offset, ltt_subbuffer_header_size());
667 //ust// atomic_long_set(<t_buf->consumed, 0);
668 //ust// atomic_long_set(<t_buf->active_readers, 0);
669 //ust// for (j = 0; j < n_subbufs; j++)
670 //ust// local_set(<t_buf->commit_count[j], 0);
671 //ust// init_waitqueue_head(<t_buf->write_wait);
672 //ust// atomic_set(<t_buf->wakeup_readers, 0);
673 //ust// spin_lock_init(<t_buf->full_lock);
675 //ust// ltt_buffer_begin_callback(buf, trace->start_tsc, 0);
676 //ust// /* atomic_add made on local variable on data that belongs to
677 //ust// * various CPUs : ok because tracing not started (for this cpu). */
678 //ust// local_add(ltt_subbuffer_header_size(), <t_buf->commit_count[0]);
680 //ust// local_set(<t_buf->events_lost, 0);
681 //ust// local_set(<t_buf->corrupted_subbuffers, 0);
686 static int ust_buffers_init_buffer(struct ust_trace
*trace
,
687 struct ust_channel
*ltt_chan
, struct ust_buffer
*buf
,
688 unsigned int n_subbufs
)
695 zmalloc(sizeof(*buf
->commit_count
) * n_subbufs
);
696 if (!buf
->commit_count
)
698 kref_get(&trace
->kref
);
699 kref_get(&trace
->ltt_transport_kref
);
700 kref_get(<t_chan
->kref
);
701 local_set(&buf
->offset
, ltt_subbuffer_header_size());
702 atomic_long_set(&buf
->consumed
, 0);
703 atomic_long_set(&buf
->active_readers
, 0);
704 for (j
= 0; j
< n_subbufs
; j
++) {
705 local_set(&buf
->commit_count
[j
].cc
, 0);
706 local_set(&buf
->commit_count
[j
].cc_sb
, 0);
708 //ust// init_waitqueue_head(&buf->write_wait);
709 //ust// atomic_set(&buf->wakeup_readers, 0);
710 //ust// spin_lock_init(&buf->full_lock);
712 ltt_buffer_begin(buf
, trace
->start_tsc
, 0);
714 local_add(ltt_subbuffer_header_size(), &buf
->commit_count
[0].cc
);
716 local_set(&buf
->events_lost
, 0);
717 local_set(&buf
->corrupted_subbuffers
, 0);
724 buf
->data_ready_fd_read
= fds
[0];
725 buf
->data_ready_fd_write
= fds
[1];
727 /* FIXME: do we actually need this? */
728 result
= fcntl(fds
[0], F_SETFL
, O_NONBLOCK
);
733 //ust// buf->commit_seq = malloc(sizeof(buf->commit_seq) * n_subbufs);
734 //ust// if(!ltt_buf->commit_seq) {
737 memset(buf
->commit_seq
, 0, sizeof(buf
->commit_seq
[0]) * n_subbufs
);
739 /* FIXME: decrementally destroy on error */
744 /* FIXME: use this function */
745 static void ust_buffers_destroy_buffer(struct ust_channel
*ltt_chan
, int cpu
)
747 struct ust_trace
*trace
= ltt_chan
->trace
;
748 struct ust_buffer
*ltt_buf
= ltt_chan
->buf
[cpu
];
750 kref_put(<t_chan
->trace
->ltt_transport_kref
,
751 ltt_release_transport
);
752 ltt_relay_print_buffer_errors(ltt_chan
, cpu
);
753 //ust// free(ltt_buf->commit_seq);
754 kfree(ltt_buf
->commit_count
);
755 ltt_buf
->commit_count
= NULL
;
756 kref_put(<t_chan
->kref
, ltt_relay_release_channel
);
757 kref_put(&trace
->kref
, ltt_release_trace
);
758 //ust// wake_up_interruptible(&trace->kref_wq);
761 static int ust_buffers_alloc_channel_buf_structs(struct ust_channel
*chan
)
768 size
= PAGE_ALIGN(1);
770 for(i
=0; i
<chan
->n_cpus
; i
++) {
772 result
= chan
->buf_struct_shmids
[i
] = shmget(getpid(), size
, IPC_CREAT
| IPC_EXCL
| 0700);
775 goto destroy_previous
;
778 /* FIXME: should have matching call to shmdt */
779 ptr
= shmat(chan
->buf_struct_shmids
[i
], NULL
, 0);
780 if(ptr
== (void *) -1) {
785 /* Already mark the shared memory for destruction. This will occur only
786 * when all users have detached.
788 result
= shmctl(chan
->buf_struct_shmids
[i
], IPC_RMID
, NULL
);
791 goto destroy_previous
;
799 /* Jumping inside this loop occurs from within the other loop above with i as
800 * counter, so it unallocates the structures for the cpu = current_i down to
804 result
= shmctl(chan
->buf_struct_shmids
[i
], IPC_RMID
, NULL
);
819 static int ust_buffers_create_channel(const char *trace_name
, struct ust_trace
*trace
,
820 const char *channel_name
, struct ust_channel
*ltt_chan
,
821 unsigned int subbuf_size
, unsigned int n_subbufs
, int overwrite
)
825 kref_init(<t_chan
->kref
);
827 ltt_chan
->trace
= trace
;
828 ltt_chan
->overwrite
= overwrite
;
829 ltt_chan
->n_subbufs_order
= get_count_order(n_subbufs
);
830 ltt_chan
->commit_count_mask
= (~0UL >> ltt_chan
->n_subbufs_order
);
831 ltt_chan
->n_cpus
= get_n_cpus();
832 //ust// ltt_chan->buf = percpu_alloc_mask(sizeof(struct ltt_channel_buf_struct), GFP_KERNEL, cpu_possible_map);
833 ltt_chan
->buf
= (void *) malloc(ltt_chan
->n_cpus
* sizeof(void *));
834 if(ltt_chan
->buf
== NULL
) {
837 ltt_chan
->buf_struct_shmids
= (int *) malloc(ltt_chan
->n_cpus
* sizeof(int));
838 if(ltt_chan
->buf_struct_shmids
== NULL
)
841 result
= ust_buffers_alloc_channel_buf_structs(ltt_chan
);
843 goto free_buf_struct_shmids
;
846 result
= ust_buffers_channel_open(ltt_chan
, subbuf_size
, n_subbufs
);
848 ERR("Cannot open channel for trace %s", trace_name
);
849 goto unalloc_buf_structs
;
855 /* FIXME: put a call here to unalloc the buf structs! */
857 free_buf_struct_shmids
:
858 free(ltt_chan
->buf_struct_shmids
);
868 * LTTng channel flush function.
870 * Must be called when no tracing is active in the channel, because of
871 * accesses across CPUs.
873 static notrace
void ltt_relay_buffer_flush(struct ust_buffer
*buf
)
877 //ust// buf->finalized = 1;
878 ltt_force_switch(buf
, FORCE_FLUSH
);
880 result
= write(buf
->data_ready_fd_write
, "1", 1);
882 PERROR("write (in ltt_relay_buffer_flush)");
883 ERR("this should never happen!");
887 static void ltt_relay_async_wakeup_chan(struct ust_channel
*ltt_channel
)
889 //ust// unsigned int i;
890 //ust// struct rchan *rchan = ltt_channel->trans_channel_data;
892 //ust// for_each_possible_cpu(i) {
893 //ust// struct ltt_channel_buf_struct *ltt_buf =
894 //ust// percpu_ptr(ltt_channel->buf, i);
896 //ust// if (atomic_read(<t_buf->wakeup_readers) == 1) {
897 //ust// atomic_set(<t_buf->wakeup_readers, 0);
898 //ust// wake_up_interruptible(&rchan->buf[i]->read_wait);
903 static void ltt_relay_finish_buffer(struct ust_channel
*channel
, unsigned int cpu
)
907 if (channel
->buf
[cpu
]) {
908 struct ust_buffer
*buf
= channel
->buf
[cpu
];
909 ltt_relay_buffer_flush(buf
);
910 //ust// ltt_relay_wake_writers(ltt_buf);
911 /* closing the pipe tells the consumer the buffer is finished */
913 //result = write(ltt_buf->data_ready_fd_write, "D", 1);
915 // PERROR("write (in ltt_relay_finish_buffer)");
916 // ERR("this should never happen!");
918 close(buf
->data_ready_fd_write
);
923 static void ltt_relay_finish_channel(struct ust_channel
*channel
)
927 for(i
=0; i
<channel
->n_cpus
; i
++) {
928 ltt_relay_finish_buffer(channel
, i
);
932 static void ltt_relay_remove_channel(struct ust_channel
*channel
)
934 ust_buffers_channel_close(channel
);
935 kref_put(&channel
->kref
, ltt_relay_release_channel
);
941 //ust// * !0 if execution must be aborted.
943 //ust// static inline int ltt_relay_try_reserve(
944 //ust// struct ust_channel *channel, struct ust_buffer *buf,
945 //ust// struct ltt_reserve_switch_offsets *offsets, size_t data_size,
946 //ust// u64 *tsc, unsigned int *rflags, int largest_align)
948 //ust// offsets->begin = local_read(&buf->offset);
949 //ust// offsets->old = offsets->begin;
950 //ust// offsets->begin_switch = 0;
951 //ust// offsets->end_switch_current = 0;
952 //ust// offsets->end_switch_old = 0;
954 //ust// *tsc = trace_clock_read64();
955 //ust// if (last_tsc_overflow(buf, *tsc))
956 //ust// *rflags = LTT_RFLAG_ID_SIZE_TSC;
958 //ust// if (SUBBUF_OFFSET(offsets->begin, buf->chan) == 0) {
959 //ust// offsets->begin_switch = 1; /* For offsets->begin */
961 //ust// offsets->size = ust_get_header_size(channel,
962 //ust// offsets->begin, data_size,
963 //ust// &offsets->before_hdr_pad, *rflags);
964 //ust// offsets->size += ltt_align(offsets->begin + offsets->size,
965 //ust// largest_align)
967 //ust// if ((SUBBUF_OFFSET(offsets->begin, buf->chan) + offsets->size)
968 //ust// > buf->chan->subbuf_size) {
969 //ust// offsets->end_switch_old = 1; /* For offsets->old */
970 //ust// offsets->begin_switch = 1; /* For offsets->begin */
973 //ust// if (offsets->begin_switch) {
974 //ust// long subbuf_index;
976 //ust// if (offsets->end_switch_old)
977 //ust// offsets->begin = SUBBUF_ALIGN(offsets->begin,
979 //ust// offsets->begin = offsets->begin + ltt_subbuffer_header_size();
980 //ust// /* Test new buffer integrity */
981 //ust// subbuf_index = SUBBUF_INDEX(offsets->begin, buf->chan);
982 //ust// offsets->reserve_commit_diff =
983 //ust// (BUFFER_TRUNC(offsets->begin, buf->chan)
984 //ust// >> channel->n_subbufs_order)
985 //ust// - (local_read(&buf->commit_count[subbuf_index])
986 //ust// & channel->commit_count_mask);
987 //ust// if (offsets->reserve_commit_diff == 0) {
988 //ust// long consumed;
990 //ust// consumed = atomic_long_read(&buf->consumed);
992 //ust// /* Next buffer not corrupted. */
993 //ust// if (!channel->overwrite &&
994 //ust// (SUBBUF_TRUNC(offsets->begin, buf->chan)
995 //ust// - SUBBUF_TRUNC(consumed, buf->chan))
996 //ust// >= channel->alloc_size) {
998 //ust// long consumed_idx = SUBBUF_INDEX(consumed, buf->chan);
999 //ust// long commit_count = local_read(&buf->commit_count[consumed_idx]);
1000 //ust// if(((commit_count - buf->chan->subbuf_size) & channel->commit_count_mask) - (BUFFER_TRUNC(consumed, buf->chan) >> channel->n_subbufs_order) != 0) {
1001 //ust// WARN("Event dropped. Caused by non-committed event.");
1004 //ust// WARN("Event dropped. Caused by non-consumed buffer.");
1007 //ust// * We do not overwrite non consumed buffers
1008 //ust// * and we are full : event is lost.
1010 //ust// local_inc(&buf->events_lost);
1014 //ust// * next buffer not corrupted, we are either in
1015 //ust// * overwrite mode or the buffer is not full.
1016 //ust// * It's safe to write in this new subbuffer.
1021 //ust// * Next subbuffer corrupted. Force pushing reader even
1022 //ust// * in normal mode. It's safe to write in this new
1023 //ust// * subbuffer.
1026 //ust// offsets->size = ust_get_header_size(channel,
1027 //ust// offsets->begin, data_size,
1028 //ust// &offsets->before_hdr_pad, *rflags);
1029 //ust// offsets->size += ltt_align(offsets->begin + offsets->size,
1030 //ust// largest_align)
1031 //ust// + data_size;
1032 //ust// if ((SUBBUF_OFFSET(offsets->begin, buf->chan) + offsets->size)
1033 //ust// > buf->chan->subbuf_size) {
1035 //ust// * Event too big for subbuffers, report error, don't
1036 //ust// * complete the sub-buffer switch.
1038 //ust// local_inc(&buf->events_lost);
1042 //ust// * We just made a successful buffer switch and the event
1043 //ust// * fits in the new subbuffer. Let's write.
1048 //ust// * Event fits in the current buffer and we are not on a switch
1049 //ust// * boundary. It's safe to write.
1052 //ust// offsets->end = offsets->begin + offsets->size;
1054 //ust// if ((SUBBUF_OFFSET(offsets->end, buf->chan)) == 0) {
1056 //ust// * The offset_end will fall at the very beginning of the next
1057 //ust// * subbuffer.
1059 //ust// offsets->end_switch_current = 1; /* For offsets->begin */
1067 //ust// * !0 if execution must be aborted.
1069 //ust// static inline int ltt_relay_try_switch(
1070 //ust// enum force_switch_mode mode,
1071 //ust// struct ust_channel *channel,
1072 //ust// struct ust_buffer *buf,
1073 //ust// struct ltt_reserve_switch_offsets *offsets,
1076 //ust// long subbuf_index;
1078 //ust// offsets->begin = local_read(&buf->offset);
1079 //ust// offsets->old = offsets->begin;
1080 //ust// offsets->begin_switch = 0;
1081 //ust// offsets->end_switch_old = 0;
1083 //ust// *tsc = trace_clock_read64();
1085 //ust// if (SUBBUF_OFFSET(offsets->begin, buf->chan) != 0) {
1086 //ust// offsets->begin = SUBBUF_ALIGN(offsets->begin, buf->chan);
1087 //ust// offsets->end_switch_old = 1;
1089 //ust// /* we do not have to switch : buffer is empty */
1092 //ust// if (mode == FORCE_ACTIVE)
1093 //ust// offsets->begin += ltt_subbuffer_header_size();
1095 //ust// * Always begin_switch in FORCE_ACTIVE mode.
1096 //ust// * Test new buffer integrity
1098 //ust// subbuf_index = SUBBUF_INDEX(offsets->begin, buf->chan);
1099 //ust// offsets->reserve_commit_diff =
1100 //ust// (BUFFER_TRUNC(offsets->begin, buf->chan)
1101 //ust// >> channel->n_subbufs_order)
1102 //ust// - (local_read(&buf->commit_count[subbuf_index])
1103 //ust// & channel->commit_count_mask);
1104 //ust// if (offsets->reserve_commit_diff == 0) {
1105 //ust// /* Next buffer not corrupted. */
1106 //ust// if (mode == FORCE_ACTIVE
1107 //ust// && !channel->overwrite
1108 //ust// && offsets->begin - atomic_long_read(&buf->consumed)
1109 //ust// >= channel->alloc_size) {
1111 //ust// * We do not overwrite non consumed buffers and we are
1112 //ust// * full : ignore switch while tracing is active.
1118 //ust// * Next subbuffer corrupted. Force pushing reader even in normal
1122 //ust// offsets->end = offsets->begin;
1126 //ust// static inline void ltt_reserve_push_reader(
1127 //ust// struct ust_channel *channel,
1128 //ust// struct ust_buffer *buf,
1129 //ust// struct ltt_reserve_switch_offsets *offsets)
1131 //ust// long consumed_old, consumed_new;
1134 //ust// consumed_old = atomic_long_read(&buf->consumed);
1136 //ust// * If buffer is in overwrite mode, push the reader consumed
1137 //ust// * count if the write position has reached it and we are not
1138 //ust// * at the first iteration (don't push the reader farther than
1139 //ust// * the writer). This operation can be done concurrently by many
1140 //ust// * writers in the same buffer, the writer being at the farthest
1141 //ust// * write position sub-buffer index in the buffer being the one
1142 //ust// * which will win this loop.
1143 //ust// * If the buffer is not in overwrite mode, pushing the reader
1144 //ust// * only happens if a sub-buffer is corrupted.
1146 //ust// if ((SUBBUF_TRUNC(offsets->end-1, buf->chan)
1147 //ust// - SUBBUF_TRUNC(consumed_old, buf->chan))
1148 //ust// >= channel->alloc_size)
1149 //ust// consumed_new = SUBBUF_ALIGN(consumed_old, buf->chan);
1151 //ust// consumed_new = consumed_old;
1154 //ust// } while (atomic_long_cmpxchg(&buf->consumed, consumed_old,
1155 //ust// consumed_new) != consumed_old);
1157 //ust// if (consumed_old != consumed_new) {
1159 //ust// * Reader pushed : we are the winner of the push, we can
1160 //ust// * therefore reequilibrate reserve and commit. Atomic increment
1161 //ust// * of the commit count permits other writers to play around
1162 //ust// * with this variable before us. We keep track of
1163 //ust// * corrupted_subbuffers even in overwrite mode :
1164 //ust// * we never want to write over a non completely committed
1165 //ust// * sub-buffer : possible causes : the buffer size is too low
1166 //ust// * compared to the unordered data input, or there is a writer
1167 //ust// * that died between the reserve and the commit.
1169 //ust// if (offsets->reserve_commit_diff) {
1171 //ust// * We have to alter the sub-buffer commit count.
1172 //ust// * We do not deliver the previous subbuffer, given it
1173 //ust// * was either corrupted or not consumed (overwrite
1176 //ust// local_add(offsets->reserve_commit_diff,
1177 //ust// &buf->commit_count[
1178 //ust// SUBBUF_INDEX(offsets->begin,
1179 //ust// buf->chan)]);
1180 //ust// if (!channel->overwrite
1181 //ust// || offsets->reserve_commit_diff
1182 //ust// != channel->subbuf_size) {
1184 //ust// * The reserve commit diff was not subbuf_size :
1185 //ust// * it means the subbuffer was partly written to
1186 //ust// * and is therefore corrupted. If it is multiple
1187 //ust// * of subbuffer size and we are in flight
1188 //ust// * recorder mode, we are skipping over a whole
1189 //ust// * subbuffer.
1191 //ust// local_inc(&buf->corrupted_subbuffers);
1198 //ust// * ltt_relay_reserve_slot - Atomic slot reservation in a LTTng buffer.
1199 //ust// * @trace: the trace structure to log to.
1200 //ust// * @ltt_channel: channel structure
1201 //ust// * @transport_data: data structure specific to ltt relay
1202 //ust// * @data_size: size of the variable length data to log.
1203 //ust// * @slot_size: pointer to total size of the slot (out)
1204 //ust// * @buf_offset : pointer to reserved buffer offset (out)
1205 //ust// * @tsc: pointer to the tsc at the slot reservation (out)
1206 //ust// * @cpu: cpuid
1208 //ust// * Return : -ENOSPC if not enough space, else returns 0.
1209 //ust// * It will take care of sub-buffer switching.
1211 //ust// static notrace int ltt_relay_reserve_slot(struct ust_trace *trace,
1212 //ust// struct ust_channel *channel, void **transport_data,
1213 //ust// size_t data_size, size_t *slot_size, long *buf_offset, u64 *tsc,
1214 //ust// unsigned int *rflags, int largest_align, int cpu)
1216 //ust// struct ust_buffer *buf = *transport_data = channel->buf[cpu];
1217 //ust// struct ltt_reserve_switch_offsets offsets;
1219 //ust// offsets.reserve_commit_diff = 0;
1220 //ust// offsets.size = 0;
1223 //ust// * Perform retryable operations.
1225 //ust// if (ltt_nesting > 4) {
1226 //ust// local_inc(&buf->events_lost);
1227 //ust// return -EPERM;
1230 //ust// if (ltt_relay_try_reserve(channel, buf, &offsets, data_size, tsc, rflags,
1231 //ust// largest_align))
1232 //ust// return -ENOSPC;
1233 //ust// } while (local_cmpxchg(&buf->offset, offsets.old,
1234 //ust// offsets.end) != offsets.old);
1237 //ust// * Atomically update last_tsc. This update races against concurrent
1238 //ust// * atomic updates, but the race will always cause supplementary full TSC
1239 //ust// * events, never the opposite (missing a full TSC event when it would be
1242 //ust// save_last_tsc(buf, *tsc);
1245 //ust// * Push the reader if necessary
1247 //ust// ltt_reserve_push_reader(channel, buf, &offsets);
1250 //ust// * Switch old subbuffer if needed.
1252 //ust// if (offsets.end_switch_old)
1253 //ust// ltt_reserve_switch_old_subbuf(channel, buf, &offsets, tsc);
1256 //ust// * Populate new subbuffer.
1258 //ust// if (offsets.begin_switch)
1259 //ust// ltt_reserve_switch_new_subbuf(channel, buf, &offsets, tsc);
1261 //ust// if (offsets.end_switch_current)
1262 //ust// ltt_reserve_end_switch_current(channel, buf, &offsets, tsc);
1264 //ust// *slot_size = offsets.size;
1265 //ust// *buf_offset = offsets.begin + offsets.before_hdr_pad;
1270 //ust// * Force a sub-buffer switch for a per-cpu buffer. This operation is
1271 //ust// * completely reentrant : can be called while tracing is active with
1272 //ust// * absolutely no lock held.
1274 //ust// * Note, however, that as a local_cmpxchg is used for some atomic
1275 //ust// * operations, this function must be called from the CPU which owns the buffer
1276 //ust// * for a ACTIVE flush.
1278 //ust// static notrace void ltt_force_switch(struct ust_buffer *buf,
1279 //ust// enum force_switch_mode mode)
1281 //ust// struct ust_channel *channel = buf->chan;
1282 //ust// struct ltt_reserve_switch_offsets offsets;
1285 //ust// offsets.reserve_commit_diff = 0;
1286 //ust// offsets.size = 0;
1289 //ust// * Perform retryable operations.
1292 //ust// if (ltt_relay_try_switch(mode, channel, buf, &offsets, &tsc))
1294 //ust// } while (local_cmpxchg(&buf->offset, offsets.old,
1295 //ust// offsets.end) != offsets.old);
1298 //ust// * Atomically update last_tsc. This update races against concurrent
1299 //ust// * atomic updates, but the race will always cause supplementary full TSC
1300 //ust// * events, never the opposite (missing a full TSC event when it would be
1303 //ust// save_last_tsc(buf, tsc);
1306 //ust// * Push the reader if necessary
1308 //ust// if (mode == FORCE_ACTIVE)
1309 //ust// ltt_reserve_push_reader(channel, buf, &offsets);
1312 //ust// * Switch old subbuffer if needed.
1314 //ust// if (offsets.end_switch_old)
1315 //ust// ltt_reserve_switch_old_subbuf(channel, buf, &offsets, &tsc);
1318 //ust// * Populate new subbuffer.
1320 //ust// if (mode == FORCE_ACTIVE)
1321 //ust// ltt_reserve_switch_new_subbuf(channel, buf, &offsets, &tsc);
1325 * ltt_reserve_switch_old_subbuf: switch old subbuffer
1327 * Concurrency safe because we are the last and only thread to alter this
1328 * sub-buffer. As long as it is not delivered and read, no other thread can
1329 * alter the offset, alter the reserve_count or call the
1330 * client_buffer_end_callback on this sub-buffer.
1332 * The only remaining threads could be the ones with pending commits. They will
1333 * have to do the deliver themselves. Not concurrency safe in overwrite mode.
1334 * We detect corrupted subbuffers with commit and reserve counts. We keep a
1335 * corrupted sub-buffers count and push the readers across these sub-buffers.
1337 * Not concurrency safe if a writer is stalled in a subbuffer and another writer
1338 * switches in, finding out it's corrupted. The result will be than the old
1339 * (uncommited) subbuffer will be declared corrupted, and that the new subbuffer
1340 * will be declared corrupted too because of the commit count adjustment.
1342 * Note : offset_old should never be 0 here.
1344 static void ltt_reserve_switch_old_subbuf(
1345 struct ust_channel
*chan
, struct ust_buffer
*buf
,
1346 struct ltt_reserve_switch_offsets
*offsets
, u64
*tsc
)
1348 long oldidx
= SUBBUF_INDEX(offsets
->old
- 1, chan
);
1349 long commit_count
, padding_size
;
1351 padding_size
= chan
->subbuf_size
1352 - (SUBBUF_OFFSET(offsets
->old
- 1, chan
) + 1);
1353 ltt_buffer_end(buf
, *tsc
, offsets
->old
, oldidx
);
1356 * Must write slot data before incrementing commit count.
1357 * This compiler barrier is upgraded into a smp_wmb() by the IPI
1358 * sent by get_subbuf() when it does its smp_rmb().
1361 local_add(padding_size
,
1362 &buf
->commit_count
[oldidx
].cc
);
1363 commit_count
= local_read(&buf
->commit_count
[oldidx
].cc
);
1364 ltt_check_deliver(chan
, buf
, offsets
->old
- 1, commit_count
, oldidx
);
1365 ltt_write_commit_counter(chan
, buf
, oldidx
,
1366 offsets
->old
, commit_count
, padding_size
);
1370 * ltt_reserve_switch_new_subbuf: Populate new subbuffer.
1372 * This code can be executed unordered : writers may already have written to the
1373 * sub-buffer before this code gets executed, caution. The commit makes sure
1374 * that this code is executed before the deliver of this sub-buffer.
1376 static void ltt_reserve_switch_new_subbuf(
1377 struct ust_channel
*chan
, struct ust_buffer
*buf
,
1378 struct ltt_reserve_switch_offsets
*offsets
, u64
*tsc
)
1380 long beginidx
= SUBBUF_INDEX(offsets
->begin
, chan
);
1383 ltt_buffer_begin(buf
, *tsc
, beginidx
);
1386 * Must write slot data before incrementing commit count.
1387 * This compiler barrier is upgraded into a smp_wmb() by the IPI
1388 * sent by get_subbuf() when it does its smp_rmb().
1391 local_add(ltt_subbuffer_header_size(),
1392 &buf
->commit_count
[beginidx
].cc
);
1393 commit_count
= local_read(&buf
->commit_count
[beginidx
].cc
);
1394 /* Check if the written buffer has to be delivered */
1395 ltt_check_deliver(chan
, buf
, offsets
->begin
, commit_count
, beginidx
);
1396 ltt_write_commit_counter(chan
, buf
, beginidx
,
1397 offsets
->begin
, commit_count
, ltt_subbuffer_header_size());
1401 * ltt_reserve_end_switch_current: finish switching current subbuffer
1403 * Concurrency safe because we are the last and only thread to alter this
1404 * sub-buffer. As long as it is not delivered and read, no other thread can
1405 * alter the offset, alter the reserve_count or call the
1406 * client_buffer_end_callback on this sub-buffer.
1408 * The only remaining threads could be the ones with pending commits. They will
1409 * have to do the deliver themselves. Not concurrency safe in overwrite mode.
1410 * We detect corrupted subbuffers with commit and reserve counts. We keep a
1411 * corrupted sub-buffers count and push the readers across these sub-buffers.
1413 * Not concurrency safe if a writer is stalled in a subbuffer and another writer
1414 * switches in, finding out it's corrupted. The result will be than the old
1415 * (uncommited) subbuffer will be declared corrupted, and that the new subbuffer
1416 * will be declared corrupted too because of the commit count adjustment.
1418 static void ltt_reserve_end_switch_current(
1419 struct ust_channel
*chan
,
1420 struct ust_buffer
*buf
,
1421 struct ltt_reserve_switch_offsets
*offsets
, u64
*tsc
)
1423 long endidx
= SUBBUF_INDEX(offsets
->end
- 1, chan
);
1424 long commit_count
, padding_size
;
1426 padding_size
= chan
->subbuf_size
1427 - (SUBBUF_OFFSET(offsets
->end
- 1, chan
) + 1);
1429 ltt_buffer_end(buf
, *tsc
, offsets
->end
, endidx
);
1432 * Must write slot data before incrementing commit count.
1433 * This compiler barrier is upgraded into a smp_wmb() by the IPI
1434 * sent by get_subbuf() when it does its smp_rmb().
1437 local_add(padding_size
,
1438 &buf
->commit_count
[endidx
].cc
);
1439 commit_count
= local_read(&buf
->commit_count
[endidx
].cc
);
1440 ltt_check_deliver(chan
, buf
,
1441 offsets
->end
- 1, commit_count
, endidx
);
1442 ltt_write_commit_counter(chan
, buf
, endidx
,
1443 offsets
->end
, commit_count
, padding_size
);
1449 * !0 if execution must be aborted.
1451 static int ltt_relay_try_switch_slow(
1452 enum force_switch_mode mode
,
1453 struct ust_channel
*chan
,
1454 struct ust_buffer
*buf
,
1455 struct ltt_reserve_switch_offsets
*offsets
,
1459 long reserve_commit_diff
;
1461 offsets
->begin
= local_read(&buf
->offset
);
1462 offsets
->old
= offsets
->begin
;
1463 offsets
->begin_switch
= 0;
1464 offsets
->end_switch_old
= 0;
1466 *tsc
= trace_clock_read64();
1468 if (SUBBUF_OFFSET(offsets
->begin
, buf
->chan
) != 0) {
1469 offsets
->begin
= SUBBUF_ALIGN(offsets
->begin
, buf
->chan
);
1470 offsets
->end_switch_old
= 1;
1472 /* we do not have to switch : buffer is empty */
1475 if (mode
== FORCE_ACTIVE
)
1476 offsets
->begin
+= ltt_subbuffer_header_size();
1478 * Always begin_switch in FORCE_ACTIVE mode.
1479 * Test new buffer integrity
1481 subbuf_index
= SUBBUF_INDEX(offsets
->begin
, buf
->chan
);
1482 reserve_commit_diff
=
1483 (BUFFER_TRUNC(offsets
->begin
, buf
->chan
)
1484 >> chan
->n_subbufs_order
)
1485 - (local_read(&buf
->commit_count
[subbuf_index
].cc_sb
)
1486 & chan
->commit_count_mask
);
1487 if (reserve_commit_diff
== 0) {
1488 /* Next buffer not corrupted. */
1489 if (mode
== FORCE_ACTIVE
1491 && offsets
->begin
- atomic_long_read(&buf
->consumed
)
1492 >= chan
->alloc_size
) {
1494 * We do not overwrite non consumed buffers and we are
1495 * full : ignore switch while tracing is active.
1501 * Next subbuffer corrupted. Force pushing reader even in normal
1505 offsets
->end
= offsets
->begin
;
1510 * Force a sub-buffer switch for a per-cpu buffer. This operation is
1511 * completely reentrant : can be called while tracing is active with
1512 * absolutely no lock held.
1514 * Note, however, that as a local_cmpxchg is used for some atomic
1515 * operations, this function must be called from the CPU which owns the buffer
1516 * for a ACTIVE flush.
1518 void ltt_force_switch_lockless_slow(struct ust_buffer
*buf
,
1519 enum force_switch_mode mode
)
1521 struct ust_channel
*chan
= buf
->chan
;
1522 struct ltt_reserve_switch_offsets offsets
;
1527 DBG("Switching (forced) %s_%d", chan
->channel_name
, buf
->cpu
);
1529 * Perform retryable operations.
1532 if (ltt_relay_try_switch_slow(mode
, chan
, buf
,
1535 } while (local_cmpxchg(&buf
->offset
, offsets
.old
,
1536 offsets
.end
) != offsets
.old
);
1539 * Atomically update last_tsc. This update races against concurrent
1540 * atomic updates, but the race will always cause supplementary full TSC
1541 * events, never the opposite (missing a full TSC event when it would be
1544 save_last_tsc(buf
, tsc
);
1547 * Push the reader if necessary
1549 if (mode
== FORCE_ACTIVE
) {
1550 ltt_reserve_push_reader(chan
, buf
, offsets
.end
- 1);
1551 //ust// ltt_clear_noref_flag(chan, buf, SUBBUF_INDEX(offsets.end - 1, chan));
1555 * Switch old subbuffer if needed.
1557 if (offsets
.end_switch_old
) {
1558 //ust// ltt_clear_noref_flag(rchan, buf, SUBBUF_INDEX(offsets.old - 1, rchan));
1559 ltt_reserve_switch_old_subbuf(chan
, buf
, &offsets
, &tsc
);
1563 * Populate new subbuffer.
1565 if (mode
== FORCE_ACTIVE
)
1566 ltt_reserve_switch_new_subbuf(chan
, buf
, &offsets
, &tsc
);
1572 * !0 if execution must be aborted.
1574 static int ltt_relay_try_reserve_slow(struct ust_channel
*chan
, struct ust_buffer
*buf
,
1575 struct ltt_reserve_switch_offsets
*offsets
, size_t data_size
,
1576 u64
*tsc
, unsigned int *rflags
, int largest_align
)
1578 long reserve_commit_diff
;
1580 offsets
->begin
= local_read(&buf
->offset
);
1581 offsets
->old
= offsets
->begin
;
1582 offsets
->begin_switch
= 0;
1583 offsets
->end_switch_current
= 0;
1584 offsets
->end_switch_old
= 0;
1586 *tsc
= trace_clock_read64();
1587 if (last_tsc_overflow(buf
, *tsc
))
1588 *rflags
= LTT_RFLAG_ID_SIZE_TSC
;
1590 if (unlikely(SUBBUF_OFFSET(offsets
->begin
, buf
->chan
) == 0)) {
1591 offsets
->begin_switch
= 1; /* For offsets->begin */
1593 offsets
->size
= ust_get_header_size(chan
,
1594 offsets
->begin
, data_size
,
1595 &offsets
->before_hdr_pad
, *rflags
);
1596 offsets
->size
+= ltt_align(offsets
->begin
+ offsets
->size
,
1599 if (unlikely((SUBBUF_OFFSET(offsets
->begin
, buf
->chan
) +
1600 offsets
->size
) > buf
->chan
->subbuf_size
)) {
1601 offsets
->end_switch_old
= 1; /* For offsets->old */
1602 offsets
->begin_switch
= 1; /* For offsets->begin */
1605 if (unlikely(offsets
->begin_switch
)) {
1609 * We are typically not filling the previous buffer completely.
1611 if (likely(offsets
->end_switch_old
))
1612 offsets
->begin
= SUBBUF_ALIGN(offsets
->begin
,
1614 offsets
->begin
= offsets
->begin
+ ltt_subbuffer_header_size();
1615 /* Test new buffer integrity */
1616 subbuf_index
= SUBBUF_INDEX(offsets
->begin
, buf
->chan
);
1617 reserve_commit_diff
=
1618 (BUFFER_TRUNC(offsets
->begin
, buf
->chan
)
1619 >> chan
->n_subbufs_order
)
1620 - (local_read(&buf
->commit_count
[subbuf_index
].cc_sb
)
1621 & chan
->commit_count_mask
);
1622 if (likely(reserve_commit_diff
== 0)) {
1623 /* Next buffer not corrupted. */
1624 if (unlikely(!chan
->overwrite
&&
1625 (SUBBUF_TRUNC(offsets
->begin
, buf
->chan
)
1626 - SUBBUF_TRUNC(atomic_long_read(
1629 >= chan
->alloc_size
)) {
1631 * We do not overwrite non consumed buffers
1632 * and we are full : event is lost.
1634 local_inc(&buf
->events_lost
);
1638 * next buffer not corrupted, we are either in
1639 * overwrite mode or the buffer is not full.
1640 * It's safe to write in this new subbuffer.
1645 * Next subbuffer corrupted. Drop event in normal and
1646 * overwrite mode. Caused by either a writer OOPS or
1647 * too many nested writes over a reserve/commit pair.
1649 local_inc(&buf
->events_lost
);
1652 offsets
->size
= ust_get_header_size(chan
,
1653 offsets
->begin
, data_size
,
1654 &offsets
->before_hdr_pad
, *rflags
);
1655 offsets
->size
+= ltt_align(offsets
->begin
+ offsets
->size
,
1658 if (unlikely((SUBBUF_OFFSET(offsets
->begin
, buf
->chan
)
1659 + offsets
->size
) > buf
->chan
->subbuf_size
)) {
1661 * Event too big for subbuffers, report error, don't
1662 * complete the sub-buffer switch.
1664 local_inc(&buf
->events_lost
);
1668 * We just made a successful buffer switch and the event
1669 * fits in the new subbuffer. Let's write.
1674 * Event fits in the current buffer and we are not on a switch
1675 * boundary. It's safe to write.
1678 offsets
->end
= offsets
->begin
+ offsets
->size
;
1680 if (unlikely((SUBBUF_OFFSET(offsets
->end
, buf
->chan
)) == 0)) {
1682 * The offset_end will fall at the very beginning of the next
1685 offsets
->end_switch_current
= 1; /* For offsets->begin */
1691 * ltt_relay_reserve_slot_lockless_slow - Atomic slot reservation in a buffer.
1692 * @trace: the trace structure to log to.
1693 * @ltt_channel: channel structure
1694 * @transport_data: data structure specific to ltt relay
1695 * @data_size: size of the variable length data to log.
1696 * @slot_size: pointer to total size of the slot (out)
1697 * @buf_offset : pointer to reserved buffer offset (out)
1698 * @tsc: pointer to the tsc at the slot reservation (out)
1701 * Return : -ENOSPC if not enough space, else returns 0.
1702 * It will take care of sub-buffer switching.
1704 int ltt_reserve_slot_lockless_slow(struct ust_trace
*trace
,
1705 struct ust_channel
*chan
, void **transport_data
,
1706 size_t data_size
, size_t *slot_size
, long *buf_offset
, u64
*tsc
,
1707 unsigned int *rflags
, int largest_align
, int cpu
)
1709 struct ust_buffer
*buf
= chan
->buf
[cpu
];
1710 struct ltt_reserve_switch_offsets offsets
;
1715 if (unlikely(ltt_relay_try_reserve_slow(chan
, buf
, &offsets
,
1716 data_size
, tsc
, rflags
, largest_align
)))
1718 } while (unlikely(local_cmpxchg(&buf
->offset
, offsets
.old
,
1719 offsets
.end
) != offsets
.old
));
1722 * Atomically update last_tsc. This update races against concurrent
1723 * atomic updates, but the race will always cause supplementary full TSC
1724 * events, never the opposite (missing a full TSC event when it would be
1727 save_last_tsc(buf
, *tsc
);
1730 * Push the reader if necessary
1732 ltt_reserve_push_reader(chan
, buf
, offsets
.end
- 1);
1735 * Clear noref flag for this subbuffer.
1737 //ust// ltt_clear_noref_flag(chan, buf, SUBBUF_INDEX(offsets.end - 1, chan));
1740 * Switch old subbuffer if needed.
1742 if (unlikely(offsets
.end_switch_old
)) {
1743 //ust// ltt_clear_noref_flag(chan, buf, SUBBUF_INDEX(offsets.old - 1, chan));
1744 ltt_reserve_switch_old_subbuf(chan
, buf
, &offsets
, tsc
);
1745 DBG("Switching %s_%d", chan
->channel_name
, cpu
);
1749 * Populate new subbuffer.
1751 if (unlikely(offsets
.begin_switch
))
1752 ltt_reserve_switch_new_subbuf(chan
, buf
, &offsets
, tsc
);
1754 if (unlikely(offsets
.end_switch_current
))
1755 ltt_reserve_end_switch_current(chan
, buf
, &offsets
, tsc
);
1757 *slot_size
= offsets
.size
;
1758 *buf_offset
= offsets
.begin
+ offsets
.before_hdr_pad
;
1762 static struct ltt_transport ust_relay_transport
= {
1765 .create_channel
= ust_buffers_create_channel
,
1766 .finish_channel
= ltt_relay_finish_channel
,
1767 .remove_channel
= ltt_relay_remove_channel
,
1768 .wakeup_channel
= ltt_relay_async_wakeup_chan
,
1772 static char initialized
= 0;
1774 void __attribute__((constructor
)) init_ustrelay_transport(void)
1777 ltt_transport_register(&ust_relay_transport
);
1782 static void __attribute__((destructor
)) ust_buffers_exit(void)
1784 ltt_transport_unregister(&ust_relay_transport
);
1787 size_t ltt_write_event_header_slow(struct ust_trace
*trace
,
1788 struct ust_channel
*channel
,
1789 struct ust_buffer
*buf
, long buf_offset
,
1790 u16 eID
, u32 event_size
,
1791 u64 tsc
, unsigned int rflags
)
1793 struct ltt_event_header header
;
1797 case LTT_RFLAG_ID_SIZE_TSC
:
1798 header
.id_time
= 29 << LTT_TSC_BITS
;
1800 case LTT_RFLAG_ID_SIZE
:
1801 header
.id_time
= 30 << LTT_TSC_BITS
;
1804 header
.id_time
= 31 << LTT_TSC_BITS
;
1808 header
.id_time
|= (u32
)tsc
& LTT_TSC_MASK
;
1809 ust_buffers_write(buf
, buf_offset
, &header
, sizeof(header
));
1810 buf_offset
+= sizeof(header
);
1813 case LTT_RFLAG_ID_SIZE_TSC
:
1814 small_size
= (u16
)min_t(u32
, event_size
, LTT_MAX_SMALL_SIZE
);
1815 ust_buffers_write(buf
, buf_offset
,
1817 buf_offset
+= sizeof(u16
);
1818 ust_buffers_write(buf
, buf_offset
,
1819 &small_size
, sizeof(u16
));
1820 buf_offset
+= sizeof(u16
);
1821 if (small_size
== LTT_MAX_SMALL_SIZE
) {
1822 ust_buffers_write(buf
, buf_offset
,
1823 &event_size
, sizeof(u32
));
1824 buf_offset
+= sizeof(u32
);
1826 buf_offset
+= ltt_align(buf_offset
, sizeof(u64
));
1827 ust_buffers_write(buf
, buf_offset
,
1829 buf_offset
+= sizeof(u64
);
1831 case LTT_RFLAG_ID_SIZE
:
1832 small_size
= (u16
)min_t(u32
, event_size
, LTT_MAX_SMALL_SIZE
);
1833 ust_buffers_write(buf
, buf_offset
,
1835 buf_offset
+= sizeof(u16
);
1836 ust_buffers_write(buf
, buf_offset
,
1837 &small_size
, sizeof(u16
));
1838 buf_offset
+= sizeof(u16
);
1839 if (small_size
== LTT_MAX_SMALL_SIZE
) {
1840 ust_buffers_write(buf
, buf_offset
,
1841 &event_size
, sizeof(u32
));
1842 buf_offset
+= sizeof(u32
);
1846 ust_buffers_write(buf
, buf_offset
,
1848 buf_offset
+= sizeof(u16
);