// SPDX-License-Identifier: GPL-2.0

/*

* Generic ring buffer

*

* Copyright (C) 2008 Steven Rostedt <srostedt@redhat.com>

*/

#include <linux/ring_buffer_types.h>

#include <linux/sched/isolation.h>

#include <linux/trace_recursion.h>

#include <linux/trace_events.h>

#include <linux/ring_buffer.h>

#include <linux/trace_clock.h>

#include <linux/sched/clock.h>

#include <linux/cacheflush.h>

#include <linux/trace_seq.h>

#include <linux/spinlock.h>

#include <linux/irq_work.h>

#include <linux/security.h>

#include <linux/uaccess.h>

#include <linux/hardirq.h>

#include <linux/kthread.h> /* for self test */

#include <linux/module.h>

#include <linux/percpu.h>

#include <linux/mutex.h>

#include <linux/delay.h>

#include <linux/slab.h>

#include <linux/init.h>

#include <linux/hash.h>

#include <linux/list.h>

#include <linux/cpu.h>

#include <linux/oom.h>

#include <linux/mm.h>

#include <asm/local64.h>

#include <asm/local.h>

#include <asm/setup.h>

#include "trace.h"

/*

* The "absolute" timestamp in the buffer is only 59 bits.

* If a clock has the 5 MSBs set, it needs to be saved and

* reinserted.

*/

#define TS_MSB (0xf8ULL << 56)

#define ABS_TS_MASK (~TS_MSB)

static void update_pages_handler(struct work_struct *work);

#define RING_BUFFER_META_MAGIC 0xBADFEED

struct ring_buffer_meta {

int magic;

int struct_sizes;

unsigned long total_size;

unsigned long buffers_offset;

};

struct ring_buffer_cpu_meta {

unsigned long first_buffer;

unsigned long head_buffer;

unsigned long commit_buffer;

__u32 subbuf_size;

__u32 nr_subbufs;

int buffers[];

};

/*

* The ring buffer header is special. We must manually up keep it.

*/

int ring_buffer_print_entry_header(struct trace_seq *s)

{

trace_seq_puts(s, "# compressed entry header\n");

trace_seq_puts(s, "\ttype_len : 5 bits\n");

trace_seq_puts(s, "\ttime_delta : 27 bits\n");

trace_seq_puts(s, "\tarray : 32 bits\n");

trace_seq_putc(s, '\n');

trace_seq_printf(s, "\tpadding : type == %d\n",

RINGBUF_TYPE_PADDING);

trace_seq_printf(s, "\ttime_extend : type == %d\n",

RINGBUF_TYPE_TIME_EXTEND);

trace_seq_printf(s, "\ttime_stamp : type == %d\n",

RINGBUF_TYPE_TIME_STAMP);

trace_seq_printf(s, "\tdata max type_len == %d\n",

RINGBUF_TYPE_DATA_TYPE_LEN_MAX);

return !trace_seq_has_overflowed(s);

}

/*

* The ring buffer is made up of a list of pages. A separate list of pages is

* allocated for each CPU. A writer may only write to a buffer that is

* associated with the CPU it is currently executing on. A reader may read

* from any per cpu buffer.

*

* The reader is special. For each per cpu buffer, the reader has its own

* reader page. When a reader has read the entire reader page, this reader

* page is swapped with another page in the ring buffer.

*

* Now, as long as the writer is off the reader page, the reader can do what

* ever it wants with that page. The writer will never write to that page

* again (as long as it is out of the ring buffer).

*

* Here's some silly ASCII art.

*

* +------+

* |reader| RING BUFFER

* |page |

* +------+ +---+ +---+ +---+

* | |-->| |-->| |

* +---+ +---+ +---+

* ^ |

* | |

* +---------------+

*

*

* +------+

* |reader| RING BUFFER

* |page |------------------v

* +------+ +---+ +---+ +---+

* | |-->| |-->| |

* +---+ +---+ +---+

* ^ |

* | |

* +---------------+

*

*

* +------+

* |reader| RING BUFFER

* |page |------------------v

* +------+ +---+ +---+ +---+

* ^ | |-->| |-->| |

* | +---+ +---+ +---+

* | |

* | |

* +------------------------------+

*

*

* +------+

* |buffer| RING BUFFER

* |page |------------------v

* +------+ +---+ +---+ +---+

* ^ | | | |-->| |

* | New +---+ +---+ +---+

* | Reader------^ |

* | page |

* +------------------------------+

*

*

* After we make this swap, the reader can hand this page off to the splice

* code and be done with it. It can even allocate a new page if it needs to

* and swap that into the ring buffer.

*

* We will be using cmpxchg soon to make all this lockless.

*

*/

/* Used for individual buffers (after the counter) */

#define RB_BUFFER_OFF (1 << 20)

/* define RINGBUF_TYPE_DATA for 'case RINGBUF_TYPE_DATA:' */

#define RINGBUF_TYPE_DATA 0 ... RINGBUF_TYPE_DATA_TYPE_LEN_MAX

enum {

RB_LEN_TIME_EXTEND = 8,

RB_LEN_TIME_STAMP = 8,

};

#define skip_time_extend(event) \

((struct ring_buffer_event *)((char *)event + RB_LEN_TIME_EXTEND))

#define extended_time(event) \

(event->type_len >= RINGBUF_TYPE_TIME_EXTEND)

static inline bool rb_null_event(struct ring_buffer_event *event)

{

return event->type_len == RINGBUF_TYPE_PADDING && !event->time_delta;

}

static void rb_event_set_padding(struct ring_buffer_event *event)

{

/* padding has a NULL time_delta */

event->type_len = RINGBUF_TYPE_PADDING;

event->time_delta = 0;

}

static unsigned

rb_event_data_length(struct ring_buffer_event *event)

{

unsigned length;

if (event->type_len)

length = event->type_len * RB_ALIGNMENT;

else

length = event->array[0];

return length + RB_EVNT_HDR_SIZE;

}

/*

* Return the length of the given event. Will return

* the length of the time extend if the event is a

* time extend.

*/

static inline unsigned

rb_event_length(struct ring_buffer_event *event)

{

switch (event->type_len) {

case RINGBUF_TYPE_PADDING:

if (rb_null_event(event))

/* undefined */

return -1;

return event->array[0] + RB_EVNT_HDR_SIZE;

case RINGBUF_TYPE_TIME_EXTEND:

return RB_LEN_TIME_EXTEND;

case RINGBUF_TYPE_TIME_STAMP:

return RB_LEN_TIME_STAMP;

case RINGBUF_TYPE_DATA:

return rb_event_data_length(event);

default:

WARN_ON_ONCE(1);

}

/* not hit */

return 0;

}

/*

* Return total length of time extend and data,

* or just the event length for all other events.

*/

static inline unsigned

rb_event_ts_length(struct ring_buffer_event *event)

{

unsigned len = 0;

if (extended_time(event)) {

/* time extends include the data event after it */

len = RB_LEN_TIME_EXTEND;

event = skip_time_extend(event);

}

return len + rb_event_length(event);

}

/**

* ring_buffer_event_length - return the length of the event

* @event: the event to get the length of

*

* Returns the size of the data load of a data event.

* If the event is something other than a data event, it

* returns the size of the event itself. With the exception

* of a TIME EXTEND, where it still returns the size of the

* data load of the data event after it.

*/

unsigned ring_buffer_event_length(struct ring_buffer_event *event)

{

unsigned length;

if (extended_time(event))

event = skip_time_extend(event);

length = rb_event_length(event);

if (event->type_len > RINGBUF_TYPE_DATA_TYPE_LEN_MAX)

return length;

length -= RB_EVNT_HDR_SIZE;

if (length > RB_MAX_SMALL_DATA + sizeof(event->array[0]))

length -= sizeof(event->array[0]);

return length;

}

EXPORT_SYMBOL_GPL(ring_buffer_event_length);

/* inline for ring buffer fast paths */

static __always_inline void *

rb_event_data(struct ring_buffer_event *event)

{

if (extended_time(event))

event = skip_time_extend(event);

WARN_ON_ONCE(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];

}

/**

* ring_buffer_event_data - return the data of the event

* @event: the event to get the data from

*/

void *ring_buffer_event_data(struct ring_buffer_event *event)

{

return rb_event_data(event);

}

EXPORT_SYMBOL_GPL(ring_buffer_event_data);

#define for_each_buffer_cpu(buffer, cpu) \

for_each_cpu(cpu, buffer->cpumask)

#define for_each_online_buffer_cpu(buffer, cpu) \

for_each_cpu_and(cpu, buffer->cpumask, cpu_online_mask)

static u64 rb_event_time_stamp(struct ring_buffer_event *event)

{

u64 ts;

ts = event->array[0];

ts <<= TS_SHIFT;

ts += event->time_delta;

return ts;

}

/* Flag when events were overwritten */

#define RB_MISSED_EVENTS (1 << 31)

/* Missed count stored at end */

#define RB_MISSED_STORED (1 << 30)

#define RB_MISSED_MASK (3 << 30)

struct buffer_data_read_page {

unsigned order; /* order of the page */

struct buffer_data_page *data; /* actual data, stored in this page */

};

/*

* Note, the buffer_page list must be first. The buffer pages

* are allocated in cache lines, which means that each buffer

* page will be at the beginning of a cache line, and thus

* the least significant bits will be zero. We use this to

* add flags in the list struct pointers, to make the ring buffer

* lockless.

*/

struct buffer_page {

struct list_head list; /* list of buffer pages */

local_t write; /* index for next write */

unsigned read; /* index for next read */

local_t entries; /* entries on this page */

unsigned long real_end; /* real end of data */

unsigned order; /* order of the page */

u32 id:30; /* ID for external mapping */

u32 range:1; /* Mapped via a range */

struct buffer_data_page *page; /* Actual data page */

};

/*

* The buffer page counters, write and entries, must be reset

* atomically when crossing page boundaries. To synchronize this

* update, two counters are inserted into the number. One is

* the actual counter for the write position or count on the page.

*

* The other is a counter of updaters. Before an update happens

* the update partition of the counter is incremented. This will

* allow the updater to update the counter atomically.

*

* The counter is 20 bits, and the state data is 12.

*/

#define RB_WRITE_MASK 0xfffff

#define RB_WRITE_INTCNT (1 << 20)

static void rb_init_page(struct buffer_data_page *bpage)

{

local_set(&bpage->commit, 0);

}

static __always_inline unsigned int rb_page_commit(struct buffer_page *bpage)

{

return local_read(&bpage->page->commit);

}

static void free_buffer_page(struct buffer_page *bpage)

{

/* Range pages are not to be freed */

if (!bpage->range)

free_pages((unsigned long)bpage->page, bpage->order);

kfree(bpage);

}

/*

* For best performance, allocate cpu buffer data cache line sized

* and per CPU.

*/

#define alloc_cpu_buffer(cpu) (struct ring_buffer_per_cpu *) \

kzalloc_node(ALIGN(sizeof(struct ring_buffer_per_cpu), \

cache_line_size()), GFP_KERNEL, cpu_to_node(cpu));

#define alloc_cpu_page(cpu) (struct buffer_page *) \

kzalloc_node(ALIGN(sizeof(struct buffer_page), \

cache_line_size()), GFP_KERNEL, cpu_to_node(cpu));

static struct buffer_data_page *alloc_cpu_data(int cpu, int order)

{

struct buffer_data_page *dpage;

struct page *page;

gfp_t mflags;

/*

* __GFP_RETRY_MAYFAIL flag makes sure that the allocation fails

* gracefully without invoking oom-killer and the system is not

* destabilized.

*/

mflags = GFP_KERNEL | __GFP_RETRY_MAYFAIL | __GFP_COMP | __GFP_ZERO;

page = alloc_pages_node(cpu_to_node(cpu), mflags, order);

if (!page)

return NULL;

dpage = page_address(page);

rb_init_page(dpage);

return dpage;

}

struct rb_irq_work {

struct irq_work work;

wait_queue_head_t waiters;

wait_queue_head_t full_waiters;

atomic_t seq;

bool waiters_pending;

bool full_waiters_pending;

bool wakeup_full;

};

/*

* Structure to hold event state and handle nested events.

*/

struct rb_event_info {

u64 ts;

u64 delta;

u64 before;

u64 after;

unsigned long length;

struct buffer_page *tail_page;

int add_timestamp;

};

/*

* Used for the add_timestamp

* NONE

* EXTEND - wants a time extend

* ABSOLUTE - the buffer requests all events to have absolute time stamps

* FORCE - force a full time stamp.

*/

enum {

RB_ADD_STAMP_NONE = 0,

RB_ADD_STAMP_EXTEND = BIT(1),

RB_ADD_STAMP_ABSOLUTE = BIT(2),

RB_ADD_STAMP_FORCE = BIT(3)

};

/*

* Used for which event context the event is in.

* TRANSITION = 0

* NMI = 1

* IRQ = 2

* SOFTIRQ = 3

* NORMAL = 4

*

* See trace_recursive_lock() comment below for more details.

*/

enum {

RB_CTX_TRANSITION,

RB_CTX_NMI,

RB_CTX_IRQ,

RB_CTX_SOFTIRQ,

RB_CTX_NORMAL,

RB_CTX_MAX

};

struct rb_time_struct {

local64_t time;

};

typedef struct rb_time_struct rb_time_t;

#define MAX_NEST 5

/*

* head_page == tail_page && head == tail then buffer is empty.

*/

struct ring_buffer_per_cpu {

int cpu;

atomic_t record_disabled;

atomic_t resize_disabled;

struct trace_buffer *buffer;

raw_spinlock_t reader_lock; /* serialize readers */

arch_spinlock_t lock;

struct lock_class_key lock_key;

struct buffer_data_page *free_page;

unsigned long nr_pages;

unsigned int current_context;

struct list_head *pages;

/* pages generation counter, incremented when the list changes */

unsigned long cnt;

struct buffer_page *head_page; /* read from head */

struct buffer_page *tail_page; /* write to tail */

struct buffer_page *commit_page; /* committed pages */

struct buffer_page *reader_page;

unsigned long lost_events;

unsigned long last_overrun;

unsigned long nest;

local_t entries_bytes;

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_lost;

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 event_stamp[MAX_NEST];

u64 read_stamp;

/* pages removed since last reset */

unsigned long pages_removed;

unsigned int mapped;

unsigned int user_mapped; /* user space mapping */

struct mutex mapping_lock;

struct buffer_page **subbuf_ids; /* ID to subbuf VA */

struct trace_buffer_meta *meta_page;

struct ring_buffer_cpu_meta *ring_meta;

struct ring_buffer_remote *remote;

/* 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;

atomic_t resizing;

cpumask_var_t cpumask;

struct lock_class_key *reader_lock_key;

struct mutex mutex;

struct ring_buffer_per_cpu **buffers;

struct ring_buffer_remote *remote;

struct hlist_node node;

u64 (*clock)(void);

struct rb_irq_work irq_work;

bool time_stamp_abs;

unsigned long range_addr_start;

unsigned long range_addr_end;

struct ring_buffer_meta *meta;

unsigned int subbuf_size;

unsigned int subbuf_order;

unsigned int max_data_size;

};

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;

unsigned long cache_pages_removed;

u64 read_stamp;

u64 page_stamp;

struct ring_buffer_event *event;

size_t event_size;

int missed_events;

};

int ring_buffer_print_page_header(struct trace_buffer *buffer, struct trace_seq *s)

{

struct buffer_data_page field;

trace_seq_printf(s, "\tfield: u64 timestamp;\t"

"offset:0;\tsize:%u;\tsigned:%u;\n",

(unsigned int)sizeof(field.time_stamp),

(unsigned int)is_signed_type(u64));

trace_seq_printf(s, "\tfield: local_t commit;\t"

"offset:%u;\tsize:%u;\tsigned:%u;\n",

(unsigned int)offsetof(typeof(field), commit),

(unsigned int)sizeof(field.commit),

(unsigned int)is_signed_type(long));

trace_seq_printf(s, "\tfield: char overwrite;\t"

"offset:%u;\tsize:%u;\tsigned:%u;\n",

(unsigned int)offsetof(typeof(field), commit),

1,

(unsigned int)is_signed_type(char));

trace_seq_printf(s, "\tfield: char data;\t"

"offset:%u;\tsize:%u;\tsigned:%u;\n",

(unsigned int)offsetof(typeof(field), data),

(unsigned int)(buffer ? buffer->subbuf_size :

PAGE_SIZE - BUF_PAGE_HDR_SIZE),

(unsigned int)is_signed_type(char));

return !trace_seq_has_overflowed(s);

}

static inline void rb_time_read(rb_time_t *t, u64 *ret)

{

*ret = local64_read(&t->time);

}

static void rb_time_set(rb_time_t *t, u64 val)

{

local64_set(&t->time, val);

}

/*

* Enable this to make sure that the event passed to

* ring_buffer_event_time_stamp() is not committed and also

* is on the buffer that it passed in.

*/

//#define RB_VERIFY_EVENT

#ifdef RB_VERIFY_EVENT

static struct list_head *rb_list_head(struct list_head *list);

static void verify_event(struct ring_buffer_per_cpu *cpu_buffer,

void *event)

{

struct buffer_page *page = cpu_buffer->commit_page;

struct buffer_page *tail_page = READ_ONCE(cpu_buffer->tail_page);

struct list_head *next;

long commit, write;

unsigned long addr = (unsigned long)event;

bool done = false;

int stop = 0;

/* Make sure the event exists and is not committed yet */

do {

if (page == tail_page || WARN_ON_ONCE(stop++ > 100))

done = true;

commit = local_read(&page->page->commit);

write = local_read(&page->write);

if (addr >= (unsigned long)&page->page->data[commit] &&

addr < (unsigned long)&page->page->data[write])

return;

next = rb_list_head(page->list.next);

page = list_entry(next, struct buffer_page, list);

} while (!done);

WARN_ON_ONCE(1);

}

#else

static inline void verify_event(struct ring_buffer_per_cpu *cpu_buffer,

void *event)

{

}

#endif

/*

* The absolute time stamp drops the 5 MSBs and some clocks may

* require them. The rb_fix_abs_ts() will take a previous full

* time stamp, and add the 5 MSB of that time stamp on to the

* saved absolute time stamp. Then they are compared in case of

* the unlikely event that the latest time stamp incremented

* the 5 MSB.

*/

static inline u64 rb_fix_abs_ts(u64 abs, u64 save_ts)

{

if (save_ts & TS_MSB) {

abs |= save_ts & TS_MSB;

/* Check for overflow */

if (unlikely(abs < save_ts))

abs += 1ULL << 59;

}

return abs;

}

static inline u64 rb_time_stamp(struct trace_buffer *buffer);

/**

* ring_buffer_event_time_stamp - return the event's current time stamp

* @buffer: The buffer that the event is on

* @event: the event to get the time stamp of

*

* Note, this must be called after @event is reserved, and before it is

* committed to the ring buffer. And must be called from the same

* context where the event was reserved (normal, softirq, irq, etc).

*

* Returns the time stamp associated with the current event.

* If the event has an extended time stamp, then that is used as

* the time stamp to return.

* In the highly unlikely case that the event was nested more than

* the max nesting, then the write_stamp of the buffer is returned,

* otherwise current time is returned, but that really neither of

* the last two cases should ever happen.

*/

u64 ring_buffer_event_time_stamp(struct trace_buffer *buffer,

struct ring_buffer_event *event)

{

struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[smp_processor_id()];

unsigned int nest;

u64 ts;

/* If the event includes an absolute time, then just use that */

if (event->type_len == RINGBUF_TYPE_TIME_STAMP) {

ts = rb_event_time_stamp(event);

return rb_fix_abs_ts(ts, cpu_buffer->tail_page->page->time_stamp);

}

nest = local_read(&cpu_buffer->committing);

verify_event(cpu_buffer, event);

if (WARN_ON_ONCE(!nest))

goto fail;

/* Read the current saved nesting level time stamp */

if (likely(--nest < MAX_NEST))

return cpu_buffer->event_stamp[nest];

/* Shouldn't happen, warn if it does */

WARN_ONCE(1, "nest (%d) greater than max", nest);

fail:

rb_time_read(&cpu_buffer->write_stamp, &ts);

return ts;

}

/**

* ring_buffer_nr_dirty_pages - 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 lost;

size_t cnt;

read = local_read(&buffer->buffers[cpu]->pages_read);

lost = local_read(&buffer->buffers[cpu]->pages_lost);

cnt = local_read(&buffer->buffers[cpu]->pages_touched);

if (WARN_ON_ONCE(cnt < lost))

return 0;

cnt -= lost;

/* 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;

}

static __always_inline bool full_hit(struct trace_buffer *buffer, int cpu, int full)

{

struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu];

size_t nr_pages;

size_t dirty;

nr_pages = cpu_buffer->nr_pages;

if (!nr_pages || !full)

return true;

/*

* Add one as dirty will never equal nr_pages, as the sub-buffer

* that the writer is on is not counted as dirty.

* This is needed if "buffer_percent" is set to 100.

*/

dirty = ring_buffer_nr_dirty_pages(buffer, cpu) + 1;

return (dirty * 100) >= (full * nr_pages);

}

/*

* 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);

/* For waiters waiting for the first wake up */

(void)atomic_fetch_inc_release(&rbwork->seq);

wake_up_all(&rbwork->waiters);

if (rbwork->full_waiters_pending || rbwork->wakeup_full) {

/* Only cpu_buffer sets the above flags */

struct ring_buffer_per_cpu *cpu_buffer =

container_of(rbwork, struct ring_buffer_per_cpu, irq_work);

/* Called from interrupt context */

raw_spin_lock(&cpu_buffer->reader_lock);

rbwork->wakeup_full = false;

rbwork->full_waiters_pending = false;

/* Waking up all waiters, they will reset the shortest full */

cpu_buffer->shortest_full = 0;

raw_spin_unlock(&cpu_buffer->reader_lock);

wake_up_all(&rbwork->full_waiters);

}

}

/**

* ring_buffer_wake_waiters - wake up any waiters on this ring buffer

* @buffer: The ring buffer to wake waiters on

* @cpu: The CPU buffer to wake waiters on

*

* In the case of a file that represents a ring buffer is closing,

* it is prudent to wake up any waiters that are on this.

*/

void ring_buffer_wake_waiters(struct trace_buffer *buffer, int cpu)

{

struct ring_buffer_per_cpu *cpu_buffer;

struct rb_irq_work *rbwork;

if (!buffer)

return;

if (cpu == RING_BUFFER_ALL_CPUS) {

/* Wake up individual ones too. One level recursion */

for_each_buffer_cpu(buffer, cpu)

ring_buffer_wake_waiters(buffer, cpu);

rbwork = &buffer->irq_work;

} else {

if (WARN_ON_ONCE(!buffer->buffers))

return;

if (WARN_ON_ONCE(cpu >= nr_cpu_ids))

return;

cpu_buffer = buffer->buffers[cpu];

/* The CPU buffer may not have been initialized yet */

if (!cpu_buffer)

return;

rbwork = &cpu_buffer->irq_work;

}

/* This can be called in any context */

irq_work_queue(&rbwork->work);

}

static bool rb_watermark_hit(struct trace_buffer *buffer, int cpu, int full)

{

struct ring_buffer_per_cpu *cpu_buffer;

bool ret = false;

/* Reads of all CPUs always waits for any data */

if (cpu == RING_BUFFER_ALL_CPUS)

return !ring_buffer_empty(buffer);

cpu_buffer = buffer->buffers[cpu];

if (!ring_buffer_empty_cpu(buffer, cpu)) {

unsigned long flags;

bool pagebusy;

if (!full)

return true;

raw_spin_lock_irqsave(&cpu_buffer->reader_lock, flags);

pagebusy = cpu_buffer->reader_page == cpu_buffer->commit_page;

ret = !pagebusy && full_hit(buffer, cpu, full);

if (!ret && (!cpu_buffer->shortest_full ||

cpu_buffer->shortest_full > full)) {

cpu_buffer->shortest_full = full;

}

raw_spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags);

}

return ret;

}

static inline bool

rb_wait_cond(struct rb_irq_work *rbwork, struct trace_buffer *buffer,

int cpu, int full, ring_buffer_cond_fn cond, void *data)

{

if (rb_watermark_hit(buffer, cpu, full))

return true;

if (cond(data))

return true;

/*

* 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)

rbwork->full_waiters_pending = true;

else

rbwork->waiters_pending = true;

return false;

}

struct rb_wait_data {

struct rb_irq_work *irq_work;

int seq;

};

/*

* The default wait condition for ring_buffer_wait() is to just to exit the

* wait loop the first time it is woken up.

*/

static bool rb_wait_once(void *data)

{

struct rb_wait_data *rdata = data;

struct rb_irq_work *rbwork = rdata->irq_work;

return atomic_read_acquire(&rbwork->seq) != rdata->seq;

}

/**

* 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 the percentage of pages are available, if @cpu != RING_BUFFER_ALL_CPUS

* @cond: condition function to break out of wait (NULL to run once)

* @data: the data to pass to @cond.

*

* 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,

ring_buffer_cond_fn cond, void *data)

{

struct ring_buffer_per_cpu *cpu_buffer;

struct wait_queue_head *waitq;

struct rb_irq_work *rbwork;

struct rb_wait_data rdata;

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) {

rbwork = &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];

rbwork = &cpu_buffer->irq_work;

}

if (full)

waitq = &rbwork->full_waiters;

else

waitq = &rbwork->waiters;

/* Set up to exit loop as soon as it is woken */

if (!cond) {

cond = rb_wait_once;

rdata.irq_work = rbwork;

rdata.seq = atomic_read_acquire(&rbwork->seq);

data = &rdata;

}

ret = wait_event_interruptible((*waitq),

rb_wait_cond(rbwork, buffer, cpu, full, cond, data));

return ret;

}

/**