Groups | Search | Server Info | Keyboard shortcuts | Login | Register [http] [https] [nntp] [nntps]

Groups > linux.kernel > #1531709 > unrolled thread

Re: [BUG] msr-trace.h:42 suspicious rcu_dereference_check() usage!

Back to article view | Back to linux.kernel

This discussion starts older than the indexed window; earlier articles aren't shown. The article labeled Started by below is the oldest one visible, not the original post.

Contents

  Re: [BUG] msr-trace.h:42 suspicious rcu_dereference_check() usage! Andi Kleen <andi@firstfloor.org> - 2016-11-28 22:50 +0100
    Re: [BUG] msr-trace.h:42 suspicious rcu_dereference_check() usage! Steven Rostedt <rostedt@goodmis.org> - 2016-11-28 23:50 +0100
      Re: [BUG] msr-trace.h:42 suspicious rcu_dereference_check() usage! Andi Kleen <andi@firstfloor.org> - 2016-11-29 00:20 +0100

#1531709 — Re: [BUG] msr-trace.h:42 suspicious rcu_dereference_check() usage!

> I took a look at this and forced some more functions to be inlined. I
> did a little tweaking here and there. Could you pull my tree and see if
> things are better?  I don't currently have the hardware to run this
> myself.
> 
> git://git.kernel.org/pub/scm/linux/kernel/git/rostedt/linux-trace.git
> 
>   branch: ftrace/core

Didn't help much unfortunately. I think you need a real fast path here.

http://halobates.de/tracepoint-trace2

$ wc -l tracepoint-trace2
595 tracepoint-trace2

Opcode Histogram: 

    201 mov
     51 push
     51 pop
     34 test
     32 cmp
     26 jz
     24 jnz
     24 and
     16 ret
     16 lea
     15 call
     12 add
     10 jmp

Functions with # of instructions:

    25 trace_event_raw_event_sched_switch
     19 trace_event_buffer_reserve
     21 trace_event_buffer_lock_reserve
     57 ring_buffer_lock_reserve
      3 trace_clock_local
      3 sched_clock
     24 native_sched_clock
      3 sched_clock
      2 trace_clock_local
     11 ring_buffer_lock_reserve
     81 __rb_reserve_next
     11 ring_buffer_lock_reserve
      6 trace_event_buffer_lock_reserve
     18 ring_buffer_event_data
      5 trace_event_buffer_lock_reserve
     38 tracing_generic_entry_update
     11 trace_event_buffer_lock_reserve
      5 trace_event_buffer_reserve
     18 ring_buffer_event_data
      4 trace_event_buffer_reserve
     28 trace_event_raw_event_sched_switch
     30 trace_event_buffer_commit
     19 trace_buffer_unlock_commit_regs
     11 ring_buffer_unlock_commit
     85 rb_commit
     18 ring_buffer_unlock_commit
     10 trace_buffer_unlock_commit_regs
      3 ftrace_trace_userstack
      7 trace_buffer_unlock_commit_regs
     11 trace_event_buffer_commit
      8 trace_event_raw_event_sched_switch

-Andi

[toc] | [next] | [standalone]

#1531761

On Mon, 28 Nov 2016 13:48:25 -0800
Andi Kleen <andi@firstfloor.org> wrote:

> > I took a look at this and forced some more functions to be inlined. I
> > did a little tweaking here and there. Could you pull my tree and see if
> > things are better?  I don't currently have the hardware to run this
> > myself.
> > 
> > git://git.kernel.org/pub/scm/linux/kernel/git/rostedt/linux-trace.git
> > 
> >   branch: ftrace/core  
> 
> Didn't help much unfortunately. I think you need a real fast path here.

Well, actually it's been optimized a lot from the first instances.
Note, tracepoints need to be quite generic, so where it can be optimized
is not obvious.

> 
> http://halobates.de/tracepoint-trace2
> 
> $ wc -l tracepoint-trace2
> 595 tracepoint-trace2
> 
> Opcode Histogram: 
> 
>     201 mov
>      51 push
>      51 pop
>      34 test
>      32 cmp
>      26 jz
>      24 jnz
>      24 and
>      16 ret
>      16 lea
>      15 call
>      12 add
>      10 jmp
> 
> Functions with # of instructions:
> 
>     25 trace_event_raw_event_sched_switch

This is from the TRACE_EVENT macro.

trace_event_raw_event_##call(void *__data, proto)                       \
{                                                                       \
        struct trace_event_file *trace_file = __data;                   \
        struct trace_event_data_offsets_##call __maybe_unused __data_offsets;\
        struct trace_event_buffer fbuffer;                              \
        struct trace_event_raw_##call *entry;                           \
        int __data_size;                                                \
                                                                        \
        if (trace_trigger_soft_disabled(trace_file))                    \
                return;                                                 \
                                                                        \
        __data_size = trace_event_get_offsets_##call(&__data_offsets, args); \
									\
        entry = trace_event_buffer_reserve(&fbuffer, trace_file,        \
                                 sizeof(*entry) + __data_size);         \


>      19 trace_event_buffer_reserve

void *trace_event_buffer_reserve(struct trace_event_buffer *fbuffer,
                                 struct trace_event_file *trace_file,
                                 unsigned long len)
{
        struct trace_event_call *event_call = trace_file->event_call;

        if ((trace_file->flags & EVENT_FILE_FL_PID_FILTER) &&
            trace_event_ignore_this_pid(trace_file))
                return NULL;

        local_save_flags(fbuffer->flags);
        fbuffer->pc = preempt_count();
        /*
         * If CONFIG_PREEMPT is enabled, then the tracepoint itself disables
         * preemption (adding one to the preempt_count). Since we are
         * interested in the preempt_count at the time the tracepoint was
         * hit, we need to subtract one to offset the increment.
         */
        if (IS_ENABLED(CONFIG_PREEMPT))
                fbuffer->pc--;
        fbuffer->trace_file = trace_file;

        fbuffer->event =
                trace_event_buffer_lock_reserve(&fbuffer->buffer, trace_file,
                                                event_call->event.type, len,
                                                fbuffer->flags, fbuffer->pc);


>      21 trace_event_buffer_lock_reserve

struct ring_buffer_event *
trace_event_buffer_lock_reserve(struct ring_buffer **current_rb,
                          struct trace_event_file *trace_file,
                          int type, unsigned long len,
                          unsigned long flags, int pc)
{
        struct ring_buffer_event *entry;
        int val;

        *current_rb = trace_file->tr->trace_buffer.buffer;

        if ((trace_file->flags &
             (EVENT_FILE_FL_SOFT_DISABLED | EVENT_FILE_FL_FILTERED)) &&
            (entry = this_cpu_read(trace_buffered_event))) {


Note: Some of these if statements can be encapsulated with jump labels, as they
are false pretty much all of the time.


                /* Try to use the per cpu buffer first */
                val = this_cpu_inc_return(trace_buffered_event_cnt);
                if (val == 1) {
                        trace_event_setup(entry, type, flags, pc);
                        entry->array[0] = len;
                        return entry;
                }
                this_cpu_dec(trace_buffered_event_cnt);
        }

        entry = trace_buffer_lock_reserve(*current_rb,
                                         type, len, flags, pc);


>      57 ring_buffer_lock_reserve

struct ring_buffer_event *
ring_buffer_lock_reserve(struct ring_buffer *buffer, unsigned long length)
{
        struct ring_buffer_per_cpu *cpu_buffer;
        struct ring_buffer_event *event;
        int cpu;

        /* If we are tracing schedule, we don't want to recurse */
        preempt_disable_notrace();

        if (unlikely(atomic_read(&buffer->record_disabled)))
                goto out;

        cpu = raw_smp_processor_id();

        if (unlikely(!cpumask_test_cpu(cpu, buffer->cpumask)))
                goto out;

        cpu_buffer = buffer->buffers[cpu];

        if (unlikely(atomic_read(&cpu_buffer->record_disabled)))
                goto out;

        if (unlikely(length > BUF_MAX_DATA_SIZE))
                goto out;

        if (unlikely(trace_recursive_lock(cpu_buffer)))
                goto out;

        event = rb_reserve_next_event(buffer, cpu_buffer, length);
        if (!event)
                goto out_unlock;

static struct ring_buffer_event *
rb_reserve_next_event(struct ring_buffer *buffer,
                      struct ring_buffer_per_cpu *cpu_buffer,
                      unsigned long length)
{
        struct ring_buffer_event *event;
        struct rb_event_info info;
        int nr_loops = 0;
        u64 diff;

        rb_start_commit(cpu_buffer);

#ifdef CONFIG_RING_BUFFER_ALLOW_SWAP
        /*
         * Due to the ability to swap a cpu buffer from a buffer
         * it is possible it was swapped before we committed.
         * (committing stops a swap). We check for it here and
         * if it happened, we have to fail the write.
         */
        barrier();
        if (unlikely(ACCESS_ONCE(cpu_buffer->buffer) != buffer)) {
                local_dec(&cpu_buffer->committing);
                local_dec(&cpu_buffer->commits);
                return NULL;
        }
#endif

        info.length = rb_calculate_event_length(length);
 again:
        info.add_timestamp = 0;
        info.delta = 0;

        /*
         * We allow for interrupts to reenter here and do a trace.
         * If one does, it will cause this original code to loop
         * back here. Even with heavy interrupts happening, this
         * should only happen a few times in a row. If this happens
         * 1000 times in a row, there must be either an interrupt
         * storm or we have something buggy.
         * Bail!
         */
        if (RB_WARN_ON(cpu_buffer, ++nr_loops > 1000))
                goto out_fail;

        info.ts = rb_time_stamp(cpu_buffer->buffer);
        diff = info.ts - cpu_buffer->write_stamp;

        /* make sure this diff is calculated here */
        barrier();

        /* Did the write stamp get updated already? */
        if (likely(info.ts >= cpu_buffer->write_stamp)) {
                info.delta = diff;
                if (unlikely(test_time_stamp(info.delta)))
                        rb_handle_timestamp(cpu_buffer, &info);
        }

        event = __rb_reserve_next(cpu_buffer, &info);


>       3 trace_clock_local
>       3 sched_clock
>      24 native_sched_clock

Not much we can do about the clock

>       3 sched_clock
>       2 trace_clock_local
>      11 ring_buffer_lock_reserve
>      81 __rb_reserve_next

static struct ring_buffer_event *
__rb_reserve_next(struct ring_buffer_per_cpu *cpu_buffer,
                  struct rb_event_info *info)
{
        struct ring_buffer_event *event;
        struct buffer_page *tail_page;
        unsigned long tail, write;

        /*
         * If the time delta since the last event is too big to
         * hold in the time field of the event, then we append a
         * TIME EXTEND event ahead of the data event.
         */
        if (unlikely(info->add_timestamp))
                info->length += RB_LEN_TIME_EXTEND;

        /* Don't let the compiler play games with cpu_buffer->tail_page */
        tail_page = info->tail_page = READ_ONCE(cpu_buffer->tail_page);
        write = local_add_return(info->length, &tail_page->write);

        /* set write to only the index of the write */
        write &= RB_WRITE_MASK;
        tail = write - info->length;

        /*
         * If this is the first commit on the page, then it has the same
         * timestamp as the page itself.
         */
        if (!tail)
                info->delta = 0;

        /* See if we shot pass the end of this buffer page */
        if (unlikely(write > BUF_PAGE_SIZE))
                return rb_move_tail(cpu_buffer, tail, info);

        /* We reserved something on the buffer */

        event = __rb_page_index(tail_page, tail);
        kmemcheck_annotate_bitfield(event, bitfield);
        rb_update_event(cpu_buffer, event, info);

        local_inc(&tail_page->entries);

        /*
         * If this is the first commit on the page, then update
         * its timestamp.
         */
        if (!tail)
                tail_page->page->time_stamp = info->ts;

        /* account for these added bytes */
        local_add(info->length, &cpu_buffer->entries_bytes);

        return event;
}

>      11 ring_buffer_lock_reserve
>       6 trace_event_buffer_lock_reserve

        if (!fbuffer->event)
                return NULL;

        fbuffer->entry = ring_buffer_event_data(fbuffer->event);
        return fbuffer->entry;


>      18 ring_buffer_event_data

static void *
rb_event_data(struct ring_buffer_event *event)
{
        if (event->type_len == RINGBUF_TYPE_TIME_EXTEND)
                event = skip_time_extend(event);
        BUG_ON(event->type_len > RINGBUF_TYPE_DATA_TYPE_LEN_MAX);
        /* If length is in len field, then array[0] has the data */
        if (event->type_len)
                return (void *)&event->array[0];
        /* Otherwise length is in array[0] and array[1] has the data */
        return (void *)&event->array[1];
}

Hmm, the above could probably be optimized better.

>       5 trace_event_buffer_lock_reserve
>      38 tracing_generic_entry_update

I tried to optimize this one, but nothing really helped.

void
tracing_generic_entry_update(struct trace_entry *entry, unsigned long flags,
                             int pc)
{
        struct task_struct *tsk = current;

        entry->preempt_count            = pc & 0xff;
        entry->pid                      = (tsk) ? tsk->pid : 0;
        entry->flags =
#ifdef CONFIG_TRACE_IRQFLAGS_SUPPORT
                (irqs_disabled_flags(flags) ? TRACE_FLAG_IRQS_OFF : 0) |
#else
                TRACE_FLAG_IRQS_NOSUPPORT |
#endif
                ((pc & NMI_MASK    ) ? TRACE_FLAG_NMI     : 0) |
                ((pc & HARDIRQ_MASK) ? TRACE_FLAG_HARDIRQ : 0) |
                ((pc & SOFTIRQ_MASK) ? TRACE_FLAG_SOFTIRQ : 0) |
                (tif_need_resched() ? TRACE_FLAG_NEED_RESCHED : 0) |
                (test_preempt_need_resched() ? TRACE_FLAG_PREEMPT_RESCHED : 0);
}

>      11 trace_event_buffer_lock_reserve
>       5 trace_event_buffer_reserve
>      18 ring_buffer_event_data

Strange, this is called twice?

>       4 trace_event_buffer_reserve
>      28 trace_event_raw_event_sched_switch
>      30 trace_event_buffer_commit
>      19 trace_buffer_unlock_commit_regs
>      11 ring_buffer_unlock_commit
>      85 rb_commit

The rb_commit is a pain due to lockless commits.

Namely this part:

static void
rb_set_commit_to_write(struct ring_buffer_per_cpu *cpu_buffer)
{
        unsigned long max_count;

        /*
         * We only race with interrupts and NMIs on this CPU.
         * If we own the commit event, then we can commit
         * all others that interrupted us, since the interruptions
         * are in stack format (they finish before they come
         * back to us). This allows us to do a simple loop to
         * assign the commit to the tail.
         */
 again:
        max_count = cpu_buffer->nr_pages * 100;

        while (cpu_buffer->commit_page != READ_ONCE(cpu_buffer->tail_page)) {
                if (RB_WARN_ON(cpu_buffer, !(--max_count)))
                        return;
                if (RB_WARN_ON(cpu_buffer,
                               rb_is_reader_page(cpu_buffer->tail_page)))
                        return;
                local_set(&cpu_buffer->commit_page->page->commit,
                          rb_page_write(cpu_buffer->commit_page));
                rb_inc_page(cpu_buffer, &cpu_buffer->commit_page);
                /* Only update the write stamp if the page has an event */
                if (rb_page_write(cpu_buffer->commit_page))
                        cpu_buffer->write_stamp =
                                cpu_buffer->commit_page->page->time_stamp;
                /* add barrier to keep gcc from optimizing too much */
                barrier();
        }
        while (rb_commit_index(cpu_buffer) !=
               rb_page_write(cpu_buffer->commit_page)) {

                local_set(&cpu_buffer->commit_page->page->commit,
                          rb_page_write(cpu_buffer->commit_page));
                RB_WARN_ON(cpu_buffer,
                           local_read(&cpu_buffer->commit_page->page->commit) &
                           ~RB_WRITE_MASK);
                barrier();
        }

        /* again, keep gcc from optimizing */
        barrier();

        /*
         * If an interrupt came in just after the first while loop
         * and pushed the tail page forward, we will be left with
         * a dangling commit that will never go forward.
         */
        if (unlikely(cpu_buffer->commit_page != READ_ONCE(cpu_buffer->tail_page)))
                goto again;
}

>      18 ring_buffer_unlock_commit
>      10 trace_buffer_unlock_commit_regs
>       3 ftrace_trace_userstack

I should make this call a jump label, as it's seldom used.

>       7 trace_buffer_unlock_commit_regs
>      11 trace_event_buffer_commit
>       8 trace_event_raw_event_sched_switch
> 


There are a few things that can still be optimized. But the issue
really is that trace_events are generic and macro created, which I need
to balance how much gets inlined vs how much is passed around to
functions.

And the ring buffer is lockless, thus it needs to do various hoops to
make sure things don't break.

-- Steve

[toc] | [prev] | [next] | [standalone]

#1531792

> Well, actually it's been optimized a lot from the first instances.
> Note, tracepoints need to be quite generic, so where it can be optimized
> is not obvious.

There is quite a bit of stuff that doesn't change from trace point to
trace point. So you could just cache all these decisions into a single
per cpu variable, and invalidate if anything changes the conditions.

> > 
> > Functions with # of instructions:
> > 
> >     25 trace_event_raw_event_sched_switch
> 
> This is from the TRACE_EVENT macro.
> 
>         if (trace_trigger_soft_disabled(trace_file))                    \
>                 return;                                                 \

Cache?

>         if ((trace_file->flags & EVENT_FILE_FL_PID_FILTER) &&
>             trace_event_ignore_this_pid(trace_file))
>                 return NULL;
> 
>         *current_rb = trace_file->tr->trace_buffer.buffer;

Indirection could be cached.

> 
>         if ((trace_file->flags &
>              (EVENT_FILE_FL_SOFT_DISABLED | EVENT_FILE_FL_FILTERED)) &&
>             (entry = this_cpu_read(trace_buffered_event))) {

Multiple tests of trace_file->flags in different functions, 
could be all folded into a single test.

> 
> 
> Note: Some of these if statements can be encapsulated with jump labels, as they
> are false pretty much all of the time.
> 
> 
>                 /* Try to use the per cpu buffer first */
>                 val = this_cpu_inc_return(trace_buffered_event_cnt);
>                 if (val == 1) {
>                         trace_event_setup(entry, type, flags, pc);
>                         entry->array[0] = len;
>                         return entry;
>                 }
>                 this_cpu_dec(trace_buffered_event_cnt);

This can be pre cached in a fast path. If true just point per cpu
variable directly to buffer and invalidate if buffer changes or is full.

>         preempt_disable_notrace();
> 
>         if (unlikely(atomic_read(&buffer->record_disabled)))
>                 goto out;

Cache. 
> 
>         cpu = raw_smp_processor_id();
> 
>         if (unlikely(!cpumask_test_cpu(cpu, buffer->cpumask)))
>                 goto out;

Not needed for per cpu buffers in a fast path.

> 
>         cpu_buffer = buffer->buffers[cpu];
> 
>         if (unlikely(atomic_read(&cpu_buffer->record_disabled)))
>                 goto out;
cache

> 
>         if (unlikely(length > BUF_MAX_DATA_SIZE))
>                 goto out;
> 
>         if (unlikely(trace_recursive_lock(cpu_buffer)))
>                 goto out;

locking should invalidate state on this cpu, so shouldn't be needed

>         u64 diff;
> 
>         rb_start_commit(cpu_buffer);
> 
> #ifdef CONFIG_RING_BUFFER_ALLOW_SWAP
>         /*
>          * Due to the ability to swap a cpu buffer from a buffer
>          * it is possible it was swapped before we committed.
>          * (committing stops a swap). We check for it here and
>          * if it happened, we have to fail the write.
>          */
>         barrier();
>         if (unlikely(ACCESS_ONCE(cpu_buffer->buffer) != buffer)) {
>                 local_dec(&cpu_buffer->committing);
>                 local_dec(&cpu_buffer->commits);
>                 return NULL;
>         }

swapping should invalidate state centrally (and then run slow path like
current path)

>         /*
>          * We allow for interrupts to reenter here and do a trace.
>          * If one does, it will cause this original code to loop
>          * back here. Even with heavy interrupts happening, this
>          * should only happen a few times in a row. If this happens
>          * 1000 times in a row, there must be either an interrupt
>          * storm or we have something buggy.
>          * Bail!
>          */
>         if (RB_WARN_ON(cpu_buffer, ++nr_loops > 1000))
>                 goto out_fail;

didn't you disable interrupts earlier?

>         /* Did the write stamp get updated already? */
>         if (likely(info.ts >= cpu_buffer->write_stamp)) {
>                 info.delta = diff;
>                 if (unlikely(test_time_stamp(info.delta)))
>                         rb_handle_timestamp(cpu_buffer, &info);
>         }

Not sure why a write stamp is needed. isn't that the last few entries?

...

-Andi

[toc] | [prev] | [standalone]

Back to top | Article view | linux.kernel

csiph-web