// SPDX-License-Identifier: GPL-2.0

/*

* Generic ring buffer

*

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

*/

#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/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 <asm/local.h>

static void update_pages_handler(struct work_struct *work);

/*

* 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 BUF_PAGE_HDR_SIZE offsetof(struct buffer_data_page, data)

#define RB_EVNT_HDR_SIZE (offsetof(struct ring_buffer_event, array))

#define RB_ALIGNMENT 4U

#define RB_MAX_SMALL_DATA (RB_ALIGNMENT * RINGBUF_TYPE_DATA_TYPE_LEN_MAX)

#define RB_EVNT_MIN_SIZE 8U /* two 32bit words */

#ifndef CONFIG_HAVE_64BIT_ALIGNED_ACCESS

# define RB_FORCE_8BYTE_ALIGNMENT 0

# define RB_ARCH_ALIGNMENT RB_ALIGNMENT

#else

# define RB_FORCE_8BYTE_ALIGNMENT 1

# define RB_ARCH_ALIGNMENT 8U

#endif

#define RB_ALIGN_DATA __aligned(RB_ARCH_ALIGNMENT)

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

#define TS_SHIFT 27

#define TS_MASK ((1ULL << TS_SHIFT) - 1)

#define TS_DELTA_TEST (~TS_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)

struct buffer_data_page {

u64 time_stamp; /* page time stamp */

local_t commit; /* write committed index */

unsigned char data[] RB_ALIGN_DATA; /* data of buffer 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 */

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

}

/*

* Also stolen from mm/slob.c. Thanks to Mathieu Desnoyers for pointing

* this issue out.

*/

static void free_buffer_page(struct buffer_page *bpage)

{

free_page((unsigned long)bpage->page);

kfree(bpage);

}

/*

* We need to fit the time_stamp delta into 27 bits.

*/

static inline int test_time_stamp(u64 delta)

{

if (delta & TS_DELTA_TEST)

return 1;

return 0;

}

#define BUF_PAGE_SIZE (PAGE_SIZE - BUF_PAGE_HDR_SIZE)

/* Max payload is BUF_PAGE_SIZE - header (8bytes) */

#define BUF_MAX_DATA_SIZE (BUF_PAGE_SIZE - (sizeof(u32) * 2))

int ring_buffer_print_page_header(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: int overwrite;\t"

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

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

1,

(unsigned int)is_signed_type(long));

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

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

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

(unsigned int)BUF_PAGE_SIZE,

(unsigned int)is_signed_type(char));

return !trace_seq_has_overflowed(s);

}

struct rb_irq_work {

struct irq_work work;

wait_queue_head_t waiters;

wait_queue_head_t full_waiters;

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

};

#if BITS_PER_LONG == 32

#define RB_TIME_32

#endif

/* To test on 64 bit machines */

//#define RB_TIME_32

#ifdef RB_TIME_32

struct rb_time_struct {

local_t cnt;

local_t top;

local_t bottom;

};

#else

#include <asm/local64.h>

struct rb_time_struct {

local64_t time;

};

#endif

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;

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

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

/*

* 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

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)

return rb_event_time_stamp(event);

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:

/* Can only fail on 32 bit */

if (!rb_time_read(&cpu_buffer->write_stamp, &ts))

/* Screw it, just read the current time */

ts = rb_time_stamp(cpu_buffer->buffer);

return ts;

}

/**

* 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 the percentage of pages are 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