From 14641debd03ba299bd06040cb62e0dbdef7fac81 Mon Sep 17 00:00:00 2001
From: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
Date: Mon, 27 Jun 2011 15:24:39 -0400
Subject: [PATCH] Port ring buffer to userspace, part 1

Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
---
 include/ust/core.h                            |  28 +-
 libprio_heap/lttng_prio_heap.c                | 206 -----
 libprio_heap/lttng_prio_heap.h                | 117 ---
 libringbuffer/Makefile.am                     |   6 +-
 libringbuffer/backend.h                       |  15 +-
 libringbuffer/backend_internal.h              |  17 +-
 libringbuffer/backend_types.h                 |   2 -
 libringbuffer/frontend.h                      |  13 +-
 libringbuffer/frontend_internal.h             |  21 +-
 libringbuffer/frontend_types.h                |  72 +-
 .../{ring_buffer_vfs.c => ring_buffer_abi.c}  |   0
 libringbuffer/ring_buffer_backend.c           | 114 +--
 libringbuffer/ring_buffer_frontend.c          |  20 +-
 libringbuffer/ring_buffer_iterator.c          | 793 ------------------
 libringbuffer/vatomic.h                       |  36 +-
 15 files changed, 129 insertions(+), 1331 deletions(-)
 delete mode 100644 libprio_heap/lttng_prio_heap.c
 delete mode 100644 libprio_heap/lttng_prio_heap.h
 rename libringbuffer/{ring_buffer_vfs.c => ring_buffer_abi.c} (100%)
 delete mode 100644 libringbuffer/ring_buffer_iterator.c

diff --git a/include/ust/core.h b/include/ust/core.h
index eee405d..8c1c490 100644
--- a/include/ust/core.h
+++ b/include/ust/core.h
@@ -82,7 +82,25 @@ static inline long IS_ERR(const void *ptr)
 
 /* MALLOCATION */
 
-#define zmalloc(s) calloc(1, s)
+#include <stdlib.h>
+
+static inline
+void *zmalloc(size_t len)
+{
+	return calloc(1, len);
+}
+
+static inline
+void *malloc_align(size_t len)
+{
+	return malloc(ALIGN(len, CAA_CACHE_LINE_SIZE));
+}
+
+static inline
+void *zmalloc_align(size_t len)
+{
+	return calloc(1, ALIGN(len, CAA_CACHE_LINE_SIZE));
+}
 
 /* MATH */
 
@@ -110,4 +128,12 @@ static __inline__ int get_count_order(unsigned int count)
         const typeof( ((type *)0)->member ) *__mptr = (ptr);    \
         (type *)( (char *)__mptr - offsetof(type,member) );})
 
+#ifndef inline_memcpy
+#define inline_memcpy memcpy
+#endif
+
+#ifndef __same_type
+#define __same_type(a, b) __builtin_types_compatible_p(typeof(a), typeof(b))
+#endif
+
 #endif /* UST_CORE_H */
diff --git a/libprio_heap/lttng_prio_heap.c b/libprio_heap/lttng_prio_heap.c
deleted file mode 100644
index 5bbd079..0000000
--- a/libprio_heap/lttng_prio_heap.c
+++ /dev/null
@@ -1,206 +0,0 @@
-/*
- * lttng_prio_heap.c
- *
- * Priority heap containing pointers. Based on CLRS, chapter 6.
- *
- * Copyright 2011 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
- *
- * Permission is hereby granted, free of charge, to any person obtaining a copy
- * of this software and associated documentation files (the "Software"), to deal
- * in the Software without restriction, including without limitation the rights
- * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
- * copies of the Software, and to permit persons to whom the Software is
- * furnished to do so, subject to the following conditions:
- *
- * The above copyright notice and this permission notice shall be included in
- * all copies or substantial portions of the Software.
- */
-
-#include <linux/slab.h>
-#include "lttng_prio_heap.h"
-
-#ifdef DEBUG_HEAP
-void lttng_check_heap(const struct lttng_ptr_heap *heap)
-{
-	size_t i;
-
-	if (!heap->len)
-		return;
-
-	for (i = 1; i < heap->len; i++)
-		WARN_ON_ONCE(!heap->gt(heap->ptrs[i], heap->ptrs[0]));
-}
-#endif
-
-static
-size_t parent(size_t i)
-{
-	return (i -1) >> 1;
-}
-
-static
-size_t left(size_t i)
-{
-	return (i << 1) + 1;
-}
-
-static
-size_t right(size_t i)
-{
-	return (i << 1) + 2;
-}
-
-/*
- * Copy of heap->ptrs pointer is invalid after heap_grow.
- */
-static
-int heap_grow(struct lttng_ptr_heap *heap, size_t new_len)
-{
-	void **new_ptrs;
-
-	if (heap->alloc_len >= new_len)
-		return 0;
-
-	heap->alloc_len = max_t(size_t, new_len, heap->alloc_len << 1);
-	new_ptrs = kmalloc(heap->alloc_len * sizeof(void *), heap->gfpmask);
-	if (!new_ptrs)
-		return -ENOMEM;
-	if (heap->ptrs)
-		memcpy(new_ptrs, heap->ptrs, heap->len * sizeof(void *));
-	kfree(heap->ptrs);
-	heap->ptrs = new_ptrs;
-	return 0;
-}
-
-static
-int heap_set_len(struct lttng_ptr_heap *heap, size_t new_len)
-{
-	int ret;
-
-	ret = heap_grow(heap, new_len);
-	if (ret)
-		return ret;
-	heap->len = new_len;
-	return 0;
-}
-
-int lttng_heap_init(struct lttng_ptr_heap *heap, size_t alloc_len,
-	      gfp_t gfpmask, int gt(void *a, void *b))
-{
-	heap->ptrs = NULL;
-	heap->len = 0;
-	heap->alloc_len = 0;
-	heap->gt = gt;
-	/*
-	 * Minimum size allocated is 1 entry to ensure memory allocation
-	 * never fails within heap_replace_max.
-	 */
-	return heap_grow(heap, max_t(size_t, 1, alloc_len));
-}
-
-void lttng_heap_free(struct lttng_ptr_heap *heap)
-{
-	kfree(heap->ptrs);
-}
-
-static void heapify(struct lttng_ptr_heap *heap, size_t i)
-{
-	void **ptrs = heap->ptrs;
-	size_t l, r, largest;
-
-	for (;;) {
-		void *tmp;
-
-		l = left(i);
-		r = right(i);
-		if (l < heap->len && heap->gt(ptrs[l], ptrs[i]))
-			largest = l;
-		else
-			largest = i;
-		if (r < heap->len && heap->gt(ptrs[r], ptrs[largest]))
-			largest = r;
-		if (largest == i)
-			break;
-		tmp = ptrs[i];
-		ptrs[i] = ptrs[largest];
-		ptrs[largest] = tmp;
-		i = largest;
-	}
-	lttng_check_heap(heap);
-}
-
-void *lttng_heap_replace_max(struct lttng_ptr_heap *heap, void *p)
-{
-	void *res;
-
-	if (!heap->len) {
-		(void) heap_set_len(heap, 1);
-		heap->ptrs[0] = p;
-		lttng_check_heap(heap);
-		return NULL;
-	}
-
-	/* Replace the current max and heapify */
-	res = heap->ptrs[0];
-	heap->ptrs[0] = p;
-	heapify(heap, 0);
-	return res;
-}
-
-int lttng_heap_insert(struct lttng_ptr_heap *heap, void *p)
-{
-	void **ptrs;
-	size_t pos;
-	int ret;
-
-	ret = heap_set_len(heap, heap->len + 1);
-	if (ret)
-		return ret;
-	ptrs = heap->ptrs;
-	pos = heap->len - 1;
-	while (pos > 0 && heap->gt(p, ptrs[parent(pos)])) {
-		/* Move parent down until we find the right spot */
-		ptrs[pos] = ptrs[parent(pos)];
-		pos = parent(pos);
-	}
-	ptrs[pos] = p;
-	lttng_check_heap(heap);
-	return 0;
-}
-
-void *lttng_heap_remove(struct lttng_ptr_heap *heap)
-{
-	switch (heap->len) {
-	case 0:
-		return NULL;
-	case 1:
-		(void) heap_set_len(heap, 0);
-		return heap->ptrs[0];
-	}
-	/* Shrink, replace the current max by previous last entry and heapify */
-	heap_set_len(heap, heap->len - 1);
-	/* len changed. previous last entry is at heap->len */
-	return lttng_heap_replace_max(heap, heap->ptrs[heap->len]);
-}
-
-void *lttng_heap_cherrypick(struct lttng_ptr_heap *heap, void *p)
-{
-	size_t pos, len = heap->len;
-
-	for (pos = 0; pos < len; pos++)
-		if (heap->ptrs[pos] == p)
-			goto found;
-	return NULL;
-found:
-	if (heap->len == 1) {
-		(void) heap_set_len(heap, 0);
-		lttng_check_heap(heap);
-		return heap->ptrs[0];
-	}
-	/* Replace p with previous last entry and heapify. */
-	heap_set_len(heap, heap->len - 1);
-	/* len changed. previous last entry is at heap->len */
-	heap->ptrs[pos] = heap->ptrs[heap->len];
-	heapify(heap, pos);
-	return p;
-}
diff --git a/libprio_heap/lttng_prio_heap.h b/libprio_heap/lttng_prio_heap.h
deleted file mode 100644
index ea8dbb8..0000000
--- a/libprio_heap/lttng_prio_heap.h
+++ /dev/null
@@ -1,117 +0,0 @@
-#ifndef _LTTNG_PRIO_HEAP_H
-#define _LTTNG_PRIO_HEAP_H
-
-/*
- * lttng_prio_heap.h
- *
- * Priority heap containing pointers. Based on CLRS, chapter 6.
- *
- * Copyright 2011 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
- *
- * Permission is hereby granted, free of charge, to any person obtaining a copy
- * of this software and associated documentation files (the "Software"), to deal
- * in the Software without restriction, including without limitation the rights
- * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
- * copies of the Software, and to permit persons to whom the Software is
- * furnished to do so, subject to the following conditions:
- *
- * The above copyright notice and this permission notice shall be included in
- * all copies or substantial portions of the Software.
- */
-
-#include <linux/gfp.h>
-
-struct lttng_ptr_heap {
-	size_t len, alloc_len;
-	void **ptrs;
-	int (*gt)(void *a, void *b);
-	gfp_t gfpmask;
-};
-
-#ifdef DEBUG_HEAP
-void lttng_check_heap(const struct lttng_ptr_heap *heap);
-#else
-static inline
-void lttng_check_heap(const struct lttng_ptr_heap *heap)
-{
-}
-#endif
-
-/**
- * lttng_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 *lttng_heap_maximum(const struct lttng_ptr_heap *heap)
-{
-	lttng_check_heap(heap);
-	return heap->len ? heap->ptrs[0] : NULL;
-}
-
-/**
- * lttng_heap_init - initialize the heap
- * @heap: the heap to initialize
- * @alloc_len: number of elements initially allocated
- * @gfp: allocation flags
- * @gt: function to compare the elements
- *
- * Returns -ENOMEM if out of memory.
- */
-extern int lttng_heap_init(struct lttng_ptr_heap *heap,
-		     size_t alloc_len, gfp_t gfpmask,
-		     int gt(void *a, void *b));
-
-/**
- * lttng_heap_free - free the heap
- * @heap: the heap to free
- */
-extern void lttng_heap_free(struct lttng_ptr_heap *heap);
-
-/**
- * lttng_heap_insert - insert an element into the heap
- * @heap: the heap to be operated on
- * @p: the element to add
- *
- * Insert an element into the heap.
- *
- * Returns -ENOMEM if out of memory.
- */
-extern int lttng_heap_insert(struct lttng_ptr_heap *heap, void *p);
-
-/**
- * lttng_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 *lttng_heap_remove(struct lttng_ptr_heap *heap);
-
-/**
- * lttng_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 *lttng_heap_cherrypick(struct lttng_ptr_heap *heap, void *p);
-
-/**
- * lttng_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. It never allocates memory.
- */
-extern void *lttng_heap_replace_max(struct lttng_ptr_heap *heap, void *p);
-
-#endif /* _LTTNG_PRIO_HEAP_H */
diff --git a/libringbuffer/Makefile.am b/libringbuffer/Makefile.am
index bc2c7a7..2266053 100644
--- a/libringbuffer/Makefile.am
+++ b/libringbuffer/Makefile.am
@@ -6,11 +6,7 @@ lib_LTLIBRARIES = libringbuffer.la
 libringbuffer_la_SOURCES = \
 	ring_buffer_backend.c \
 	ring_buffer_frontend.c \
-	ring_buffer_iterator.c \
-	ring_buffer_vfs.c \
-	ring_buffer_splice.c \
-	ring_buffer_mmap.c \
-	../libprio_heap/lttng_prio_heap.c
+	ring_buffer_abi.c
 
 libringbuffer_la_LDFLAGS = -no-undefined -version-info 0:0:0
 
diff --git a/libringbuffer/backend.h b/libringbuffer/backend.h
index b50bae1..61d2f32 100644
--- a/libringbuffer/backend.h
+++ b/libringbuffer/backend.h
@@ -14,6 +14,10 @@
  * the reader in flight recorder mode.
  */
 
+#include <unistd.h>
+
+#include "ust/core.h"
+
 /* Internal helpers */
 #include "backend_internal.h"
 #include "frontend_internal.h"
@@ -25,10 +29,6 @@
 extern size_t lib_ring_buffer_read(struct lib_ring_buffer_backend *bufb,
 				   size_t offset, void *dest, size_t len);
 
-extern int __lib_ring_buffer_copy_to_user(struct lib_ring_buffer_backend *bufb,
-					  size_t offset, void __user *dest,
-					  size_t len);
-
 extern int lib_ring_buffer_read_cstr(struct lib_ring_buffer_backend *bufb,
 				     size_t offset, void *dest, size_t len);
 
@@ -76,7 +76,7 @@ void lib_ring_buffer_write(const struct lib_ring_buffer_config *config,
 
 	offset &= chanb->buf_size - 1;
 	sbidx = offset >> chanb->subbuf_size_order;
-	index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
+	index = (offset & (chanb->subbuf_size - 1)) >> get_count_order(PAGE_SIZE);
 	pagecpy = min_t(size_t, len, (-offset) & ~PAGE_MASK);
 	id = bufb->buf_wsb[sbidx].id;
 	sb_bindex = subbuffer_id_get_index(config, id);
@@ -124,9 +124,4 @@ unsigned long lib_ring_buffer_get_records_unread(
 	return records_unread;
 }
 
-ssize_t lib_ring_buffer_file_splice_read(struct file *in, loff_t *ppos,
-					 struct pipe_inode_info *pipe,
-					 size_t len, unsigned int flags);
-loff_t lib_ring_buffer_no_llseek(struct file *file, loff_t offset, int origin);
-
 #endif /* _LINUX_RING_BUFFER_BACKEND_H */
diff --git a/libringbuffer/backend_internal.h b/libringbuffer/backend_internal.h
index 2c3de67..c5f3362 100644
--- a/libringbuffer/backend_internal.h
+++ b/libringbuffer/backend_internal.h
@@ -11,6 +11,9 @@
  * Dual LGPL v2.1/GPL v2 license.
  */
 
+#include <unistd.h>
+#include <urcu/compiler.h>
+
 #include "config.h"
 #include "backend_types.h"
 #include "frontend_types.h"
@@ -52,7 +55,7 @@ extern void _lib_ring_buffer_write(struct lib_ring_buffer_backend *bufb,
  * sampling and subbuffer ID exchange).
  */
 
-#define HALF_ULONG_BITS		(BITS_PER_LONG >> 1)
+#define HALF_ULONG_BITS		(CAA_BITS_PER_LONG >> 1)
 
 #define SB_ID_OFFSET_SHIFT	(HALF_ULONG_BITS + 1)
 #define SB_ID_OFFSET_COUNT	(1UL << SB_ID_OFFSET_SHIFT)
@@ -145,7 +148,7 @@ void subbuffer_id_set_noref_offset(const struct lib_ring_buffer_config *config,
 		tmp |= offset << SB_ID_OFFSET_SHIFT;
 		tmp |= SB_ID_NOREF_MASK;
 		/* Volatile store, read concurrently by readers. */
-		ACCESS_ONCE(*id) = tmp;
+		CMM_ACCESS_ONCE(*id) = tmp;
 	}
 }
 
@@ -300,7 +303,7 @@ void lib_ring_buffer_clear_noref(const struct lib_ring_buffer_config *config,
 	 * Performing a volatile access to read the sb_pages, because we want to
 	 * read a coherent version of the pointer and the associated noref flag.
 	 */
-	id = ACCESS_ONCE(bufb->buf_wsb[idx].id);
+	id = CMM_ACCESS_ONCE(bufb->buf_wsb[idx].id);
 	for (;;) {
 		/* This check is called on the fast path for each record. */
 		if (likely(!subbuffer_id_is_noref(config, id))) {
@@ -314,7 +317,7 @@ void lib_ring_buffer_clear_noref(const struct lib_ring_buffer_config *config,
 		}
 		new_id = id;
 		subbuffer_id_clear_noref(config, &new_id);
-		new_id = cmpxchg(&bufb->buf_wsb[idx].id, id, new_id);
+		new_id = uatomic_cmpxchg(&bufb->buf_wsb[idx].id, id, new_id);
 		if (likely(new_id == id))
 			break;
 		id = new_id;
@@ -350,7 +353,7 @@ void lib_ring_buffer_set_noref_offset(const struct lib_ring_buffer_config *confi
 	 * Memory barrier that ensures counter stores are ordered before set
 	 * noref and offset.
 	 */
-	smp_mb();
+	cmm_smp_mb();
 	subbuffer_id_set_noref_offset(config, &bufb->buf_wsb[idx].id, offset);
 }
 
@@ -369,7 +372,7 @@ int update_read_sb_index(const struct lib_ring_buffer_config *config,
 	if (config->mode == RING_BUFFER_OVERWRITE) {
 		/*
 		 * Exchange the target writer subbuffer with our own unused
-		 * subbuffer. No need to use ACCESS_ONCE() here to read the
+		 * subbuffer. No need to use CMM_ACCESS_ONCE() here to read the
 		 * old_wpage, because the value read will be confirmed by the
 		 * following cmpxchg().
 		 */
@@ -387,7 +390,7 @@ int update_read_sb_index(const struct lib_ring_buffer_config *config,
 			     !subbuffer_id_is_noref(config, bufb->buf_rsb.id));
 		subbuffer_id_set_noref_offset(config, &bufb->buf_rsb.id,
 					      consumed_count);
-		new_id = cmpxchg(&bufb->buf_wsb[consumed_idx].id, old_id,
+		new_id = uatomic_cmpxchg(&bufb->buf_wsb[consumed_idx].id, old_id,
 				 bufb->buf_rsb.id);
 		if (unlikely(old_id != new_id))
 			return -EAGAIN;
diff --git a/libringbuffer/backend_types.h b/libringbuffer/backend_types.h
index 99de9f1..cfbe59c 100644
--- a/libringbuffer/backend_types.h
+++ b/libringbuffer/backend_types.h
@@ -68,9 +68,7 @@ struct channel_backend {
 	unsigned long num_subbuf;	/* Number of sub-buffers for writer */
 	u64 start_tsc;			/* Channel creation TSC value */
 	void *priv;			/* Client-specific information */
-	struct notifier_block cpu_hp_notifier;	 /* CPU hotplug notifier */
 	const struct lib_ring_buffer_config *config; /* Ring buffer configuration */
-	cpumask_var_t cpumask;		/* Allocated per-cpu buffers cpumask */
 	char name[NAME_MAX];		/* Channel name */
 };
 
diff --git a/libringbuffer/frontend.h b/libringbuffer/frontend.h
index 1eb4ae9..9d73da1 100644
--- a/libringbuffer/frontend.h
+++ b/libringbuffer/frontend.h
@@ -16,6 +16,9 @@
  * Dual LGPL v2.1/GPL v2 license.
  */
 
+#include <urcu/compiler.h>
+#include <urcu/uatomic.h>
+
 /* Internal helpers */
 #include "frontend_internal.h"
 
@@ -60,7 +63,7 @@ void *channel_destroy(struct channel *chan);
 #define for_each_channel_cpu(cpu, chan)					\
 	for ((cpu) = -1;						\
 		({ (cpu) = cpumask_next(cpu, (chan)->backend.cpumask);	\
-		   smp_read_barrier_depends(); (cpu) < nr_cpu_ids; });)
+		   cmm_smp_read_barrier_depends(); (cpu) < nr_cpu_ids; });)
 
 extern struct lib_ring_buffer *channel_get_ring_buffer(
 				const struct lib_ring_buffer_config *config,
@@ -118,7 +121,7 @@ static inline
 unsigned long lib_ring_buffer_get_consumed(const struct lib_ring_buffer_config *config,
 					   struct lib_ring_buffer *buf)
 {
-	return atomic_long_read(&buf->consumed);
+	return uatomic_read(&buf->consumed);
 }
 
 /*
@@ -129,11 +132,11 @@ static inline
 int lib_ring_buffer_is_finalized(const struct lib_ring_buffer_config *config,
 				 struct lib_ring_buffer *buf)
 {
-	int finalized = ACCESS_ONCE(buf->finalized);
+	int finalized = CMM_ACCESS_ONCE(buf->finalized);
 	/*
 	 * Read finalized before counters.
 	 */
-	smp_rmb();
+	cmm_smp_rmb();
 	return finalized;
 }
 
@@ -146,7 +149,7 @@ int lib_ring_buffer_channel_is_finalized(const struct channel *chan)
 static inline
 int lib_ring_buffer_channel_is_disabled(const struct channel *chan)
 {
-	return atomic_read(&chan->record_disabled);
+	return uatomic_read(&chan->record_disabled);
 }
 
 static inline
diff --git a/libringbuffer/frontend_internal.h b/libringbuffer/frontend_internal.h
index 37e046b..f758a68 100644
--- a/libringbuffer/frontend_internal.h
+++ b/libringbuffer/frontend_internal.h
@@ -16,10 +16,11 @@
  * Dual LGPL v2.1/GPL v2 license.
  */
 
+#include <urcu/compiler.h>
+
 #include "config.h"
 #include "backend_types.h"
 #include "frontend_types.h"
-#include "../lib_prio_heap/lttng_prio_heap.h"	/* For per-CPU read-side iterator */
 
 /* Buffer offset macros */
 
@@ -81,7 +82,7 @@ unsigned long subbuf_index(unsigned long offset, struct channel *chan)
  * last_tsc atomically.
  */
 
-#if (BITS_PER_LONG == 32)
+#if (CAA_BITS_PER_LONG == 32)
 static inline
 void save_last_tsc(const struct lib_ring_buffer_config *config,
 		   struct lib_ring_buffer *buf, u64 tsc)
@@ -154,7 +155,7 @@ void lib_ring_buffer_reserve_push_reader(struct lib_ring_buffer *buf,
 	unsigned long consumed_old, consumed_new;
 
 	do {
-		consumed_old = atomic_long_read(&buf->consumed);
+		consumed_old = uatomic_read(&buf->consumed);
 		/*
 		 * If buffer is in overwrite mode, push the reader consumed
 		 * count if the write position has reached it and we are not
@@ -170,7 +171,7 @@ void lib_ring_buffer_reserve_push_reader(struct lib_ring_buffer *buf,
 			consumed_new = subbuf_align(consumed_old, chan);
 		else
 			return;
-	} while (unlikely(atomic_long_cmpxchg(&buf->consumed, consumed_old,
+	} while (unlikely(uatomic_cmpxchg(&buf->consumed, consumed_old,
 					      consumed_new) != consumed_old));
 }
 
@@ -191,7 +192,7 @@ int lib_ring_buffer_poll_deliver(const struct lib_ring_buffer_config *config,
 {
 	unsigned long consumed_old, consumed_idx, commit_count, write_offset;
 
-	consumed_old = atomic_long_read(&buf->consumed);
+	consumed_old = uatomic_read(&buf->consumed);
 	consumed_idx = subbuf_index(consumed_old, chan);
 	commit_count = v_read(config, &buf->commit_cold[consumed_idx].cc_sb);
 	/*
@@ -354,7 +355,7 @@ void lib_ring_buffer_check_deliver(const struct lib_ring_buffer_config *config,
 			 * respect to writers coming into the subbuffer after
 			 * wrap around, and also order wrt concurrent readers.
 			 */
-			smp_mb();
+			cmm_smp_mb();
 			/* End of exclusive subbuffer access */
 			v_set(config, &buf->commit_cold[idx].cc_sb,
 			      commit_count);
@@ -365,10 +366,10 @@ void lib_ring_buffer_check_deliver(const struct lib_ring_buffer_config *config,
 			 * RING_BUFFER_WAKEUP_BY_WRITER wakeup is not lock-free.
 			 */
 			if (config->wakeup == RING_BUFFER_WAKEUP_BY_WRITER
-			    && atomic_long_read(&buf->active_readers)
+			    && uatomic_read(&buf->active_readers)
 			    && lib_ring_buffer_poll_deliver(config, buf, chan)) {
-				wake_up_interruptible(&buf->read_wait);
-				wake_up_interruptible(&chan->read_wait);
+				//wake_up_interruptible(&buf->read_wait);
+				//wake_up_interruptible(&chan->read_wait);
 			}
 
 		}
@@ -419,6 +420,6 @@ extern int lib_ring_buffer_create(struct lib_ring_buffer *buf,
 extern void lib_ring_buffer_free(struct lib_ring_buffer *buf);
 
 /* Keep track of trap nesting inside ring buffer code */
-DECLARE_PER_CPU(unsigned int, lib_ring_buffer_nesting);
+extern __thread unsigned int lib_ring_buffer_nesting;
 
 #endif /* _LINUX_RING_BUFFER_FRONTEND_INTERNAL_H */
diff --git a/libringbuffer/frontend_types.h b/libringbuffer/frontend_types.h
index f89d803..c9f98cb 100644
--- a/libringbuffer/frontend_types.h
+++ b/libringbuffer/frontend_types.h
@@ -16,9 +16,15 @@
  * Dual LGPL v2.1/GPL v2 license.
  */
 
+#include <urcu/list.h>
+#include <urcu/uatomic.h>
+#include <urcu/ref.h>
+
+#include "ust/core.h"
+
+#include "usterr_signal_safe.h"
 #include "config.h"
 #include "backend_types.h"
-#include "../lib_prio_heap/lttng_prio_heap.h"	/* For per-CPU read-side iterator */
 
 /*
  * A switch is done during tracing or as a final flush after tracing (so it
@@ -26,24 +32,9 @@
  */
 enum switch_mode { SWITCH_ACTIVE, SWITCH_FLUSH };
 
-/* channel-level read-side iterator */
-struct channel_iter {
-	/* Prio heap of buffers. Lowest timestamps at the top. */
-	struct lttng_ptr_heap heap;	/* Heap of struct lib_ring_buffer ptrs */
-	struct list_head empty_head;	/* Empty buffers linked-list head */
-	int read_open;			/* Opened for reading ? */
-	u64 last_qs;			/* Last quiescent state timestamp */
-	u64 last_timestamp;		/* Last timestamp (for WARN_ON) */
-	int last_cpu;			/* Last timestamp cpu */
-	/*
-	 * read() file operation state.
-	 */
-	unsigned long len_left;
-};
-
 /* channel: collection of per-cpu ring buffers. */
 struct channel {
-	atomic_t record_disabled;
+	int record_disabled;
 	unsigned long commit_count_mask;	/*
 						 * Commit count mask, removing
 						 * the MSBs corresponding to
@@ -55,16 +46,9 @@ struct channel {
 
 	unsigned long switch_timer_interval;	/* Buffer flush (jiffies) */
 	unsigned long read_timer_interval;	/* Reader wakeup (jiffies) */
-	struct notifier_block cpu_hp_notifier;	/* CPU hotplug notifier */
-	struct notifier_block tick_nohz_notifier; /* CPU nohz notifier */
-	struct notifier_block hp_iter_notifier;	/* hotplug iterator notifier */
-	int cpu_hp_enable:1;			/* Enable CPU hotplug notif. */
-	int hp_iter_enable:1;			/* Enable hp iter notif. */
-	wait_queue_head_t read_wait;		/* reader wait queue */
-	wait_queue_head_t hp_wait;		/* CPU hotplug wait queue */
+	//wait_queue_head_t read_wait;		/* reader wait queue */
 	int finalized;				/* Has channel been finalized */
-	struct channel_iter iter;		/* Channel read-side iterator */
-	struct kref ref;			/* Reference count */
+	struct urcu_ref ref;			/* Reference count */
 };
 
 /* Per-subbuffer commit counters used on the hot path */
@@ -78,35 +62,17 @@ struct commit_counters_cold {
 	union v_atomic cc_sb;		/* Incremented _once_ at sb switch */
 };
 
-/* Per-buffer read iterator */
-struct lib_ring_buffer_iter {
-	u64 timestamp;			/* Current record timestamp */
-	size_t header_len;		/* Current record header length */
-	size_t payload_len;		/* Current record payload length */
-
-	struct list_head empty_node;	/* Linked list of empty buffers */
-	unsigned long consumed, read_offset, data_size;
-	enum {
-		ITER_GET_SUBBUF = 0,
-		ITER_TEST_RECORD,
-		ITER_NEXT_RECORD,
-		ITER_PUT_SUBBUF,
-	} state;
-	int allocated:1;
-	int read_open:1;		/* Opened for reading ? */
-};
-
 /* ring buffer state */
 struct lib_ring_buffer {
 	/* First 32 bytes cache-hot cacheline */
 	union v_atomic offset;		/* Current offset in the buffer */
 	struct commit_counters_hot *commit_hot;
 					/* Commit count per sub-buffer */
-	atomic_long_t consumed;		/*
+	long consumed;			/*
 					 * Current offset in the buffer
 					 * standard atomic access (shared)
 					 */
-	atomic_t record_disabled;
+	int record_disabled;
 	/* End of first 32 bytes cacheline */
 	union v_atomic last_tsc;	/*
 					 * Last timestamp written in the buffer.
@@ -116,7 +82,7 @@ struct lib_ring_buffer {
 
 	struct commit_counters_cold *commit_cold;
 					/* Commit count per sub-buffer */
-	atomic_long_t active_readers;	/*
+	long active_readers;		/*
 					 * Active readers count
 					 * standard atomic access (shared)
 					 */
@@ -126,12 +92,10 @@ struct lib_ring_buffer {
 	union v_atomic records_lost_big;	/* Events too big */
 	union v_atomic records_count;	/* Number of records written */
 	union v_atomic records_overrun;	/* Number of overwritten records */
-	wait_queue_head_t read_wait;	/* reader buffer-level wait queue */
+	//wait_queue_head_t read_wait;	/* reader buffer-level wait queue */
 	int finalized;			/* buffer has been finalized */
-	struct timer_list switch_timer;	/* timer for periodical switch */
-	struct timer_list read_timer;	/* timer for read poll */
-	raw_spinlock_t raw_tick_nohz_spinlock;	/* nohz entry lock/trylock */
-	struct lib_ring_buffer_iter iter;	/* read-side iterator */
+	//struct timer_list switch_timer;	/* timer for periodical switch */
+	//struct timer_list read_timer;	/* timer for read poll */
 	unsigned long get_subbuf_consumed;	/* Read-side consumed */
 	unsigned long prod_snapshot;	/* Producer count snapshot */
 	unsigned long cons_snapshot;	/* Consumer count snapshot */
@@ -157,14 +121,14 @@ void *channel_get_private(struct channel *chan)
 		int _____ret = unlikely(cond);				\
 		if (_____ret) {						\
 			if (__same_type(*(c), struct channel_backend))	\
-				__chan = container_of((void *) (c),	\
+				__chan = caa_container_of((void *) (c),	\
 							struct channel, \
 							backend);	\
 			else if (__same_type(*(c), struct channel))	\
 				__chan = (void *) (c);			\
 			else						\
 				BUG_ON(1);				\
-			atomic_inc(&__chan->record_disabled);		\
+			uatomic_inc(&__chan->record_disabled);		\
 			WARN_ON(1);					\
 		}							\
 		_____ret;						\
diff --git a/libringbuffer/ring_buffer_vfs.c b/libringbuffer/ring_buffer_abi.c
similarity index 100%
rename from libringbuffer/ring_buffer_vfs.c
rename to libringbuffer/ring_buffer_abi.c
diff --git a/libringbuffer/ring_buffer_backend.c b/libringbuffer/ring_buffer_backend.c
index 5ff38f0..861acf7 100644
--- a/libringbuffer/ring_buffer_backend.c
+++ b/libringbuffer/ring_buffer_backend.c
@@ -6,6 +6,10 @@
  * Dual LGPL v2.1/GPL v2 license.
  */
 
+#include <urcu/arch.h>
+
+#include "ust/core.h"
+
 #include "config.h"
 #include "backend.h"
 #include "frontend.h"
@@ -32,7 +36,7 @@ int lib_ring_buffer_backend_allocate(const struct lib_ring_buffer_config *config
 	void **virt;
 	unsigned long i;
 
-	num_pages = size >> PAGE_SHIFT;
+	num_pages = size >> get_count_order(PAGE_SIZE);
 	num_pages_per_subbuf = num_pages >> get_count_order(num_subbuf);
 	subbuf_size = chanb->subbuf_size;
 	num_subbuf_alloc = num_subbuf;
@@ -42,22 +46,15 @@ int lib_ring_buffer_backend_allocate(const struct lib_ring_buffer_config *config
 		num_subbuf_alloc++;
 	}
 
-	pages = kmalloc_node(ALIGN(sizeof(*pages) * num_pages,
-				   1 << INTERNODE_CACHE_SHIFT),
-			GFP_KERNEL, cpu_to_node(max(bufb->cpu, 0)));
+	pages = malloc_align(sizeof(*pages) * num_pages);
 	if (unlikely(!pages))
 		goto pages_error;
 
-	virt = kmalloc_node(ALIGN(sizeof(*virt) * num_pages,
-				  1 << INTERNODE_CACHE_SHIFT),
-			GFP_KERNEL, cpu_to_node(max(bufb->cpu, 0)));
+	virt = malloc_align(sizeof(*virt) * num_pages);
 	if (unlikely(!virt))
 		goto virt_error;
 
-	bufb->array = kmalloc_node(ALIGN(sizeof(*bufb->array)
-					 * num_subbuf_alloc,
-				  1 << INTERNODE_CACHE_SHIFT),
-			GFP_KERNEL, cpu_to_node(max(bufb->cpu, 0)));
+	bufb->array = malloc_align(sizeof(*bufb->array) * num_subbuf_alloc);
 	if (unlikely(!bufb->array))
 		goto array_error;
 
@@ -73,22 +70,18 @@ int lib_ring_buffer_backend_allocate(const struct lib_ring_buffer_config *config
 	/* Allocate backend pages array elements */
 	for (i = 0; i < num_subbuf_alloc; i++) {
 		bufb->array[i] =
-			kzalloc_node(ALIGN(
+			zmalloc_align(
 				sizeof(struct lib_ring_buffer_backend_pages) +
 				sizeof(struct lib_ring_buffer_backend_page)
-				* num_pages_per_subbuf,
-				1 << INTERNODE_CACHE_SHIFT),
-				GFP_KERNEL, cpu_to_node(max(bufb->cpu, 0)));
+				* num_pages_per_subbuf);
 		if (!bufb->array[i])
 			goto free_array;
 	}
 
 	/* Allocate write-side subbuffer table */
-	bufb->buf_wsb = kzalloc_node(ALIGN(
+	bufb->buf_wsb = zmalloc_align(
 				sizeof(struct lib_ring_buffer_backend_subbuffer)
-				* num_subbuf,
-				1 << INTERNODE_CACHE_SHIFT),
-				GFP_KERNEL, cpu_to_node(max(bufb->cpu, 0)));
+				* num_subbuf);
 	if (unlikely(!bufb->buf_wsb))
 		goto free_array;
 
@@ -116,11 +109,6 @@ int lib_ring_buffer_backend_allocate(const struct lib_ring_buffer_config *config
 		}
 	}
 
-	/*
-	 * If kmalloc ever uses vmalloc underneath, make sure the buffer pages
-	 * will not fault.
-	 */
-	wrapper_vmalloc_sync_all();
 	kfree(virt);
 	kfree(pages);
 	return 0;
@@ -146,7 +134,7 @@ int lib_ring_buffer_backend_create(struct lib_ring_buffer_backend *bufb,
 {
 	const struct lib_ring_buffer_config *config = chanb->config;
 
-	bufb->chan = container_of(chanb, struct channel, backend);
+	bufb->chan = caa_container_of(chanb, struct channel, backend);
 	bufb->cpu = cpu;
 
 	return lib_ring_buffer_backend_allocate(config, bufb, chanb->buf_size,
@@ -209,7 +197,7 @@ void lib_ring_buffer_backend_reset(struct lib_ring_buffer_backend *bufb)
  */
 void channel_backend_reset(struct channel_backend *chanb)
 {
-	struct channel *chan = container_of(chanb, struct channel, backend);
+	struct channel *chan = caa_container_of(chanb, struct channel, backend);
 	const struct lib_ring_buffer_config *config = chanb->config;
 
 	/*
@@ -235,7 +223,7 @@ int __cpuinit lib_ring_buffer_cpu_hp_callback(struct notifier_block *nb,
 					      void *hcpu)
 {
 	unsigned int cpu = (unsigned long)hcpu;
-	struct channel_backend *chanb = container_of(nb, struct channel_backend,
+	struct channel_backend *chanb = caa_container_of(nb, struct channel_backend,
 						     cpu_hp_notifier);
 	const struct lib_ring_buffer_config *config = chanb->config;
 	struct lib_ring_buffer *buf;
@@ -289,7 +277,7 @@ int channel_backend_init(struct channel_backend *chanb,
 			 const struct lib_ring_buffer_config *config,
 			 void *priv, size_t subbuf_size, size_t num_subbuf)
 {
-	struct channel *chan = container_of(chanb, struct channel, backend);
+	struct channel *chan = caa_container_of(chanb, struct channel, backend);
 	unsigned int i;
 	int ret;
 
@@ -468,7 +456,7 @@ void _lib_ring_buffer_write(struct lib_ring_buffer_backend *bufb, size_t offset,
 		src += pagecpy;
 		offset += pagecpy;
 		sbidx = offset >> chanb->subbuf_size_order;
-		index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
+		index = (offset & (chanb->subbuf_size - 1)) >> get_count_order(PAGE_SIZE);
 
 		/*
 		 * Underlying layer should never ask for writes across
@@ -512,7 +500,7 @@ size_t lib_ring_buffer_read(struct lib_ring_buffer_backend *bufb, size_t offset,
 
 	orig_len = len;
 	offset &= chanb->buf_size - 1;
-	index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
+	index = (offset & (chanb->subbuf_size - 1)) >> get_count_order(PAGE_SIZE);
 	if (unlikely(!len))
 		return 0;
 	for (;;) {
@@ -529,7 +517,7 @@ size_t lib_ring_buffer_read(struct lib_ring_buffer_backend *bufb, size_t offset,
 			break;
 		dest += pagecpy;
 		offset += pagecpy;
-		index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
+		index = (offset & (chanb->subbuf_size - 1)) >> get_count_order(PAGE_SIZE);
 		/*
 		 * Underlying layer should never ask for reads across
 		 * subbuffers.
@@ -540,60 +528,6 @@ size_t lib_ring_buffer_read(struct lib_ring_buffer_backend *bufb, size_t offset,
 }
 EXPORT_SYMBOL_GPL(lib_ring_buffer_read);
 
-/**
- * __lib_ring_buffer_copy_to_user - read data from ring_buffer to userspace
- * @bufb : buffer backend
- * @offset : offset within the buffer
- * @dest : destination userspace address
- * @len : length to copy to destination
- *
- * Should be protected by get_subbuf/put_subbuf.
- * access_ok() must have been performed on dest addresses prior to call this
- * function.
- * Returns -EFAULT on error, 0 if ok.
- */
-int __lib_ring_buffer_copy_to_user(struct lib_ring_buffer_backend *bufb,
-				   size_t offset, void __user *dest, size_t len)
-{
-	struct channel_backend *chanb = &bufb->chan->backend;
-	const struct lib_ring_buffer_config *config = chanb->config;
-	size_t index;
-	ssize_t pagecpy, orig_len;
-	struct lib_ring_buffer_backend_pages *rpages;
-	unsigned long sb_bindex, id;
-
-	orig_len = len;
-	offset &= chanb->buf_size - 1;
-	index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
-	if (unlikely(!len))
-		return 0;
-	for (;;) {
-		pagecpy = min_t(size_t, len, PAGE_SIZE - (offset & ~PAGE_MASK));
-		id = bufb->buf_rsb.id;
-		sb_bindex = subbuffer_id_get_index(config, id);
-		rpages = bufb->array[sb_bindex];
-		CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
-			     && subbuffer_id_is_noref(config, id));
-		if (__copy_to_user(dest,
-			       rpages->p[index].virt + (offset & ~PAGE_MASK),
-			       pagecpy))
-			return -EFAULT;
-		len -= pagecpy;
-		if (likely(!len))
-			break;
-		dest += pagecpy;
-		offset += pagecpy;
-		index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
-		/*
-		 * Underlying layer should never ask for reads across
-		 * subbuffers.
-		 */
-		CHAN_WARN_ON(chanb, offset >= chanb->buf_size);
-	}
-	return 0;
-}
-EXPORT_SYMBOL_GPL(__lib_ring_buffer_copy_to_user);
-
 /**
  * lib_ring_buffer_read_cstr - read a C-style string from ring_buffer.
  * @bufb : buffer backend
@@ -616,7 +550,7 @@ int lib_ring_buffer_read_cstr(struct lib_ring_buffer_backend *bufb, size_t offse
 	unsigned long sb_bindex, id;
 
 	offset &= chanb->buf_size - 1;
-	index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
+	index = (offset & (chanb->subbuf_size - 1)) >> get_count_order(PAGE_SIZE);
 	orig_offset = offset;
 	for (;;) {
 		id = bufb->buf_rsb.id;
@@ -636,7 +570,7 @@ int lib_ring_buffer_read_cstr(struct lib_ring_buffer_backend *bufb, size_t offse
 			len -= pagecpy;
 		}
 		offset += strpagelen;
-		index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
+		index = (offset & (chanb->subbuf_size - 1)) >> get_count_order(PAGE_SIZE);
 		if (strpagelen < pagelen)
 			break;
 		/*
@@ -670,7 +604,7 @@ struct page **lib_ring_buffer_read_get_page(struct lib_ring_buffer_backend *bufb
 	unsigned long sb_bindex, id;
 
 	offset &= chanb->buf_size - 1;
-	index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
+	index = (offset & (chanb->subbuf_size - 1)) >> get_count_order(PAGE_SIZE);
 	id = bufb->buf_rsb.id;
 	sb_bindex = subbuffer_id_get_index(config, id);
 	rpages = bufb->array[sb_bindex];
@@ -701,7 +635,7 @@ void *lib_ring_buffer_read_offset_address(struct lib_ring_buffer_backend *bufb,
 	unsigned long sb_bindex, id;
 
 	offset &= chanb->buf_size - 1;
-	index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
+	index = (offset & (chanb->subbuf_size - 1)) >> get_count_order(PAGE_SIZE);
 	id = bufb->buf_rsb.id;
 	sb_bindex = subbuffer_id_get_index(config, id);
 	rpages = bufb->array[sb_bindex];
@@ -732,7 +666,7 @@ void *lib_ring_buffer_offset_address(struct lib_ring_buffer_backend *bufb,
 
 	offset &= chanb->buf_size - 1;
 	sbidx = offset >> chanb->subbuf_size_order;
-	index = (offset & (chanb->subbuf_size - 1)) >> PAGE_SHIFT;
+	index = (offset & (chanb->subbuf_size - 1)) >> get_count_order(PAGE_SIZE);
 	id = bufb->buf_wsb[sbidx].id;
 	sb_bindex = subbuffer_id_get_index(config, id);
 	rpages = bufb->array[sb_bindex];
diff --git a/libringbuffer/ring_buffer_frontend.c b/libringbuffer/ring_buffer_frontend.c
index a3f14f3..5ceda87 100644
--- a/libringbuffer/ring_buffer_frontend.c
+++ b/libringbuffer/ring_buffer_frontend.c
@@ -38,6 +38,8 @@
  * Dual LGPL v2.1/GPL v2 license.
  */
 
+#include <urcu/compiler.h>
+
 #include "config.h"
 #include "backend.h"
 #include "frontend.h"
@@ -158,7 +160,7 @@ int lib_ring_buffer_create(struct lib_ring_buffer *buf,
 			   struct channel_backend *chanb, int cpu)
 {
 	const struct lib_ring_buffer_config *config = chanb->config;
-	struct channel *chan = container_of(chanb, struct channel, backend);
+	struct channel *chan = caa_container_of(chanb, struct channel, backend);
 	void *priv = chanb->priv;
 	unsigned int num_subbuf;
 	size_t subbuf_header_size;
@@ -391,7 +393,7 @@ int __cpuinit lib_ring_buffer_cpu_hp_callback(struct notifier_block *nb,
 					      void *hcpu)
 {
 	unsigned int cpu = (unsigned long)hcpu;
-	struct channel *chan = container_of(nb, struct channel,
+	struct channel *chan = caa_container_of(nb, struct channel,
 					    cpu_hp_notifier);
 	struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, cpu);
 	const struct lib_ring_buffer_config *config = chan->backend.config;
@@ -447,7 +449,7 @@ static int notrace ring_buffer_tick_nohz_callback(struct notifier_block *nb,
 						  unsigned long val,
 						  void *data)
 {
-	struct channel *chan = container_of(nb, struct channel,
+	struct channel *chan = caa_container_of(nb, struct channel,
 					    tick_nohz_notifier);
 	const struct lib_ring_buffer_config *config = chan->backend.config;
 	struct lib_ring_buffer *buf;
@@ -687,7 +689,7 @@ EXPORT_SYMBOL_GPL(channel_create);
 static
 void channel_release(struct kref *kref)
 {
-	struct channel *chan = container_of(kref, struct channel, ref);
+	struct channel *chan = caa_container_of(kref, struct channel, ref);
 	channel_free(chan);
 }
 
@@ -729,7 +731,7 @@ void *channel_destroy(struct channel *chan)
 			 * Perform flush before writing to finalized.
 			 */
 			smp_wmb();
-			ACCESS_ONCE(buf->finalized) = 1;
+			CMM_ACCESS_ONCE(buf->finalized) = 1;
 			wake_up_interruptible(&buf->read_wait);
 		}
 	} else {
@@ -743,10 +745,10 @@ void *channel_destroy(struct channel *chan)
 		 * Perform flush before writing to finalized.
 		 */
 		smp_wmb();
-		ACCESS_ONCE(buf->finalized) = 1;
+		CMM_ACCESS_ONCE(buf->finalized) = 1;
 		wake_up_interruptible(&buf->read_wait);
 	}
-	ACCESS_ONCE(chan->finalized) = 1;
+	CMM_ACCESS_ONCE(chan->finalized) = 1;
 	wake_up_interruptible(&chan->hp_wait);
 	wake_up_interruptible(&chan->read_wait);
 	kref_put(&chan->ref, channel_release);
@@ -820,7 +822,7 @@ int lib_ring_buffer_snapshot(struct lib_ring_buffer *buf,
 	int finalized;
 
 retry:
-	finalized = ACCESS_ONCE(buf->finalized);
+	finalized = CMM_ACCESS_ONCE(buf->finalized);
 	/*
 	 * Read finalized before counters.
 	 */
@@ -908,7 +910,7 @@ int lib_ring_buffer_get_subbuf(struct lib_ring_buffer *buf,
 	int finalized;
 
 retry:
-	finalized = ACCESS_ONCE(buf->finalized);
+	finalized = CMM_ACCESS_ONCE(buf->finalized);
 	/*
 	 * Read finalized before counters.
 	 */
diff --git a/libringbuffer/ring_buffer_iterator.c b/libringbuffer/ring_buffer_iterator.c
deleted file mode 100644
index ff8e582..0000000
--- a/libringbuffer/ring_buffer_iterator.c
+++ /dev/null
@@ -1,793 +0,0 @@
-/*
- * ring_buffer_iterator.c
- *
- * (C) Copyright 2010 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
- *
- * 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 <mathieu.desnoyers@efficios.com>
- *
- * Dual LGPL v2.1/GPL v2 license.
- */
-
-#include "iterator.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)
-
-/**
- * lib_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 lib_ring_buffer_get_next_record(struct channel *chan,
-					struct lib_ring_buffer *buf)
-{
-	const struct lib_ring_buffer_config *config = chan->backend.config;
-	struct lib_ring_buffer_iter *iter = &buf->iter;
-	int ret;
-
-restart:
-	switch (iter->state) {
-	case ITER_GET_SUBBUF:
-		ret = lib_ring_buffer_get_next_subbuf(buf);
-		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.
-			 */
-			lib_ring_buffer_switch_slow(buf, SWITCH_ACTIVE);
-			ret = lib_ring_buffer_get_next_subbuf(buf);
-		}
-		if (ret)
-			return ret;
-		iter->consumed = buf->cons_snapshot;
-		iter->data_size = lib_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:
-		lib_ring_buffer_put_next_subbuf(buf);
-		iter->state = ITER_GET_SUBBUF;
-		goto restart;
-	default:
-		CHAN_WARN_ON(chan, 1);	/* Should not happen */
-		return -EPERM;
-	}
-}
-EXPORT_SYMBOL_GPL(lib_ring_buffer_get_next_record);
-
-static int buf_is_higher(void *a, void *b)
-{
-	struct lib_ring_buffer *bufa = a;
-	struct lib_ring_buffer *bufb = b;
-
-	/* Consider lowest timestamps to be at the top of the heap */
-	return (bufa->iter.timestamp < bufb->iter.timestamp);
-}
-
-static
-void lib_ring_buffer_get_empty_buf_records(const struct lib_ring_buffer_config *config,
-					   struct channel *chan)
-{
-	struct lttng_ptr_heap *heap = &chan->iter.heap;
-	struct lib_ring_buffer *buf, *tmp;
-	ssize_t len;
-
-	list_for_each_entry_safe(buf, tmp, &chan->iter.empty_head,
-				 iter.empty_node) {
-		len = lib_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.
-			 */
-			CHAN_WARN_ON(chan, len < 0);
-			list_del(&buf->iter.empty_node);
-			CHAN_WARN_ON(chan, lttng_heap_insert(heap, buf));
-		}
-	}
-}
-
-static
-void lib_ring_buffer_wait_for_qs(const struct lib_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.
-	 */
-	lib_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);
-	lib_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 lib_ring_buffer **ret_buf)
-{
-	const struct lib_ring_buffer_config *config = chan->backend.config;
-	struct lib_ring_buffer *buf;
-	struct lttng_ptr_heap *heap;
-	ssize_t len;
-
-	if (config->alloc == RING_BUFFER_ALLOC_GLOBAL) {
-		*ret_buf = channel_get_ring_buffer(config, chan, 0);
-		return lib_ring_buffer_get_next_record(chan, *ret_buf);
-	}
-
-	heap = &chan->iter.heap;
-
-	/*
-	 * get next record for topmost buffer.
-	 */
-	buf = lttng_heap_maximum(heap);
-	if (buf) {
-		len = lib_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, lttng_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, lttng_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
-			 * lttng_heap_replace_max().
-			 */
-			CHAN_WARN_ON(chan, len < 0);
-			CHAN_WARN_ON(chan, lttng_heap_replace_max(heap, buf) != buf);
-			break;
-		}
-	}
-
-	buf = lttng_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.
-		 */
-		lib_ring_buffer_wait_for_qs(config, chan);
-	}
-
-	*ret_buf = buf = lttng_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 lib_ring_buffer_iterator_init(struct channel *chan, struct lib_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, lib_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 lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, cpu);
-	const struct lib_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:
-		lib_ring_buffer_iterator_init(chan, buf);
-		return NOTIFY_OK;
-	default:
-		return NOTIFY_DONE;
-	}
-}
-#endif
-
-int channel_iterator_init(struct channel *chan)
-{
-	const struct lib_ring_buffer_config *config = chan->backend.config;
-	struct lib_ring_buffer *buf;
-
-	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
-		int cpu, ret;
-
-		INIT_LIST_HEAD(&chan->iter.empty_head);
-		ret = lttng_heap_init(&chan->iter.heap,
-				num_possible_cpus(),
-				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);
-			lib_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);
-			lib_ring_buffer_iterator_init(chan, buf);
-		}
-#endif
-	} else {
-		buf = channel_get_ring_buffer(config, chan, 0);
-		lib_ring_buffer_iterator_init(chan, buf);
-	}
-	return 0;
-}
-
-void channel_iterator_unregister_notifiers(struct channel *chan)
-{
-	const struct lib_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 lib_ring_buffer_config *config = chan->backend.config;
-
-	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
-		lttng_heap_free(&chan->iter.heap);
-}
-
-int lib_ring_buffer_iterator_open(struct lib_ring_buffer *buf)
-{
-	struct channel *chan = buf->backend.chan;
-	const struct lib_ring_buffer_config *config = chan->backend.config;
-	CHAN_WARN_ON(chan, config->output != RING_BUFFER_ITERATOR);
-	return lib_ring_buffer_open_read(buf);
-}
-EXPORT_SYMBOL_GPL(lib_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 lib_ring_buffer_iterator_release(struct lib_ring_buffer *buf)
-{
-	lib_ring_buffer_release_read(buf);
-}
-EXPORT_SYMBOL_GPL(lib_ring_buffer_iterator_release);
-
-int channel_iterator_open(struct channel *chan)
-{
-	const struct lib_ring_buffer_config *config = chan->backend.config;
-	struct lib_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 = lib_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 = lib_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 lib_ring_buffer_config *config = chan->backend.config;
-	struct lib_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) {
-				lib_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);
-		lib_ring_buffer_iterator_release(buf);
-	}
-}
-EXPORT_SYMBOL_GPL(channel_iterator_release);
-
-void lib_ring_buffer_iterator_reset(struct lib_ring_buffer *buf)
-{
-	struct channel *chan = buf->backend.chan;
-
-	if (buf->iter.state != ITER_GET_SUBBUF)
-		lib_ring_buffer_put_next_subbuf(buf);
-	buf->iter.state = ITER_GET_SUBBUF;
-	/* Remove from heap (if present). */
-	if (lttng_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 lib_ring_buffer_config *config = chan->backend.config;
-	struct lib_ring_buffer *buf;
-	int cpu;
-
-	/* Empty heap, put into empty_head */
-	while ((buf = lttng_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);
-		lib_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 lib_ring_buffer *buf,
-				      int fusionmerge)
-{
-	const struct lib_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 = lttng_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 = lib_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 = lib_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 (__lib_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;
-}
-
-/**
- * lib_ring_buffer_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 lib_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 lib_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 lib_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 lib_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 lib_ring_buffer_file_open(struct inode *inode, struct file *file)
-{
-	struct lib_ring_buffer *buf = inode->i_private;
-	int ret;
-
-	ret = lib_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:
-	lib_ring_buffer_iterator_release(buf);
-	return ret;
-}
-
-static
-int lib_ring_buffer_file_release(struct inode *inode, struct file *file)
-{
-	struct lib_ring_buffer *buf = inode->i_private;
-
-	lib_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 = lib_ring_buffer_no_llseek,
-};
-EXPORT_SYMBOL_GPL(channel_payload_file_operations);
-
-const struct file_operations lib_ring_buffer_payload_file_operations = {
-	.open = lib_ring_buffer_file_open,
-	.release = lib_ring_buffer_file_release,
-	.read = lib_ring_buffer_file_read,
-	.llseek = lib_ring_buffer_no_llseek,
-};
-EXPORT_SYMBOL_GPL(lib_ring_buffer_payload_file_operations);
diff --git a/libringbuffer/vatomic.h b/libringbuffer/vatomic.h
index eea6d04..831ea04 100644
--- a/libringbuffer/vatomic.h
+++ b/libringbuffer/vatomic.h
@@ -9,6 +9,9 @@
  * Dual LGPL v2.1/GPL v2 license.
  */
 
+#include <assert.h>
+#include <urcu/uatomic.h>
+
 /*
  * Same data type (long) accessed differently depending on configuration.
  * v field is for non-atomic access (protected by mutual exclusion).
@@ -18,46 +21,37 @@
  * atomic_long_t is used for globally shared buffers.
  */
 union v_atomic {
-	local_t l;
-	atomic_long_t a;
+	long a;	/* accessed through uatomic */
 	long v;
 };
 
 static inline
 long v_read(const struct lib_ring_buffer_config *config, union v_atomic *v_a)
 {
-	if (config->sync == RING_BUFFER_SYNC_PER_CPU)
-		return local_read(&v_a->l);
-	else
-		return atomic_long_read(&v_a->a);
+	assert(config->sync != RING_BUFFER_SYNC_PER_CPU);
+	return uatomic_read(&v_a->a);
 }
 
 static inline
 void v_set(const struct lib_ring_buffer_config *config, union v_atomic *v_a,
 	   long v)
 {
-	if (config->sync == RING_BUFFER_SYNC_PER_CPU)
-		local_set(&v_a->l, v);
-	else
-		atomic_long_set(&v_a->a, v);
+	assert(config->sync != RING_BUFFER_SYNC_PER_CPU);
+	uatomic_set(&v_a->a, v);
 }
 
 static inline
 void v_add(const struct lib_ring_buffer_config *config, long v, union v_atomic *v_a)
 {
-	if (config->sync == RING_BUFFER_SYNC_PER_CPU)
-		local_add(v, &v_a->l);
-	else
-		atomic_long_add(v, &v_a->a);
+	assert(config->sync != RING_BUFFER_SYNC_PER_CPU);
+	uatomic_add(&v_a->a, v);
 }
 
 static inline
 void v_inc(const struct lib_ring_buffer_config *config, union v_atomic *v_a)
 {
-	if (config->sync == RING_BUFFER_SYNC_PER_CPU)
-		local_inc(&v_a->l);
-	else
-		atomic_long_inc(&v_a->a);
+	assert(config->sync != RING_BUFFER_SYNC_PER_CPU);
+	uatomic_inc(&v_a->a);
 }
 
 /*
@@ -73,10 +67,8 @@ static inline
 long v_cmpxchg(const struct lib_ring_buffer_config *config, union v_atomic *v_a,
 	       long old, long _new)
 {
-	if (config->sync == RING_BUFFER_SYNC_PER_CPU)
-		return local_cmpxchg(&v_a->l, old, _new);
-	else
-		return atomic_long_cmpxchg(&v_a->a, old, _new);
+	assert(config->sync != RING_BUFFER_SYNC_PER_CPU);
+	return uatomic_cmpxchg(&v_a->a, old, _new);
 }
 
 #endif /* _LINUX_RING_BUFFER_VATOMIC_H */
-- 
2.34.1