local_t entries; local_t overrun; local_t commit_overrun; local_t dropped_events; local_t committing; local_t commits; local_t pages_touched; local_t pages_read; long last_pages_touch; size_t shortest_full; unsigned long read; unsigned long read_bytes; rb_time_t write_stamp; rb_time_t before_stamp; u64 read_stamp; /* ring buffer pages to update, > 0 to add, < 0 to remove */ long nr_pages_to_update; struct list_head new_pages; /* new pages to add */ struct work_struct update_pages_work; struct completion update_done; struct rb_irq_work irq_work;}; struct trace_buffer { unsigned flags; int cpus; atomic_t record_disabled; cpumask_var_t cpumask; struct lock_class_key *reader_lock_key; struct mutex mutex; struct ring_buffer_per_cpu **buffers; struct hlist_node node; u64 (*clock)(void); struct rb_irq_work irq_work; bool time_stamp_abs;}; struct ring_buffer_iter { struct ring_buffer_per_cpu *cpu_buffer; unsigned long head; unsigned long next_event; struct buffer_page *head_page; struct buffer_page *cache_reader_page; unsigned long cache_read; u64 read_stamp; u64 page_stamp; struct ring_buffer_event *event; int missed_events;}; #ifdef RB_TIME_32 /* * On 32 bit machines, local64_t is very expensive. As the ring * buffer doesn't need all the features of a true 64 bit atomic, * on 32 bit, it uses these functions (64 still uses local64_t). * * For the ring buffer, 64 bit required operations for the time is * the following: * * - Only need 59 bits (uses 60 to make it even). * - Reads may fail if it interrupted a modification of the time stamp. * It will succeed if it did not interrupt another write even if * the read itself is interrupted by a write. * It returns whether it was successful or not. * * - Writes always succeed and will overwrite other writes and writes * that were done by events interrupting the current write. * * - A write followed by a read of the same time stamp will always succeed, * but may not contain the same value. * * - A cmpxchg will fail if it interrupted another write or cmpxchg. * Other than that, it acts like a normal cmpxchg. * * The 60 bit time stamp is broken up by 30 bits in a top and bottom half * (bottom being the least significant 30 bits of the 60 bit time stamp). * * The two most significant bits of each half holds a 2 bit counter (0-3). * Each update will increment this counter by one. * When reading the top and bottom, if the two counter bits match then the * top and bottom together make a valid 60 bit number. */#define RB_TIME_SHIFT 30#define RB_TIME_VAL_MASK ((1 << RB_TIME_SHIFT) - 1) static inline int rb_time_cnt(unsigned long val){ return (val >> RB_TIME_SHIFT) & 3;} static inline u64 rb_time_val(unsigned long top, unsigned long bottom){ u64 val; val = top & RB_TIME_VAL_MASK; val <<= RB_TIME_SHIFT; val |= bottom & RB_TIME_VAL_MASK; return val;} static inline bool __rb_time_read(rb_time_t *t, u64 *ret, unsigned long *cnt){ unsigned long top, bottom; unsigned long c; /* * If the read is interrupted by a write, then the cnt will * be different. Loop until both top and bottom have been read * without interruption. */ do { c = local_read(&t->cnt); top = local_read(&t->top); bottom = local_read(&t->bottom); } while (c != local_read(&t->cnt)); *cnt = rb_time_cnt(top); /* If top and bottom counts don't match, this interrupted a write */ if (*cnt != rb_time_cnt(bottom)) return false; *ret = rb_time_val(top, bottom); return true;} static bool rb_time_read(rb_time_t *t, u64 *ret){ unsigned long cnt; return __rb_time_read(t, ret, &cnt);} static inline unsigned long rb_time_val_cnt(unsigned long val, unsigned long cnt){ return (val & RB_TIME_VAL_MASK) | ((cnt & 3) << RB_TIME_SHIFT);} static inline void rb_time_split(u64 val, unsigned long *top, unsigned long *bottom){ *top = (unsigned long)((val >> RB_TIME_SHIFT) & RB_TIME_VAL_MASK); *bottom = (unsigned long)(val & RB_TIME_VAL_MASK);} static inline void rb_time_val_set(local_t *t, unsigned long val, unsigned long cnt){ val = rb_time_val_cnt(val, cnt); local_set(t, val);} static void rb_time_set(rb_time_t *t, u64 val){ unsigned long cnt, top, bottom; rb_time_split(val, &top, &bottom); /* Writes always succeed with a valid number even if it gets interrupted. */ do { cnt = local_inc_return(&t->cnt); rb_time_val_set(&t->top, top, cnt); rb_time_val_set(&t->bottom, bottom, cnt); } while (cnt != local_read(&t->cnt));} static inline boolrb_time_read_cmpxchg(local_t *l, unsigned long expect, unsigned long set){ unsigned long ret; ret = local_cmpxchg(l, expect, set); return ret == expect;} static int rb_time_cmpxchg(rb_time_t *t, u64 expect, u64 set){ unsigned long cnt, top, bottom; unsigned long cnt2, top2, bottom2; u64 val; /* The cmpxchg always fails if it interrupted an update */ if (!__rb_time_read(t, &val, &cnt2)) return false; if (val != expect) return false; cnt = local_read(&t->cnt); if ((cnt & 3) != cnt2) return false; cnt2 = cnt + 1; rb_time_split(val, &top, &bottom); top = rb_time_val_cnt(top, cnt); bottom = rb_time_val_cnt(bottom, cnt); rb_time_split(set, &top2, &bottom2); top2 = rb_time_val_cnt(top2, cnt2); bottom2 = rb_time_val_cnt(bottom2, cnt2); if (!rb_time_read_cmpxchg(&t->cnt, cnt, cnt2)) return false; if (!rb_time_read_cmpxchg(&t->top, top, top2)) return false; if (!rb_time_read_cmpxchg(&t->bottom, bottom, bottom2)) return false; return true;} #else /* 64 bits */ /* local64_t always succeeds */ static inline bool rb_time_read(rb_time_t *t, u64 *ret){ *ret = local64_read(&t->time); return true;}static void rb_time_set(rb_time_t *t, u64 val){ local64_set(&t->time, val);} static bool rb_time_cmpxchg(rb_time_t *t, u64 expect, u64 set){ u64 val; val = local64_cmpxchg(&t->time, expect, set); return val == expect;}#endif /** * ring_buffer_nr_pages - get the number of buffer pages in the ring buffer * @buffer: The ring_buffer to get the number of pages from * @cpu: The cpu of the ring_buffer to get the number of pages from * * Returns the number of pages used by a per_cpu buffer of the ring buffer. */size_t ring_buffer_nr_pages(struct trace_buffer *buffer, int cpu){ return buffer->buffers[cpu]->nr_pages;} /** * ring_buffer_nr_pages_dirty - get the number of used pages in the ring buffer * @buffer: The ring_buffer to get the number of pages from * @cpu: The cpu of the ring_buffer to get the number of pages from * * Returns the number of pages that have content in the ring buffer. */size_t ring_buffer_nr_dirty_pages(struct trace_buffer *buffer, int cpu){ size_t read; size_t cnt; read = local_read(&buffer->buffers[cpu]->pages_read); cnt = local_read(&buffer->buffers[cpu]->pages_touched); /* The reader can read an empty page, but not more than that */ if (cnt < read) { WARN_ON_ONCE(read > cnt + 1); return 0; } return cnt - read;} /* * rb_wake_up_waiters - wake up tasks waiting for ring buffer input * * Schedules a delayed work to wake up any task that is blocked on the * ring buffer waiters queue. */static void rb_wake_up_waiters(struct irq_work *work){ struct rb_irq_work *rbwork = container_of(work, struct rb_irq_work, work); wake_up_all(&rbwork->waiters); if (rbwork->wakeup_full) { rbwork->wakeup_full = false; wake_up_all(&rbwork->full_waiters); }} /** * ring_buffer_wait - wait for input to the ring buffer * @buffer: buffer to wait on * @cpu: the cpu buffer to wait on * @full: wait until a full page is available, if @cpu != RING_BUFFER_ALL_CPUS * * If @cpu == RING_BUFFER_ALL_CPUS then the task will wake up as soon * as data is added to any of the @buffer's cpu buffers. Otherwise * it will wait for data to be added to a specific cpu buffer. */int ring_buffer_wait(struct trace_buffer *buffer, int cpu, int full){ struct ring_buffer_per_cpu *cpu_buffer; DEFINE_WAIT(wait); struct rb_irq_work *work; int ret = 0; /* * Depending on what the caller is waiting for, either any * data in any cpu buffer, or a specific buffer, put the * caller on the appropriate wait queue. */ if (cpu == RING_BUFFER_ALL_CPUS) { work = &buffer->irq_work; /* Full only makes sense on per cpu reads */ full = 0; } else { if (!cpumask_test_cpu(cpu, buffer->cpumask)) return -ENODEV; cpu_buffer = buffer->buffers[cpu]; work = &cpu_buffer->irq_work; } while (true) { if (full) prepare_to_wait(&work->full_waiters, &wait, TASK_INTERRUPTIBLE); else prepare_to_wait(&work->waiters, &wait, TASK_INTERRUPTIBLE); /* * The events can happen in critical sections where * checking a work queue can cause deadlocks. * After adding a task to the queue, this flag is set * only to notify events to try to wake up the queue * using irq_work. * * We don't clear it even if the buffer is no longer * empty. The flag only causes the next event to run * irq_work to do the work queue wake up. The worse * that can happen if we race with !trace_empty() is that * an event will cause an irq_work to try to wake up * an empty queue. * * There's no reason to protect this flag either, as * the work queue and irq_work logic will do the necessary * synchronization for the wake ups. The only thing * that is necessary is that the wake up happens after * a task has been queued. It's OK for spurious wake ups. */ if (full) work->full_waiters_pending = true; else work->waiters_pending = true; if (signal_pending(current)) { ret = -EINTR; break; } if (cpu == RING_BUFFER_ALL_CPUS && !ring_buffer_empty(buffer)) break; if (cpu != RING_BUFFER_ALL_CPUS && !ring_buffer_empty_cpu(buffer, cpu)) { unsigned long flags; bool pagebusy; size_t nr_pages; size_t dirty; if (!full) break; raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); pagebusy = cpu_buffer->reader_page == cpu_buffer->commit_page; nr_pages = cpu_buffer->nr_pages; dirty = ring_buffer_nr_dirty_pages(buffer, cpu); if (!cpu_buffer->shortest_full || cpu_buffer->shortest_full < full) cpu_buffer->shortest_full = full; raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); if (!pagebusy && (!nr_pages || (dirty * 100) > full * nr_pages)) break; } schedule(); } if (full) finish_wait(&work->full_waiters, &wait); else finish_wait(&work->waiters, &wait); return ret;} /** * ring_buffer_poll_wait - poll on buffer input * @buffer: buffer to wait on * @cpu: the cpu buffer to wait on * @filp: the file descriptor * @poll_table: The poll descriptor * * If @cpu == RING_BUFFER_ALL_CPUS then the task will wake up as soon * as data is added to any of the @buffer's cpu buffers. Otherwise * it will wait for data to be added to a specific cpu buffer. * * Returns EPOLLIN | EPOLLRDNORM if data exists in the buffers, * zero otherwise. */__poll_t ring_buffer_poll_wait(struct trace_buffer *buffer, int cpu, struct file *filp, poll_table *poll_table){ struct ring_buffer_per_cpu *cpu_buffer; struct rb_irq_work *work; if (cpu == RING_BUFFER_ALL_CPUS) work = &buffer->irq_work; else { if (!cpumask_test_cpu(cpu, buffer->cpumask)) return -EINVAL; cpu_buffer = buffer->buffers[cpu]; work = &cpu_buffer->irq_work; } poll_wait(filp, &work->waiters, poll_table); work->waiters_pending = true; /* * There's a tight race between setting the waiters_pending and * checking if the ring buffer is empty. Once the waiters_pending bit * is set, the next event will wake the task up, but we can get stuck * if there's only a single event in. * * FIXME: Ideally, we need a memory barrier on the writer side as well, * but adding a memory barrier to all events will cause too much of a * performance hit in the fast path. We only need a memory barrier when * the buffer goes from empty to having content. But as this race is * extremely small, and it's not a problem if another event comes in, we * will fix it later. */ smp_mb(); if ((cpu == RING_BUFFER_ALL_CPUS && !ring_buffer_empty(buffer)) || (cpu != RING_BUFFER_ALL_CPUS && !ring_buffer_empty_cpu(buffer, cpu))) return EPOLLIN | EPOLLRDNORM; return 0;} /* buffer may be either ring_buffer or ring_buffer_per_cpu */#define RB_WARN_ON(b, cond) \ ({ \ int _____ret = unlikely(cond); \ if (_____ret) { \ if (__same_type(*(b), struct ring_buffer_per_cpu)) { \ struct ring_buffer_per_cpu *__b = \ (void *)b; \ atomic_inc(&__b->buffer->record_disabled); \ } else \ atomic_inc(&b->record_disabled); \ WARN_ON(1); \ } \ _____ret; \ }) /* Up this if you want to test the TIME_EXTENTS and normalization */#define DEBUG_SHIFT 0 static inline u64 rb_time_stamp(struct trace_buffer *buffer){ u64 ts; /* Skip retpolines :-( */ if (IS_ENABLED(CONFIG_RETPOLINE) && likely(buffer->clock == trace_clock_local)) ts = trace_clock_local(); else ts = buffer->clock(); /* shift to debug/test normalization and TIME_EXTENTS */ return ts << DEBUG_SHIFT;} u64 ring_buffer_time_stamp(struct trace_buffer *buffer, int cpu){ u64 time; preempt_disable_notrace(); time = rb_time_stamp(buffer); preempt_enable_notrace(); return time;}EXPORT_SYMBOL_GPL(ring_buffer_time_stamp); void ring_buffer_normalize_time_stamp(struct trace_buffer *buffer, int cpu, u64 *ts){ /* Just stupid testing the normalize function and deltas */Lines 501–1000 of 5723. Previous lines Load more
local_t entries; local_t overrun; local_t commit_overrun; local_t dropped_events; local_t committing; local_t commits; local_t pages_touched; local_t pages_read; long last_pages_touch; size_t shortest_full; unsigned long read; unsigned long read_bytes; rb_time_t write_stamp; rb_time_t before_stamp; u64 read_stamp; /* ring buffer pages to update, > 0 to add, < 0 to remove */ long nr_pages_to_update; struct list_head new_pages; /* new pages to add */ struct work_struct update_pages_work; struct completion update_done; struct rb_irq_work irq_work; }; struct trace_buffer { unsigned flags; int cpus; atomic_t record_disabled; cpumask_var_t cpumask; struct lock_class_key *reader_lock_key; struct mutex mutex; struct ring_buffer_per_cpu **buffers; struct hlist_node node; u64 (*clock)(void); struct rb_irq_work irq_work; bool time_stamp_abs; }; struct ring_buffer_iter { struct ring_buffer_per_cpu *cpu_buffer; unsigned long head; unsigned long next_event; struct buffer_page *head_page; struct buffer_page *cache_reader_page; unsigned long cache_read; u64 read_stamp; u64 page_stamp; struct ring_buffer_event *event; int missed_events; }; #ifdef RB_TIME_32 /* * On 32 bit machines, local64_t is very expensive. As the ring * buffer doesn't need all the features of a true 64 bit atomic, * on 32 bit, it uses these functions (64 still uses local64_t). * * For the ring buffer, 64 bit required operations for the time is * the following: * * - Only need 59 bits (uses 60 to make it even). * - Reads may fail if it interrupted a modification of the time stamp. * It will succeed if it did not interrupt another write even if * the read itself is interrupted by a write. * It returns whether it was successful or not. * * - Writes always succeed and will overwrite other writes and writes * that were done by events interrupting the current write. * * - A write followed by a read of the same time stamp will always succeed, * but may not contain the same value. * * - A cmpxchg will fail if it interrupted another write or cmpxchg. * Other than that, it acts like a normal cmpxchg. * * The 60 bit time stamp is broken up by 30 bits in a top and bottom half * (bottom being the least significant 30 bits of the 60 bit time stamp). * * The two most significant bits of each half holds a 2 bit counter (0-3). * Each update will increment this counter by one. * When reading the top and bottom, if the two counter bits match then the * top and bottom together make a valid 60 bit number. */ #define RB_TIME_SHIFT 30 #define RB_TIME_VAL_MASK ((1 << RB_TIME_SHIFT) - 1) static inline int rb_time_cnt(unsigned long val) { return (val >> RB_TIME_SHIFT) & 3; } static inline u64 rb_time_val(unsigned long top, unsigned long bottom) { u64 val; val = top & RB_TIME_VAL_MASK; val <<= RB_TIME_SHIFT; val |= bottom & RB_TIME_VAL_MASK; return val; } static inline bool __rb_time_read(rb_time_t *t, u64 *ret, unsigned long *cnt) { unsigned long top, bottom; unsigned long c; /* * If the read is interrupted by a write, then the cnt will * be different. Loop until both top and bottom have been read * without interruption. */ do { c = local_read(&t->cnt); top = local_read(&t->top); bottom = local_read(&t->bottom); } while (c != local_read(&t->cnt)); *cnt = rb_time_cnt(top); /* If top and bottom counts don't match, this interrupted a write */ if (*cnt != rb_time_cnt(bottom)) return false; *ret = rb_time_val(top, bottom); return true; } static bool rb_time_read(rb_time_t *t, u64 *ret) { unsigned long cnt; return __rb_time_read(t, ret, &cnt); } static inline unsigned long rb_time_val_cnt(unsigned long val, unsigned long cnt) { return (val & RB_TIME_VAL_MASK) | ((cnt & 3) << RB_TIME_SHIFT); } static inline void rb_time_split(u64 val, unsigned long *top, unsigned long *bottom) { *top = (unsigned long)((val >> RB_TIME_SHIFT) & RB_TIME_VAL_MASK); *bottom = (unsigned long)(val & RB_TIME_VAL_MASK); } static inline void rb_time_val_set(local_t *t, unsigned long val, unsigned long cnt) { val = rb_time_val_cnt(val, cnt); local_set(t, val); } static void rb_time_set(rb_time_t *t, u64 val) { unsigned long cnt, top, bottom; rb_time_split(val, &top, &bottom); /* Writes always succeed with a valid number even if it gets interrupted. */ do { cnt = local_inc_return(&t->cnt); rb_time_val_set(&t->top, top, cnt); rb_time_val_set(&t->bottom, bottom, cnt); } while (cnt != local_read(&t->cnt)); } static inline bool rb_time_read_cmpxchg(local_t *l, unsigned long expect, unsigned long set) { unsigned long ret; ret = local_cmpxchg(l, expect, set); return ret == expect; } static int rb_time_cmpxchg(rb_time_t *t, u64 expect, u64 set) { unsigned long cnt, top, bottom; unsigned long cnt2, top2, bottom2; u64 val; /* The cmpxchg always fails if it interrupted an update */ if (!__rb_time_read(t, &val, &cnt2)) return false; if (val != expect) return false; cnt = local_read(&t->cnt); if ((cnt & 3) != cnt2) return false; cnt2 = cnt + 1; rb_time_split(val, &top, &bottom); top = rb_time_val_cnt(top, cnt); bottom = rb_time_val_cnt(bottom, cnt); rb_time_split(set, &top2, &bottom2); top2 = rb_time_val_cnt(top2, cnt2); bottom2 = rb_time_val_cnt(bottom2, cnt2); if (!rb_time_read_cmpxchg(&t->cnt, cnt, cnt2)) return false; if (!rb_time_read_cmpxchg(&t->top, top, top2)) return false; if (!rb_time_read_cmpxchg(&t->bottom, bottom, bottom2)) return false; return true; } #else /* 64 bits */ /* local64_t always succeeds */ static inline bool rb_time_read(rb_time_t *t, u64 *ret) { *ret = local64_read(&t->time); return true; } static void rb_time_set(rb_time_t *t, u64 val) { local64_set(&t->time, val); } static bool rb_time_cmpxchg(rb_time_t *t, u64 expect, u64 set) { u64 val; val = local64_cmpxchg(&t->time, expect, set); return val == expect; } #endif /** * ring_buffer_nr_pages - get the number of buffer pages in the ring buffer * @buffer: The ring_buffer to get the number of pages from * @cpu: The cpu of the ring_buffer to get the number of pages from * * Returns the number of pages used by a per_cpu buffer of the ring buffer. */ size_t ring_buffer_nr_pages(struct trace_buffer *buffer, int cpu) { return buffer->buffers[cpu]->nr_pages; } /** * ring_buffer_nr_pages_dirty - get the number of used pages in the ring buffer * @buffer: The ring_buffer to get the number of pages from * @cpu: The cpu of the ring_buffer to get the number of pages from * * Returns the number of pages that have content in the ring buffer. */ size_t ring_buffer_nr_dirty_pages(struct trace_buffer *buffer, int cpu) { size_t read; size_t cnt; read = local_read(&buffer->buffers[cpu]->pages_read); cnt = local_read(&buffer->buffers[cpu]->pages_touched); /* The reader can read an empty page, but not more than that */ if (cnt < read) { WARN_ON_ONCE(read > cnt + 1); return 0; } return cnt - read; } /* * rb_wake_up_waiters - wake up tasks waiting for ring buffer input * * Schedules a delayed work to wake up any task that is blocked on the * ring buffer waiters queue. */ static void rb_wake_up_waiters(struct irq_work *work) { struct rb_irq_work *rbwork = container_of(work, struct rb_irq_work, work); wake_up_all(&rbwork->waiters); if (rbwork->wakeup_full) { rbwork->wakeup_full = false; wake_up_all(&rbwork->full_waiters); } } /** * ring_buffer_wait - wait for input to the ring buffer * @buffer: buffer to wait on * @cpu: the cpu buffer to wait on * @full: wait until a full page is available, if @cpu != RING_BUFFER_ALL_CPUS * * If @cpu == RING_BUFFER_ALL_CPUS then the task will wake up as soon * as data is added to any of the @buffer's cpu buffers. Otherwise * it will wait for data to be added to a specific cpu buffer. */ int ring_buffer_wait(struct trace_buffer *buffer, int cpu, int full) { struct ring_buffer_per_cpu *cpu_buffer; DEFINE_WAIT(wait); struct rb_irq_work *work; int ret = 0; /* * Depending on what the caller is waiting for, either any * data in any cpu buffer, or a specific buffer, put the * caller on the appropriate wait queue. */ if (cpu == RING_BUFFER_ALL_CPUS) { work = &buffer->irq_work; /* Full only makes sense on per cpu reads */ full = 0; } else { if (!cpumask_test_cpu(cpu, buffer->cpumask)) return -ENODEV; cpu_buffer = buffer->buffers[cpu]; work = &cpu_buffer->irq_work; } while (true) { if (full) prepare_to_wait(&work->full_waiters, &wait, TASK_INTERRUPTIBLE); else prepare_to_wait(&work->waiters, &wait, TASK_INTERRUPTIBLE); /* * The events can happen in critical sections where * checking a work queue can cause deadlocks. * After adding a task to the queue, this flag is set * only to notify events to try to wake up the queue * using irq_work. * * We don't clear it even if the buffer is no longer * empty. The flag only causes the next event to run * irq_work to do the work queue wake up. The worse * that can happen if we race with !trace_empty() is that * an event will cause an irq_work to try to wake up * an empty queue. * * There's no reason to protect this flag either, as * the work queue and irq_work logic will do the necessary * synchronization for the wake ups. The only thing * that is necessary is that the wake up happens after * a task has been queued. It's OK for spurious wake ups. */ if (full) work->full_waiters_pending = true; else work->waiters_pending = true; if (signal_pending(current)) { ret = -EINTR; break; } if (cpu == RING_BUFFER_ALL_CPUS && !ring_buffer_empty(buffer)) break; if (cpu != RING_BUFFER_ALL_CPUS && !ring_buffer_empty_cpu(buffer, cpu)) { unsigned long flags; bool pagebusy; size_t nr_pages; size_t dirty; if (!full) break; raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags); pagebusy = cpu_buffer->reader_page == cpu_buffer->commit_page; nr_pages = cpu_buffer->nr_pages; dirty = ring_buffer_nr_dirty_pages(buffer, cpu); if (!cpu_buffer->shortest_full || cpu_buffer->shortest_full < full) cpu_buffer->shortest_full = full; raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); if (!pagebusy && (!nr_pages || (dirty * 100) > full * nr_pages)) break; } schedule(); } if (full) finish_wait(&work->full_waiters, &wait); else finish_wait(&work->waiters, &wait); return ret; } /** * ring_buffer_poll_wait - poll on buffer input * @buffer: buffer to wait on * @cpu: the cpu buffer to wait on * @filp: the file descriptor * @poll_table: The poll descriptor * * If @cpu == RING_BUFFER_ALL_CPUS then the task will wake up as soon * as data is added to any of the @buffer's cpu buffers. Otherwise * it will wait for data to be added to a specific cpu buffer. * * Returns EPOLLIN | EPOLLRDNORM if data exists in the buffers, * zero otherwise. */ __poll_t ring_buffer_poll_wait(struct trace_buffer *buffer, int cpu, struct file *filp, poll_table *poll_table) { struct ring_buffer_per_cpu *cpu_buffer; struct rb_irq_work *work; if (cpu == RING_BUFFER_ALL_CPUS) work = &buffer->irq_work; else { if (!cpumask_test_cpu(cpu, buffer->cpumask)) return -EINVAL; cpu_buffer = buffer->buffers[cpu]; work = &cpu_buffer->irq_work; } poll_wait(filp, &work->waiters, poll_table); work->waiters_pending = true; /* * There's a tight race between setting the waiters_pending and * checking if the ring buffer is empty. Once the waiters_pending bit * is set, the next event will wake the task up, but we can get stuck * if there's only a single event in. * * FIXME: Ideally, we need a memory barrier on the writer side as well, * but adding a memory barrier to all events will cause too much of a * performance hit in the fast path. We only need a memory barrier when * the buffer goes from empty to having content. But as this race is * extremely small, and it's not a problem if another event comes in, we * will fix it later. */ smp_mb(); if ((cpu == RING_BUFFER_ALL_CPUS && !ring_buffer_empty(buffer)) || (cpu != RING_BUFFER_ALL_CPUS && !ring_buffer_empty_cpu(buffer, cpu))) return EPOLLIN | EPOLLRDNORM; return 0; } /* buffer may be either ring_buffer or ring_buffer_per_cpu */ #define RB_WARN_ON(b, cond) \ ({ \ int _____ret = unlikely(cond); \ if (_____ret) { \ if (__same_type(*(b), struct ring_buffer_per_cpu)) { \ struct ring_buffer_per_cpu *__b = \ (void *)b; \ atomic_inc(&__b->buffer->record_disabled); \ } else \ atomic_inc(&b->record_disabled); \ WARN_ON(1); \ } \ _____ret; \ }) /* Up this if you want to test the TIME_EXTENTS and normalization */ #define DEBUG_SHIFT 0 static inline u64 rb_time_stamp(struct trace_buffer *buffer) { u64 ts; /* Skip retpolines :-( */ if (IS_ENABLED(CONFIG_RETPOLINE) && likely(buffer->clock == trace_clock_local)) ts = trace_clock_local(); else ts = buffer->clock(); /* shift to debug/test normalization and TIME_EXTENTS */ return ts << DEBUG_SHIFT; } u64 ring_buffer_time_stamp(struct trace_buffer *buffer, int cpu) { u64 time; preempt_disable_notrace(); time = rb_time_stamp(buffer); preempt_enable_notrace(); return time; } EXPORT_SYMBOL_GPL(ring_buffer_time_stamp); void ring_buffer_normalize_time_stamp(struct trace_buffer *buffer, int cpu, u64 *ts) { /* Just stupid testing the normalize function and deltas */