/*

* Generic ring buffer

*

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

*/

#include <linux/ring_buffer.h>

#include <linux/trace_clock.h>

#include <linux/spinlock.h>

#include <linux/debugfs.h>

#include <linux/uaccess.h>

#include <linux/hardirq.h>

#include <linux/kmemcheck.h>

#include <linux/module.h>

#include <linux/percpu.h>

#include <linux/mutex.h>

#include <linux/slab.h>

#include <linux/init.h>

#include <linux/hash.h>

#include <linux/list.h>

#include <linux/cpu.h>

#include <linux/fs.h>

#include <asm/local.h>

#include "trace.h"

/*

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

*/

int ring_buffer_print_entry_header(struct trace_seq *s)

{

int ret;

ret = trace_seq_printf(s, "# compressed entry header\n");

ret = trace_seq_printf(s, "\ttype_len : 5 bits\n");

ret = trace_seq_printf(s, "\ttime_delta : 27 bits\n");

ret = trace_seq_printf(s, "\tarray : 32 bits\n");

ret = trace_seq_printf(s, "\n");

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

RINGBUF_TYPE_PADDING);

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

RINGBUF_TYPE_TIME_EXTEND);

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

RINGBUF_TYPE_DATA_TYPE_LEN_MAX);

return ret;

}

/*

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

*

*/

/*

* A fast way to enable or disable all ring buffers is to

* call tracing_on or tracing_off. Turning off the ring buffers

* prevents all ring buffers from being recorded to.

* Turning this switch on, makes it OK to write to the

* ring buffer, if the ring buffer is enabled itself.

*

* There's three layers that must be on in order to write

* to the ring buffer.

*

* 1) This global flag must be set.

* 2) The ring buffer must be enabled for recording.

* 3) The per cpu buffer must be enabled for recording.

*

* In case of an anomaly, this global flag has a bit set that

* will permantly disable all ring buffers.

*/

/*

* Global flag to disable all recording to ring buffers

* This has two bits: ON, DISABLED

*

* ON DISABLED

* ---- ----------

* 0 0 : ring buffers are off

* 1 0 : ring buffers are on

* X 1 : ring buffers are permanently disabled

*/

enum {

RB_BUFFERS_ON_BIT = 0,

RB_BUFFERS_DISABLED_BIT = 1,

};

enum {

RB_BUFFERS_ON = 1 << RB_BUFFERS_ON_BIT,

RB_BUFFERS_DISABLED = 1 << RB_BUFFERS_DISABLED_BIT,

};

static unsigned long ring_buffer_flags __read_mostly = RB_BUFFERS_ON;

#define BUF_PAGE_HDR_SIZE offsetof(struct buffer_data_page, data)

/**

* tracing_on - enable all tracing buffers

*

* This function enables all tracing buffers that may have been

* disabled with tracing_off.

*/

void tracing_on(void)

{

set_bit(RB_BUFFERS_ON_BIT, &ring_buffer_flags);

}

EXPORT_SYMBOL_GPL(tracing_on);

/**

* tracing_off - turn off all tracing buffers

*

* This function stops all tracing buffers from recording data.

* It does not disable any overhead the tracers themselves may

* be causing. This function simply causes all recording to

* the ring buffers to fail.

*/

void tracing_off(void)

{

clear_bit(RB_BUFFERS_ON_BIT, &ring_buffer_flags);

}

EXPORT_SYMBOL_GPL(tracing_off);

/**

* tracing_off_permanent - permanently disable ring buffers

*

* This function, once called, will disable all ring buffers

* permanently.

*/

void tracing_off_permanent(void)

{

set_bit(RB_BUFFERS_DISABLED_BIT, &ring_buffer_flags);

}

/**

* tracing_is_on - show state of ring buffers enabled

*/

int tracing_is_on(void)

{

return ring_buffer_flags == RB_BUFFERS_ON;

}

EXPORT_SYMBOL_GPL(tracing_is_on);

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

#if !defined(CONFIG_64BIT) || defined(CONFIG_HAVE_EFFICIENT_UNALIGNED_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 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 = 16,

};

#define skip_time_extend(event) \

((struct ring_buffer_event *)((char *)event + RB_LEN_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:

BUG();

}

/* 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 (event->type_len == RINGBUF_TYPE_TIME_EXTEND) {

/* 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 (event->type_len == RINGBUF_TYPE_TIME_EXTEND)

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

}

/**

* 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 TS_SHIFT 27

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

#define TS_DELTA_TEST (~TS_MASK)

/* 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[]; /* 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);

}

/**

* ring_buffer_page_len - the size of data on the page.

* @page: The page to read

*

* Returns the amount of data on the page, including buffer page header.

*/

size_t ring_buffer_page_len(void *page)

{

return local_read(&((struct buffer_data_page *)page)->commit)

+ BUF_PAGE_HDR_SIZE;

}

/*

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

int ret;

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

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

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

ret = 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 ret;

}

/*

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

*/

struct ring_buffer_per_cpu {

int cpu;

atomic_t record_disabled;

struct ring_buffer *buffer;

spinlock_t reader_lock; /* serialize readers */

arch_spinlock_t lock;

struct lock_class_key lock_key;

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;

local_t commit_overrun;

local_t overrun;

local_t entries;

local_t committing;

local_t commits;

unsigned long read;

u64 write_stamp;

u64 read_stamp;

};

struct ring_buffer {

unsigned pages;

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;

#ifdef CONFIG_HOTPLUG_CPU

struct notifier_block cpu_notify;

#endif

u64 (*clock)(void);

};

struct ring_buffer_iter {

struct ring_buffer_per_cpu *cpu_buffer;

unsigned long head;

struct buffer_page *head_page;

struct buffer_page *cache_reader_page;

unsigned long cache_read;

u64 read_stamp;

};

/* 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 ring_buffer *buffer)

{

/* shift to debug/test normalization and TIME_EXTENTS */

return buffer->clock() << DEBUG_SHIFT;

}

u64 ring_buffer_time_stamp(struct ring_buffer *buffer, int cpu)

{

u64 time;

preempt_disable_notrace();

time = rb_time_stamp(buffer);

preempt_enable_no_resched_notrace();

return time;

}

EXPORT_SYMBOL_GPL(ring_buffer_time_stamp);

void ring_buffer_normalize_time_stamp(struct ring_buffer *buffer,

int cpu, u64 *ts)

{

/* Just stupid testing the normalize function and deltas */

*ts >>= DEBUG_SHIFT;

}

EXPORT_SYMBOL_GPL(ring_buffer_normalize_time_stamp);

/*

* Making the ring buffer lockless makes things tricky.

* Although writes only happen on the CPU that they are on,

* and they only need to worry about interrupts. Reads can

* happen on any CPU.

*

* The reader page is always off the ring buffer, but when the

* reader finishes with a page, it needs to swap its page with

* a new one from the buffer. The reader needs to take from

* the head (writes go to the tail). But if a writer is in overwrite

* mode and wraps, it must push the head page forward.

*

* Here lies the problem.

*

* The reader must be careful to replace only the head page, and

* not another one. As described at the top of the file in the

* ASCII art, the reader sets its old page to point to the next

* page after head. It then sets the page after head to point to

* the old reader page. But if the writer moves the head page

* during this operation, the reader could end up with the tail.

*

* We use cmpxchg to help prevent this race. We also do something

* special with the page before head. We set the LSB to 1.

*

* When the writer must push the page forward, it will clear the

* bit that points to the head page, move the head, and then set

* the bit that points to the new head page.

*

* We also don't want an interrupt coming in and moving the head

* page on another writer. Thus we use the second LSB to catch

* that too. Thus:

*

* head->list->prev->next bit 1 bit 0

* ------- -------

* Normal page 0 0

* Points to head page 0 1

* New head page 1 0

*

* Note we can not trust the prev pointer of the head page, because:

*

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

* | |------>| T |---X--->| N |

* | |<------| | | |

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

* ^ ^ |

* | +-----+ | |

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

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

* +-----+

*

* Key: ---X--> HEAD flag set in pointer

* T Tail page

* R Reader page

* N Next page

*

* (see __rb_reserve_next() to see where this happens)

*

* What the above shows is that the reader just swapped out

* the reader page with a page in the buffer, but before it

* could make the new header point back to the new page added

* it was preempted by a writer. The writer moved forward onto

* the new page added by the reader and is about to move forward

* again.

*

* You can see, it is legitimate for the previous pointer of

* the head (or any page) not to point back to itself. But only

* temporarially.

*/

#define RB_PAGE_NORMAL 0UL

#define RB_PAGE_HEAD 1UL

#define RB_PAGE_UPDATE 2UL

#define RB_FLAG_MASK 3UL

/* PAGE_MOVED is not part of the mask */

#define RB_PAGE_MOVED 4UL

/*

* rb_list_head - remove any bit

*/

static struct list_head *rb_list_head(struct list_head *list)

{

unsigned long val = (unsigned long)list;

return (struct list_head *)(val & ~RB_FLAG_MASK);

}

/*

* rb_is_head_page - test if the given page is the head page

*

* Because the reader may move the head_page pointer, we can

* not trust what the head page is (it may be pointing to

* the reader page). But if the next page is a header page,

* its flags will be non zero.

*/

static inline int

rb_is_head_page(struct ring_buffer_per_cpu *cpu_buffer,

struct buffer_page *page, struct list_head *list)

{

unsigned long val;

val = (unsigned long)list->next;

if ((val & ~RB_FLAG_MASK) != (unsigned long)&page->list)

return RB_PAGE_MOVED;

return val & RB_FLAG_MASK;

}

/*

* rb_is_reader_page

*

* The unique thing about the reader page, is that, if the

* writer is ever on it, the previous pointer never points

* back to the reader page.

*/

static int rb_is_reader_page(struct buffer_page *page)

{

struct list_head *list = page->list.prev;

return rb_list_head(list->next) != &page->list;

}

/*

* rb_set_list_to_head - set a list_head to be pointing to head.

*/

static void rb_set_list_to_head(struct ring_buffer_per_cpu *cpu_buffer,

struct list_head *list)

{

unsigned long *ptr;

ptr = (unsigned long *)&list->next;

*ptr |= RB_PAGE_HEAD;

*ptr &= ~RB_PAGE_UPDATE;

}

/*

* rb_head_page_activate - sets up head page

*/

static void rb_head_page_activate(struct ring_buffer_per_cpu *cpu_buffer)

{

struct buffer_page *head;

head = cpu_buffer->head_page;

if (!head)

return;

/*

* Set the previous list pointer to have the HEAD flag.

*/

rb_set_list_to_head(cpu_buffer, head->list.prev);

}

static void rb_list_head_clear(struct list_head *list)

{

unsigned long *ptr = (unsigned long *)&list->next;

*ptr &= ~RB_FLAG_MASK;

}

/*

* rb_head_page_dactivate - clears head page ptr (for free list)

*/

static void

rb_head_page_deactivate(struct ring_buffer_per_cpu *cpu_buffer)

{

struct list_head *hd;

/* Go through the whole list and clear any pointers found. */

rb_list_head_clear(cpu_buffer->pages);

list_for_each(hd, cpu_buffer->pages)

rb_list_head_clear(hd);

}

static int rb_head_page_set(struct ring_buffer_per_cpu *cpu_buffer,

struct buffer_page *head,

struct buffer_page *prev,

int old_flag, int new_flag)

{

struct list_head *list;

unsigned long val = (unsigned long)&head->list;

unsigned long ret;

list = &prev->list;

val &= ~RB_FLAG_MASK;

ret = cmpxchg((unsigned long *)&list->next,

val | old_flag, val | new_flag);

/* check if the reader took the page */

if ((ret & ~RB_FLAG_MASK) != val)

return RB_PAGE_MOVED;

return ret & RB_FLAG_MASK;

}

static int rb_head_page_set_update(struct ring_buffer_per_cpu *cpu_buffer,

struct buffer_page *head,

struct buffer_page *prev,

int old_flag)

{

return rb_head_page_set(cpu_buffer, head, prev,

old_flag, RB_PAGE_UPDATE);

}

static int rb_head_page_set_head(struct ring_buffer_per_cpu *cpu_buffer,

struct buffer_page *head,

struct buffer_page *prev,

int old_flag)

{

return rb_head_page_set(cpu_buffer, head, prev,

old_flag, RB_PAGE_HEAD);

}

static int rb_head_page_set_normal(struct ring_buffer_per_cpu *cpu_buffer,

struct buffer_page *head,

struct buffer_page *prev,

int old_flag)

{

return rb_head_page_set(cpu_buffer, head, prev,

old_flag, RB_PAGE_NORMAL);

}

static inline void rb_inc_page(struct ring_buffer_per_cpu *cpu_buffer,

struct buffer_page **bpage)

{

struct list_head *p = rb_list_head((*bpage)->list.next);

*bpage = list_entry(p, struct buffer_page, list);

}

static struct buffer_page *

rb_set_head_page(struct ring_buffer_per_cpu *cpu_buffer)

{

struct buffer_page *head;

struct buffer_page *page;

struct list_head *list;

int i;

if (RB_WARN_ON(cpu_buffer, !cpu_buffer->head_page))

return NULL;

/* sanity check */

list = cpu_buffer->pages;

if (RB_WARN_ON(cpu_buffer, rb_list_head(list->prev->next) != list))

return NULL;

page = head = cpu_buffer->head_page;

/*

* It is possible that the writer moves the header behind

* where we started, and we miss in one loop.

* A second loop should grab the header, but we'll do

* three loops just because I'm paranoid.

*/

for (i = 0; i < 3; i++) {

do {

if (rb_is_head_page(cpu_buffer, page, page->list.prev)) {

cpu_buffer->head_page = page;

return page;

}

rb_inc_page(cpu_buffer, &page);

} while (page != head);

}

RB_WARN_ON(cpu_buffer, 1);

return NULL;

}

static int rb_head_page_replace(struct buffer_page *old,

struct buffer_page *new)

{

unsigned long *ptr = (unsigned long *)&old->list.prev->next;

unsigned long val;

unsigned long ret;

val = *ptr & ~RB_FLAG_MASK;

val |= RB_PAGE_HEAD;

ret = cmpxchg(ptr, val, (unsigned long)&new->list);

return ret == val;

}

/*

* rb_tail_page_update - move the tail page forward

*

* Returns 1 if moved tail page, 0 if someone else did.

*/

static int rb_tail_page_update(struct ring_buffer_per_cpu *cpu_buffer,

struct buffer_page *tail_page,

struct buffer_page *next_page)

{

struct buffer_page *old_tail;

unsigned long old_entries;

unsigned long old_write;

int ret = 0;

/*

* The tail page now needs to be moved forward.

*

* We need to reset the tail page, but without messing

* with possible erasing of data brought in by interrupts

* that have moved the tail page and are currently on it.

*

* We add a counter to the write field to denote this.

*/

old_write = local_add_return(RB_WRITE_INTCNT, &next_page->write);

old_entries = local_add_return(RB_WRITE_INTCNT, &next_page->entries);

/*

* Just make sure we have seen our old_write and synchronize

* with any interrupts that come in.

*/

barrier();

/*

* If the tail page is still the same as what we think

* it is, then it is up to us to update the tail

* pointer.

*/

if (tail_page == cpu_buffer->tail_page) {

/* Zero the write counter */

unsigned long val = old_write & ~RB_WRITE_MASK;

unsigned long eval = old_entries & ~RB_WRITE_MASK;

/*

* This will only succeed if an interrupt did

* not come in and change it. In which case, we

* do not want to modify it.

*

* We add (void) to let the compiler know that we do not care

* about the return value of these functions. We use the

* cmpxchg to only update if an interrupt did not already

* do it for us. If the cmpxchg fails, we don't care.

*/

(void)local_cmpxchg(&next_page->write, old_write, val);

(void)local_cmpxchg(&next_page->entries, old_entries, eval);

/*

* No need to worry about races with clearing out the commit.

* it only can increment when a commit takes place. But that

* only happens in the outer most nested commit.

*/

local_set(&next_page->page->commit, 0);

old_tail = cmpxchg(&cpu_buffer->tail_page,

tail_page, next_page);

if (old_tail == tail_page)

ret = 1;

}

return ret;

}

static int rb_check_bpage(struct ring_buffer_per_cpu *cpu_buffer,

struct buffer_page *bpage)

{

unsigned long val = (unsigned long)bpage;

if (RB_WARN_ON(cpu_buffer, val & RB_FLAG_MASK))

return 1;

return 0;

}

/**

* rb_check_list - make sure a pointer to a list has the last bits zero

*/

static int rb_check_list(struct ring_buffer_per_cpu *cpu_buffer,

struct list_head *list)

{

if (RB_WARN_ON(cpu_buffer, rb_list_head(list->prev) != list->prev))

return 1;

if (RB_WARN_ON(cpu_buffer, rb_list_head(list->next) != list->next))

return 1;

return 0;

}

/**

* check_pages - integrity check of buffer pages

* @cpu_buffer: CPU buffer with pages to test

*

* As a safety measure we check to make sure the data pages have not

* been corrupted.

*/

static int rb_check_pages(struct ring_buffer_per_cpu *cpu_buffer)

{

struct list_head *head = cpu_buffer->pages;

struct buffer_page *bpage, *tmp;

rb_head_page_deactivate(cpu_buffer);

if (RB_WARN_ON(cpu_buffer, head->next->prev != head))

return -1;

if (RB_WARN_ON(cpu_buffer, head->prev->next != head))

return -1;

if (rb_check_list(cpu_buffer, head))

return -1;

list_for_each_entry_safe(bpage, tmp, head, list) {

if (RB_WARN_ON(cpu_buffer,

bpage->list.next->prev != &bpage->list))

return -1;

if (RB_WARN_ON(cpu_buffer,

bpage->list.prev->next != &bpage->list))

return -1;

if (rb_check_list(cpu_buffer, &bpage->list))

return -1;

}

rb_head_page_activate(cpu_buffer);

return 0;

}

static int rb_allocate_pages(struct ring_buffer_per_cpu *cpu_buffer,

unsigned nr_pages)

{

struct buffer_page *bpage, *tmp;

LIST_HEAD(pages);