Generic Ring Buffer Library

Index: linux-2.6-lttng/kernel/notifier.c
===================================================================
@@ -148,7 +148,7 @@ int atomic_notifier_chain_unregister(str
 	spin_lock_irqsave(&nh->lock, flags);
 	ret = notifier_chain_unregister(&nh->head, n);
 	spin_unlock_irqrestore(&nh->lock, flags);
-	synchronize_rcu();
+	synchronize_sched();
 	return ret;
 }
 EXPORT_SYMBOL_GPL(atomic_notifier_chain_unregister);

@@ -178,9 +178,9 @@ int __kprobes __atomic_notifier_call_cha
 {
 	int ret;
 
-	rcu_read_lock();
+	rcu_read_lock_sched_notrace();
 	ret = notifier_call_chain(&nh->head, val, v, nr_to_call, nr_calls);
-	rcu_read_unlock();
+	rcu_read_unlock_sched_notrace();
 	return ret;
 }
 EXPORT_SYMBOL_GPL(__atomic_notifier_call_chain);

Index: linux.trees.git/arch/x86/kernel/process_64.c
===================================================================
@@ -35,6 +35,7 @@
 #include <linux/tick.h>
 #include <linux/prctl.h>
 #include <linux/uaccess.h>
+#include <linux/idle.h>
 #include <linux/io.h>
 #include <linux/ftrace.h>
 

@@ -58,31 +59,17 @@ asmlinkage extern void ret_from_fork(voi
 DEFINE_PER_CPU(unsigned long, old_rsp);
 static DEFINE_PER_CPU(unsigned char, is_idle);
 
-static ATOMIC_NOTIFIER_HEAD(idle_notifier);
-
-void idle_notifier_register(struct notifier_block *n)
-{
-	atomic_notifier_chain_register(&idle_notifier, n);
-}
-EXPORT_SYMBOL_GPL(idle_notifier_register);
-
-void idle_notifier_unregister(struct notifier_block *n)
-{
-	atomic_notifier_chain_unregister(&idle_notifier, n);
-}
-EXPORT_SYMBOL_GPL(idle_notifier_unregister);
-
 void enter_idle(void)
 {
 	percpu_write(is_idle, 1);
-	atomic_notifier_call_chain(&idle_notifier, IDLE_START, NULL);
+	notify_idle(IDLE_START);
 }
 
 static void __exit_idle(void)
 {
 	if (x86_test_and_clear_bit_percpu(0, is_idle) == 0)
 		return;
-	atomic_notifier_call_chain(&idle_notifier, IDLE_END, NULL);
+	notify_idle(IDLE_END);
 }
 
 /* Called from interrupts to signify idle end */

Index: linux.trees.git/arch/x86/include/asm/idle.h
===================================================================
@@ -1,13 +1,6 @@
 #ifndef _ASM_X86_IDLE_H
 #define _ASM_X86_IDLE_H
 
-#define IDLE_START 1
-#define IDLE_END 2
-
-struct notifier_block;
-void idle_notifier_register(struct notifier_block *n);
-void idle_notifier_unregister(struct notifier_block *n);
-
 #ifdef CONFIG_X86_64
 void enter_idle(void);
 void exit_idle(void);

Index: linux.trees.git/include/linux/idle.h
===================================================================
@@ -0,0 +1,19 @@
+/*
+ * include/linux/idle.h - generic idle definition
+ *
+ */
+#ifndef _LINUX_IDLE_H_
+#define _LINUX_IDLE_H_
+
+#include <linux/notifier.h>
+
+enum idle_val {
+	IDLE_START = 1,
+	IDLE_END = 2,
+};
+
+int notify_idle(enum idle_val val);
+void register_idle_notifier(struct notifier_block *n);
+void unregister_idle_notifier(struct notifier_block *n);
+
+#endif /* _LINUX_IDLE_H_ */

Index: linux.trees.git/kernel/notifier.c
===================================================================
@@ -5,6 +5,7 @@
 #include <linux/rcupdate.h>
 #include <linux/vmalloc.h>
 #include <linux/reboot.h>
+#include <linux/idle.h>
 
 /*
  *	Notifier list for kernel code which wants to be called

@@ -584,3 +585,27 @@ int unregister_die_notifier(struct notif
 	return atomic_notifier_chain_unregister(&die_chain, nb);
 }
 EXPORT_SYMBOL_GPL(unregister_die_notifier);
+
+static ATOMIC_NOTIFIER_HEAD(idle_notifier);
+
+/*
+ * Trace last event before calling notifiers. Notifiers flush data from buffers
+ * before going to idle.
+ */
+int notrace notify_idle(enum idle_val val)
+{
+	return atomic_notifier_call_chain(&idle_notifier, val, NULL);
+}
+EXPORT_SYMBOL_GPL(notify_idle);
+
+void register_idle_notifier(struct notifier_block *n)
+{
+	atomic_notifier_chain_register(&idle_notifier, n);
+}
+EXPORT_SYMBOL_GPL(register_idle_notifier);
+
+void unregister_idle_notifier(struct notifier_block *n)
+{
+	atomic_notifier_chain_unregister(&idle_notifier, n);
+}
+EXPORT_SYMBOL_GPL(unregister_idle_notifier);

Index: linux.trees.git/drivers/idle/i7300_idle.c
===================================================================
@@ -27,6 +27,7 @@
 #include <linux/debugfs.h>
 #include <linux/stop_machine.h>
 #include <linux/i7300_idle.h>
+#include <linux/idle.h>
 
 #include <asm/idle.h>
 

@@ -583,7 +584,7 @@ static int __init i7300_idle_init(void)
 		}
 	}
 
-	idle_notifier_register(&i7300_idle_nb);
+	register_idle_notifier(&i7300_idle_nb);
 
 	printk(KERN_INFO "i7300_idle: loaded v%s\n", I7300_IDLE_DRIVER_VERSION);
 	return 0;

@@ -591,7 +592,7 @@ static int __init i7300_idle_init(void)
 
 static void __exit i7300_idle_exit(void)
 {
-	idle_notifier_unregister(&i7300_idle_nb);
+	unregister_idle_notifier(&i7300_idle_nb);
 	free_cpumask_var(idle_cpumask);
 
 	if (debugfs_dir) {

Index: linux.trees.git/arch/x86/kernel/process_32.c
===================================================================
@@ -38,6 +38,8 @@
 #include <linux/uaccess.h>
 #include <linux/io.h>
 #include <linux/kdebug.h>
+#include <linux/notifier.h>
+#include <linux/idle.h>
 
 #include <asm/pgtable.h>
 #include <asm/system.h>

@@ -61,6 +63,30 @@
 
 asmlinkage void ret_from_fork(void) __asm__("ret_from_fork");
 
+static DEFINE_PER_CPU(unsigned char, is_idle);
+
+void enter_idle(void)
+{
+	percpu_write(is_idle, 1);
+	notify_idle(IDLE_START);
+}
+
+static void __exit_idle(void)
+{
+	if (x86_test_and_clear_bit_percpu(0, is_idle) == 0)
+		return;
+	notify_idle(IDLE_END);
+}
+
+/* Called from interrupts to signify idle end */
+void exit_idle(void)
+{
+	/* idle loop has pid 0 */
+	if (current->pid)
+		return;
+	__exit_idle();
+}
+
 /*
  * Return saved PC of a blocked thread.
  */

@@ -109,10 +135,17 @@ void cpu_idle(void)
 				play_dead();
 
 			local_irq_disable();
+			enter_idle();
 			/* Don't trace irqs off for idle */
 			stop_critical_timings();
 			pm_idle();
 			start_critical_timings();
+			/*
+			 * In many cases the interrupt that ended idle
+			 * has already called exit_idle. But some idle loops can
+			 * be woken up without interrupt.
+			 */
+			__exit_idle();
 
 			trace_power_end(smp_processor_id());
 		}

Index: linux.trees.git/arch/x86/include/asm/idle.h
===================================================================
@@ -1,13 +1,8 @@
 #ifndef _ASM_X86_IDLE_H
 #define _ASM_X86_IDLE_H
 
-#ifdef CONFIG_X86_64
 void enter_idle(void);
 void exit_idle(void);
-#else /* !CONFIG_X86_64 */
-static inline void enter_idle(void) { }
-static inline void exit_idle(void) { }
-#endif /* CONFIG_X86_64 */
 
 void c1e_remove_cpu(int cpu);
 

Index: linux.trees.git/arch/x86/kernel/apm_32.c
===================================================================
@@ -227,6 +227,7 @@
 #include <linux/suspend.h>
 #include <linux/kthread.h>
 #include <linux/jiffies.h>
+#include <linux/idle.h>
 
 #include <asm/system.h>
 #include <asm/uaccess.h>

@@ -947,10 +948,15 @@ recalc:
 				break;
 			}
 		}
+		enter_idle();
 		if (original_pm_idle)
 			original_pm_idle();
 		else
 			default_idle();
+		/* In many cases the interrupt that ended idle
+		   has already called exit_idle. But some idle
+		   loops can be woken up without interrupt. */
+		__exit_idle();
 		local_irq_disable();
 		jiffies_since_last_check = jiffies - last_jiffies;
 		if (jiffies_since_last_check > idle_period)

Index: linux.trees.git/fs/select.c
===================================================================
@@ -112,6 +112,9 @@ struct poll_table_page {
  */
 static void __pollwait(struct file *filp, wait_queue_head_t *wait_address,
 		       poll_table *p);
+static void __pollwait_exclusive(struct file *filp,
+				 wait_queue_head_t *wait_address,
+				 poll_table *p);
 
 void poll_initwait(struct poll_wqueues *pwq)
 {

@@ -152,6 +155,20 @@ void poll_freewait(struct poll_wqueues *
 }
 EXPORT_SYMBOL(poll_freewait);
 
+/**
+ * poll_wait_set_exclusive - set poll wait queue to exclusive
+ *
+ * Sets up a poll wait queue to use exclusive wakeups. This is useful to
+ * wake up only one waiter at each wakeup. Used to work-around "thundering herd"
+ * problem.
+ */
+void poll_wait_set_exclusive(poll_table *p)
+{
+	if (p)
+		init_poll_funcptr(p, __pollwait_exclusive);
+}
+EXPORT_SYMBOL(poll_wait_set_exclusive);
+
 static struct poll_table_entry *poll_get_entry(struct poll_wqueues *p)
 {
 	struct poll_table_page *table = p->table;

@@ -213,8 +230,10 @@ static int pollwake(wait_queue_t *wait,
 }
 
 /* Add a new entry */
-static void __pollwait(struct file *filp, wait_queue_head_t *wait_address,
-				poll_table *p)
+static void __pollwait_common(struct file *filp,
+			      wait_queue_head_t *wait_address,
+			      poll_table *p,
+			      int exclusive)
 {
 	struct poll_wqueues *pwq = container_of(p, struct poll_wqueues, pt);
 	struct poll_table_entry *entry = poll_get_entry(pwq);

@@ -226,7 +245,23 @@ static void __pollwait(struct file *filp
 	entry->key = p->key;
 	init_waitqueue_func_entry(&entry->wait, pollwake);
 	entry->wait.private = pwq;
-	add_wait_queue(wait_address, &entry->wait);
+	if (!exclusive)
+		add_wait_queue(wait_address, &entry->wait);
+	else
+		add_wait_queue_exclusive(wait_address, &entry->wait);
+}
+
+static void __pollwait(struct file *filp, wait_queue_head_t *wait_address,
+				poll_table *p)
+{
+	__pollwait_common(filp, wait_address, p, 0);
+}
+
+static void __pollwait_exclusive(struct file *filp,
+				 wait_queue_head_t *wait_address,
+				 poll_table *p)
+{
+	__pollwait_common(filp, wait_address, p, 1);
 }
 
 int poll_schedule_timeout(struct poll_wqueues *pwq, int state,

Index: linux.trees.git/include/linux/poll.h
===================================================================
@@ -79,6 +79,8 @@ static inline int poll_schedule(struct p
 	return poll_schedule_timeout(pwq, state, NULL, 0);
 }
 
+extern void poll_wait_set_exclusive(poll_table *p);
+
 /*
  * Scaleable version of the fd_set.
  */

Index: linux.trees.git/include/linux/prio_heap.h
===================================================================
@@ -23,6 +23,18 @@ struct ptr_heap {
 };
 
 /**
+ * heap_maximum - return the largest element in the heap
+ * @heap: the heap to be operated on
+ *
+ * Returns the largest element in the heap, without performing any modification
+ * to the heap structure. Returns NULL if the heap is empty.
+ */
+static inline void *heap_maximum(const struct ptr_heap *heap)
+{
+	return heap->size ? heap->ptrs[0] : NULL;
+}
+
+/**
  * heap_init - initialize an empty heap with a given memory size
  * @heap: the heap structure to be initialized
  * @size: amount of memory to use in bytes

@@ -53,6 +65,38 @@ void heap_free(struct ptr_heap *heap);
  */
 extern void *heap_insert(struct ptr_heap *heap, void *p);
 
+/**
+ * heap_remove - remove the largest element from the heap
+ * @heap: the heap to be operated on
+ *
+ * Returns the largest element in the heap. It removes this element from the
+ * heap. Returns NULL if the heap is empty.
+ */
+extern void *heap_remove(struct ptr_heap *heap);
 
+/**
+ * heap_cherrypick - remove a given element from the heap
+ * @heap: the heap to be operated on
+ * @p: the element
+ *
+ * Remove the given element from the heap. Return the element if present, else
+ * return NULL. This algorithm has a complexity of O(n), which is higher than
+ * O(log(n)) provided by the rest of this API.
+ */
+extern void *heap_cherrypick(struct ptr_heap *heap, void *p);
+
+/**
+ * heap_replace_max - replace the the largest element from the heap
+ * @heap: the heap to be operated on
+ * @p: the pointer to be inserted as topmost element replacement
+ *
+ * Returns the largest element in the heap. It removes this element from the
+ * heap. The heap is rebalanced only once after the insertion. Returns NULL if
+ * the heap is empty.
+ *
+ * This is the equivalent of calling heap_remove() and then heap_insert(), but
+ * it only rebalances the heap once.
+ */
+extern void *heap_replace_max(struct ptr_heap *heap, void *p);
 
 #endif /* _LINUX_PRIO_HEAP_H */

108 lines not on this page · raw message

Index: linux.trees.git/include/linux/kthread.h
===================================================================
@@ -29,6 +29,7 @@ struct task_struct *kthread_create(int (
 
 void kthread_bind(struct task_struct *k, unsigned int cpu);
 int kthread_stop(struct task_struct *k);
+int kthread_kill_stop(struct task_struct *k, int signo);
 int kthread_should_stop(void);
 
 int kthreadd(void *unused);

Index: linux.trees.git/kernel/kthread.c
===================================================================
@@ -211,6 +211,46 @@ int kthread_stop(struct task_struct *k)
 }
 EXPORT_SYMBOL(kthread_stop);
 
+/**
+ * kthread_kill_stop - kill and stop a thread created by kthread_create().
+ * @k: thread created by kthread_create().
+ * @signo: signal number to send.
+ *
+ * Sets kthread_should_stop() for @k to return true, sends a signal, and
+ * waits for it to exit. This can also be called after kthread_create()
+ * instead of calling wake_up_process(): the thread will exit without
+ * calling threadfn().
+ *
+ * If threadfn() may call do_exit() itself, the caller must ensure
+ * task_struct can't go away.
+ *
+ * Returns the result of threadfn(), or %-EINTR if wake_up_process()
+ * was never called.
+ */
+int kthread_kill_stop(struct task_struct *k, int signo)
+{
+	struct kthread *kthread;
+	int ret;
+
+	trace_sched_kthread_stop(k);
+	get_task_struct(k);
+
+	kthread = to_kthread(k);
+	barrier(); /* it might have exited */
+	if (k->vfork_done != NULL) {
+		kthread->should_stop = 1;
+		force_sig(signo, k);
+		wait_for_completion(&kthread->exited);
+	}
+	ret = k->exit_code;
+
+	put_task_struct(k);
+	trace_sched_kthread_stop_ret(ret);
+
+	return ret;
+}
+EXPORT_SYMBOL(kthread_kill_stop);
+
 int kthreadd(void *unused)
 {
 	struct task_struct *tsk = current;

Index: linux.trees.git/include/linux/string.h
===================================================================
@@ -102,6 +102,9 @@ extern void * memset(void *,int,__kernel
 #ifndef __HAVE_ARCH_MEMCPY
 extern void * memcpy(void *,const void *,__kernel_size_t);
 #endif
+#ifndef __HAVE_ARCH_INLINE_MEMCPY
+#define inline_memcpy memcpy
+#endif
 #ifndef __HAVE_ARCH_MEMMOVE
 extern void * memmove(void *,const void *,__kernel_size_t);
 #endif

Index: linux.trees.git/arch/x86/include/asm/string_32.h
===================================================================
@@ -44,6 +44,13 @@ static __always_inline void *__memcpy(vo
 	return to;
 }
 
+#define __HAVE_ARCH_INLINE_MEMCPY
+static __always_inline void *inline_memcpy(void *to, const void *from,
+					   size_t n)
+{
+	return __memcpy(to, from, n);
+}
+
 /*
  * This looks ugly, but the compiler can optimize it totally,
  * as the count is constant.

Index: linux.trees.git/arch/x86/include/asm/string_64.h
===================================================================
@@ -23,6 +23,13 @@ static __always_inline void *__inline_me
 	return to;
 }
 
+#define __HAVE_ARCH_INLINE_MEMCPY
+static __always_inline void *inline_memcpy(void *to, const void *from,
+					   size_t n)
+{
+	return __inline_memcpy(to, from, n);
+}
+
 /* Even with __builtin_ the compiler may decide to use the out of line
    function. */
 

Index: linux.trees.git/kernel/trace/trace_clock.c
===================================================================
@@ -44,6 +44,7 @@ u64 notrace trace_clock_local(void)
 
 	return clock;
 }
+EXPORT_SYMBOL_GPL(trace_clock_local);
 
 /*
  * trace_clock(): 'inbetween' trace clock. Not completely serialized,

@@ -57,6 +58,7 @@ u64 notrace trace_clock(void)
 {
 	return local_clock();
 }
+EXPORT_SYMBOL_GPL(trace_clock);
 
 
 /*

@@ -113,3 +115,4 @@ u64 notrace trace_clock_global(void)
 
 	return now;
 }
+EXPORT_SYMBOL_GPL(trace_clock_global);

Index: linux.trees.git/include/linux/ftrace_event.h
===================================================================
@@ -1,7 +1,7 @@
 #ifndef _LINUX_FTRACE_EVENT_H
 #define _LINUX_FTRACE_EVENT_H
 
-#include <linux/ring_buffer.h>
+#include <linux/ftrace_ring_buffer.h>
 #include <linux/trace_seq.h>
 #include <linux/percpu.h>
 #include <linux/hardirq.h>

@@ -55,7 +55,7 @@ struct trace_iterator {
 	void			*private;
 	int			cpu_file;
 	struct mutex		mutex;
-	struct ring_buffer_iter	*buffer_iter[NR_CPUS];
+	struct ftrace_ring_buffer_iter	*buffer_iter[NR_CPUS];
 	unsigned long		iter_flags;
 
 	/* The below is zeroed out in pipe_read */

@@ -106,18 +106,18 @@ enum print_line_t {
 void tracing_generic_entry_update(struct trace_entry *entry,
 				  unsigned long flags,
 				  int pc);
-struct ring_buffer_event *
-trace_current_buffer_lock_reserve(struct ring_buffer **current_buffer,
+struct ftrace_ring_buffer_event *
+trace_current_buffer_lock_reserve(struct ftrace_ring_buffer **current_buffer,
 				  int type, unsigned long len,
 				  unsigned long flags, int pc);
-void trace_current_buffer_unlock_commit(struct ring_buffer *buffer,
-					struct ring_buffer_event *event,
+void trace_current_buffer_unlock_commit(struct ftrace_ring_buffer *buffer,
+					struct ftrace_ring_buffer_event *event,
 					unsigned long flags, int pc);
-void trace_nowake_buffer_unlock_commit(struct ring_buffer *buffer,
-				       struct ring_buffer_event *event,
+void trace_nowake_buffer_unlock_commit(struct ftrace_ring_buffer *buffer,
+				       struct ftrace_ring_buffer_event *event,
 					unsigned long flags, int pc);
-void trace_current_buffer_discard_commit(struct ring_buffer *buffer,
-					 struct ring_buffer_event *event);
+void trace_current_buffer_discard_commit(struct ftrace_ring_buffer *buffer,
+					 struct ftrace_ring_buffer_event *event);
 
 void tracing_record_cmdline(struct task_struct *tsk);
 

@@ -199,10 +199,10 @@ struct ftrace_event_call {
 
 extern void destroy_preds(struct ftrace_event_call *call);
 extern int filter_match_preds(struct event_filter *filter, void *rec);
-extern int filter_current_check_discard(struct ring_buffer *buffer,
+extern int filter_current_check_discard(struct ftrace_ring_buffer *buffer,
 					struct ftrace_event_call *call,
 					void *rec,
-					struct ring_buffer_event *event);
+					struct ftrace_ring_buffer_event *event);
 
 enum {
 	FILTER_OTHER = 0,

196 lines not on this page · raw message

196 lines not on this page · raw message

2,496 lines not on this page · raw message

4,022 lines not on this page · raw message

Index: linux.trees.git/include/linux/oprofile.h
===================================================================

Lines of this file are not shown. View the raw message.

Index: linux.trees.git/include/trace/ftrace.h
===================================================================

Lines of this file are not shown. View the raw message.

Index: linux.trees.git/kernel/trace/blktrace.c
===================================================================

Lines of this file are not shown. View the raw message.

488 lines not on this page · raw message

488 lines not on this page · raw message

262 lines not on this page · raw message

Index: linux.trees.git/kernel/trace/trace.h
===================================================================

Lines of this file are not shown. View the raw message.

Index: linux.trees.git/kernel/trace/trace_branch.c
===================================================================

Lines of this file are not shown. View the raw message.

Index: linux.trees.git/kernel/trace/trace_events.c
===================================================================

Lines of this file are not shown. View the raw message.

Index: linux.trees.git/kernel/trace/trace_functions_graph.c
===================================================================

Lines of this file are not shown. View the raw message.

Index: linux.trees.git/kernel/trace/trace_kprobe.c
===================================================================

Lines of this file are not shown. View the raw message.

Index: linux.trees.git/kernel/trace/trace_ksym.c
===================================================================

Lines of this file are not shown. View the raw message.

Index: linux.trees.git/kernel/trace/trace_mmiotrace.c
===================================================================

Lines of this file are not shown. View the raw message.

Index: linux.trees.git/kernel/trace/trace_sched_switch.c
===================================================================

Lines of this file are not shown. View the raw message.

Index: linux.trees.git/kernel/trace/trace_selftest.c
===================================================================

Lines of this file are not shown. View the raw message.

Index: linux.trees.git/kernel/trace/trace_syscalls.c
===================================================================

Lines of this file are not shown. View the raw message.

Index: linux.trees.git/drivers/oprofile/cpu_buffer.c
===================================================================

Lines of this file are not shown. View the raw message.

Index: linux.trees.git/drivers/oprofile/cpu_buffer.h
===================================================================

Lines of this file are not shown. View the raw message.

Index: linux.trees.git/include/linux/kernel.h
===================================================================

Lines of this file are not shown. View the raw message.

Index: linux.trees.git/kernel/trace/trace_functions.c
===================================================================

Lines of this file are not shown. View the raw message.

141 lines not on this page · raw message

418 lines not on this page · raw message

755 lines not on this page · raw message

80 lines not on this page · raw message

Index: linux.trees.git/lib/ringbuffer/ring_buffer_frontend.c
===================================================================
@@ -0,0 +1,1510 @@
+/*
+ * ring_buffer_frontend.c
+ *
+ * (C) Copyright 2005-2010 - Mathieu Desnoyers <[email protected]>
+ *
+ * Ring buffer wait-free buffer synchronization. Producer-consumer and flight
+ * recorder (overwrite) modes. See thesis:
+ *
+ * Desnoyers, Mathieu (2009), "Low-Impact Operating System Tracing", Ph.D.
+ * dissertation, Ecole Polytechnique de Montreal.
+ * http://www.lttng.org/pub/thesis/desnoyers-dissertation-2009-12.pdf
+ *
+ * - Algorithm presentation in Chapter 5:
+ *     "Lockless Multi-Core High-Throughput Buffering".
+ * - Algorithm formal verification in Section 8.6:
+ *     "Formal verification of LTTng"
+ *
+ * Author:
+ *	Mathieu Desnoyers <[email protected]>
+ *
+ * Inspired from LTT and RelayFS:
+ *  Karim Yaghmour <[email protected]>
+ *  Tom Zanussi <[email protected]>
+ *  Bob Wisniewski <[email protected]>
+ * And from K42 :
+ *  Bob Wisniewski <[email protected]>
+ *
+ * Buffer reader semantic :
+ *
+ * - get_subbuf_size
+ * while buffer is not finalized and empty
+ *   - get_subbuf
+ *     - if return value != 0, continue
+ *   - splice one subbuffer worth of data to a pipe
+ *   - splice the data from pipe to disk/network
+ *   - put_subbuf
+ *
+ * Dual LGPL v2.1/GPL v2 license.
+ */
+
+#include <linux/idle.h>
+#include <linux/delay.h>
+#include <linux/module.h>
+#include <linux/percpu.h>
+
+#include <linux/ringbuffer/config.h>
+#include <linux/ringbuffer/backend.h>
+#include <linux/ringbuffer/frontend.h>
+#include <linux/ringbuffer/iterator.h>
+
+/*
+ * Internal structure representing offsets to use at a sub-buffer switch.
+ */
+struct switch_offsets {
+	unsigned long begin, end, old;
+	size_t pre_header_padding, size;
+	unsigned int switch_new_start:1, switch_new_end:1, switch_old_start:1,
+		     switch_old_end:1;
+};
+
+DEFINE_PER_CPU(unsigned int, ring_buffer_nesting);
+EXPORT_PER_CPU_SYMBOL(ring_buffer_nesting);
+
+static
+void ring_buffer_print_errors(struct channel *chan,
+			      struct ring_buffer *buf,
+			      int cpu);
+
+static const struct file_operations ring_buffer_file_operations;
+
+/*
+ * Must be called under cpu hotplug protection.
+ */
+void ring_buffer_free(struct ring_buffer *buf)
+{
+	struct channel *chan = buf->backend.chan;
+
+	ring_buffer_print_errors(chan, buf, buf->backend.cpu);
+	kfree(buf->commit_hot);
+	kfree(buf->commit_cold);
+
+	ring_buffer_backend_free(&buf->backend);
+}
+
+/**
+ * ring_buffer_reset - Reset ring buffer to initial values.
+ * @buf: Ring buffer.
+ *
+ * Effectively empty the ring buffer. Should be called when the buffer is not
+ * used for writing. The ring buffer can be opened for reading, but the reader
+ * should not be using the iterator concurrently with reset. The previous
+ * current iterator record is reset.
+ */
+void ring_buffer_reset(struct ring_buffer *buf)
+{
+	struct channel *chan = buf->backend.chan;
+	const struct ring_buffer_config *config = chan->backend.config;
+	unsigned int i;
+
+	/*
+	 * Reset iterator first. It will put the subbuffer if it currently holds
+	 * it.
+	 */
+	ring_buffer_iterator_reset(buf);
+	v_set(config, &buf->offset, 0);
+	for (i = 0; i < chan->backend.num_subbuf; i++) {
+		v_set(config, &buf->commit_hot[i].cc, 0);
+		v_set(config, &buf->commit_hot[i].seq, 0);
+		v_set(config, &buf->commit_cold[i].cc_sb, 0);
+	}
+	atomic_long_set(&buf->consumed, 0);
+	atomic_set(&buf->record_disabled, 0);
+	v_set(config, &buf->last_tsc, 0);
+	ring_buffer_backend_reset(&buf->backend);
+	/* Don't reset number of active readers */
+	v_set(config, &buf->records_lost_full, 0);
+	v_set(config, &buf->records_lost_wrap, 0);
+	v_set(config, &buf->records_lost_big, 0);
+	v_set(config, &buf->records_count, 0);
+	v_set(config, &buf->records_overrun, 0);
+	buf->finalized = 0;
+}
+EXPORT_SYMBOL_GPL(ring_buffer_reset);
+
+/**
+ * channel_reset - Reset channel to initial values.
+ * @chan: Channel.
+ *
+ * Effectively empty the channel. Should be called when the channel is not used
+ * for writing. The channel can be opened for reading, but the reader should not
+ * be using the iterator concurrently with reset. The previous current iterator
+ * record is reset.
+ */
+void channel_reset(struct channel *chan)
+{
+	/*
+	 * Reset iterators first. Will put the subbuffer if held for reading.
+	 */
+	channel_iterator_reset(chan);
+	atomic_set(&chan->record_disabled, 0);
+	/* Don't reset commit_count_mask, still valid */
+	channel_backend_reset(&chan->backend);
+	/* Don't reset switch/read timer interval */
+	/* Don't reset notifiers and notifier enable bits */
+	/* Don't reset reader reference count */
+}
+EXPORT_SYMBOL_GPL(channel_reset);
+
+/*
+ * Must be called under cpu hotplug protection.
+ */
+int ring_buffer_create(struct ring_buffer *buf,
+		       struct channel_backend *chanb,
+		       int cpu)
+{
+	const struct ring_buffer_config *config = chanb->config;
+	struct channel *chan = container_of(chanb, struct channel, backend);
+	void *priv = chanb->priv;
+	unsigned int j, num_subbuf;
+	size_t subbuf_header_size;
+	u64 tsc;
+	int ret;
+
+	/* Test for cpu hotplug */
+	if (buf->backend.allocated)
+		return 0;
+
+	ret = ring_buffer_backend_create(&buf->backend, &chan->backend, cpu);
+	if (ret)
+		return ret;
+
+	buf->commit_hot =
+		kzalloc_node(ALIGN(sizeof(*buf->commit_hot)
+				   * chan->backend.num_subbuf,
+				   1 << INTERNODE_CACHE_SHIFT),
+			GFP_KERNEL, cpu_to_node(max(cpu, 0)));
+	if (!buf->commit_hot) {
+		ret = -ENOMEM;
+		goto free_chanbuf;
+	}
+
+	buf->commit_cold =
+		kzalloc_node(ALIGN(sizeof(*buf->commit_cold)
+				   * chan->backend.num_subbuf,
+				   1 << INTERNODE_CACHE_SHIFT),
+			GFP_KERNEL, cpu_to_node(max(cpu, 0)));
+	if (!buf->commit_cold) {
+		ret = -ENOMEM;
+		goto free_commit;
+	}
+
+	atomic_long_set(&buf->consumed, 0);
+	atomic_long_set(&buf->active_readers, 0);
+	num_subbuf = chan->backend.num_subbuf;
+	for (j = 0; j < num_subbuf; j++) {
+		v_set(config, &buf->commit_hot[j].cc, 0);
+		v_set(config, &buf->commit_hot[j].seq, 0);
+		v_set(config, &buf->commit_cold[j].cc_sb, 0);
+	}
+	init_waitqueue_head(&buf->read_wait);
+	raw_spin_lock_init(&buf->raw_idle_spinlock);
+
+	v_set(config, &buf->records_lost_full, 0);
+	v_set(config, &buf->records_lost_wrap, 0);
+	v_set(config, &buf->records_lost_big, 0);
+	v_set(config, &buf->records_count, 0);
+	v_set(config, &buf->records_overrun, 0);
+	buf->finalized = 0;
+
+	/*
+	 * Write the subbuffer header for first subbuffer so we know the total
+	 * duration of data gathering.
+	 */
+	subbuf_header_size = config->cb.subbuffer_header_size();
+	v_set(config, &buf->offset, subbuf_header_size);
+	subbuffer_id_clear_noref(config, &buf->backend.buf_wsb[0].id);
+	tsc = config->cb.ring_buffer_clock_read(buf->backend.chan);
+	config->cb.buffer_begin(buf, tsc, 0);
+	v_add(config, subbuf_header_size, &buf->commit_hot[0].cc);
+
+	if (config->cb.buffer_create) {
+		ret = config->cb.buffer_create(buf, priv, cpu, chanb->name);
+		if (ret)
+			goto free_init;
+	}
+
+	/*
+	 * Ensure the buffer is ready before setting it to allocated and setting
+	 * the cpumask.
+	 * Used for cpu hotplug vs cpumask iteration.
+	 */
+	smp_wmb();
+	buf->backend.allocated = 1;
+
+	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
+		CHAN_WARN_ON(chan, cpumask_test_cpu(cpu,
+			     chan->backend.cpumask));
+		cpumask_set_cpu(cpu, chan->backend.cpumask);
+	}
+
+	return 0;
+
+	/* Error handling */
+free_init:
+	kfree(buf->commit_cold);
+free_commit:
+	kfree(buf->commit_hot);
+free_chanbuf:
+	ring_buffer_backend_free(&buf->backend);
+	return ret;
+}
+
+static void switch_buffer_timer(unsigned long data)
+{
+	struct ring_buffer *buf = (struct ring_buffer *)data;
+	struct channel *chan = buf->backend.chan;
+	const struct ring_buffer_config *config = chan->backend.config;
+
+	/*
+	 * Only flush buffers periodically if readers are active.
+	 */
+	if (atomic_long_read(&buf->active_readers))
+		ring_buffer_switch_slow(buf, SWITCH_ACTIVE);
+
+	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
+		mod_timer_pinned(&buf->switch_timer,
+				 jiffies + chan->switch_timer_interval);
+	else
+		mod_timer(&buf->switch_timer,
+			  jiffies + chan->switch_timer_interval);
+}
+
+static void ring_buffer_start_switch_timer(struct ring_buffer *buf)
+{
+	struct channel *chan = buf->backend.chan;
+	const struct ring_buffer_config *config = chan->backend.config;
+
+	if (!chan->switch_timer_interval)
+		return;
+
+	init_timer_deferrable(&buf->switch_timer);
+	buf->switch_timer.function = switch_buffer_timer;
+	buf->switch_timer.expires = jiffies + chan->switch_timer_interval;
+	buf->switch_timer.data = (unsigned long)buf;
+	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
+		add_timer_on(&buf->switch_timer, buf->backend.cpu);
+	else
+		add_timer(&buf->switch_timer);
+}
+
+static void ring_buffer_stop_switch_timer(struct ring_buffer *buf)
+{
+	struct channel *chan = buf->backend.chan;
+
+	if (!chan->switch_timer_interval)
+		return;
+
+	del_timer_sync(&buf->switch_timer);
+}
+
+/*
+ * Polling timer to check the channels for data.
+ */
+static void read_buffer_timer(unsigned long data)
+{
+	struct ring_buffer *buf = (struct ring_buffer *)data;
+	struct channel *chan = buf->backend.chan;
+	const struct ring_buffer_config *config = chan->backend.config;
+
+	CHAN_WARN_ON(chan, !buf->backend.allocated);
+
+	if (atomic_long_read(&buf->active_readers)
+	    && ring_buffer_poll_deliver(config, buf, chan)) {
+		wake_up_interruptible(&buf->read_wait);
+		wake_up_interruptible(&chan->read_wait);
+	}
+
+	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
+		mod_timer_pinned(&buf->read_timer,
+				 jiffies + chan->read_timer_interval);
+	else
+		mod_timer(&buf->read_timer,
+			  jiffies + chan->read_timer_interval);
+}
+
+static void ring_buffer_start_read_timer(struct ring_buffer *buf)
+{
+	struct channel *chan = buf->backend.chan;
+	const struct ring_buffer_config *config = chan->backend.config;
+
+	if (config->wakeup != RING_BUFFER_WAKEUP_BY_TIMER
+	    || !chan->read_timer_interval)
+		return;
+
+	init_timer_deferrable(&buf->read_timer);
+	buf->read_timer.function = read_buffer_timer;
+	buf->read_timer.expires = jiffies + chan->read_timer_interval;
+	buf->read_timer.data = (unsigned long)buf;
+
+	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
+		add_timer_on(&buf->read_timer, buf->backend.cpu);
+	else
+		add_timer(&buf->read_timer);
+}
+
+static void ring_buffer_stop_read_timer(struct ring_buffer *buf)
+{
+	struct channel *chan = buf->backend.chan;
+	const struct ring_buffer_config *config = chan->backend.config;
+
+	if (config->wakeup != RING_BUFFER_WAKEUP_BY_TIMER
+	    || !chan->read_timer_interval)
+		return;
+
+	del_timer_sync(&buf->read_timer);
+	/*
+	 * do one more check to catch data that has been written in the last
+	 * timer period.
+	 */
+	if (ring_buffer_poll_deliver(config, buf, chan)) {
+		wake_up_interruptible(&buf->read_wait);
+		wake_up_interruptible(&chan->read_wait);
+	}
+}
+
+#ifdef CONFIG_HOTPLUG_CPU
+/**
+ *	ring_buffer_cpu_hp_callback - CPU hotplug callback
+ *	@nb: notifier block
+ *	@action: hotplug action to take
+ *	@hcpu: CPU number
+ *
+ *	Returns the success/failure of the operation. (%NOTIFY_OK, %NOTIFY_BAD)
+ */
+static
+int __cpuinit ring_buffer_cpu_hp_callback(struct notifier_block *nb,
+					  unsigned long action,
+					  void *hcpu)
+{
+	unsigned int cpu = (unsigned long)hcpu;
+	struct channel *chan = container_of(nb, struct channel,
+					    cpu_hp_notifier);
+	struct ring_buffer *buf = per_cpu_ptr(chan->backend.buf, cpu);
+	const struct ring_buffer_config *config = chan->backend.config;
+
+	if (!chan->cpu_hp_enable)
+		return NOTIFY_DONE;
+
+	CHAN_WARN_ON(chan, config->alloc == RING_BUFFER_ALLOC_GLOBAL);
+
+	switch (action) {
+	case CPU_DOWN_FAILED:
+	case CPU_DOWN_FAILED_FROZEN:
+	case CPU_ONLINE:
+	case CPU_ONLINE_FROZEN:
+		ring_buffer_start_switch_timer(buf);
+		ring_buffer_start_read_timer(buf);
+		return NOTIFY_OK;
+
+	case CPU_DOWN_PREPARE:
+	case CPU_DOWN_PREPARE_FROZEN:
+		ring_buffer_stop_switch_timer(buf);
+		ring_buffer_stop_read_timer(buf);
+		return NOTIFY_OK;
+
+	case CPU_DEAD:
+	case CPU_DEAD_FROZEN:
+		/*
+		 * Performing a buffer switch on a remote CPU. Performed by
+		 * the CPU responsible for doing the hotunplug after the target
+		 * CPU stopped running completely. Ensures that all data
+		 * from that remote CPU is flushed.
+		 */
+		ring_buffer_switch_slow(buf, SWITCH_ACTIVE);
+		return NOTIFY_OK;
+
+	default:
+		return NOTIFY_DONE;
+	}
+}
+#endif
+
+
+/*
+ * For per-cpu buffers, call the reader wakeups before switching the buffer, so
+ * that wake-up-tracing generated events are flushed before going idle. We test
+ * if the spinlock is locked to deal with the race where readers try to sample
+ * the ring buffer before we perform the switch. We let the readers retry in
+ * that case. If there is data in the buffer, the wake up is going to forbid the
+ * CPU running the reader thread from going idle.
+ *
+ * For a global buffer, if the client requested a reader timer, then chances are
+ * we are going to keep the system from going idle anyway, so just bite the
+ * bullet and do the wake up. We have no way to know if we are the last CPU
+ * going to idle, so just switch the buffer. Use a spinlock to ensure we send
+ * the wakeup before performing the buffer switch, in case wakeup is
+ * instrumented and writes data in the buffer.
+ */
+static int ring_buffer_idle_callback(struct notifier_block *nb,
+				  unsigned long val,
+				  void *data)
+{
+	struct channel *chan = container_of(nb, struct channel,
+					    idle_notifier);
+	const struct ring_buffer_config *config = chan->backend.config;
+	struct ring_buffer *buf;
+
+	if (val != IDLE_START)
+		return 0;
+
+	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
+		buf = channel_get_ring_buffer(config, chan, smp_processor_id());
+	else
+		buf = channel_get_ring_buffer(config, chan, 0);
+
+	raw_spin_lock(&buf->raw_idle_spinlock);
+	if (config->wakeup == RING_BUFFER_WAKEUP_BY_TIMER
+	    && chan->read_timer_interval
+	    && atomic_long_read(&buf->active_readers)
+	    && (ring_buffer_poll_deliver(config, buf, chan)
+		|| ring_buffer_pending_data(config, buf, chan))) {
+		wake_up_interruptible(&buf->read_wait);
+		wake_up_interruptible(&chan->read_wait);
+	}
+	if (chan->switch_timer_interval)
+		ring_buffer_switch_slow(buf, SWITCH_ACTIVE);
+	raw_spin_unlock(&buf->raw_idle_spinlock);
+
+	return 0;
+}
+
+/*
+ * Holds CPU hotplug.
+ */
+static void channel_unregister_notifiers(struct channel *chan)
+{
+	const struct ring_buffer_config *config = chan->backend.config;
+	int cpu;
+
+	channel_iterator_unregister_notifiers(chan);
+	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
+#ifdef CONFIG_HOTPLUG_CPU
+		get_online_cpus();
+		chan->cpu_hp_enable = 0;
+		for_each_online_cpu(cpu) {
+			struct ring_buffer *buf = per_cpu_ptr(chan->backend.buf,
+							      cpu);
+			ring_buffer_stop_switch_timer(buf);
+			ring_buffer_stop_read_timer(buf);
+		}
+		put_online_cpus();
+		unregister_cpu_notifier(&chan->cpu_hp_notifier);
+#else
+		for_each_possible_cpu(cpu) {
+			struct ring_buffer *buf = per_cpu_ptr(chan->backend.buf,
+							      cpu);
+			ring_buffer_stop_switch_timer(buf);
+			ring_buffer_stop_read_timer(buf);
+		}
+#endif
+	} else {
+		struct ring_buffer *buf = chan->backend.buf;
+
+		ring_buffer_stop_switch_timer(buf);
+		ring_buffer_stop_read_timer(buf);
+	}
+	unregister_idle_notifier(&chan->idle_notifier);
+	channel_backend_unregister_notifiers(&chan->backend);
+}
+
+static void channel_free(struct channel *chan)
+{
+	channel_iterator_free(chan);
+	channel_backend_free(&chan->backend);
+	kfree(chan);
+}
+
+/**
+ * channel_create - Create channel.
+ * @config: ring buffer instance configuration
+ * @name: name of the channel
+ * @priv: ring buffer client private data
+ * @buf_addr: pointer the the beginning of the preallocated buffer contiguous
+ *            address mapping. It is used only by RING_BUFFER_STATIC
+ *            configuration. It can be set to NULL for other backends.
+ * @subbuf_size: subbuffer size
+ * @num_subbuf: number of subbuffers
+ * @switch_timer_interval: Time interval (in us) to fill sub-buffers with
+ *                         padding to let readers get those sub-buffers.
+ *                         Used for live streaming.
+ * @read_timer_interval: Time interval (in us) to wake up pending readers.
+ *
+ * Holds cpu hotplug.
+ * Returns NULL on failure.
+ */
+struct channel *channel_create(const struct ring_buffer_config *config,
+		   const char *name, void *priv, void *buf_addr,
+		   size_t subbuf_size,
+		   size_t num_subbuf, unsigned int switch_timer_interval,
+		   unsigned int read_timer_interval)
+{
+	int ret, cpu;
+	struct channel *chan;
+
+	if (ring_buffer_check_config(config,
+				     switch_timer_interval,
+				     read_timer_interval))
+		return NULL;
+
+	chan = kzalloc(sizeof(struct channel), GFP_KERNEL);
+	if (!chan)
+		return NULL;
+
+	ret = channel_backend_init(&chan->backend, name, config, priv,
+				   subbuf_size, num_subbuf);
+	if (ret)
+		goto error;
+
+	ret = channel_iterator_init(chan);
+	if (ret)
+		goto error_free_backend;
+
+	chan->commit_count_mask = (~0UL >> chan->backend.num_subbuf_order);
+	chan->switch_timer_interval = usecs_to_jiffies(switch_timer_interval);
+	chan->read_timer_interval = usecs_to_jiffies(read_timer_interval);
+	init_waitqueue_head(&chan->read_wait);
+
+	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
+		/*
+		 * 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
+		chan->cpu_hp_notifier.notifier_call =
+				ring_buffer_cpu_hp_callback;
+		chan->cpu_hp_notifier.priority = 6;
+		register_cpu_notifier(&chan->cpu_hp_notifier);
+
+		get_online_cpus();
+		for_each_online_cpu(cpu) {
+			struct ring_buffer *buf = per_cpu_ptr(chan->backend.buf,
+							       cpu);
+			ring_buffer_start_switch_timer(buf);
+			ring_buffer_start_read_timer(buf);
+		}
+		chan->cpu_hp_enable = 1;
+		put_online_cpus();
+#else
+		for_each_possible_cpu(cpu) {
+			struct ring_buffer *buf = per_cpu_ptr(chan->backend.buf,
+							      cpu);
+			ring_buffer_start_switch_timer(buf);
+			ring_buffer_start_read_timer(buf);
+		}
+#endif
+	} else {
+		struct ring_buffer *buf = chan->backend.buf;
+
+		ring_buffer_start_switch_timer(buf);
+		ring_buffer_start_read_timer(buf);
+	}
+
+	chan->idle_notifier.notifier_call = ring_buffer_idle_callback;
+	/*
+	 * smallest prio, run after any tracing activity, right before sleeping.
+	 */
+	chan->idle_notifier.priority = ~0U;
+	register_idle_notifier(&chan->idle_notifier);
+
+	return chan;
+
+error_free_backend:
+	channel_backend_free(&chan->backend);
+error:
+	kfree(chan);
+	return NULL;
+}
+EXPORT_SYMBOL_GPL(channel_create);
+
+/**
+ * channel_destroy - Finalize, wait for q.s. and destroy channel.
+ * @chan: channel to destroy
+ *
+ * Holds cpu hotplug.
+ * Call "destroy" callback, finalize channels, wait for readers to release their
+ * reference, then destroy ring buffer data. Note that when readers have
+ * completed data consumption of finalized channels, get_subbuf() will return
+ * -ENODATA. They should release their handle at that point.
+ * Returns the private data pointer.
+ */
+void *channel_destroy(struct channel *chan)
+{
+	int cpu;
+	const struct ring_buffer_config *config = chan->backend.config;
+	void *priv;
+
+	channel_unregister_notifiers(chan);
+
+	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
+		/*
+		 * No need to hold cpu hotplug, because all notifiers have been
+		 * unregistered.
+		 */
+		for_each_channel_cpu(cpu, chan) {
+			struct ring_buffer *buf = per_cpu_ptr(chan->backend.buf,
+							      cpu);
+
+			if (config->cb.buffer_finalize)
+				config->cb.buffer_finalize(buf,
+							   chan->backend.priv,
+							   cpu);
+			if (buf->backend.allocated)
+				ring_buffer_switch_slow(buf, SWITCH_FLUSH);
+			/*
+			 * Perform flush before writing to finalized.
+			 */
+			smp_wmb();
+			ACCESS_ONCE(buf->finalized) = 1;
+			wake_up_interruptible(&buf->read_wait);
+		}
+	} else {
+		struct ring_buffer *buf = chan->backend.buf;
+
+		if (config->cb.buffer_finalize)
+			config->cb.buffer_finalize(buf, chan->backend.priv, -1);
+		if (buf->backend.allocated)
+			ring_buffer_switch_slow(buf, SWITCH_FLUSH);
+		/*
+		 * Perform flush before writing to finalized.
+		 */
+		smp_wmb();
+		ACCESS_ONCE(buf->finalized) = 1;
+		wake_up_interruptible(&buf->read_wait);
+	}
+	wake_up_interruptible(&chan->read_wait);
+
+	while (atomic_long_read(&chan->read_ref) > 0)
+		msleep(100);
+	/* Finish waiting for refcount before free */
+	smp_mb();
+	priv = chan->backend.priv;
+	channel_free(chan);
+	return priv;
+}
+EXPORT_SYMBOL_GPL(channel_destroy);
+
+struct ring_buffer *channel_get_ring_buffer(
+					const struct ring_buffer_config *config,
+					struct channel *chan, int cpu)
+{
+	if (config->alloc == RING_BUFFER_ALLOC_GLOBAL)
+		return chan->backend.buf;
+	else
+		return per_cpu_ptr(chan->backend.buf, cpu);
+}
+EXPORT_SYMBOL_GPL(channel_get_ring_buffer);
+
+int ring_buffer_open_read(struct ring_buffer *buf)
+{
+	struct channel *chan = buf->backend.chan;
+
+	if (!atomic_long_add_unless(&buf->active_readers, 1, 1))
+		return -EBUSY;
+	atomic_long_inc(&chan->read_ref);
+	smp_mb__after_atomic_inc();
+	return 0;
+}
+EXPORT_SYMBOL_GPL(ring_buffer_open_read);
+
+void ring_buffer_release_read(struct ring_buffer *buf)
+{
+	struct channel *chan = buf->backend.chan;
+
+	CHAN_WARN_ON(chan, atomic_long_read(&buf->active_readers) != 1);
+	smp_mb__before_atomic_dec();
+	atomic_long_dec(&chan->read_ref);
+	atomic_long_dec(&buf->active_readers);
+}
+EXPORT_SYMBOL_GPL(ring_buffer_release_read);
+
+/*
+ * Promote compiler barrier to a smp_mb().
+ * For the specific ring buffer case, this IPI call should be removed if the
+ * architecture does not reorder writes.  This should eventually be provided by
+ * a separate architecture-specific infrastructure.
+ */
+static void remote_mb(void *info)
+{
+	smp_mb();
+}
+
+/**
+ * ring_buffer_get_subbuf - get exclusive access to subbuffer for reading
+ * @buf: ring buffer
+ * @consumed: pointer where to save the consumed count value (output)
+ *
+ * Returns -ENODATA if buffer is finalized, -EAGAIN if there is currently no
+ * data ready, or 0 if the get operation succeeds.
+ * Busy-loop trying to get data if the idle sequence lock is held.
+ */
+int ring_buffer_get_subbuf(struct ring_buffer *buf, unsigned long *consumed)
+{
+	struct channel *chan = buf->backend.chan;
+	const struct ring_buffer_config *config = chan->backend.config;
+	unsigned long consumed_old, consumed_idx, commit_count, write_offset;
+	int ret;
+	int finalized;
+
+retry:
+	finalized = ACCESS_ONCE(buf->finalized);
+	/*
+	 * Read finalized before counters.
+	 */
+	smp_rmb();
+	consumed_old = atomic_long_read(&buf->consumed);
+	consumed_idx = subbuf_index(consumed_old, chan);
+	commit_count = v_read(config, &buf->commit_cold[consumed_idx].cc_sb);
+	/*
+	 * Make sure we read the commit count before reading the buffer
+	 * data and the write offset. Correct consumed offset ordering
+	 * wrt commit count is insured by the use of cmpxchg to update
+	 * the consumed offset.
+	 * smp_call_function_single can fail if the remote CPU is offline,
+	 * this is OK because then there is no wmb to execute there.
+	 * If our thread is executing on the same CPU as the on the buffers
+	 * belongs to, we don't have to synchronize it at all. If we are
+	 * migrated, the scheduler will take care of the memory barriers.
+	 * Normally, smp_call_function_single() should ensure program order when
+	 * executing the remote function, which implies that it surrounds the
+	 * function execution with :
+	 * smp_mb()
+	 * send IPI
+	 * csd_lock_wait
+	 *                recv IPI
+	 *                smp_mb()
+	 *                exec. function
+	 *                smp_mb()
+	 *                csd unlock
+	 * smp_mb()
+	 *
+	 * However, smp_call_function_single() does not seem to clearly execute
+	 * such barriers. It depends on spinlock semantic to provide the barrier
+	 * before executing the IPI and, when busy-looping, csd_lock_wait only
+	 * executes smp_mb() when it has to wait for the other CPU.
+	 *
+	 * I don't trust this code. Therefore, let's add the smp_mb() sequence
+	 * required ourself, even if duplicated. It has no performance impact
+	 * anyway.
+	 *
+	 * smp_mb() is needed because smp_rmb() and smp_wmb() only order read vs
+	 * read and write vs write. They do not ensure core synchronization. We
+	 * really have to ensure total order between the 3 barriers running on
+	 * the 2 CPUs.
+	 */
+	if (config->ipi == RING_BUFFER_IPI_BARRIER) {
+		if (config->sync == RING_BUFFER_SYNC_PER_CPU
+		    && config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
+			if (raw_smp_processor_id() != buf->backend.cpu) {
+				/* Total order with IPI handler smp_mb() */
+				smp_mb();
+				smp_call_function_single(buf->backend.cpu,
+							 remote_mb, NULL, 1);
+				/* Total order with IPI handler smp_mb() */
+				smp_mb();
+			}
+		} else {
+			/* Total order with IPI handler smp_mb() */
+			smp_mb();
+			smp_call_function(remote_mb, NULL, 1);
+			/* Total order with IPI handler smp_mb() */
+			smp_mb();
+		}
+	} else {
+		/*
+		 * Local rmb to match the remote wmb to read the commit count
+		 * before the buffer data and the write offset.
+		 */
+		smp_rmb();
+	}
+
+	write_offset = v_read(config, &buf->offset);
+
+	/*
+	 * Check that the subbuffer we are trying to consume has been
+	 * already fully committed.
+	 */
+	if (((commit_count - chan->backend.subbuf_size)
+	     & chan->commit_count_mask)
+	    - (buf_trunc(consumed_old, chan)
+	       >> chan->backend.num_subbuf_order)
+	    != 0)
+		goto nodata;
+
+	/*
+	 * Check that we are not about to read the same subbuffer in
+	 * which the writer head is.
+	 */
+	if ((subbuf_trunc(write_offset, chan)
+	   - subbuf_trunc(consumed_old, chan))
+	   == 0)
+		goto nodata;
+
+	/*
+	 * Failure to get the subbuffer causes a busy-loop retry without going
+	 * to a wait queue. These are caused by short-lived race windows where
+	 * the writer is getting access to a subbuffer we were trying to get
+	 * access to. Also checks that the "consumed" buffer count we are
+	 * looking for matches the one contained in the subbuffer id.
+	 */
+	ret = update_read_sb_index(config, &buf->backend, &chan->backend,
+				   consumed_idx,
+				   buf_trunc_val(consumed_old, chan));
+	if (ret)
+		goto retry;
+
+	*consumed = consumed_old;
+	return 0;
+
+nodata:
+	/*
+	 * The memory barriers __wait_event()/wake_up_interruptible() take care
+	 * of "raw_spin_is_locked" memory ordering.
+	 */
+	if (finalized)
+		return -ENODATA;
+	else if (raw_spin_is_locked(&buf->raw_idle_spinlock))
+		goto retry;
+	else
+		return -EAGAIN;
+}
+EXPORT_SYMBOL_GPL(ring_buffer_get_subbuf);
+
+/**
+ * ring_buffer_put_subbuf - release exclusive subbuffer access
+ * @buf: ring buffer
+ * @consumed: consumed count value (output)
+ */
+
+void ring_buffer_put_subbuf(struct ring_buffer *buf, unsigned long consumed)
+{
+	struct ring_buffer_backend *bufb = &buf->backend;
+	struct channel *chan = bufb->chan;
+	const struct ring_buffer_config *config = chan->backend.config;
+	unsigned long consumed_new, consumed_old, sb_bindex;
+
+	CHAN_WARN_ON(chan, atomic_long_read(&buf->active_readers) != 1);
+
+	consumed_old = consumed;
+	consumed_new = subbuf_align(consumed_old, chan);
+
+	/*
+	 * Clear the records_unread counter. (overruns counter)
+	 * Can still be non-zero if a file reader simply grabbed the data
+	 * without using iterators.
+	 */
+	sb_bindex = subbuffer_id_get_index(config, bufb->buf_rsb.id);
+	v_add(config, v_read(config, &bufb->array[sb_bindex]->records_unread),
+	      &bufb->records_read);
+	v_set(config, &bufb->array[sb_bindex]->records_unread, 0);
+	CHAN_WARN_ON(chan, config->mode == RING_BUFFER_OVERWRITE
+		     && subbuffer_id_is_noref(config, bufb->buf_rsb.id));
+	subbuffer_id_set_noref(config, &bufb->buf_rsb.id);
+
+	/*
+	 * We exchange the subbuffer pages. No corruption possible even if the
+	 * writer did push us, because our subbuffer pages were owned by the
+	 * reader. If the consumed cmpxchg fails, this is because we have been
+	 * pushed by the writer in flight recorder mode. Don't retry, just keep
+	 * the new consumed offset given by the writer.
+	 */
+	atomic_long_cmpxchg(&buf->consumed, consumed_old, consumed_new);
+}
+EXPORT_SYMBOL_GPL(ring_buffer_put_subbuf);
+
+/*
+ * cons_offset is an iterator on all subbuffer offsets between the reader
+ * position and the writer position. (inclusive)
+ */
+static
+void ring_buffer_print_subbuffer_errors(struct ring_buffer *buf,
+					struct channel *chan,
+					unsigned long cons_offset,
+					int cpu)
+{
+	const struct ring_buffer_config *config = chan->backend.config;
+	unsigned long cons_idx, commit_count, commit_count_sb;
+
+	cons_idx = subbuf_index(cons_offset, chan);
+	commit_count = v_read(config, &buf->commit_hot[cons_idx].cc);
+	commit_count_sb = v_read(config, &buf->commit_cold[cons_idx].cc_sb);
+
+	if (subbuf_offset(commit_count, chan) != 0)
+		printk(KERN_WARNING
+		       "ring buffer %s, cpu %d: "
+		       "commit count in subbuffer %lu,\n"
+		       "expecting multiples of %lu bytes\n"
+		       "  [ %lu bytes committed, %lu bytes reader-visible ]\n",
+		       chan->backend.name, cpu, cons_idx,
+		       chan->backend.subbuf_size,
+		       commit_count, commit_count_sb);
+
+	printk(KERN_DEBUG "ring buffer: %s, cpu %d: %lu bytes committed\n",
+	       chan->backend.name, cpu, commit_count);
+}
+
+static
+void ring_buffer_print_buffer_errors(struct ring_buffer *buf,
+				     struct channel *chan,
+				     void *priv, int cpu)
+{
+	const struct ring_buffer_config *config = chan->backend.config;
+	unsigned long write_offset, cons_offset;
+
+	/*
+	 * Can be called in the error path of allocation when
+	 * trans_channel_data is not yet set.
+	 */
+	if (!chan)
+		return;
+	/*
+	 * No need to order commit_count, write_offset and cons_offset reads
+	 * because we execute at teardown when no more writer nor reader
+	 * references are left.
+	 */
+	write_offset = v_read(config, &buf->offset);
+	cons_offset = atomic_long_read(&buf->consumed);
+	if (write_offset != cons_offset)
+		printk(KERN_WARNING
+		       "ring buffer %s, cpu %d: "
+		       "non-consumed data\n"
+		       "  [ %lu bytes written, %lu bytes read ]\n",
+		       chan->backend.name, cpu, write_offset, cons_offset);
+
+	for (cons_offset = atomic_long_read(&buf->consumed);
+	     (long) (subbuf_trunc((unsigned long) v_read(config, &buf->offset),
+				  chan)
+		     - cons_offset) > 0;
+	     cons_offset = subbuf_align(cons_offset, chan))
+		ring_buffer_print_subbuffer_errors(buf, chan, cons_offset,
+						   cpu);
+}
+
+static
+void ring_buffer_print_errors(struct channel *chan,
+			      struct ring_buffer *buf,
+			      int cpu)
+{
+	const struct ring_buffer_config *config = chan->backend.config;
+	void *priv = chan->backend.priv;
+
+	printk(KERN_DEBUG "ring buffer %s, cpu %d: %lu records written, "
+			  "%lu records overrun\n",
+			  chan->backend.name, cpu,
+			  v_read(config, &buf->records_count),
+			  v_read(config, &buf->records_overrun));
+
+	if (v_read(config, &buf->records_lost_full)
+	    || v_read(config, &buf->records_lost_wrap)
+	    || v_read(config, &buf->records_lost_big))
+		printk(KERN_WARNING
+		       "ring buffer %s, cpu %d: records were lost. Caused by:\n"
+		       "  [ %lu buffer full, %lu nest buffer wrap-around, "
+		       "%lu event too big ]\n",
+		       chan->backend.name, cpu,
+		       v_read(config, &buf->records_lost_full),
+		       v_read(config, &buf->records_lost_wrap),
+		       v_read(config, &buf->records_lost_big));
+
+	ring_buffer_print_buffer_errors(buf, chan, priv, cpu);
+}
+
+/*
+ * ring_buffer_switch_old_start: Populate old subbuffer header.
+ *
+ * Only executed when the buffer is finalized, in SWITCH_FLUSH.
+ */
+static
+void ring_buffer_switch_old_start(struct ring_buffer *buf,
+				  struct channel *chan,
+				  struct switch_offsets *offsets,
+				  u64 tsc)
+{
+	const struct ring_buffer_config *config = chan->backend.config;
+	unsigned long oldidx = subbuf_index(offsets->old, chan);
+	unsigned long commit_count;
+
+	config->cb.buffer_begin(buf, tsc, oldidx);
+
+	/*
+	 * Order all writes to buffer before the commit count update that will
+	 * determine that the subbuffer is full.
+	 */
+	if (config->ipi == RING_BUFFER_IPI_BARRIER) {
+		/*
+		 * Must write slot data before incrementing commit count.  This
+		 * compiler barrier is upgraded into a smp_mb() by the IPI sent
+		 * by get_subbuf().
+		 */
+		barrier();
+	} else
+		smp_wmb();
+	v_add(config, config->cb.subbuffer_header_size(),
+	      &buf->commit_hot[oldidx].cc);
+	commit_count = v_read(config, &buf->commit_hot[oldidx].cc);
+	/* Check if the written buffer has to be delivered */
+	ring_buffer_check_deliver(config, buf, chan, offsets->old, commit_count,
+				  oldidx);
+	ring_buffer_write_commit_counter(config, buf, chan, oldidx,
+					 offsets->old, commit_count,
+					 config->cb.subbuffer_header_size());
+}
+
+/*
+ * ring_buffer_switch_old_end: switch old subbuffer
+ *
+ * Note : offset_old should never be 0 here. It is ok, because we never perform
+ * buffer switch on an empty subbuffer in SWITCH_ACTIVE mode. The caller
+ * increments the offset_old value when doing a SWITCH_FLUSH on an empty
+ * subbuffer.
+ */
+static
+void ring_buffer_switch_old_end(struct ring_buffer *buf,
+					   struct channel *chan,
+					   struct switch_offsets *offsets,
+					   u64 tsc)
+{
+	const struct ring_buffer_config *config = chan->backend.config;
+	unsigned long oldidx = subbuf_index(offsets->old - 1, chan);
+	unsigned long commit_count, padding_size, data_size;
+
+	data_size = subbuf_offset(offsets->old - 1, chan) + 1;
+	padding_size = chan->backend.subbuf_size - data_size;
+	subbuffer_set_data_size(config, &buf->backend, oldidx, data_size);
+
+	/*
+	 * Order all writes to buffer before the commit count update that will
+	 * determine that the subbuffer is full.
+	 */
+	if (config->ipi == RING_BUFFER_IPI_BARRIER) {
+		/*
+		 * Must write slot data before incrementing commit count.  This
+		 * compiler barrier is upgraded into a smp_mb() by the IPI sent
+		 * by get_subbuf().
+		 */
+		barrier();
+	} else
+		smp_wmb();
+	v_add(config, padding_size, &buf->commit_hot[oldidx].cc);
+	commit_count = v_read(config, &buf->commit_hot[oldidx].cc);
+	ring_buffer_check_deliver(config, buf, chan, offsets->old - 1,
+				  commit_count, oldidx);
+	ring_buffer_write_commit_counter(config, buf, chan, oldidx,
+					 offsets->old, commit_count,
+					 padding_size);
+}
+
+/*
+ * ring_buffer_switch_new_start: Populate new subbuffer.
+ *
+ * This code can be executed unordered : writers may already have written to the
+ * sub-buffer before this code gets executed, caution.  The commit makes sure
+ * that this code is executed before the deliver of this sub-buffer.
+ */
+static
+void ring_buffer_switch_new_start(struct ring_buffer *buf,
+					   struct channel *chan,
+					   struct switch_offsets *offsets,
+					   u64 tsc)
+{
+	const struct ring_buffer_config *config = chan->backend.config;
+	unsigned long beginidx = subbuf_index(offsets->begin, chan);
+	unsigned long commit_count;
+
+	config->cb.buffer_begin(buf, tsc, beginidx);
+
+	/*
+	 * Order all writes to buffer before the commit count update that will
+	 * determine that the subbuffer is full.
+	 */
+	if (config->ipi == RING_BUFFER_IPI_BARRIER) {
+		/*
+		 * Must write slot data before incrementing commit count.  This
+		 * compiler barrier is upgraded into a smp_mb() by the IPI sent
+		 * by get_subbuf().
+		 */
+		barrier();
+	} else
+		smp_wmb();
+	v_add(config, config->cb.subbuffer_header_size(),
+	      &buf->commit_hot[beginidx].cc);
+	commit_count = v_read(config, &buf->commit_hot[beginidx].cc);
+	/* Check if the written buffer has to be delivered */
+	ring_buffer_check_deliver(config, buf, chan, offsets->begin,
+				  commit_count, beginidx);
+	ring_buffer_write_commit_counter(config, buf, chan, beginidx,
+					 offsets->begin, commit_count,
+					 config->cb.subbuffer_header_size());
+}
+
+/*
+ * ring_buffer_switch_new_end: finish switching current subbuffer
+ *
+ * The only remaining threads could be the ones with pending commits. They will
+ * have to do the deliver themselves.
+ */
+static
+void ring_buffer_switch_new_end(struct ring_buffer *buf,
+					    struct channel *chan,
+					    struct switch_offsets *offsets,
+					    u64 tsc)
+{
+	const struct ring_buffer_config *config = chan->backend.config;
+	unsigned long endidx = subbuf_index(offsets->end - 1, chan);
+	unsigned long commit_count, padding_size, data_size;
+
+	data_size = subbuf_offset(offsets->end - 1, chan) + 1;
+	padding_size = chan->backend.subbuf_size - data_size;
+	subbuffer_set_data_size(config, &buf->backend, endidx, data_size);
+
+	/*
+	 * Order all writes to buffer before the commit count update that will
+	 * determine that the subbuffer is full.
+	 */
+	if (config->ipi == RING_BUFFER_IPI_BARRIER) {
+		/*
+		 * Must write slot data before incrementing commit count.  This
+		 * compiler barrier is upgraded into a smp_mb() by the IPI sent
+		 * by get_subbuf().
+		 */
+		barrier();
+	} else
+		smp_wmb();
+	v_add(config, padding_size, &buf->commit_hot[endidx].cc);
+	commit_count = v_read(config, &buf->commit_hot[endidx].cc);
+	ring_buffer_check_deliver(config, buf, chan, offsets->end - 1,
+				  commit_count, endidx);
+	ring_buffer_write_commit_counter(config, buf, chan, endidx,
+					 offsets->end, commit_count,
+					 padding_size);
+}
+
+/*
+ * Returns :
+ * 0 if ok
+ * !0 if execution must be aborted.
+ */
+static
+int ring_buffer_try_switch_slow(enum switch_mode mode,
+				struct ring_buffer *buf,
+				struct channel *chan,
+				struct switch_offsets *offsets,
+				u64 *tsc)
+{
+	const struct ring_buffer_config *config = chan->backend.config;
+	unsigned long off;
+
+	offsets->begin = v_read(config, &buf->offset);
+	offsets->old = offsets->begin;
+	offsets->switch_old_start = 0;
+	off = subbuf_offset(offsets->begin, chan);
+
+	*tsc = config->cb.ring_buffer_clock_read(chan);
+
+	/*
+	 * Ensure we flush the header of an empty subbuffer when doing the
+	 * finalize (SWITCH_FLUSH). This ensures that we end up knowing the
+	 * total data gathering duration even if there were no records saved
+	 * after the last buffer switch.
+	 * In SWITCH_ACTIVE mode, switch the buffer when it contains events.
+	 * SWITCH_ACTIVE only flushes the current subbuffer, dealing with end of
+	 * subbuffer header as appropriate.
+	 * The next record that reserves space will be responsible for
+	 * populating the following subbuffer header. We choose not to populate
+	 * the next subbuffer header here because we want to be able to use
+	 * SWITCH_ACTIVE for periodical buffer flush and CPU idle entry buffer
+	 * flush, which must guarantee that all the buffer content (records and
+	 * header timestamps) are visible to the reader. This is required for
+	 * quiescence guarantees for the fusion merge.
+	 */
+	if (mode == SWITCH_FLUSH || off > 0) {
+		if (unlikely(off == 0)) {
+			/*
+			 * The client does not save any header information.
+			 * Don't switch empty subbuffer on finalize, because it
+			 * is invalid to deliver a completely empty subbuffer.
+			 */
+			if (!config->cb.subbuffer_header_size())
+				return -1;
+			/*
+			 * Need to write the subbuffer start header on finalize.
+			 */
+			offsets->switch_old_start = 1;
+		}
+		offsets->begin = subbuf_align(offsets->begin, chan);
+	} else
+		return -1;	/* we do not have to switch : buffer is empty */
+	/* Note: old points to the next subbuf at offset 0 */
+	offsets->end = offsets->begin;
+	return 0;
+}
+
+/*
+ * Force a sub-buffer switch. This operation is completely reentrant : can be
+ * called while tracing is active with absolutely no lock held.
+ *
+ * Note, however, that as a v_cmpxchg is used for some atomic
+ * operations, this function must be called from the CPU which owns the buffer
+ * for a ACTIVE flush.
+ */
+void ring_buffer_switch_slow(struct ring_buffer *buf, enum switch_mode mode)
+{
+	struct channel *chan = buf->backend.chan;
+	const struct ring_buffer_config *config = chan->backend.config;
+	struct switch_offsets offsets;
+	unsigned long oldidx;
+	u64 tsc;
+
+	offsets.size = 0;
+
+	/*
+	 * Perform retryable operations.
+	 */
+	do {
+		if (ring_buffer_try_switch_slow(mode, buf, chan, &offsets,
+						&tsc))
+			return;	/* Switch not needed */
+	} while (v_cmpxchg(config, &buf->offset, offsets.old, offsets.end)
+		 != offsets.old);
+
+	/*
+	 * Atomically update last_tsc. This update races against concurrent
+	 * atomic updates, but the race will always cause supplementary full TSC
+	 * records, never the opposite (missing a full TSC record when it would
+	 * be needed).
+	 */
+	save_last_tsc(config, buf, tsc);
+
+	/*
+	 * Push the reader if necessary
+	 */
+	ring_buffer_reserve_push_reader(buf, chan, offsets.old);
+
+	oldidx = subbuf_index(offsets.old, chan);
+	ring_buffer_clear_noref(config, &buf->backend, oldidx);
+
+	/*
+	 * May need to populate header start on SWITCH_FLUSH.
+	 */
+	if (offsets.switch_old_start) {
+		ring_buffer_switch_old_start(buf, chan, &offsets, tsc);
+		offsets.old += config->cb.subbuffer_header_size();
+	}
+
+	/*
+	 * Switch old subbuffer.
+	 */
+	ring_buffer_switch_old_end(buf, chan, &offsets, tsc);
+}
+EXPORT_SYMBOL_GPL(ring_buffer_switch_slow);
+
+/*
+ * Returns :
+ * 0 if ok
+ * !0 if execution must be aborted.
+ */
+static
+int ring_buffer_try_reserve_slow(struct ring_buffer *buf, struct channel *chan,
+				 struct switch_offsets *offsets,
+				 struct ring_buffer_ctx *ctx)
+{
+	const struct ring_buffer_config *config = chan->backend.config;
+	unsigned long reserve_commit_diff;
+
+	offsets->begin = v_read(config, &buf->offset);
+	offsets->old = offsets->begin;
+	offsets->switch_new_start = 0;
+	offsets->switch_new_end = 0;
+	offsets->switch_old_end = 0;
+	offsets->pre_header_padding = 0;
+
+	ctx->tsc = config->cb.ring_buffer_clock_read(chan);
+
+	if (last_tsc_overflow(config, buf, ctx->tsc))
+		ctx->rflags = RING_BUFFER_RFLAG_FULL_TSC;
+
+	if (unlikely(subbuf_offset(offsets->begin, ctx->chan) == 0)) {
+		offsets->switch_new_start = 1;		/* For offsets->begin */
+	} else {
+		offsets->size = config->cb.record_header_size(config, chan,
+						offsets->begin,
+						ctx->data_size,
+						&offsets->pre_header_padding,
+						ctx->rflags, ctx);
+		offsets->size +=
+			ring_buffer_align(config,
+					  offsets->begin + offsets->size,
+					  ctx->largest_align)
+			+ ctx->data_size;
+		if (unlikely((subbuf_offset(offsets->begin, chan) +
+			     offsets->size) > chan->backend.subbuf_size)) {
+			offsets->switch_old_end = 1;	/* For offsets->old */
+			offsets->switch_new_start = 1;	/* For offsets->begin */
+		}
+	}
+	if (unlikely(offsets->switch_new_start)) {
+		unsigned long sb_index;
+
+		/*
+		 * We are typically not filling the previous buffer completely.
+		 */
+		if (likely(offsets->switch_old_end))
+			offsets->begin = subbuf_align(offsets->begin, chan);
+		offsets->begin = offsets->begin
+				 + config->cb.subbuffer_header_size();
+		/* Test new buffer integrity */
+		sb_index = subbuf_index(offsets->begin, chan);
+		reserve_commit_diff =
+		  (buf_trunc(offsets->begin, chan)
+		   >> chan->backend.num_subbuf_order)
+		  - ((unsigned long) v_read(config,
+					    &buf->commit_cold[sb_index].cc_sb)
+		     & chan->commit_count_mask);
+		if (likely(reserve_commit_diff == 0)) {
+			/* Next subbuffer not being written to. */
+			if (unlikely(config->mode != RING_BUFFER_OVERWRITE &&
+				(subbuf_trunc(offsets->begin, chan)
+				 - subbuf_trunc((unsigned long)
+				     atomic_long_read(&buf->consumed), chan))
+				>= chan->backend.buf_size)) {
+				/*
+				 * We do not overwrite non consumed buffers
+				 * and we are full : record is lost.
+				 */
+				v_inc(config, &buf->records_lost_full);
+				return -1;
+			} else {
+				/*
+				 * Next subbuffer not being written to, and we
+				 * are either in overwrite mode or the buffer is
+				 * not full. It's safe to write in this new
+				 * subbuffer.
+				 */
+			}
+		} else {
+			/*
+			 * Next subbuffer reserve offset does not match the
+			 * commit offset. Drop record in producer-consumer and
+			 * overwrite mode. Caused by either a writer OOPS or too
+			 * many nested writes over a reserve/commit pair.
+			 */
+			v_inc(config, &buf->records_lost_wrap);
+			return -1;
+		}
+		offsets->size =
+			config->cb.record_header_size(config, chan,
+						offsets->begin,
+						ctx->data_size,
+						&offsets->pre_header_padding,
+						ctx->rflags, ctx);
+		offsets->size +=
+			ring_buffer_align(config,
+					  offsets->begin + offsets->size,
+					  ctx->largest_align)
+			+ ctx->data_size;
+		if (unlikely((subbuf_offset(offsets->begin, chan)
+			     + offsets->size) > chan->backend.subbuf_size)) {
+			/*
+			 * Record too big for subbuffers, report error, don't
+			 * complete the sub-buffer switch.
+			 */
+			v_inc(config, &buf->records_lost_big);
+			return -1;
+		} else {
+			/*
+			 * We just made a successful buffer switch and the
+			 * record fits in the new subbuffer. Let's write.
+			 */
+		}
+	} else {
+		/*
+		 * Record fits in the current buffer and we are not on a switch
+		 * boundary. It's safe to write.
+		 */
+	}
+	offsets->end = offsets->begin + offsets->size;
+
+	if (unlikely((subbuf_offset(offsets->end, chan)) == 0)) {
+		/*
+		 * The offset_end will fall at the very beginning of the next
+		 * subbuffer.
+		 */
+		offsets->switch_new_end = 1;	/* For offsets->begin */
+	}
+	return 0;
+}
+
+/**
+ * ring_buffer_reserve_slow - Atomic slot reservation in a buffer.
+ * @ctx: ring buffer context.
+ *
+ * Return : -ENOSPC if not enough space, else returns 0.
+ * It will take care of sub-buffer switching.
+ */
+int ring_buffer_reserve_slow(struct ring_buffer_ctx *ctx)
+{
+	struct channel *chan = ctx->chan;
+	const struct ring_buffer_config *config = chan->backend.config;
+	struct ring_buffer *buf;
+	struct switch_offsets offsets;
+
+	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
+		buf = per_cpu_ptr(chan->backend.buf, ctx->cpu);
+	else
+		buf = chan->backend.buf;
+	ctx->buf = buf;
+
+	offsets.size = 0;
+
+	do {
+		if (unlikely(ring_buffer_try_reserve_slow(buf, chan, &offsets,
+							  ctx)))
+			return -ENOSPC;
+	} while (unlikely(v_cmpxchg(config, &buf->offset, offsets.old,
+				    offsets.end)
+			  != offsets.old));
+
+	/*
+	 * Atomically update last_tsc. This update races against concurrent
+	 * atomic updates, but the race will always cause supplementary full TSC
+	 * records, never the opposite (missing a full TSC record when it would
+	 * be needed).
+	 */
+	save_last_tsc(config, buf, ctx->tsc);
+
+	/*
+	 * Push the reader if necessary
+	 */
+	ring_buffer_reserve_push_reader(buf, chan, offsets.end - 1);
+
+	/*
+	 * Clear noref flag for this subbuffer.
+	 */
+	ring_buffer_clear_noref(config, &buf->backend,
+				subbuf_index(offsets.end - 1, chan));
+
+	/*
+	 * Switch old subbuffer if needed.
+	 */
+	if (unlikely(offsets.switch_old_end)) {
+		ring_buffer_clear_noref(config, &buf->backend,
+					subbuf_index(offsets.old - 1, chan));
+		ring_buffer_switch_old_end(buf, chan, &offsets, ctx->tsc);
+	}
+
+	/*
+	 * Populate new subbuffer.
+	 */
+	if (unlikely(offsets.switch_new_start))
+		ring_buffer_switch_new_start(buf, chan, &offsets, ctx->tsc);
+
+	if (unlikely(offsets.switch_new_end))
+		ring_buffer_switch_new_end(buf, chan, &offsets, ctx->tsc);
+
+	ctx->slot_size = offsets.size;
+	ctx->pre_offset = offsets.begin;
+	ctx->buf_offset = offsets.begin + offsets.pre_header_padding;
+	return 0;
+}
+EXPORT_SYMBOL_GPL(ring_buffer_reserve_slow);

Index: linux.trees.git/include/linux/ringbuffer/api.h
===================================================================
@@ -0,0 +1,25 @@
+#ifndef _LINUX_RING_BUFFER_API_H
+#define _LINUX_RING_BUFFER_API_H
+
+/*
+ * linux/ringbuffer/api.h
+ *
+ * Copyright (C) 2010 - Mathieu Desnoyers <[email protected]>
+ *
+ * Ring Buffer API.
+ *
+ * Dual LGPL v2.1/GPL v2 license.
+ */
+
+#include <linux/ringbuffer/backend.h>
+#include <linux/ringbuffer/frontend.h>
+#include <linux/ringbuffer/vfs.h>
+
+/*
+ * ring_buffer_frontend_api.h contains static inline functions that depend on
+ * client static inlines. Hence the inclusion of this "api" header only
+ * within the client.
+ */
+#include <linux/ringbuffer/frontend_api.h>
+
+#endif /* _LINUX_RING_BUFFER_API_H */

309 lines not on this page · raw message

352 lines not on this page · raw message

424 lines not on this page · raw message

191 lines not on this page · raw message

158 lines not on this page · raw message

3 lines not on this page · raw message

190 lines not on this page · raw message

115 lines not on this page · raw message

57 lines not on this page · raw message

257 lines not on this page · raw message

273 lines not on this page · raw message

100 lines not on this page · raw message

Index: linux.trees.git/kernel/trace/lib_ring_buffer_benchmark.c
===================================================================
@@ -0,0 +1,658 @@
+/*
+ * ring buffer library tester and benchmark
+ *
+ * Copyright (C) 2009 Steven Rostedt <[email protected]>
+ * Copyright (C) 2010 Mathieu Desnoyers <[email protected]>
+ */
+
+#define RING_BUFFER_ALIGN
+
+#include <linux/ringbuffer/config.h>
+#include <linux/trace_clock.h>
+#include <linux/completion.h>
+#include <linux/kmemcheck.h>
+#include <linux/kthread.h>
+#include <linux/module.h>
+#include <linux/time.h>
+#include <asm/local.h>
+
+struct subbuffer_header {
+	u64		ts;
+	unsigned long	commit;
+	uint8_t		header_end[0];
+};
+
+/*
+ * Only 27-bit tsc support, TODO: extended header to support 27-bit overflow.
+ */
+struct event_header {
+	kmemcheck_bitfield_begin(bitfield);
+	u32		type_len:5, tsc:27;
+	kmemcheck_bitfield_end(bitfield);
+	uint8_t		header_end[0];
+};
+
+struct payload {
+	int cpuid;
+} RING_BUFFER_ALIGN_ATTR;
+
+static inline notrace u64 ring_buffer_clock_read(struct channel *chan)
+{
+	return trace_clock_local();
+}
+
+static inline notrace
+unsigned char record_header_size(const struct ring_buffer_config *config,
+				 struct channel *chan, size_t offset,
+				 size_t data_size, size_t *pre_header_padding,
+				 unsigned int rflags,
+				 struct ring_buffer_ctx *ctx)
+{
+	size_t orig_offset = offset;
+	size_t padding;
+
+	padding = ring_buffer_align(config, offset,
+				    offsetof(struct event_header, header_end));
+	offset += padding;
+	offset += sizeof(struct event_header);
+
+	*pre_header_padding = padding;
+	return offset - orig_offset;
+}
+
+#include <linux/ringbuffer/api.h>
+
+static struct channel *channel;
+static const struct ring_buffer_config client_config;
+
+static u64 client_ring_buffer_clock_read(struct channel *chan)
+{
+	return ring_buffer_clock_read(chan);
+}
+
+static
+size_t client_record_header_size(const struct ring_buffer_config *config,
+				struct channel *chan, size_t offset,
+				size_t data_size,
+				size_t *pre_header_padding,
+				unsigned int rflags,
+				struct ring_buffer_ctx *ctx)
+{
+	return record_header_size(config, chan, offset, data_size,
+				  pre_header_padding, rflags, ctx);
+}
+
+static size_t client_subbuffer_header_size(void)
+{
+	return offsetof(struct subbuffer_header, header_end);
+}
+
+static void client_buffer_begin(struct ring_buffer *buf, u64 tsc,
+				unsigned int subbuf_idx)
+{
+	struct channel *chan = buf->backend.chan;
+	struct subbuffer_header *header =
+		(struct subbuffer_header *)
+			ring_buffer_offset_address(&buf->backend,
+				subbuf_idx * chan->backend.subbuf_size);
+	header->ts = tsc;
+}
+
+static void client_buffer_end(struct ring_buffer *buf, u64 tsc,
+			    unsigned int subbuf_idx, unsigned long data_size)
+{
+	struct channel *chan = buf->backend.chan;
+	struct subbuffer_header *header =
+		(struct subbuffer_header *)
+			ring_buffer_offset_address(&buf->backend,
+				subbuf_idx * chan->backend.subbuf_size);
+	header->commit = data_size;
+}
+
+static const struct ring_buffer_config client_config = {
+	.cb.ring_buffer_clock_read = client_ring_buffer_clock_read,
+	.cb.record_header_size = client_record_header_size,
+	.cb.subbuffer_header_size = client_subbuffer_header_size,
+	.cb.buffer_begin = client_buffer_begin,
+	.cb.buffer_end = client_buffer_end,
+	.cb.buffer_create = NULL,
+	.cb.buffer_finalize = NULL,
+
+	.tsc_bits = 27,
+	.alloc = RING_BUFFER_ALLOC_PER_CPU,
+	/* .alloc = RING_BUFFER_ALLOC_GLOBAL, */
+	.sync = RING_BUFFER_SYNC_PER_CPU,
+	/* .sync = RING_BUFFER_SYNC_GLOBAL, */
+	.mode = RING_BUFFER_OVERWRITE,
+	/* .mode = RING_BUFFER_DISCARD, */
+	.align = RING_BUFFER_NATURAL,
+	.backend = RING_BUFFER_PAGE,
+	.output = RING_BUFFER_NONE,
+	.oops = RING_BUFFER_NO_OOPS_CONSISTENCY,
+	.ipi = RING_BUFFER_IPI_BARRIER,
+	.wakeup = RING_BUFFER_WAKEUP_BY_TIMER,
+};
+
+static void write_event_header(const struct ring_buffer_config *config,
+			       struct ring_buffer_ctx *ctx)
+{
+	struct event_header header;
+
+	header.type_len = ctx->data_size;
+	header.tsc = ctx->tsc;
+	/*
+	 * eventually check rflags to know if a 27 bit tsc overflow is detected
+	 * between consecutive events.
+	 */
+	ring_buffer_write(config, ctx, &header, sizeof(header));
+}
+
+/* run time and sleep time in seconds */
+#define RUN_TIME	10
+#define SLEEP_TIME	10
+
+#ifndef CONFIG_PREEMPT
+/* number of events for writer to give up the cpu */
+static int resched_interval = 5000;
+#endif
+
+static struct completion read_start;
+
+static struct task_struct *producer;
+static struct task_struct *consumer;
+static unsigned long iter_read;
+static unsigned long long global_read;
+
+static int disable_reader;
+module_param(disable_reader, uint, 0644);
+MODULE_PARM_DESC(disable_reader, "only run producer");
+
+static int write_iteration = 50;
+module_param(write_iteration, uint, 0644);
+MODULE_PARM_DESC(write_iteration, "# of writes between timestamp readings");
+
+static int producer_nice = 19;
+static int consumer_nice = 19;
+
+static int producer_fifo = -1;
+static int consumer_fifo = -1;
+
+module_param(producer_nice, uint, 0644);
+MODULE_PARM_DESC(producer_nice, "nice prio for producer");
+
+module_param(consumer_nice, uint, 0644);
+MODULE_PARM_DESC(consumer_nice, "nice prio for consumer");
+
+module_param(producer_fifo, uint, 0644);
+MODULE_PARM_DESC(producer_fifo, "fifo prio for producer");
+
+module_param(consumer_fifo, uint, 0644);
+MODULE_PARM_DESC(consumer_fifo, "fifo prio for consumer");
+
+static int kill_test;
+
+#define KILL_TEST()				\
+	do {					\
+		if (!kill_test) {		\
+			kill_test = 1;		\
+			WARN_ON(1);		\
+		}				\
+	} while (0)
+
+enum event_status {
+	EVENT_FOUND,
+	EVENT_DROPPED,
+};
+
+static int read_subbuffer(struct ring_buffer *buf, int cpu)
+{
+	unsigned long consumed, offset, data_size;
+	int ret;
+
+	ret = ring_buffer_get_subbuf(buf, &consumed);
+	if (ret && !ACCESS_ONCE(buf->finalized)
+	    && client_config.alloc == RING_BUFFER_ALLOC_GLOBAL) {
+		/*
+		 * Use "pull" scheme for global buffers. The reader
+		 * itself flushes the buffer to "pull" data not visible
+		 * to readers yet. Flush current subbuffer and re-try.
+		 *
+		 * Per-CPU buffers rather use a "push" scheme because
+		 * the IPI needed to flush all CPU's buffers is too
+		 * costly. In the "push" scheme, the reader waits for
+		 * the writer periodic deferrable timer to flush the
+		 * buffers (keeping track of a quiescent state
+		 * timestamp). Therefore, the writer "pushes" data out
+		 * of the buffers rather than letting the reader "pull"
+		 * data from the buffer.
+		 */
+		ring_buffer_switch(&client_config, buf, SWITCH_ACTIVE);
+		ret = ring_buffer_get_subbuf(buf, &consumed);
+	}
+	if (ret)
+		goto get_fail;
+
+	data_size = ring_buffer_get_read_data_size(&client_config, buf);
+	offset = consumed;
+
+	/* Skip header */
+	offset += client_subbuffer_header_size();
+
+	/* Read events */
+	while (offset < consumed + data_size) {
+		unsigned int cpuid;
+
+		offset += ring_buffer_align(&client_config, offset,
+					    offsetof(struct event_header,
+						     header_end));
+		offset += offsetof(struct event_header, header_end);
+		offset += ring_buffer_align(&client_config, offset,
+					    sizeof(cpuid));
+		ring_buffer_read(&buf->backend, offset, &cpuid, sizeof(cpuid));
+		if (client_config.alloc == RING_BUFFER_ALLOC_GLOBAL)
+			WARN_ON_ONCE(cpuid > num_online_cpus());
+		else
+			WARN_ON_ONCE(cpuid != cpu);
+		offset += sizeof(cpuid);
+		iter_read++;
+		global_read++;
+	}
+
+	/* Put subbuffer */
+	ring_buffer_put_subbuf(buf, consumed);
+get_fail:
+
+	return ret;
+}
+
+static int read_channel_events(void)
+{
+	int cpu;
+	int all_finalized = 1;
+	int all_empty = 1;
+
+	for_each_channel_cpu(cpu, channel) {
+		struct ring_buffer *buf;
+		int ret;
+
+		buf = channel_get_ring_buffer(&client_config, channel, cpu);
+		ret = read_subbuffer(buf, cpu);
+		if (ret != -ENODATA)
+			all_finalized = 0;
+		if (ret == 0)
+			all_empty = 0;
+	}
+
+	if (all_finalized)
+		return -ENODATA;
+	if (all_empty)
+		return -EAGAIN;
+	else
+		return 0;
+}
+
+static void ring_buffer_consumer(void)
+{
+	int ret;
+
+	do {
+		if (client_config.alloc == RING_BUFFER_ALLOC_GLOBAL) {
+			struct ring_buffer *buf;
+
+			buf = channel_get_ring_buffer(
+				&client_config, channel, 0);
+			ret = read_subbuffer(buf, 0);
+			if (ret == -EAGAIN)
+				wait_event_interruptible(channel->read_wait,
+					(ret = read_subbuffer(buf, 0),
+					 ret != -EAGAIN));
+		} else {
+			ret = read_channel_events();
+			if (ret == -EAGAIN)
+				wait_event_interruptible(channel->read_wait,
+					(ret = read_channel_events(),
+					 ret != -EAGAIN));
+		}
+	} while (!kill_test && ret != -ENODATA);
+}
+
+static void ring_buffer_producer(void)
+{
+	struct timeval start_tv;
+	struct timeval end_tv;
+	unsigned long long time;
+
+	unsigned long long hit = 0;
+	unsigned long long missed = 0;
+
+	unsigned long long written = 0;
+	unsigned long long lost_full = 0, lost_wrap = 0, lost_big = 0;
+	unsigned long long entries = 0;
+	unsigned long long overruns = 0;
+	unsigned long long read = 0;
+	unsigned long avg;
+	int cnt = 0;
+	int ret, cpu;
+	struct ring_buffer_ctx ctx;
+
+	/*
+	 * Hammer the buffer for 10 secs (this may
+	 * make the system stall)
+	 */
+	trace_printk("Starting ring buffer hammer\n");
+	do_gettimeofday(&start_tv);
+	do {
+		int i;
+
+		for (i = 0; i < write_iteration; i++) {
+			cpu = ring_buffer_get_cpu(&client_config);
+			if (cpu < 0)
+				continue;
+			ring_buffer_ctx_init(&ctx, channel, NULL,
+					     sizeof(struct payload),
+					     sizeof(cpuid), cpu);
+			ret = ring_buffer_reserve(&client_config, &ctx);
+			if (ret) {
+				missed++;
+			} else {
+				int cpuid = smp_processor_id();
+
+				hit++;
+				write_event_header(&client_config, &ctx);
+				ring_buffer_align_ctx(&client_config, &ctx,
+						      sizeof(cpuid));
+				ring_buffer_write(&client_config, &ctx,
+						  &cpuid, sizeof(cpuid));
+				ring_buffer_commit(&client_config, &ctx);
+			}
+			ring_buffer_put_cpu(&client_config);
+		}
+		do_gettimeofday(&end_tv);
+
+		cnt++;
+
+#ifndef CONFIG_PREEMPT
+		/*
+		 * If we are a non preempt kernel, the 10 second run will
+		 * stop everything while it runs. Instead, we will call
+		 * cond_resched and also add any time that was lost by a
+		 * rescedule.
+		 *
+		 * Do a cond resched at the same frequency we would wake up
+		 * the reader.
+		 */
+		if (cnt % resched_interval)
+			cond_resched();
+#endif
+
+	} while (end_tv.tv_sec < (start_tv.tv_sec + RUN_TIME) && !kill_test);
+	trace_printk("End ring buffer hammer\n");
+
+	time = end_tv.tv_sec - start_tv.tv_sec;
+	time *= USEC_PER_SEC;
+	time += (long long)((long)end_tv.tv_usec - (long)start_tv.tv_usec);
+
+	if (client_config.alloc == RING_BUFFER_ALLOC_GLOBAL) {
+		struct ring_buffer *buf =
+			channel_get_ring_buffer(&client_config,
+						channel, 0);
+
+		/*
+		 * These values only take into account flushed subbuffers.
+		 */
+		written += ring_buffer_get_records_count(&client_config, buf);
+		lost_full += ring_buffer_get_records_lost_full(&client_config, buf);
+		lost_wrap += ring_buffer_get_records_lost_wrap(&client_config, buf);
+		lost_big += ring_buffer_get_records_lost_big(&client_config, buf);
+		overruns += ring_buffer_get_records_overrun(&client_config, buf);
+		entries += ring_buffer_get_records_unread(&client_config, buf);
+		read += ring_buffer_get_records_read(&client_config, buf);
+	} else {
+		for_each_channel_cpu(cpu, channel) {
+			struct ring_buffer *buf =
+				channel_get_ring_buffer(&client_config,
+							channel, cpu);
+
+			/*
+			 * These values only take into account flushed
+			 * subbuffers.
+			 */
+			written += ring_buffer_get_records_count(&client_config, buf);
+			lost_full += ring_buffer_get_records_lost_full(&client_config, buf);
+			lost_wrap += ring_buffer_get_records_lost_wrap(&client_config, buf);
+			lost_big += ring_buffer_get_records_lost_big(&client_config, buf);
+			overruns += ring_buffer_get_records_overrun(&client_config, buf);
+			entries += ring_buffer_get_records_unread(&client_config, buf);
+			read += ring_buffer_get_records_read(&client_config, buf);
+		}
+	}
+
+	if (kill_test)
+		trace_printk("ERROR!\n");
+
+	if (!disable_reader) {
+		if (consumer_fifo < 0)
+			trace_printk("Running Consumer at nice: %d\n",
+				     consumer_nice);
+		else
+			trace_printk("Running Consumer at SCHED_FIFO %d\n",
+				     consumer_fifo);
+	}
+	if (producer_fifo < 0)
+		trace_printk("Running Producer at nice: %d\n",
+			     producer_nice);
+	else
+		trace_printk("Running Producer at SCHED_FIFO %d\n",
+			     producer_fifo);
+
+	/* Let the user know that the test is running at low priority */
+	if (producer_fifo < 0 && consumer_fifo < 0 &&
+	    producer_nice == 19 && consumer_nice == 19)
+		trace_printk("WARNING!!! This test is running at lowest priority.\n");
+
+	trace_printk("This iteration:           %llu (usecs)\n", time);
+	trace_printk("  Time:                   %llu (usecs)\n", time);
+	trace_printk("  Data production:\n");
+	trace_printk("    Written:              %llu\n", hit);
+	trace_printk("    Lost:                 %llu\n", missed);
+	trace_printk("  Data consumption:\n");
+	if (disable_reader)
+		trace_printk("    Read:                 (reader disabled)\n");
+	else
+		trace_printk("    Read:                 %lu\n",
+			     iter_read);
+	trace_printk("\n");
+	trace_printk("Global (only flushed subbuffers):\n");
+	trace_printk("  Data production:\n");
+	trace_printk("    Written:              %llu\n", written);
+	trace_printk("    Lost (buffer full)    %llu\n", lost_full);
+	trace_printk("    Lost (wrap around)    %llu\n", lost_wrap);
+	trace_printk("    Lost (event too big)  %llu\n", lost_big);
+	trace_printk("  Data consumption:\n");
+	if (disable_reader)
+		trace_printk("    Read:                 (reader disabled)\n");
+	else
+		trace_printk("    Read:                 %llu (%llu read-side)\n",
+			     read, global_read);
+	trace_printk("    Overruns:             %llu\n", overruns);
+	trace_printk("    Non-consumed entries: %llu\n", entries);
+	trace_printk("    Consumption total:    %llu\n",
+		     entries + overruns + read);
+
+	/* Convert time from usecs to millisecs */
+	do_div(time, USEC_PER_MSEC);
+	if (time)
+		hit /= (long)time;
+	else
+		trace_printk("TIME IS ZERO??\n");
+
+	trace_printk("Entries per millisec: %llu\n", hit);
+
+	if (hit) {
+		/* Calculate the average time in nanosecs */
+		avg = NSEC_PER_MSEC / hit;
+		trace_printk("%lu ns per entry written\n", avg);
+	}
+
+	if (missed) {
+		if (time)
+			missed /= (long)time;
+
+		trace_printk("Total iterations per millisec: %llu\n",
+			     hit + missed);
+
+		/* it is possible that hit + missed will overflow and be zero */
+		if (!(hit + missed)) {
+			trace_printk("hit + missed overflowed and totalled zero!\n");
+			hit--; /* make it non zero */
+		}
+
+		/* Caculate the average time in nanosecs */
+		avg = NSEC_PER_MSEC / (hit + missed);
+		trace_printk("%lu ns per entry (written+lost)\n", avg);
+	}
+	iter_read = 0;
+}
+
+static void wait_to_die(void)
+{
+	set_current_state(TASK_INTERRUPTIBLE);
+	while (!kthread_should_stop()) {
+		schedule();
+		set_current_state(TASK_INTERRUPTIBLE);
+	}
+	__set_current_state(TASK_RUNNING);
+}
+
+static int ring_buffer_consumer_thread(void *arg)
+{
+	struct ring_buffer *buf;
+	int cpu;
+
+	wait_for_completion_interruptible(&read_start);
+
+	if (client_config.alloc == RING_BUFFER_ALLOC_GLOBAL) {
+		buf = channel_get_ring_buffer(&client_config, channel, 0);
+		WARN_ON_ONCE(ring_buffer_open_read(buf));
+	} else {
+		/* TODO: benchmark does not take cpu hotplug into account. */
+		for_each_channel_cpu(cpu, channel) {
+			buf = channel_get_ring_buffer(&client_config, channel,
+						      cpu);
+			WARN_ON_ONCE(ring_buffer_open_read(buf));
+		}
+	}
+
+	ring_buffer_consumer();
+
+	if (client_config.alloc == RING_BUFFER_ALLOC_GLOBAL)
+		ring_buffer_release_read(buf);
+	else {
+		/* TODO: benchmark does not take cpu hotplug into account. */
+		for_each_channel_cpu(cpu, channel) {
+			buf = channel_get_ring_buffer(&client_config, channel,
+						      cpu);
+			ring_buffer_release_read(buf);
+		}
+	}
+
+	wait_to_die();
+
+	return 0;
+}
+
+static int ring_buffer_producer_thread(void *arg)
+{
+	if (consumer)
+		complete(&read_start);
+
+	while (!kthread_should_stop() && !kill_test) {
+		ring_buffer_producer();
+
+		trace_printk("Sleeping for 10 secs\n");
+		set_current_state(TASK_INTERRUPTIBLE);
+		schedule_timeout(HZ * SLEEP_TIME);
+		__set_current_state(TASK_RUNNING);
+	}
+
+	if (kill_test)
+		wait_to_die();
+
+	return 0;
+}
+
+static int __init ring_buffer_benchmark_init(void)
+{
+	int ret;
+
+	/* make a one meg buffer in overwite mode */
+	/* altern. ftrace equivalent: 4k subbuffers: 4096 * 256. */
+	channel = channel_create(&client_config, "benchmark", NULL, NULL,
+				 /* 4096, 256, */
+				 131072, 8,
+				 /* 524288, 2, */
+				 100000, 100000);
+	if (!channel)
+		return -EINVAL;
+
+	if (!disable_reader) {
+		init_completion(&read_start);
+		consumer = kthread_run(ring_buffer_consumer_thread,
+				       NULL, "rb_consumer");
+		ret = PTR_ERR(consumer);
+		if (IS_ERR(consumer))
+			goto out_fail;
+	}
+
+	producer = kthread_run(ring_buffer_producer_thread,
+			       NULL, "rb_producer");
+	ret = PTR_ERR(producer);
+
+	if (IS_ERR(producer))
+		goto out_kill;
+
+	/*
+	 * Run them as low-prio background tasks by default:
+	 */
+	if (!disable_reader) {
+		if (consumer_fifo >= 0) {
+			struct sched_param param = {
+				.sched_priority = consumer_fifo
+			};
+			sched_setscheduler(consumer, SCHED_FIFO, &param);
+		} else
+			set_user_nice(consumer, consumer_nice);
+	}
+
+	if (producer_fifo >= 0) {
+		struct sched_param param = {
+			.sched_priority = consumer_fifo
+		};
+		sched_setscheduler(producer, SCHED_FIFO, &param);
+	} else
+		set_user_nice(producer, producer_nice);
+
+	return 0;
+
+out_kill:
+	if (consumer)
+		kthread_kill_stop(consumer, SIGKILL);
+out_fail:
+	channel_destroy(channel);
+	return ret;
+}
+
+static void __exit ring_buffer_benchmark_exit(void)
+{
+	kthread_kill_stop(producer, SIGKILL);
+	channel_destroy(channel);
+	if (consumer)
+		kthread_kill_stop(consumer, SIGKILL);
+}
+
+module_init(ring_buffer_benchmark_init);
+module_exit(ring_buffer_benchmark_exit);
+
+MODULE_AUTHOR("Mathieu Desnoyers");
+MODULE_DESCRIPTION("lib_ring_buffer_benchmark");
+MODULE_LICENSE("GPL");

Index: linux.trees.git/lib/ringbuffer/ring_buffer_iterator.c
===================================================================
@@ -0,0 +1,795 @@
+/*
+ * ring_buffer_iterator.c
+ *
+ * (C) Copyright 2010 - Mathieu Desnoyers <[email protected]>
+ *
+ * Ring buffer and channel iterators. Get each event of a channel in order. Uses
+ * a prio heap for per-cpu buffers, giving a O(log(NR_CPUS)) algorithmic
+ * complexity for the "get next event" operation.
+ *
+ * Author:
+ *	Mathieu Desnoyers <[email protected]>
+ *
+ * Dual LGPL v2.1/GPL v2 license.
+ */
+
+#include <linux/ringbuffer/iterator.h>
+#include <linux/jiffies.h>
+#include <linux/module.h>
+
+/*
+ * Safety factor taking into account internal kernel interrupt latency.
+ * Assuming 250ms worse-case latency.
+ */
+#define MAX_SYSTEM_LATENCY	250
+
+/*
+ * Maximum delta expected between trace clocks. At most 1 jiffy delta.
+ */
+#define MAX_CLOCK_DELTA		(jiffies_to_usecs(1) * 1000)
+
+/**
+ * ring_buffer_get_next_record - Get the next record in a buffer.
+ * @chan: channel
+ * @buf: buffer
+ *
+ * Returns the size of the event read, -EAGAIN if buffer is empty, -ENODATA if
+ * buffer is empty and finalized. The buffer must already be opened for reading.
+ */
+ssize_t ring_buffer_get_next_record(struct channel *chan,
+				    struct ring_buffer *buf)
+{
+	const struct ring_buffer_config *config = chan->backend.config;
+	struct ring_buffer_iter *iter = &buf->iter;
+	int ret;
+
+restart:
+	switch (iter->state) {
+	case ITER_GET_SUBBUF:
+		ret = ring_buffer_get_subbuf(buf, &iter->consumed);
+		if (ret && !ACCESS_ONCE(buf->finalized)
+		    && config->alloc == RING_BUFFER_ALLOC_GLOBAL) {
+			/*
+			 * Use "pull" scheme for global buffers. The reader
+			 * itself flushes the buffer to "pull" data not visible
+			 * to readers yet. Flush current subbuffer and re-try.
+			 *
+			 * Per-CPU buffers rather use a "push" scheme because
+			 * the IPI needed to flush all CPU's buffers is too
+			 * costly. In the "push" scheme, the reader waits for
+			 * the writer periodic deferrable timer to flush the
+			 * buffers (keeping track of a quiescent state
+			 * timestamp). Therefore, the writer "pushes" data out
+			 * of the buffers rather than letting the reader "pull"
+			 * data from the buffer.
+			 */
+			ring_buffer_switch_slow(buf, SWITCH_ACTIVE);
+			ret = ring_buffer_get_subbuf(buf, &iter->consumed);
+		}
+		if (ret)
+			return ret;
+		iter->data_size = ring_buffer_get_read_data_size(config, buf);
+		iter->read_offset = iter->consumed;
+		/* skip header */
+		iter->read_offset += config->cb.subbuffer_header_size();
+		iter->state = ITER_TEST_RECORD;
+		goto restart;
+	case ITER_TEST_RECORD:
+		if (iter->read_offset - iter->consumed >= iter->data_size) {
+			iter->state = ITER_PUT_SUBBUF;
+		} else {
+			CHAN_WARN_ON(chan, !config->cb.record_get);
+			config->cb.record_get(config, chan, buf,
+					      iter->read_offset,
+					      &iter->header_len,
+					      &iter->payload_len,
+					      &iter->timestamp);
+			iter->read_offset += iter->header_len;
+			subbuffer_consume_record(config, &buf->backend);
+			iter->state = ITER_NEXT_RECORD;
+			return iter->payload_len;
+		}
+		goto restart;
+	case ITER_NEXT_RECORD:
+		iter->read_offset += iter->payload_len;
+		iter->state = ITER_TEST_RECORD;
+		goto restart;
+	case ITER_PUT_SUBBUF:
+		ring_buffer_put_subbuf(buf, iter->consumed);
+		iter->state = ITER_GET_SUBBUF;
+		goto restart;
+	default:
+		CHAN_WARN_ON(chan, 1);	/* Should not happen */
+		return -EPERM;
+	}
+}
+EXPORT_SYMBOL_GPL(ring_buffer_get_next_record);
+
+static int buf_is_higher(void *a, void *b)
+{
+	struct ring_buffer *bufa = a;
+	struct ring_buffer *bufb = b;
+
+	/* Consider lowest timestamps to be at the top of the heap */
+	return (bufa->iter.timestamp < bufb->iter.timestamp);
+}
+
+static
+void ring_buffer_get_empty_buf_records(const struct ring_buffer_config *config,
+				       struct channel *chan)
+{
+	struct ptr_heap *heap = &chan->iter.heap;
+	struct ring_buffer *buf, *tmp;
+	ssize_t len;
+
+	list_for_each_entry_safe(buf, tmp, &chan->iter.empty_head,
+				 iter.empty_node) {
+		len = ring_buffer_get_next_record(chan, buf);
+
+		/*
+		 * Deal with -EAGAIN and -ENODATA.
+		 * len >= 0 means record contains data.
+		 * -EBUSY should never happen, because we support only one
+		 * reader.
+		 */
+		switch (len) {
+		case -EAGAIN:
+			/* Keep node in empty list */
+			break;
+		case -ENODATA:
+			/*
+			 * Buffer is finalized. Don't add to list of empty
+			 * buffer, because it has no more data to provide, ever.
+			 */
+			list_del(&buf->iter.empty_node);
+			break;
+		case -EBUSY:
+			CHAN_WARN_ON(chan, 1);
+			break;
+		default:
+			/*
+			 * Insert buffer into the heap, remove from empty buffer
+			 * list. The heap should never overflow.
+			 */
+			CHAN_WARN_ON(chan, len < 0);
+			list_del(&buf->iter.empty_node);
+			CHAN_WARN_ON(chan, heap_insert(heap, buf) != NULL);
+		}
+	}
+}
+
+static
+void ring_buffer_wait_for_qs(const struct ring_buffer_config *config,
+			     struct channel *chan)
+{
+	u64 timestamp_qs;
+	unsigned long wait_msecs;
+
+	/*
+	 * No need to wait if no empty buffers are present.
+	 */
+	if (list_empty(&chan->iter.empty_head))
+		return;
+
+	timestamp_qs = config->cb.ring_buffer_clock_read(chan);
+	/*
+	 * We need to consider previously empty buffers.
+	 * Do a get next buf record on each of them. Add them to
+	 * the heap if they have data. If at least one of them
+	 * don't have data, we need to wait for
+	 * switch_timer_interval + MAX_SYSTEM_LATENCY (so we are sure the
+	 * buffers have been switched either by the timer or idle entry) and
+	 * check them again, adding them if they have data.
+	 */
+	ring_buffer_get_empty_buf_records(config, chan);
+
+	/*
+	 * No need to wait if no empty buffers are present.
+	 */
+	if (list_empty(&chan->iter.empty_head))
+		return;
+
+	/*
+	 * We need to wait for the buffer switch timer to run. If the
+	 * CPU is idle, idle entry performed the switch.
+	 * TODO: we could optimize further by skipping the sleep if all
+	 * empty buffers belong to idle or offline cpus.
+	 */
+	wait_msecs = jiffies_to_msecs(chan->switch_timer_interval);
+	wait_msecs += MAX_SYSTEM_LATENCY;
+	msleep(wait_msecs);
+	ring_buffer_get_empty_buf_records(config, chan);
+	/*
+	 * Any buffer still in the empty list here cannot possibly
+	 * contain an event with a timestamp prior to "timestamp_qs".
+	 * The new quiescent state timestamp is the one we grabbed
+	 * before waiting for buffer data.  It is therefore safe to
+	 * ignore empty buffers up to last_qs timestamp for fusion
+	 * merge.
+	 */
+	chan->iter.last_qs = timestamp_qs;
+}
+
+/**
+ * channel_get_next_record - Get the next record in a channel.
+ * @chan: channel
+ * @ret_buf: the buffer in which the event is located (output)
+ *
+ * Returns the size of new current event, -EAGAIN if all buffers are empty,
+ * -ENODATA if all buffers are empty and finalized. The channel must already be
+ * opened for reading.
+ */
+
+ssize_t channel_get_next_record(struct channel *chan,
+				struct ring_buffer **ret_buf)
+{
+	const struct ring_buffer_config *config = chan->backend.config;
+	struct ring_buffer *buf;
+	struct ptr_heap *heap;
+	ssize_t len;
+
+	if (config->alloc == RING_BUFFER_ALLOC_GLOBAL) {
+		*ret_buf = channel_get_ring_buffer(config, chan, 0);
+		return ring_buffer_get_next_record(chan, *ret_buf);
+	}
+
+	heap = &chan->iter.heap;
+
+	/*
+	 * get next record for topmost buffer.
+	 */
+	buf = heap_maximum(heap);
+	if (buf) {
+		len = ring_buffer_get_next_record(chan, buf);
+		/*
+		 * Deal with -EAGAIN and -ENODATA.
+		 * len >= 0 means record contains data.
+		 */
+		switch (len) {
+		case -EAGAIN:
+			buf->iter.timestamp = 0;
+			list_add(&buf->iter.empty_node, &chan->iter.empty_head);
+			/* Remove topmost buffer from the heap */
+			CHAN_WARN_ON(chan, heap_remove(heap) != buf);
+			break;
+		case -ENODATA:
+			/*
+			 * Buffer is finalized. Remove buffer from heap and
+			 * don't add to list of empty buffer, because it has no
+			 * more data to provide, ever.
+			 */
+			CHAN_WARN_ON(chan, heap_remove(heap) != buf);
+			break;
+		case -EBUSY:
+			CHAN_WARN_ON(chan, 1);
+			break;
+		default:
+			/*
+			 * Reinsert buffer into the heap. Note that heap can be
+			 * partially empty, so we need to use
+			 * heap_replace_max().
+			 */
+			CHAN_WARN_ON(chan, len < 0);
+			CHAN_WARN_ON(chan, heap_replace_max(heap, buf) != buf);
+			break;
+		}
+	}
+
+	buf = heap_maximum(heap);
+	if (!buf || buf->iter.timestamp > chan->iter.last_qs) {
+		/*
+		 * Deal with buffers previously showing no data.
+		 * Add buffers containing data to the heap, update
+		 * last_qs.
+		 */
+		ring_buffer_wait_for_qs(config, chan);
+	}
+
+	*ret_buf = buf = heap_maximum(heap);
+	if (buf) {
+		/*
+		 * If this warning triggers, you probably need to check your
+		 * system interrupt latency. Typical causes: too many printk()
+		 * output going to a serial console with interrupts off.
+		 * Allow for MAX_CLOCK_DELTA ns timestamp delta going backward.
+		 * Observed on SMP KVM setups with trace_clock().
+		 */
+		if (chan->iter.last_timestamp
+		    > (buf->iter.timestamp + MAX_CLOCK_DELTA)) {
+			printk(KERN_WARNING "ring_buffer: timestamps going "
+			       "backward. Last time %llu ns, cpu %d, "
+			       "current time %llu ns, cpu %d, "
+			       "delta %llu ns.\n",
+			       chan->iter.last_timestamp, chan->iter.last_cpu,
+			       buf->iter.timestamp, buf->backend.cpu,
+			       chan->iter.last_timestamp - buf->iter.timestamp);
+			CHAN_WARN_ON(chan, 1);
+		}
+		chan->iter.last_timestamp = buf->iter.timestamp;
+		chan->iter.last_cpu = buf->backend.cpu;
+		return buf->iter.payload_len;
+	} else {
+		/* Heap is empty */
+		if (list_empty(&chan->iter.empty_head))
+			return -ENODATA;	/* All buffers finalized */
+		else
+			return -EAGAIN;		/* Temporarily empty */
+	}
+}
+EXPORT_SYMBOL_GPL(channel_get_next_record);
+
+static
+void ring_buffer_iterator_init(struct channel *chan, struct ring_buffer *buf)
+{
+	if (buf->iter.allocated)
+		return;
+
+	buf->iter.allocated = 1;
+	if (chan->iter.read_open && !buf->iter.read_open) {
+		CHAN_WARN_ON(chan, ring_buffer_open_read(buf) != 0);
+		buf->iter.read_open = 1;
+	}
+
+	/* Add to list of buffers without any current record */
+	if (chan->backend.config->alloc == RING_BUFFER_ALLOC_PER_CPU)
+		list_add(&buf->iter.empty_node, &chan->iter.empty_head);
+}
+
+#ifdef CONFIG_HOTPLUG_CPU
+static
+int __cpuinit channel_iterator_cpu_hotplug(struct notifier_block *nb,
+					   unsigned long action,
+					   void *hcpu)
+{
+	unsigned int cpu = (unsigned long)hcpu;
+	struct channel *chan = container_of(nb, struct channel,
+					    hp_iter_notifier);
+	struct ring_buffer *buf = per_cpu_ptr(chan->backend.buf, cpu);
+	const struct ring_buffer_config *config = chan->backend.config;
+
+	if (!chan->hp_iter_enable)
+		return NOTIFY_DONE;
+
+	CHAN_WARN_ON(chan, config->alloc == RING_BUFFER_ALLOC_GLOBAL);
+
+	switch (action) {
+	case CPU_DOWN_FAILED:
+	case CPU_DOWN_FAILED_FROZEN:
+	case CPU_ONLINE:
+	case CPU_ONLINE_FROZEN:
+		ring_buffer_iterator_init(chan, buf);
+		return NOTIFY_OK;
+	default:
+		return NOTIFY_DONE;
+	}
+}
+#endif
+
+int channel_iterator_init(struct channel *chan)
+{
+	const struct ring_buffer_config *config = chan->backend.config;
+	struct ring_buffer *buf;
+
+	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
+		int cpu, ret;
+
+		INIT_LIST_HEAD(&chan->iter.empty_head);
+		ret = heap_init(&chan->iter.heap,
+				num_possible_cpus()
+				* sizeof(struct ring_buffer *),
+				GFP_KERNEL, buf_is_higher);
+		if (ret)
+			return ret;
+		/*
+		 * 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
+		chan->hp_iter_notifier.notifier_call =
+			channel_iterator_cpu_hotplug;
+		chan->hp_iter_notifier.priority = 10;
+		register_cpu_notifier(&chan->hp_iter_notifier);
+		get_online_cpus();
+		for_each_online_cpu(cpu) {
+			buf = per_cpu_ptr(chan->backend.buf, cpu);
+			ring_buffer_iterator_init(chan, buf);
+		}
+		chan->hp_iter_enable = 1;
+		put_online_cpus();
+#else
+		for_each_possible_cpu(cpu) {
+			buf = per_cpu_ptr(chan->backend.buf, cpu);
+			ring_buffer_iterator_init(chan, buf);
+		}
+#endif
+	} else {
+		buf = channel_get_ring_buffer(config, chan, 0);
+		ring_buffer_iterator_init(chan, buf);
+	}
+	return 0;
+}
+
+void channel_iterator_unregister_notifiers(struct channel *chan)
+{
+	const struct ring_buffer_config *config = chan->backend.config;
+
+	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
+		chan->hp_iter_enable = 0;
+		unregister_cpu_notifier(&chan->hp_iter_notifier);
+	}
+}
+
+void channel_iterator_free(struct channel *chan)
+{
+	const struct ring_buffer_config *config = chan->backend.config;
+
+	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
+		heap_free(&chan->iter.heap);
+}
+
+int ring_buffer_iterator_open(struct ring_buffer *buf)
+{
+	struct channel *chan = buf->backend.chan;
+	const struct ring_buffer_config *config = chan->backend.config;
+	CHAN_WARN_ON(chan, config->output != RING_BUFFER_ITERATOR);
+	return ring_buffer_open_read(buf);
+}
+EXPORT_SYMBOL_GPL(ring_buffer_iterator_open);
+
+/*
+ * Note: Iterators must not be mixed with other types of outputs, because an
+ * iterator can leave the buffer in "GET" state, which is not consistent with
+ * other types of output (mmap, splice, raw data read).
+ */
+void ring_buffer_iterator_release(struct ring_buffer *buf)
+{
+	ring_buffer_release_read(buf);
+}
+EXPORT_SYMBOL_GPL(ring_buffer_iterator_release);
+
+int channel_iterator_open(struct channel *chan)
+{
+	const struct ring_buffer_config *config = chan->backend.config;
+	struct ring_buffer *buf;
+	int ret = 0, cpu;
+
+	CHAN_WARN_ON(chan, config->output != RING_BUFFER_ITERATOR);
+
+	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
+		get_online_cpus();
+		/* Allow CPU hotplug to keep track of opened reader */
+		chan->iter.read_open = 1;
+		for_each_channel_cpu(cpu, chan) {
+			buf = channel_get_ring_buffer(config, chan, cpu);
+			ret = ring_buffer_iterator_open(buf);
+			if (ret)
+				goto error;
+			buf->iter.read_open = 1;
+		}
+		put_online_cpus();
+	} else {
+		buf = channel_get_ring_buffer(config, chan, 0);
+		ret = ring_buffer_iterator_open(buf);
+	}
+	return ret;
+error:
+	/* Error should always happen on CPU 0, hence no close is required. */
+	CHAN_WARN_ON(chan, cpu != 0);
+	put_online_cpus();
+	return ret;
+}
+EXPORT_SYMBOL_GPL(channel_iterator_open);
+
+void channel_iterator_release(struct channel *chan)
+{
+	const struct ring_buffer_config *config = chan->backend.config;
+	struct ring_buffer *buf;
+	int cpu;
+
+	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
+		get_online_cpus();
+		for_each_channel_cpu(cpu, chan) {
+			buf = channel_get_ring_buffer(config, chan, cpu);
+			if (buf->iter.read_open) {
+				ring_buffer_iterator_release(buf);
+				buf->iter.read_open = 0;
+			}
+		}
+		chan->iter.read_open = 0;
+		put_online_cpus();
+	} else {
+		buf = channel_get_ring_buffer(config, chan, 0);
+		ring_buffer_iterator_release(buf);
+	}
+}
+EXPORT_SYMBOL_GPL(channel_iterator_release);
+
+void ring_buffer_iterator_reset(struct ring_buffer *buf)
+{
+	struct channel *chan = buf->backend.chan;
+
+	if (buf->iter.state != ITER_GET_SUBBUF)
+		ring_buffer_put_subbuf(buf, buf->iter.consumed);
+	buf->iter.state = ITER_GET_SUBBUF;
+	/* Remove from heap (if present). */
+	if (heap_cherrypick(&chan->iter.heap, buf))
+		list_add(&buf->iter.empty_node, &chan->iter.empty_head);
+	buf->iter.timestamp = 0;
+	buf->iter.header_len = 0;
+	buf->iter.payload_len = 0;
+	buf->iter.consumed = 0;
+	buf->iter.read_offset = 0;
+	buf->iter.data_size = 0;
+	/* Don't reset allocated and read_open */
+}
+
+void channel_iterator_reset(struct channel *chan)
+{
+	const struct ring_buffer_config *config = chan->backend.config;
+	struct ring_buffer *buf;
+	int cpu;
+
+	/* Empty heap, put into empty_head */
+	while ((buf = heap_remove(&chan->iter.heap)) != NULL)
+		list_add(&buf->iter.empty_node, &chan->iter.empty_head);
+
+	for_each_channel_cpu(cpu, chan) {
+		buf = channel_get_ring_buffer(config, chan, cpu);
+		ring_buffer_iterator_reset(buf);
+	}
+	/* Don't reset read_open */
+	chan->iter.last_qs = 0;
+	chan->iter.last_timestamp = 0;
+	chan->iter.last_cpu = 0;
+	chan->iter.len_left = 0;
+}
+
+/*
+ * Ring buffer payload extraction read() implementation.
+ */
+static
+ssize_t channel_ring_buffer_file_read(struct file *filp,
+				      char __user *user_buf,
+				      size_t count,
+				      loff_t *ppos,
+				      struct channel *chan,
+				      struct ring_buffer *buf,
+				      int fusionmerge)
+{
+	const struct ring_buffer_config *config = chan->backend.config;
+	size_t read_count = 0, read_offset;
+	ssize_t len;
+
+	might_sleep();
+	if (!access_ok(VERIFY_WRITE, user_buf, count))
+		return -EFAULT;
+
+	/* Finish copy of previous record */
+	if (*ppos != 0) {
+		if (read_count < count) {
+			len = chan->iter.len_left;
+			read_offset = *ppos;
+			if (config->alloc == RING_BUFFER_ALLOC_PER_CPU
+			    && fusionmerge)
+				buf = heap_maximum(&chan->iter.heap);
+			CHAN_WARN_ON(chan, !buf);
+			goto skip_get_next;
+		}
+	}
+
+	while (read_count < count) {
+		size_t copy_len, space_left;
+
+		if (fusionmerge)
+			len = channel_get_next_record(chan, &buf);
+		else
+			len = ring_buffer_get_next_record(chan, buf);
+len_test:
+		if (len < 0) {
+			/*
+			 * Check if buffer is finalized (end of file).
+			 */
+			if (len == -ENODATA) {
+				/* A 0 read_count will tell about end of file */
+				goto nodata;
+			}
+			if (filp->f_flags & O_NONBLOCK) {
+				if (!read_count)
+					read_count = -EAGAIN;
+				goto nodata;
+			} else {
+				int error;
+
+				/*
+				 * No data available at the moment, return what
+				 * we got.
+				 */
+				if (read_count)
+					goto nodata;
+
+				/*
+				 * Wait for returned len to be >= 0 or -ENODATA.
+				 */
+				if (fusionmerge)
+					error = wait_event_interruptible(
+					  chan->read_wait,
+					  ((len = channel_get_next_record(chan,
+						&buf)), len != -EAGAIN));
+				else
+					error = wait_event_interruptible(
+					  buf->read_wait,
+					  ((len = ring_buffer_get_next_record(
+						  chan, buf)), len != -EAGAIN));
+				CHAN_WARN_ON(chan, len == -EBUSY);
+				if (error) {
+					read_count = error;
+					goto nodata;
+				}
+				CHAN_WARN_ON(chan, len < 0 && len != -ENODATA);
+				goto len_test;
+			}
+		}
+		read_offset = buf->iter.read_offset;
+skip_get_next:
+		space_left = count - read_count;
+		if (len <= space_left) {
+			copy_len = len;
+			chan->iter.len_left = 0;
+			*ppos = 0;
+		} else {
+			copy_len = space_left;
+			chan->iter.len_left = len - copy_len;
+			*ppos = read_offset + copy_len;
+		}
+		if (__ring_buffer_copy_to_user(&buf->backend, read_offset,
+					       &user_buf[read_count],
+					       copy_len)) {
+			/*
+			 * Leave the len_left and ppos values at their current
+			 * state, as we currently have a valid event to read.
+			 */
+			return -EFAULT;
+		}
+		read_count += copy_len;
+	};
+	return read_count;
+
+nodata:
+	*ppos = 0;
+	chan->iter.len_left = 0;
+	return read_count;
+}
+
+/**
+ * ring_buffer_sp_file_read - Read buffer record payload.
+ * @filp: file structure pointer.
+ * @buffer: user buffer to read data into.
+ * @count: number of bytes to read.
+ * @ppos: file read position.
+ *
+ * Returns a negative value on error, or the number of bytes read on success.
+ * ppos is used to save the position _within the current record_ between calls
+ * to read().
+ */
+static
+ssize_t ring_buffer_file_read(struct file *filp,
+			      char __user *user_buf,
+			      size_t count,
+			      loff_t *ppos)
+{
+	struct inode *inode = filp->f_dentry->d_inode;
+	struct ring_buffer *buf = inode->i_private;
+	struct channel *chan = buf->backend.chan;
+
+	return channel_ring_buffer_file_read(filp, user_buf, count, ppos,
+					     chan, buf, 0);
+}
+
+/**
+ * channel_file_read - Read channel record payload.
+ * @filp: file structure pointer.
+ * @buffer: user buffer to read data into.
+ * @count: number of bytes to read.
+ * @ppos: file read position.
+ *
+ * Returns a negative value on error, or the number of bytes read on success.
+ * ppos is used to save the position _within the current record_ between calls
+ * to read().
+ */
+static
+ssize_t channel_file_read(struct file *filp,
+			  char __user *user_buf,
+			  size_t count,
+			  loff_t *ppos)
+{
+	struct inode *inode = filp->f_dentry->d_inode;
+	struct channel *chan = inode->i_private;
+	const struct ring_buffer_config *config = chan->backend.config;
+
+	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
+		return channel_ring_buffer_file_read(filp, user_buf, count,
+						     ppos, chan, NULL, 1);
+	else {
+		struct ring_buffer *buf =
+			channel_get_ring_buffer(config, chan, 0);
+		return channel_ring_buffer_file_read(filp, user_buf, count,
+						     ppos, chan, buf, 0);
+	}
+}
+
+static
+int ring_buffer_file_open(struct inode *inode, struct file *file)
+{
+	struct ring_buffer *buf = inode->i_private;
+	int ret;
+
+	ret = ring_buffer_iterator_open(buf);
+	if (ret)
+		return ret;
+
+	file->private_data = buf;
+	ret = nonseekable_open(inode, file);
+	if (ret)
+		goto release_iter;
+	return 0;
+
+release_iter:
+	ring_buffer_iterator_release(buf);
+	return ret;
+}
+
+static
+int ring_buffer_file_release(struct inode *inode, struct file *file)
+{
+	struct ring_buffer *buf = inode->i_private;
+
+	ring_buffer_iterator_release(buf);
+	return 0;
+}
+
+static
+int channel_file_open(struct inode *inode, struct file *file)
+{
+	struct channel *chan = inode->i_private;
+	int ret;
+
+	ret = channel_iterator_open(chan);
+	if (ret)
+		return ret;
+
+	file->private_data = chan;
+	ret = nonseekable_open(inode, file);
+	if (ret)
+		goto release_iter;
+	return 0;
+
+release_iter:
+	channel_iterator_release(chan);
+	return ret;
+}
+
+static
+int channel_file_release(struct inode *inode, struct file *file)
+{
+	struct channel *chan = inode->i_private;
+
+	channel_iterator_release(chan);
+	return 0;
+}
+
+const struct file_operations channel_payload_file_operations = {
+	.open = channel_file_open,
+	.release = channel_file_release,
+	.read = channel_file_read,
+	.llseek = ring_buffer_no_llseek,
+};
+EXPORT_SYMBOL_GPL(channel_payload_file_operations);
+
+const struct file_operations ring_buffer_payload_file_operations = {
+	.open = ring_buffer_file_open,
+	.release = ring_buffer_file_release,
+	.read = ring_buffer_file_read,
+	.llseek = ring_buffer_no_llseek,
+};
+EXPORT_SYMBOL_GPL(ring_buffer_payload_file_operations);