/*
   * 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
  ");
  	ret = trace_seq_printf(s, "\ttype_len    :    5 bits
  ");

  	ret = trace_seq_printf(s, "\ttime_delta  :   27 bits
  ");
  	ret = trace_seq_printf(s, "\tarray       :   32 bits
  ");
  	ret = trace_seq_printf(s, "
  ");
  	ret = trace_seq_printf(s, "\tpadding     : type == %d
  ",
  			       RINGBUF_TYPE_PADDING);
  	ret = trace_seq_printf(s, "\ttime_extend : type == %d
  ",
  			       RINGBUF_TYPE_TIME_EXTEND);

  	ret = trace_seq_printf(s, "\tdata max type_len  == %d
  ",
  			       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;
  ",
  			       (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;
  ",

  			       (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;
  ",
  			       (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;
  ",

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

  	raw_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				entries_bytes;

  	local_t				commit_overrun;
  	local_t				overrun;

  	local_t				entries;

  	local_t				committing;
  	local_t				commits;

  	unsigned long			read;

  	unsigned long			read_bytes;

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

  	WARN_ON(!nr_pages);

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

  		struct page *page;

  		/*
  		 * __GFP_NORETRY flag makes sure that the allocation fails
  		 * gracefully without invoking oom-killer and the system is
  		 * not destabilized.
  		 */

  		bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()),

  				    GFP_KERNEL | __GFP_NORETRY,
  				    cpu_to_node(cpu_buffer->cpu));

  		if (!bpage)

  			goto free_pages;

  
  		rb_check_bpage(cpu_buffer, bpage);

  		list_add(&bpage->list, &pages);

  		page = alloc_pages_node(cpu_to_node(cpu_buffer->cpu),

  					GFP_KERNEL | __GFP_NORETRY, 0);

  		if (!page)

  			goto free_pages;

  		bpage->page = page_address(page);

  		rb_init_page(bpage->page);

  	}

  	/*
  	 * The ring buffer page list is a circular list that does not
  	 * start and end with a list head. All page list items point to
  	 * other pages.
  	 */
  	cpu_buffer->pages = pages.next;
  	list_del(&pages);

  
  	rb_check_pages(cpu_buffer);
  
  	return 0;
  
   free_pages:

  	list_for_each_entry_safe(bpage, tmp, &pages, list) {
  		list_del_init(&bpage->list);
  		free_buffer_page(bpage);

  	}
  	return -ENOMEM;
  }
  
  static struct ring_buffer_per_cpu *
  rb_allocate_cpu_buffer(struct ring_buffer *buffer, int cpu)
  {
  	struct ring_buffer_per_cpu *cpu_buffer;

  	struct buffer_page *bpage;

  	struct page *page;

  	int ret;
  
  	cpu_buffer = kzalloc_node(ALIGN(sizeof(*cpu_buffer), cache_line_size()),
  				  GFP_KERNEL, cpu_to_node(cpu));
  	if (!cpu_buffer)
  		return NULL;
  
  	cpu_buffer->cpu = cpu;
  	cpu_buffer->buffer = buffer;

  	raw_spin_lock_init(&cpu_buffer->reader_lock);

  	lockdep_set_class(&cpu_buffer->reader_lock, buffer->reader_lock_key);

  	cpu_buffer->lock = (arch_spinlock_t)__ARCH_SPIN_LOCK_UNLOCKED;

  	bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()),

  			    GFP_KERNEL, cpu_to_node(cpu));

  	if (!bpage)

  		goto fail_free_buffer;

  	rb_check_bpage(cpu_buffer, bpage);

  	cpu_buffer->reader_page = bpage;

  	page = alloc_pages_node(cpu_to_node(cpu), GFP_KERNEL, 0);
  	if (!page)

  		goto fail_free_reader;

  	bpage->page = page_address(page);

  	rb_init_page(bpage->page);

  	INIT_LIST_HEAD(&cpu_buffer->reader_page->list);

  	ret = rb_allocate_pages(cpu_buffer, buffer->pages);
  	if (ret < 0)

  		goto fail_free_reader;

  
  	cpu_buffer->head_page

  		= list_entry(cpu_buffer->pages, struct buffer_page, list);

  	cpu_buffer->tail_page = cpu_buffer->commit_page = cpu_buffer->head_page;

  	rb_head_page_activate(cpu_buffer);

  	return cpu_buffer;

   fail_free_reader:
  	free_buffer_page(cpu_buffer->reader_page);

   fail_free_buffer:
  	kfree(cpu_buffer);
  	return NULL;
  }
  
  static void rb_free_cpu_buffer(struct ring_buffer_per_cpu *cpu_buffer)
  {

  	struct list_head *head = cpu_buffer->pages;

  	struct buffer_page *bpage, *tmp;

  	free_buffer_page(cpu_buffer->reader_page);

  	rb_head_page_deactivate(cpu_buffer);

  	if (head) {
  		list_for_each_entry_safe(bpage, tmp, head, list) {
  			list_del_init(&bpage->list);
  			free_buffer_page(bpage);
  		}
  		bpage = list_entry(head, struct buffer_page, list);

  		free_buffer_page(bpage);

  	}

  	kfree(cpu_buffer);
  }

  #ifdef CONFIG_HOTPLUG_CPU

  static int rb_cpu_notify(struct notifier_block *self,
  			 unsigned long action, void *hcpu);

  #endif

  /**
   * ring_buffer_alloc - allocate a new ring_buffer

   * @size: the size in bytes per cpu that is needed.

   * @flags: attributes to set for the ring buffer.
   *
   * Currently the only flag that is available is the RB_FL_OVERWRITE
   * flag. This flag means that the buffer will overwrite old data
   * when the buffer wraps. If this flag is not set, the buffer will
   * drop data when the tail hits the head.
   */

  struct ring_buffer *__ring_buffer_alloc(unsigned long size, unsigned flags,
  					struct lock_class_key *key)

  {
  	struct ring_buffer *buffer;
  	int bsize;
  	int cpu;
  
  	/* keep it in its own cache line */
  	buffer = kzalloc(ALIGN(sizeof(*buffer), cache_line_size()),
  			 GFP_KERNEL);
  	if (!buffer)
  		return NULL;

  	if (!alloc_cpumask_var(&buffer->cpumask, GFP_KERNEL))
  		goto fail_free_buffer;

  	buffer->pages = DIV_ROUND_UP(size, BUF_PAGE_SIZE);
  	buffer->flags = flags;

  	buffer->clock = trace_clock_local;

  	buffer->reader_lock_key = key;

  
  	/* need at least two pages */

  	if (buffer->pages < 2)
  		buffer->pages = 2;

  	/*
  	 * In case of non-hotplug cpu, if the ring-buffer is allocated
  	 * in early initcall, it will not be notified of secondary cpus.
  	 * In that off case, we need to allocate for all possible cpus.
  	 */
  #ifdef CONFIG_HOTPLUG_CPU

  	get_online_cpus();
  	cpumask_copy(buffer->cpumask, cpu_online_mask);

  #else
  	cpumask_copy(buffer->cpumask, cpu_possible_mask);
  #endif

  	buffer->cpus = nr_cpu_ids;
  
  	bsize = sizeof(void *) * nr_cpu_ids;
  	buffer->buffers = kzalloc(ALIGN(bsize, cache_line_size()),
  				  GFP_KERNEL);
  	if (!buffer->buffers)

  		goto fail_free_cpumask;

  
  	for_each_buffer_cpu(buffer, cpu) {
  		buffer->buffers[cpu] =
  			rb_allocate_cpu_buffer(buffer, cpu);
  		if (!buffer->buffers[cpu])
  			goto fail_free_buffers;
  	}

  #ifdef CONFIG_HOTPLUG_CPU

  	buffer->cpu_notify.notifier_call = rb_cpu_notify;
  	buffer->cpu_notify.priority = 0;
  	register_cpu_notifier(&buffer->cpu_notify);
  #endif
  
  	put_online_cpus();

  	mutex_init(&buffer->mutex);
  
  	return buffer;
  
   fail_free_buffers:
  	for_each_buffer_cpu(buffer, cpu) {
  		if (buffer->buffers[cpu])
  			rb_free_cpu_buffer(buffer->buffers[cpu]);
  	}
  	kfree(buffer->buffers);

   fail_free_cpumask:
  	free_cpumask_var(buffer->cpumask);

  	put_online_cpus();

   fail_free_buffer:
  	kfree(buffer);
  	return NULL;
  }

  EXPORT_SYMBOL_GPL(__ring_buffer_alloc);

  
  /**
   * ring_buffer_free - free a ring buffer.
   * @buffer: the buffer to free.
   */
  void
  ring_buffer_free(struct ring_buffer *buffer)
  {
  	int cpu;

  	get_online_cpus();

  #ifdef CONFIG_HOTPLUG_CPU

  	unregister_cpu_notifier(&buffer->cpu_notify);
  #endif

  	for_each_buffer_cpu(buffer, cpu)
  		rb_free_cpu_buffer(buffer->buffers[cpu]);

  	put_online_cpus();

  	kfree(buffer->buffers);

  	free_cpumask_var(buffer->cpumask);

  	kfree(buffer);
  }

  EXPORT_SYMBOL_GPL(ring_buffer_free);

  void ring_buffer_set_clock(struct ring_buffer *buffer,
  			   u64 (*clock)(void))
  {
  	buffer->clock = clock;
  }

  static void rb_reset_cpu(struct ring_buffer_per_cpu *cpu_buffer);
  
  static void
  rb_remove_pages(struct ring_buffer_per_cpu *cpu_buffer, unsigned nr_pages)
  {

  	struct buffer_page *bpage;

  	struct list_head *p;
  	unsigned i;

  	raw_spin_lock_irq(&cpu_buffer->reader_lock);

  	rb_head_page_deactivate(cpu_buffer);

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

  		if (RB_WARN_ON(cpu_buffer, list_empty(cpu_buffer->pages)))

  			goto out;

  		p = cpu_buffer->pages->next;

  		bpage = list_entry(p, struct buffer_page, list);
  		list_del_init(&bpage->list);
  		free_buffer_page(bpage);

  	}

  	if (RB_WARN_ON(cpu_buffer, list_empty(cpu_buffer->pages)))

  		goto out;

  
  	rb_reset_cpu(cpu_buffer);

  	rb_check_pages(cpu_buffer);

  out:

  	raw_spin_unlock_irq(&cpu_buffer->reader_lock);

  }
  
  static void
  rb_insert_pages(struct ring_buffer_per_cpu *cpu_buffer,
  		struct list_head *pages, unsigned nr_pages)
  {

  	struct buffer_page *bpage;

  	struct list_head *p;
  	unsigned i;

  	raw_spin_lock_irq(&cpu_buffer->reader_lock);

  	rb_head_page_deactivate(cpu_buffer);

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

  		if (RB_WARN_ON(cpu_buffer, list_empty(pages)))

  			goto out;

  		p = pages->next;

  		bpage = list_entry(p, struct buffer_page, list);
  		list_del_init(&bpage->list);

  		list_add_tail(&bpage->list, cpu_buffer->pages);

  	}
  	rb_reset_cpu(cpu_buffer);

  	rb_check_pages(cpu_buffer);

  out:

  	raw_spin_unlock_irq(&cpu_buffer->reader_lock);

  }
  
  /**
   * ring_buffer_resize - resize the ring buffer
   * @buffer: the buffer to resize.
   * @size: the new size.
   *

   * Minimum size is 2 * BUF_PAGE_SIZE.
   *
   * Returns -1 on failure.
   */
  int ring_buffer_resize(struct ring_buffer *buffer, unsigned long size)
  {
  	struct ring_buffer_per_cpu *cpu_buffer;
  	unsigned nr_pages, rm_pages, new_pages;

  	struct buffer_page *bpage, *tmp;

  	unsigned long buffer_size;

  	LIST_HEAD(pages);
  	int i, cpu;

  	/*
  	 * Always succeed at resizing a non-existent buffer:
  	 */
  	if (!buffer)
  		return size;

  	size = DIV_ROUND_UP(size, BUF_PAGE_SIZE);
  	size *= BUF_PAGE_SIZE;
  	buffer_size = buffer->pages * BUF_PAGE_SIZE;
  
  	/* we need a minimum of two pages */
  	if (size < BUF_PAGE_SIZE * 2)
  		size = BUF_PAGE_SIZE * 2;
  
  	if (size == buffer_size)
  		return size;

  	atomic_inc(&buffer->record_disabled);
  
  	/* Make sure all writers are done with this buffer. */
  	synchronize_sched();

  	mutex_lock(&buffer->mutex);

  	get_online_cpus();

  
  	nr_pages = DIV_ROUND_UP(size, BUF_PAGE_SIZE);
  
  	if (size < buffer_size) {
  
  		/* easy case, just free pages */

  		if (RB_WARN_ON(buffer, nr_pages >= buffer->pages))
  			goto out_fail;

  
  		rm_pages = buffer->pages - nr_pages;
  
  		for_each_buffer_cpu(buffer, cpu) {
  			cpu_buffer = buffer->buffers[cpu];
  			rb_remove_pages(cpu_buffer, rm_pages);
  		}
  		goto out;
  	}
  
  	/*
  	 * This is a bit more difficult. We only want to add pages
  	 * when we can allocate enough for all CPUs. We do this
  	 * by allocating all the pages and storing them on a local
  	 * link list. If we succeed in our allocation, then we
  	 * add these pages to the cpu_buffers. Otherwise we just free
  	 * them all and return -ENOMEM;
  	 */

  	if (RB_WARN_ON(buffer, nr_pages <= buffer->pages))
  		goto out_fail;

  	new_pages = nr_pages - buffer->pages;
  
  	for_each_buffer_cpu(buffer, cpu) {
  		for (i = 0; i < new_pages; i++) {

  			struct page *page;

  			/*
  			 * __GFP_NORETRY flag makes sure that the allocation
  			 * fails gracefully without invoking oom-killer and
  			 * the system is not destabilized.
  			 */

  			bpage = kzalloc_node(ALIGN(sizeof(*bpage),

  						  cache_line_size()),

  					    GFP_KERNEL | __GFP_NORETRY,
  					    cpu_to_node(cpu));

  			if (!bpage)

  				goto free_pages;

  			list_add(&bpage->list, &pages);

  			page = alloc_pages_node(cpu_to_node(cpu),
  						GFP_KERNEL | __GFP_NORETRY, 0);

  			if (!page)

  				goto free_pages;

  			bpage->page = page_address(page);

  			rb_init_page(bpage->page);

  		}
  	}
  
  	for_each_buffer_cpu(buffer, cpu) {
  		cpu_buffer = buffer->buffers[cpu];
  		rb_insert_pages(cpu_buffer, &pages, new_pages);
  	}

  	if (RB_WARN_ON(buffer, !list_empty(&pages)))
  		goto out_fail;

  
   out:
  	buffer->pages = nr_pages;

  	put_online_cpus();

  	mutex_unlock(&buffer->mutex);

  	atomic_dec(&buffer->record_disabled);

  	return size;
  
   free_pages:

  	list_for_each_entry_safe(bpage, tmp, &pages, list) {
  		list_del_init(&bpage->list);
  		free_buffer_page(bpage);

  	}

  	put_online_cpus();

  	mutex_unlock(&buffer->mutex);

  	atomic_dec(&buffer->record_disabled);

  	return -ENOMEM;

  
  	/*
  	 * Something went totally wrong, and we are too paranoid
  	 * to even clean up the mess.
  	 */
   out_fail:
  	put_online_cpus();
  	mutex_unlock(&buffer->mutex);

  	atomic_dec(&buffer->record_disabled);

  	return -1;

  }

  EXPORT_SYMBOL_GPL(ring_buffer_resize);

  void ring_buffer_change_overwrite(struct ring_buffer *buffer, int val)
  {
  	mutex_lock(&buffer->mutex);
  	if (val)
  		buffer->flags |= RB_FL_OVERWRITE;
  	else
  		buffer->flags &= ~RB_FL_OVERWRITE;
  	mutex_unlock(&buffer->mutex);
  }
  EXPORT_SYMBOL_GPL(ring_buffer_change_overwrite);

  static inline void *

  __rb_data_page_index(struct buffer_data_page *bpage, unsigned index)

  {

  	return bpage->data + index;

  }

  static inline void *__rb_page_index(struct buffer_page *bpage, unsigned index)

  {

  	return bpage->page->data + index;

  }
  
  static inline struct ring_buffer_event *

  rb_reader_event(struct ring_buffer_per_cpu *cpu_buffer)

  {

  	return __rb_page_index(cpu_buffer->reader_page,
  			       cpu_buffer->reader_page->read);
  }
  
  static inline struct ring_buffer_event *

  rb_iter_head_event(struct ring_buffer_iter *iter)
  {

  	return __rb_page_index(iter->head_page, iter->head);

  }

  static inline unsigned long rb_page_write(struct buffer_page *bpage)

  {

  	return local_read(&bpage->write) & RB_WRITE_MASK;

  }
  
  static inline unsigned rb_page_commit(struct buffer_page *bpage)
  {

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

  }

  static inline unsigned long rb_page_entries(struct buffer_page *bpage)
  {
  	return local_read(&bpage->entries) & RB_WRITE_MASK;
  }

  /* Size is determined by what has been committed */

  static inline unsigned rb_page_size(struct buffer_page *bpage)
  {
  	return rb_page_commit(bpage);
  }
  
  static inline unsigned
  rb_commit_index(struct ring_buffer_per_cpu *cpu_buffer)
  {
  	return rb_page_commit(cpu_buffer->commit_page);
  }

  static inline unsigned
  rb_event_index(struct ring_buffer_event *event)
  {
  	unsigned long addr = (unsigned long)event;

  	return (addr & ~PAGE_MASK) - BUF_PAGE_HDR_SIZE;

  }

  static inline int

  rb_event_is_commit(struct ring_buffer_per_cpu *cpu_buffer,
  		   struct ring_buffer_event *event)

  {
  	unsigned long addr = (unsigned long)event;
  	unsigned long index;
  
  	index = rb_event_index(event);
  	addr &= PAGE_MASK;
  
  	return cpu_buffer->commit_page->page == (void *)addr &&
  		rb_commit_index(cpu_buffer) == index;
  }

  static void

  rb_set_commit_to_write(struct ring_buffer_per_cpu *cpu_buffer)

  {

  	unsigned long max_count;

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

   again:

  	max_count = cpu_buffer->buffer->pages * 100;

  	while (cpu_buffer->commit_page != cpu_buffer->tail_page) {

  		if (RB_WARN_ON(cpu_buffer, !(--max_count)))
  			return;
  		if (RB_WARN_ON(cpu_buffer,
  			       rb_is_reader_page(cpu_buffer->tail_page)))
  			return;
  		local_set(&cpu_buffer->commit_page->page->commit,
  			  rb_page_write(cpu_buffer->commit_page));

  		rb_inc_page(cpu_buffer, &cpu_buffer->commit_page);

  		cpu_buffer->write_stamp =
  			cpu_buffer->commit_page->page->time_stamp;

  		/* add barrier to keep gcc from optimizing too much */
  		barrier();
  	}
  	while (rb_commit_index(cpu_buffer) !=
  	       rb_page_write(cpu_buffer->commit_page)) {

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

  		barrier();
  	}

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

  }

  static void rb_reset_reader_page(struct ring_buffer_per_cpu *cpu_buffer)

  {

  	cpu_buffer->read_stamp = cpu_buffer->reader_page->page->time_stamp;

  	cpu_buffer->reader_page->read = 0;

  }

  static void rb_inc_iter(struct ring_buffer_iter *iter)

  {
  	struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer;
  
  	/*
  	 * The iterator could be on the reader page (it starts there).
  	 * But the head could have moved, since the reader was
  	 * found. Check for this case and assign the iterator
  	 * to the head page instead of next.
  	 */
  	if (iter->head_page == cpu_buffer->reader_page)

  		iter->head_page = rb_set_head_page(cpu_buffer);

  	else
  		rb_inc_page(cpu_buffer, &iter->head_page);

  	iter->read_stamp = iter->head_page->page->time_stamp;

  	iter->head = 0;
  }

  /* Slow path, do not inline */
  static noinline struct ring_buffer_event *
  rb_add_time_stamp(struct ring_buffer_event *event, u64 delta)
  {
  	event->type_len = RINGBUF_TYPE_TIME_EXTEND;
  
  	/* Not the first event on the page? */
  	if (rb_event_index(event)) {
  		event->time_delta = delta & TS_MASK;
  		event->array[0] = delta >> TS_SHIFT;
  	} else {
  		/* nope, just zero it */
  		event->time_delta = 0;
  		event->array[0] = 0;
  	}
  
  	return skip_time_extend(event);
  }

  /**
   * ring_buffer_update_event - update event type and data
   * @event: the even to update
   * @type: the type of event
   * @length: the size of the event field in the ring buffer
   *
   * Update the type and data fields of the event. The length
   * is the actual size that is written to the ring buffer,
   * and with this, we can determine what to place into the
   * data field.
   */

  static void

  rb_update_event(struct ring_buffer_per_cpu *cpu_buffer,
  		struct ring_buffer_event *event, unsigned length,
  		int add_timestamp, u64 delta)

  {

  	/* Only a commit updates the timestamp */
  	if (unlikely(!rb_event_is_commit(cpu_buffer, event)))
  		delta = 0;

  	/*
  	 * If we need to add a timestamp, then we
  	 * add it to the start of the resevered space.
  	 */
  	if (unlikely(add_timestamp)) {
  		event = rb_add_time_stamp(event, delta);
  		length -= RB_LEN_TIME_EXTEND;
  		delta = 0;

  	}

  
  	event->time_delta = delta;
  	length -= RB_EVNT_HDR_SIZE;
  	if (length > RB_MAX_SMALL_DATA || RB_FORCE_8BYTE_ALIGNMENT) {
  		event->type_len = 0;
  		event->array[0] = length;
  	} else
  		event->type_len = DIV_ROUND_UP(length, RB_ALIGNMENT);

  }

  /*
   * rb_handle_head_page - writer hit the head page
   *
   * Returns: +1 to retry page
   *           0 to continue
   *          -1 on error
   */
  static int
  rb_handle_head_page(struct ring_buffer_per_cpu *cpu_buffer,
  		    struct buffer_page *tail_page,
  		    struct buffer_page *next_page)
  {
  	struct buffer_page *new_head;
  	int entries;
  	int type;
  	int ret;
  
  	entries = rb_page_entries(next_page);
  
  	/*
  	 * The hard part is here. We need to move the head
  	 * forward, and protect against both readers on
  	 * other CPUs and writers coming in via interrupts.
  	 */
  	type = rb_head_page_set_update(cpu_buffer, next_page, tail_page,
  				       RB_PAGE_HEAD);
  
  	/*
  	 * type can be one of four:
  	 *  NORMAL - an interrupt already moved it for us
  	 *  HEAD   - we are the first to get here.
  	 *  UPDATE - we are the interrupt interrupting
  	 *           a current move.
  	 *  MOVED  - a reader on another CPU moved the next
  	 *           pointer to its reader page. Give up
  	 *           and try again.
  	 */
  
  	switch (type) {
  	case RB_PAGE_HEAD:
  		/*
  		 * We changed the head to UPDATE, thus
  		 * it is our responsibility to update
  		 * the counters.
  		 */
  		local_add(entries, &cpu_buffer->overrun);

  		local_sub(BUF_PAGE_SIZE, &cpu_buffer->entries_bytes);

  
  		/*
  		 * The entries will be zeroed out when we move the
  		 * tail page.
  		 */
  
  		/* still more to do */
  		break;
  
  	case RB_PAGE_UPDATE:
  		/*
  		 * This is an interrupt that interrupt the
  		 * previous update. Still more to do.
  		 */
  		break;
  	case RB_PAGE_NORMAL:
  		/*
  		 * An interrupt came in before the update
  		 * and processed this for us.
  		 * Nothing left to do.
  		 */
  		return 1;
  	case RB_PAGE_MOVED:
  		/*
  		 * The reader is on another CPU and just did
  		 * a swap with our next_page.
  		 * Try again.
  		 */
  		return 1;
  	default:
  		RB_WARN_ON(cpu_buffer, 1); /* WTF??? */
  		return -1;
  	}
  
  	/*
  	 * Now that we are here, the old head pointer is
  	 * set to UPDATE. This will keep the reader from
  	 * swapping the head page with the reader page.
  	 * The reader (on another CPU) will spin till
  	 * we are finished.
  	 *
  	 * We just need to protect against interrupts
  	 * doing the job. We will set the next pointer
  	 * to HEAD. After that, we set the old pointer
  	 * to NORMAL, but only if it was HEAD before.
  	 * otherwise we are an interrupt, and only
  	 * want the outer most commit to reset it.
  	 */
  	new_head = next_page;
  	rb_inc_page(cpu_buffer, &new_head);
  
  	ret = rb_head_page_set_head(cpu_buffer, new_head, next_page,
  				    RB_PAGE_NORMAL);
  
  	/*
  	 * Valid returns are:
  	 *  HEAD   - an interrupt came in and already set it.
  	 *  NORMAL - One of two things:
  	 *            1) We really set it.
  	 *            2) A bunch of interrupts came in and moved
  	 *               the page forward again.
  	 */
  	switch (ret) {
  	case RB_PAGE_HEAD:
  	case RB_PAGE_NORMAL:
  		/* OK */
  		break;
  	default:
  		RB_WARN_ON(cpu_buffer, 1);
  		return -1;
  	}
  
  	/*
  	 * It is possible that an interrupt came in,
  	 * set the head up, then more interrupts came in
  	 * and moved it again. When we get back here,
  	 * the page would have been set to NORMAL but we
  	 * just set it back to HEAD.
  	 *
  	 * How do you detect this? Well, if that happened
  	 * the tail page would have moved.
  	 */
  	if (ret == RB_PAGE_NORMAL) {
  		/*
  		 * If the tail had moved passed next, then we need
  		 * to reset the pointer.
  		 */
  		if (cpu_buffer->tail_page != tail_page &&
  		    cpu_buffer->tail_page != next_page)
  			rb_head_page_set_normal(cpu_buffer, new_head,
  						next_page,
  						RB_PAGE_HEAD);
  	}
  
  	/*
  	 * If this was the outer most commit (the one that
  	 * changed the original pointer from HEAD to UPDATE),
  	 * then it is up to us to reset it to NORMAL.
  	 */
  	if (type == RB_PAGE_HEAD) {
  		ret = rb_head_page_set_normal(cpu_buffer, next_page,
  					      tail_page,
  					      RB_PAGE_UPDATE);
  		if (RB_WARN_ON(cpu_buffer,
  			       ret != RB_PAGE_UPDATE))
  			return -1;
  	}
  
  	return 0;
  }

  static unsigned rb_calculate_event_length(unsigned length)

  {
  	struct ring_buffer_event event; /* Used only for sizeof array */
  
  	/* zero length can cause confusions */
  	if (!length)
  		length = 1;

  	if (length > RB_MAX_SMALL_DATA || RB_FORCE_8BYTE_ALIGNMENT)

  		length += sizeof(event.array[0]);
  
  	length += RB_EVNT_HDR_SIZE;

  	length = ALIGN(length, RB_ARCH_ALIGNMENT);

  
  	return length;
  }

  static inline void
  rb_reset_tail(struct ring_buffer_per_cpu *cpu_buffer,
  	      struct buffer_page *tail_page,
  	      unsigned long tail, unsigned long length)
  {
  	struct ring_buffer_event *event;
  
  	/*
  	 * Only the event that crossed the page boundary
  	 * must fill the old tail_page with padding.
  	 */
  	if (tail >= BUF_PAGE_SIZE) {

  		/*
  		 * If the page was filled, then we still need
  		 * to update the real_end. Reset it to zero
  		 * and the reader will ignore it.
  		 */
  		if (tail == BUF_PAGE_SIZE)
  			tail_page->real_end = 0;

  		local_sub(length, &tail_page->write);
  		return;
  	}
  
  	event = __rb_page_index(tail_page, tail);

  	kmemcheck_annotate_bitfield(event, bitfield);

  	/* account for padding bytes */
  	local_add(BUF_PAGE_SIZE - tail, &cpu_buffer->entries_bytes);

  	/*

  	 * Save the original length to the meta data.
  	 * This will be used by the reader to add lost event
  	 * counter.
  	 */
  	tail_page->real_end = tail;
  
  	/*

  	 * If this event is bigger than the minimum size, then
  	 * we need to be careful that we don't subtract the
  	 * write counter enough to allow another writer to slip
  	 * in on this page.
  	 * We put in a discarded commit instead, to make sure
  	 * that this space is not used again.
  	 *
  	 * If we are less than the minimum size, we don't need to
  	 * worry about it.
  	 */
  	if (tail > (BUF_PAGE_SIZE - RB_EVNT_MIN_SIZE)) {
  		/* No room for any events */
  
  		/* Mark the rest of the page with padding */
  		rb_event_set_padding(event);
  
  		/* Set the write back to the previous setting */
  		local_sub(length, &tail_page->write);
  		return;
  	}
  
  	/* Put in a discarded event */
  	event->array[0] = (BUF_PAGE_SIZE - tail) - RB_EVNT_HDR_SIZE;
  	event->type_len = RINGBUF_TYPE_PADDING;
  	/* time delta must be non zero */
  	event->time_delta = 1;

  
  	/* Set write to end of buffer */
  	length = (tail + length) - BUF_PAGE_SIZE;
  	local_sub(length, &tail_page->write);
  }

  /*
   * This is the slow path, force gcc not to inline it.
   */
  static noinline struct ring_buffer_event *

  rb_move_tail(struct ring_buffer_per_cpu *cpu_buffer,
  	     unsigned long length, unsigned long tail,

  	     struct buffer_page *tail_page, u64 ts)

  {

  	struct buffer_page *commit_page = cpu_buffer->commit_page;

  	struct ring_buffer *buffer = cpu_buffer->buffer;

  	struct buffer_page *next_page;
  	int ret;

  
  	next_page = tail_page;

  	rb_inc_page(cpu_buffer, &next_page);

  	/*
  	 * If for some reason, we had an interrupt storm that made
  	 * it all the way around the buffer, bail, and warn
  	 * about it.
  	 */
  	if (unlikely(next_page == commit_page)) {

  		local_inc(&cpu_buffer->commit_overrun);

  		goto out_reset;
  	}

  	/*
  	 * This is where the fun begins!
  	 *
  	 * We are fighting against races between a reader that
  	 * could be on another CPU trying to swap its reader
  	 * page with the buffer head.
  	 *
  	 * We are also fighting against interrupts coming in and
  	 * moving the head or tail on us as well.
  	 *
  	 * If the next page is the head page then we have filled
  	 * the buffer, unless the commit page is still on the
  	 * reader page.
  	 */
  	if (rb_is_head_page(cpu_buffer, next_page, &tail_page->list)) {

  		/*
  		 * If the commit is not on the reader page, then
  		 * move the header page.
  		 */
  		if (!rb_is_reader_page(cpu_buffer->commit_page)) {
  			/*
  			 * If we are not in overwrite mode,
  			 * this is easy, just stop here.
  			 */
  			if (!(buffer->flags & RB_FL_OVERWRITE))
  				goto out_reset;
  
  			ret = rb_handle_head_page(cpu_buffer,
  						  tail_page,
  						  next_page);
  			if (ret < 0)
  				goto out_reset;
  			if (ret)
  				goto out_again;
  		} else {
  			/*
  			 * We need to be careful here too. The
  			 * commit page could still be on the reader
  			 * page. We could have a small buffer, and
  			 * have filled up the buffer with events
  			 * from interrupts and such, and wrapped.
  			 *
  			 * Note, if the tail page is also the on the
  			 * reader_page, we let it move out.
  			 */
  			if (unlikely((cpu_buffer->commit_page !=
  				      cpu_buffer->tail_page) &&
  				     (cpu_buffer->commit_page ==
  				      cpu_buffer->reader_page))) {
  				local_inc(&cpu_buffer->commit_overrun);
  				goto out_reset;
  			}

  		}
  	}

  	ret = rb_tail_page_update(cpu_buffer, tail_page, next_page);
  	if (ret) {
  		/*
  		 * Nested commits always have zero deltas, so
  		 * just reread the time stamp
  		 */

  		ts = rb_time_stamp(buffer);
  		next_page->page->time_stamp = ts;

  	}

   out_again:

  	rb_reset_tail(cpu_buffer, tail_page, tail, length);

  
  	/* fail and let the caller try again */
  	return ERR_PTR(-EAGAIN);

   out_reset:

  	/* reset write */

  	rb_reset_tail(cpu_buffer, tail_page, tail, length);

  	return NULL;

  }

  static struct ring_buffer_event *
  __rb_reserve_next(struct ring_buffer_per_cpu *cpu_buffer,

  		  unsigned long length, u64 ts,
  		  u64 delta, int add_timestamp)

  {

  	struct buffer_page *tail_page;

  	struct ring_buffer_event *event;
  	unsigned long tail, write;

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

  	tail_page = cpu_buffer->tail_page;
  	write = local_add_return(length, &tail_page->write);

  
  	/* set write to only the index of the write */
  	write &= RB_WRITE_MASK;

  	tail = write - length;
  
  	/* See if we shot pass the end of this buffer page */

  	if (unlikely(write > BUF_PAGE_SIZE))

  		return rb_move_tail(cpu_buffer, length, tail,

  				    tail_page, ts);

  
  	/* We reserved something on the buffer */

  	event = __rb_page_index(tail_page, tail);

  	kmemcheck_annotate_bitfield(event, bitfield);

  	rb_update_event(cpu_buffer, event, length, add_timestamp, delta);

  	local_inc(&tail_page->entries);

  
  	/*

  	 * If this is the first commit on the page, then update
  	 * its timestamp.

  	 */

  	if (!tail)

  		tail_page->page->time_stamp = ts;

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

  	return event;
  }

  static inline int
  rb_try_to_discard(struct ring_buffer_per_cpu *cpu_buffer,
  		  struct ring_buffer_event *event)
  {
  	unsigned long new_index, old_index;
  	struct buffer_page *bpage;
  	unsigned long index;
  	unsigned long addr;
  
  	new_index = rb_event_index(event);

  	old_index = new_index + rb_event_ts_length(event);

  	addr = (unsigned long)event;
  	addr &= PAGE_MASK;
  
  	bpage = cpu_buffer->tail_page;
  
  	if (bpage->page == (void *)addr && rb_page_write(bpage) == old_index) {

  		unsigned long write_mask =
  			local_read(&bpage->write) & ~RB_WRITE_MASK;

  		unsigned long event_length = rb_event_length(event);

  		/*
  		 * This is on the tail page. It is possible that
  		 * a write could come in and move the tail page
  		 * and write to the next page. That is fine
  		 * because we just shorten what is on this page.
  		 */

  		old_index += write_mask;
  		new_index += write_mask;

  		index = local_cmpxchg(&bpage->write, old_index, new_index);

  		if (index == old_index) {
  			/* update counters */
  			local_sub(event_length, &cpu_buffer->entries_bytes);

  			return 1;

  		}

  	}
  
  	/* could not discard */
  	return 0;
  }

  static void rb_start_commit(struct ring_buffer_per_cpu *cpu_buffer)
  {
  	local_inc(&cpu_buffer->committing);
  	local_inc(&cpu_buffer->commits);
  }

  static inline void rb_end_commit(struct ring_buffer_per_cpu *cpu_buffer)

  {
  	unsigned long commits;
  
  	if (RB_WARN_ON(cpu_buffer,
  		       !local_read(&cpu_buffer->committing)))
  		return;
  
   again:
  	commits = local_read(&cpu_buffer->commits);
  	/* synchronize with interrupts */
  	barrier();
  	if (local_read(&cpu_buffer->committing) == 1)
  		rb_set_commit_to_write(cpu_buffer);
  
  	local_dec(&cpu_buffer->committing);
  
  	/* synchronize with interrupts */
  	barrier();
  
  	/*
  	 * Need to account for interrupts coming in between the
  	 * updating of the commit page and the clearing of the
  	 * committing counter.
  	 */
  	if (unlikely(local_read(&cpu_buffer->commits) != commits) &&
  	    !local_read(&cpu_buffer->committing)) {
  		local_inc(&cpu_buffer->committing);
  		goto again;
  	}
  }

  static struct ring_buffer_event *

  rb_reserve_next_event(struct ring_buffer *buffer,
  		      struct ring_buffer_per_cpu *cpu_buffer,

  		      unsigned long length)

  {
  	struct ring_buffer_event *event;

  	u64 ts, delta;

  	int nr_loops = 0;

  	int add_timestamp;

  	u64 diff;

  	rb_start_commit(cpu_buffer);

  #ifdef CONFIG_RING_BUFFER_ALLOW_SWAP

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

  #endif

  	length = rb_calculate_event_length(length);

   again:

  	add_timestamp = 0;
  	delta = 0;

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

  	if (RB_WARN_ON(cpu_buffer, ++nr_loops > 1000))

  		goto out_fail;

  	ts = rb_time_stamp(cpu_buffer->buffer);

  	diff = ts - cpu_buffer->write_stamp;

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

  	/* Did the write stamp get updated already? */
  	if (likely(ts >= cpu_buffer->write_stamp)) {

  		delta = diff;
  		if (unlikely(test_time_stamp(delta))) {

  			int local_clock_stable = 1;
  #ifdef CONFIG_HAVE_UNSTABLE_SCHED_CLOCK
  			local_clock_stable = sched_clock_stable;
  #endif

  			WARN_ONCE(delta > (1ULL << 59),

  				  KERN_WARNING "Delta way too big! %llu ts=%llu write stamp = %llu
  %s",

  				  (unsigned long long)delta,
  				  (unsigned long long)ts,

  				  (unsigned long long)cpu_buffer->write_stamp,
  				  local_clock_stable ? "" :
  				  "If you just came from a suspend/resume,
  "
  				  "please switch to the trace global clock:
  "
  				  "  echo global > /sys/kernel/debug/tracing/trace_clock
  ");

  			add_timestamp = 1;

  		}

  	}

  	event = __rb_reserve_next(cpu_buffer, length, ts,
  				  delta, add_timestamp);

  	if (unlikely(PTR_ERR(event) == -EAGAIN))

  		goto again;

  	if (!event)
  		goto out_fail;

  	return event;

  
   out_fail:
  	rb_end_commit(cpu_buffer);
  	return NULL;

  }

  #ifdef CONFIG_TRACING

  #define TRACE_RECURSIVE_DEPTH 16

  /* Keep this code out of the fast path cache */
  static noinline void trace_recursive_fail(void)

  {

  	/* Disable all tracing before we do anything else */
  	tracing_off_permanent();

  	printk_once(KERN_WARNING "Tracing recursion: depth[%ld]:"

  		    "HC[%lu]:SC[%lu]:NMI[%lu]
  ",

  		    trace_recursion_buffer(),

  		    hardirq_count() >> HARDIRQ_SHIFT,
  		    softirq_count() >> SOFTIRQ_SHIFT,
  		    in_nmi());

  	WARN_ON_ONCE(1);

  }
  
  static inline int trace_recursive_lock(void)
  {

  	trace_recursion_inc();

  	if (likely(trace_recursion_buffer() < TRACE_RECURSIVE_DEPTH))

  		return 0;
  
  	trace_recursive_fail();

  	return -1;

  }

  static inline void trace_recursive_unlock(void)

  {

  	WARN_ON_ONCE(!trace_recursion_buffer());

  	trace_recursion_dec();

  }

  #else
  
  #define trace_recursive_lock()		(0)
  #define trace_recursive_unlock()	do { } while (0)
  
  #endif

  /**
   * ring_buffer_lock_reserve - reserve a part of the buffer
   * @buffer: the ring buffer to reserve from
   * @length: the length of the data to reserve (excluding event header)

   *
   * Returns a reseverd event on the ring buffer to copy directly to.
   * The user of this interface will need to get the body to write into
   * and can use the ring_buffer_event_data() interface.
   *
   * The length is the length of the data needed, not the event length
   * which also includes the event header.
   *
   * Must be paired with ring_buffer_unlock_commit, unless NULL is returned.
   * If NULL is returned, then nothing has been allocated or locked.
   */
  struct ring_buffer_event *

  ring_buffer_lock_reserve(struct ring_buffer *buffer, unsigned long length)

  {
  	struct ring_buffer_per_cpu *cpu_buffer;
  	struct ring_buffer_event *event;

  	int cpu;

  	if (ring_buffer_flags != RB_BUFFERS_ON)

  		return NULL;

  	/* If we are tracing schedule, we don't want to recurse */

  	preempt_disable_notrace();

  	if (atomic_read(&buffer->record_disabled))
  		goto out_nocheck;

  	if (trace_recursive_lock())
  		goto out_nocheck;

  	cpu = raw_smp_processor_id();

  	if (!cpumask_test_cpu(cpu, buffer->cpumask))

  		goto out;

  
  	cpu_buffer = buffer->buffers[cpu];

  
  	if (atomic_read(&cpu_buffer->record_disabled))

  		goto out;

  	if (length > BUF_MAX_DATA_SIZE)

  		goto out;

  	event = rb_reserve_next_event(buffer, cpu_buffer, length);

  	if (!event)

  		goto out;

  
  	return event;

   out:

  	trace_recursive_unlock();
  
   out_nocheck:

  	preempt_enable_notrace();

  	return NULL;
  }

  EXPORT_SYMBOL_GPL(ring_buffer_lock_reserve);

  static void
  rb_update_write_stamp(struct ring_buffer_per_cpu *cpu_buffer,

  		      struct ring_buffer_event *event)
  {

  	u64 delta;

  	/*
  	 * The event first in the commit queue updates the
  	 * time stamp.
  	 */