X-Git-Url: http://git.liburcu.org/?a=blobdiff_plain;f=libringbuffer%2Fring_buffer_frontend.c;h=ea73bcc676351ac76f25eeae85282c1aa17bc25a;hb=d1a996f5524540ebc8caa7c3c1bd3049018d212c;hp=0996ecefc0313ff5ef8b955ce7dffd3a6a793849;hpb=64493e4fa019e9bdfe0d9c6a4738c9552f250f35;p=lttng-ust.git

diff --git a/libringbuffer/ring_buffer_frontend.c b/libringbuffer/ring_buffer_frontend.c
index 0996ecef..ea73bcc6 100644
--- a/libringbuffer/ring_buffer_frontend.c
+++ b/libringbuffer/ring_buffer_frontend.c
@@ -1,7 +1,22 @@
 /*
  * ring_buffer_frontend.c
  *
- * (C) Copyright 2005-2010 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+ * Copyright (C) 2005-2012 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; only
+ * version 2.1 of the License.
+ *
+ * This library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this library; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ *
  *
  * Ring buffer wait-free buffer synchronization. Producer-consumer and flight
  * recorder (overwrite) modes. See thesis:
@@ -34,17 +49,19 @@
  *   - 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.
  */
 
 #define _GNU_SOURCE
 #include <sys/types.h>
 #include <sys/mman.h>
 #include <sys/stat.h>
+#include <unistd.h>
 #include <fcntl.h>
+#include <signal.h>
+#include <time.h>
 #include <urcu/compiler.h>
 #include <urcu/ref.h>
+#include <urcu/tls-compat.h>
 #include <helper.h>
 
 #include "smp.h"
@@ -53,6 +70,8 @@
 #include "backend.h"
 #include "frontend.h"
 #include "shm.h"
+#include "tlsfixup.h"
+#include "../liblttng-ust/compat.h"	/* For ENODATA */
 
 #ifndef max
 #define max(a, b)	((a) > (b) ? (a) : (b))
@@ -61,6 +80,11 @@
 /* Print DBG() messages about events lost only every 1048576 hits */
 #define DBG_PRINT_NR_LOST	(1UL << 20)
 
+#define LTTNG_UST_RB_SIG_FLUSH		SIGRTMIN
+#define LTTNG_UST_RB_SIG_READ		SIGRTMIN + 1
+#define LTTNG_UST_RB_SIG_TEARDOWN	SIGRTMIN + 2
+#define CLOCKID		CLOCK_MONOTONIC
+
 /*
  * Use POSIX SHM: shm_open(3) and shm_unlink(3).
  * close(2) to close the fd returned by shm_open.
@@ -87,17 +111,32 @@ struct switch_offsets {
 		     switch_old_end:1;
 };
 
-__thread unsigned int lib_ring_buffer_nesting;
+DEFINE_URCU_TLS(unsigned int, lib_ring_buffer_nesting);
 
 /*
- * TODO: this is unused. Errors are saved within the ring buffer.
- * Eventually, allow consumerd to print these errors.
+ * wakeup_fd_mutex protects wakeup fd use by timer from concurrent
+ * close.
  */
+static pthread_mutex_t wakeup_fd_mutex = PTHREAD_MUTEX_INITIALIZER;
+
 static
 void lib_ring_buffer_print_errors(struct channel *chan,
-				  struct lttng_ust_lib_ring_buffer *buf, int cpu,
-				  struct lttng_ust_shm_handle *handle)
-	__attribute__((unused));
+				struct lttng_ust_lib_ring_buffer *buf, int cpu,
+				struct lttng_ust_shm_handle *handle);
+
+/*
+ * Handle timer teardown race wrt memory free of private data by
+ * ring buffer signals are handled by a single thread, which permits
+ * a synchronization point between handling of each signal.
+ * Protected by the ust mutex.
+ */
+struct timer_signal_data {
+	pthread_t tid;	/* thread id managing signals */
+	int setup_done;
+	int qs_done;
+};
+
+static struct timer_signal_data timer_signal;
 
 /**
  * lib_ring_buffer_reset - Reset ring buffer to initial values.
@@ -230,167 +269,409 @@ free_chanbuf:
 	return ret;
 }
 
-#if 0
-static void switch_buffer_timer(unsigned long data)
+static
+void lib_ring_buffer_channel_switch_timer(int sig, siginfo_t *si, void *uc)
 {
-	struct lttng_ust_lib_ring_buffer *buf = (struct lttng_ust_lib_ring_buffer *)data;
-	struct channel *chan = shmp(handle, buf->backend.chan);
-	const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
+	const struct lttng_ust_lib_ring_buffer_config *config;
+	struct lttng_ust_shm_handle *handle;
+	struct channel *chan;
+	int cpu;
+
+	assert(CMM_LOAD_SHARED(timer_signal.tid) == pthread_self());
+
+	chan = si->si_value.sival_ptr;
+	handle = chan->handle;
+	config = &chan->backend.config;
+
+	DBG("Switch timer for channel %p\n", chan);
 
 	/*
 	 * Only flush buffers periodically if readers are active.
 	 */
-	if (uatomic_read(&buf->active_readers) || uatomic_read(&buf->active_shadow_readers))
-		lib_ring_buffer_switch_slow(buf, SWITCH_ACTIVE, handle);
-
-	//TODO timers
-	//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);
+	pthread_mutex_lock(&wakeup_fd_mutex);
+	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
+		for_each_possible_cpu(cpu) {
+			struct lttng_ust_lib_ring_buffer *buf =
+				shmp(handle, chan->backend.buf[cpu].shmp);
+			if (uatomic_read(&buf->active_readers))
+				lib_ring_buffer_switch_slow(buf, SWITCH_ACTIVE,
+					chan->handle);
+		}
+	} else {
+		struct lttng_ust_lib_ring_buffer *buf =
+			shmp(handle, chan->backend.buf[0].shmp);
+
+		if (uatomic_read(&buf->active_readers))
+			lib_ring_buffer_switch_slow(buf, SWITCH_ACTIVE,
+				chan->handle);
+	}
+	pthread_mutex_unlock(&wakeup_fd_mutex);
+	return;
 }
-#endif //0
 
-static void lib_ring_buffer_start_switch_timer(struct lttng_ust_lib_ring_buffer *buf,
-			   struct lttng_ust_shm_handle *handle)
+static
+void lib_ring_buffer_channel_do_read(struct channel *chan)
 {
-	struct channel *chan = shmp(handle, buf->backend.chan);
-	//const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
+	const struct lttng_ust_lib_ring_buffer_config *config;
+	struct lttng_ust_shm_handle *handle;
+	int cpu;
+
+	handle = chan->handle;
+	config = &chan->backend.config;
+
+	/*
+	 * Only flush buffers periodically if readers are active.
+	 */
+	pthread_mutex_lock(&wakeup_fd_mutex);
+	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
+		for_each_possible_cpu(cpu) {
+			struct lttng_ust_lib_ring_buffer *buf =
+				shmp(handle, chan->backend.buf[cpu].shmp);
+
+			if (uatomic_read(&buf->active_readers)
+			    && lib_ring_buffer_poll_deliver(config, buf,
+					chan, handle)) {
+				lib_ring_buffer_wakeup(buf, handle);
+			}
+		}
+	} else {
+		struct lttng_ust_lib_ring_buffer *buf =
+			shmp(handle, chan->backend.buf[0].shmp);
+
+		if (uatomic_read(&buf->active_readers)
+		    && lib_ring_buffer_poll_deliver(config, buf,
+				chan, handle)) {
+			lib_ring_buffer_wakeup(buf, handle);
+		}
+	}
+	pthread_mutex_unlock(&wakeup_fd_mutex);
+}
+
+static
+void lib_ring_buffer_channel_read_timer(int sig, siginfo_t *si, void *uc)
+{
+	struct channel *chan;
+
+	assert(CMM_LOAD_SHARED(timer_signal.tid) == pthread_self());
+	chan = si->si_value.sival_ptr;
+	DBG("Read timer for channel %p\n", chan);
+	lib_ring_buffer_channel_do_read(chan);
+	return;
+}
+
+static
+void rb_setmask(sigset_t *mask)
+{
+	int ret;
+
+	ret = sigemptyset(mask);
+	if (ret) {
+		PERROR("sigemptyset");
+	}
+	ret = sigaddset(mask, LTTNG_UST_RB_SIG_FLUSH);
+	if (ret) {
+		PERROR("sigaddset");
+	}
+	ret = sigaddset(mask, LTTNG_UST_RB_SIG_READ);
+	if (ret) {
+		PERROR("sigaddset");
+	}
+	ret = sigaddset(mask, LTTNG_UST_RB_SIG_TEARDOWN);
+	if (ret) {
+		PERROR("sigaddset");
+	}
+}
+
+static
+void *sig_thread(void *arg)
+{
+	sigset_t mask;
+	siginfo_t info;
+	int signr;
+
+	/* Only self thread will receive signal mask. */
+	rb_setmask(&mask);
+	CMM_STORE_SHARED(timer_signal.tid, pthread_self());
+
+	for (;;) {
+		signr = sigwaitinfo(&mask, &info);
+		if (signr == -1) {
+			if (errno != EINTR)
+				PERROR("sigwaitinfo");
+			continue;
+		}
+		if (signr == LTTNG_UST_RB_SIG_FLUSH) {
+			lib_ring_buffer_channel_switch_timer(info.si_signo,
+					&info, NULL);
+		} else if (signr == LTTNG_UST_RB_SIG_READ) {
+			lib_ring_buffer_channel_read_timer(info.si_signo,
+					&info, NULL);
+		} else if (signr == LTTNG_UST_RB_SIG_TEARDOWN) {
+			cmm_smp_mb();
+			CMM_STORE_SHARED(timer_signal.qs_done, 1);
+			cmm_smp_mb();
+		} else {
+			ERR("Unexptected signal %d\n", info.si_signo);
+		}
+	}
+	return NULL;
+}
+
+/*
+ * Called with ust_lock() held.
+ * Ensure only a single thread listens on the timer signal.
+ */
+static
+void lib_ring_buffer_setup_timer_thread(void)
+{
+	pthread_t thread;
+	int ret;
 
-	if (!chan->switch_timer_interval || buf->switch_timer_enabled)
+	if (timer_signal.setup_done)
 		return;
-	//TODO
-	//init_timer(&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);
-	buf->switch_timer_enabled = 1;
+
+	ret = pthread_create(&thread, NULL, &sig_thread, NULL);
+	if (ret) {
+		errno = ret;
+		PERROR("pthread_create");
+	}
+	ret = pthread_detach(thread);
+	if (ret) {
+		errno = ret;
+		PERROR("pthread_detach");
+	}
+	timer_signal.setup_done = 1;
 }
 
-static void lib_ring_buffer_stop_switch_timer(struct lttng_ust_lib_ring_buffer *buf,
-			   struct lttng_ust_shm_handle *handle)
+/*
+ * Called with ust_lock() held.
+ */
+static
+void lib_ring_buffer_channel_switch_timer_start(struct channel *chan)
 {
-	struct channel *chan = shmp(handle, buf->backend.chan);
+	struct sigevent sev;
+	struct itimerspec its;
+	int ret;
 
-	if (!chan->switch_timer_interval || !buf->switch_timer_enabled)
+	if (!chan->switch_timer_interval || chan->switch_timer_enabled)
 		return;
 
-	//TODO
-	//del_timer_sync(&buf->switch_timer);
-	buf->switch_timer_enabled = 0;
+	chan->switch_timer_enabled = 1;
+
+	lib_ring_buffer_setup_timer_thread();
+
+	sev.sigev_notify = SIGEV_SIGNAL;
+	sev.sigev_signo = LTTNG_UST_RB_SIG_FLUSH;
+	sev.sigev_value.sival_ptr = chan;
+	ret = timer_create(CLOCKID, &sev, &chan->switch_timer);
+	if (ret == -1) {
+		PERROR("timer_create");
+	}
+
+	its.it_value.tv_sec = chan->switch_timer_interval / 1000000;
+	its.it_value.tv_nsec = chan->switch_timer_interval % 1000000;
+	its.it_interval.tv_sec = its.it_value.tv_sec;
+	its.it_interval.tv_nsec = its.it_value.tv_nsec;
+
+	ret = timer_settime(chan->switch_timer, 0, &its, NULL);
+	if (ret == -1) {
+		PERROR("timer_settime");
+	}
 }
 
-#if 0
 /*
- * Polling timer to check the channels for data.
+ * Called with ust_lock() held.
  */
-static void read_buffer_timer(unsigned long data)
+static
+void lib_ring_buffer_channel_switch_timer_stop(struct channel *chan)
 {
-	struct lttng_ust_lib_ring_buffer *buf = (struct lttng_ust_lib_ring_buffer *)data;
-	struct channel *chan = shmp(handle, buf->backend.chan);
-	const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
+	sigset_t pending_set;
+	int ret;
+
+	if (!chan->switch_timer_interval || !chan->switch_timer_enabled)
+		return;
 
-	CHAN_WARN_ON(chan, !buf->backend.allocated);
+	ret = timer_delete(chan->switch_timer);
+	if (ret == -1) {
+		PERROR("timer_delete");
+	}
 
-	if (uatomic_read(&buf->active_readers) || uatomic_read(&buf->active_shadow_readers))
-	    && lib_ring_buffer_poll_deliver(config, buf, chan)) {
-		//TODO
-		//wake_up_interruptible(&buf->read_wait);
-		//wake_up_interruptible(&chan->read_wait);
+	/*
+	 * Ensure we don't have any signal queued for this channel.
+	 */
+	for (;;) {
+		ret = sigemptyset(&pending_set);
+		if (ret == -1) {
+			PERROR("sigemptyset");
+		}
+		ret = sigpending(&pending_set);
+		if (ret == -1) {
+			PERROR("sigpending");
+		}
+		if (!sigismember(&pending_set, LTTNG_UST_RB_SIG_FLUSH))
+			break;
+		caa_cpu_relax();
 	}
 
-	//TODO
-	//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);
+	/*
+	 * From this point, no new signal handler will be fired that
+	 * would try to access "chan". However, we still need to wait
+	 * for any currently executing handler to complete.
+	 */
+	cmm_smp_mb();
+	CMM_STORE_SHARED(timer_signal.qs_done, 0);
+	cmm_smp_mb();
+
+	/*
+	 * Kill with LTTNG_UST_RB_SIG_TEARDOWN, so signal management
+	 * thread wakes up.
+	 */
+	kill(getpid(), LTTNG_UST_RB_SIG_TEARDOWN);
+
+	while (!CMM_LOAD_SHARED(timer_signal.qs_done))
+		caa_cpu_relax();
+	cmm_smp_mb();
+
+	chan->switch_timer = 0;
+	chan->switch_timer_enabled = 0;
 }
-#endif //0
 
-static void lib_ring_buffer_start_read_timer(struct lttng_ust_lib_ring_buffer *buf,
-			   struct lttng_ust_shm_handle *handle)
+/*
+ * Called with ust_lock() held.
+ */
+static
+void lib_ring_buffer_channel_read_timer_start(struct channel *chan)
 {
-	struct channel *chan = shmp(handle, buf->backend.chan);
 	const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
+	struct sigevent sev;
+	struct itimerspec its;
+	int ret;
 
 	if (config->wakeup != RING_BUFFER_WAKEUP_BY_TIMER
-	    || !chan->read_timer_interval
-	    || buf->read_timer_enabled)
+			|| !chan->read_timer_interval || chan->read_timer_enabled)
 		return;
 
-	//TODO
-	//init_timer(&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);
-	buf->read_timer_enabled = 1;
+	chan->read_timer_enabled = 1;
+
+	lib_ring_buffer_setup_timer_thread();
+
+	sev.sigev_notify = SIGEV_SIGNAL;
+	sev.sigev_signo = LTTNG_UST_RB_SIG_READ;
+	sev.sigev_value.sival_ptr = chan;
+	ret = timer_create(CLOCKID, &sev, &chan->read_timer);
+	if (ret == -1) {
+		PERROR("timer_create");
+	}
+
+	its.it_value.tv_sec = chan->read_timer_interval / 1000000;
+	its.it_value.tv_nsec = chan->read_timer_interval % 1000000;
+	its.it_interval.tv_sec = its.it_value.tv_sec;
+	its.it_interval.tv_nsec = its.it_value.tv_nsec;
+
+	ret = timer_settime(chan->read_timer, 0, &its, NULL);
+	if (ret == -1) {
+		PERROR("timer_settime");
+	}
 }
 
-static void lib_ring_buffer_stop_read_timer(struct lttng_ust_lib_ring_buffer *buf,
-			   struct lttng_ust_shm_handle *handle)
+/*
+ * Called with ust_lock() held.
+ */
+static
+void lib_ring_buffer_channel_read_timer_stop(struct channel *chan)
 {
-	struct channel *chan = shmp(handle, buf->backend.chan);
 	const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
+	sigset_t pending_set;
+	int ret;
 
 	if (config->wakeup != RING_BUFFER_WAKEUP_BY_TIMER
-	    || !chan->read_timer_interval
-	    || !buf->read_timer_enabled)
+			|| !chan->read_timer_interval || !chan->read_timer_enabled)
 		return;
 
-	//TODO
-	//del_timer_sync(&buf->read_timer);
+	ret = timer_delete(chan->read_timer);
+	if (ret == -1) {
+		PERROR("timer_delete");
+	}
+
 	/*
 	 * do one more check to catch data that has been written in the last
 	 * timer period.
 	 */
-	if (lib_ring_buffer_poll_deliver(config, buf, chan, handle)) {
-		//TODO
-		//wake_up_interruptible(&buf->read_wait);
-		//wake_up_interruptible(&chan->read_wait);
+	lib_ring_buffer_channel_do_read(chan);
+
+
+	/*
+	 * Ensure we don't have any signal queued for this channel.
+	 */
+	for (;;) {
+		ret = sigemptyset(&pending_set);
+		if (ret == -1) {
+			PERROR("sigemptyset");
+		}
+		ret = sigpending(&pending_set);
+		if (ret == -1) {
+			PERROR("sigpending");
+		}
+		if (!sigismember(&pending_set, LTTNG_UST_RB_SIG_READ))
+			break;
+		caa_cpu_relax();
 	}
-	buf->read_timer_enabled = 0;
+
+	/*
+	 * From this point, no new signal handler will be fired that
+	 * would try to access "chan". However, we still need to wait
+	 * for any currently executing handler to complete.
+	 */
+	cmm_smp_mb();
+	CMM_STORE_SHARED(timer_signal.qs_done, 0);
+	cmm_smp_mb();
+
+	/*
+	 * Kill with LTTNG_UST_RB_SIG_TEARDOWN, so signal management
+	 * thread wakes up.
+	 */
+	kill(getpid(), LTTNG_UST_RB_SIG_TEARDOWN);
+
+	while (!CMM_LOAD_SHARED(timer_signal.qs_done))
+		caa_cpu_relax();
+	cmm_smp_mb();
+
+	chan->read_timer = 0;
+	chan->read_timer_enabled = 0;
 }
 
 static void channel_unregister_notifiers(struct channel *chan,
 			   struct lttng_ust_shm_handle *handle)
 {
-	const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
+	lib_ring_buffer_channel_switch_timer_stop(chan);
+	lib_ring_buffer_channel_read_timer_stop(chan);
+}
+
+static void channel_print_errors(struct channel *chan,
+		struct lttng_ust_shm_handle *handle)
+{
+	const struct lttng_ust_lib_ring_buffer_config *config =
+			&chan->backend.config;
 	int cpu;
 
 	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
 		for_each_possible_cpu(cpu) {
-			struct lttng_ust_lib_ring_buffer *buf = shmp(handle, chan->backend.buf[cpu].shmp);
-
-			lib_ring_buffer_stop_switch_timer(buf, handle);
-			lib_ring_buffer_stop_read_timer(buf, handle);
+			struct lttng_ust_lib_ring_buffer *buf =
+				shmp(handle, chan->backend.buf[cpu].shmp);
+			lib_ring_buffer_print_errors(chan, buf, cpu, handle);
 		}
 	} else {
-		struct lttng_ust_lib_ring_buffer *buf = shmp(handle, chan->backend.buf[0].shmp);
+		struct lttng_ust_lib_ring_buffer *buf =
+			shmp(handle, chan->backend.buf[0].shmp);
 
-		lib_ring_buffer_stop_switch_timer(buf, handle);
-		lib_ring_buffer_stop_read_timer(buf, handle);
+		lib_ring_buffer_print_errors(chan, buf, -1, handle);
 	}
-	//channel_backend_unregister_notifiers(&chan->backend);
 }
 
-static void channel_free(struct channel *chan, struct lttng_ust_shm_handle *handle,
-		int shadow)
+static void channel_free(struct channel *chan,
+		struct lttng_ust_shm_handle *handle)
 {
-	if (!shadow)
-		channel_backend_free(&chan->backend, handle);
+	channel_backend_free(&chan->backend, handle);
 	/* chan is freed by shm teardown */
 	shm_object_table_destroy(handle->table);
 	free(handle);
@@ -424,15 +705,19 @@ struct lttng_ust_shm_handle *channel_create(const struct lttng_ust_lib_ring_buff
 		   void *priv_data_init,
 		   void *buf_addr, size_t subbuf_size,
 		   size_t num_subbuf, unsigned int switch_timer_interval,
-		   unsigned int read_timer_interval,
-		   int **shm_fd, int **wait_fd, uint64_t **memory_map_size)
+		   unsigned int read_timer_interval)
 {
-	int ret, cpu;
+	int ret;
 	size_t shmsize, chansize;
 	struct channel *chan;
 	struct lttng_ust_shm_handle *handle;
 	struct shm_object *shmobj;
-	struct shm_ref *ref;
+	unsigned int nr_streams;
+
+	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
+		nr_streams = num_possible_cpus();
+	else
+		nr_streams = 1;
 
 	if (lib_ring_buffer_check_config(config, switch_timer_interval,
 					 read_timer_interval))
@@ -450,15 +735,14 @@ struct lttng_ust_shm_handle *channel_create(const struct lttng_ust_lib_ring_buff
 	/* Calculate the shm allocation layout */
 	shmsize = sizeof(struct channel);
 	shmsize += offset_align(shmsize, __alignof__(struct lttng_ust_lib_ring_buffer_shmp));
-	if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
-		shmsize += sizeof(struct lttng_ust_lib_ring_buffer_shmp) * num_possible_cpus();
-	else
-		shmsize += sizeof(struct lttng_ust_lib_ring_buffer_shmp);
+	shmsize += sizeof(struct lttng_ust_lib_ring_buffer_shmp) * nr_streams;
 	chansize = shmsize;
-	shmsize += offset_align(shmsize, priv_data_align);
+	if (priv_data_align)
+		shmsize += offset_align(shmsize, priv_data_align);
 	shmsize += priv_data_size;
 
-	shmobj = shm_object_table_append(handle->table, shmsize);
+	/* Allocate normal memory for channel (not shared) */
+	shmobj = shm_object_table_alloc(handle->table, shmsize, SHM_OBJECT_MEM);
 	if (!shmobj)
 		goto error_append;
 	/* struct channel is at object 0, offset 0 (hardcoded) */
@@ -468,6 +752,7 @@ struct lttng_ust_shm_handle *channel_create(const struct lttng_ust_lib_ring_buff
 	chan = shmp(handle, handle->chan);
 	if (!chan)
 		goto error_append;
+	chan->nr_streams = nr_streams;
 
 	/* space for private data */
 	if (priv_data_size) {
@@ -482,7 +767,8 @@ struct lttng_ust_shm_handle *channel_create(const struct lttng_ust_lib_ring_buff
 		memcpy(*priv_data, priv_data_init, priv_data_size);
 	} else {
 		chan->priv_data_offset = -1;
-		*priv_data = NULL;
+		if (priv_data)
+			*priv_data = NULL;
 	}
 
 	ret = channel_backend_init(&chan->backend, name, config,
@@ -490,33 +776,14 @@ struct lttng_ust_shm_handle *channel_create(const struct lttng_ust_lib_ring_buff
 	if (ret)
 		goto error_backend_init;
 
+	chan->handle = handle;
 	chan->commit_count_mask = (~0UL >> chan->backend.num_subbuf_order);
-	//TODO
-	//chan->switch_timer_interval = usecs_to_jiffies(switch_timer_interval);
-	//chan->read_timer_interval = usecs_to_jiffies(read_timer_interval);
-	//TODO
-	//init_waitqueue_head(&chan->read_wait);
-	//init_waitqueue_head(&chan->hp_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.
-		 */
-		for_each_possible_cpu(cpu) {
-			struct lttng_ust_lib_ring_buffer *buf = shmp(handle, chan->backend.buf[cpu].shmp);
-			lib_ring_buffer_start_switch_timer(buf, handle);
-			lib_ring_buffer_start_read_timer(buf, handle);
-		}
-	} else {
-		struct lttng_ust_lib_ring_buffer *buf = shmp(handle, chan->backend.buf[0].shmp);
+	chan->switch_timer_interval = switch_timer_interval;
+	chan->read_timer_interval = read_timer_interval;
+	lib_ring_buffer_channel_switch_timer_start(chan);
+	lib_ring_buffer_channel_read_timer_start(chan);
 
-		lib_ring_buffer_start_switch_timer(buf, handle);
-		lib_ring_buffer_start_read_timer(buf, handle);
-	}
-	ref = &handle->chan._ref;
-	shm_get_object_data(handle, ref, shm_fd, wait_fd, memory_map_size);
 	return handle;
 
 error_backend_init:
@@ -527,8 +794,9 @@ error_table_alloc:
 	return NULL;
 }
 
-struct lttng_ust_shm_handle *channel_handle_create(int shm_fd, int wait_fd,
-					uint64_t memory_map_size)
+struct lttng_ust_shm_handle *channel_handle_create(void *data,
+					uint64_t memory_map_size,
+					int wakeup_fd)
 {
 	struct lttng_ust_shm_handle *handle;
 	struct shm_object *object;
@@ -542,8 +810,8 @@ struct lttng_ust_shm_handle *channel_handle_create(int shm_fd, int wait_fd,
 	if (!handle->table)
 		goto error_table_alloc;
 	/* Add channel object */
-	object = shm_object_table_append_shadow(handle->table,
-			shm_fd, wait_fd, memory_map_size);
+	object = shm_object_table_append_mem(handle->table, data,
+			memory_map_size, wakeup_fd);
 	if (!object)
 		goto error_table_object;
 	/* struct channel is at object 0, offset 0 (hardcoded) */
@@ -559,23 +827,30 @@ error_table_alloc:
 }
 
 int channel_handle_add_stream(struct lttng_ust_shm_handle *handle,
-		int shm_fd, int wait_fd, uint64_t memory_map_size)
+		int shm_fd, int wakeup_fd, uint32_t stream_nr,
+		uint64_t memory_map_size)
 {
 	struct shm_object *object;
 
 	/* Add stream object */
-	object = shm_object_table_append_shadow(handle->table,
-			shm_fd, wait_fd, memory_map_size);
+	object = shm_object_table_append_shm(handle->table,
+			shm_fd, wakeup_fd, stream_nr,
+			memory_map_size);
 	if (!object)
-		return -1;
+		return -EINVAL;
 	return 0;
 }
 
+unsigned int channel_handle_get_nr_streams(struct lttng_ust_shm_handle *handle)
+{
+	assert(handle->table);
+	return handle->table->allocated_len - 1;
+}
+
 static
-void channel_release(struct channel *chan, struct lttng_ust_shm_handle *handle,
-		int shadow)
+void channel_release(struct channel *chan, struct lttng_ust_shm_handle *handle)
 {
-	channel_free(chan, handle, shadow);
+	channel_free(chan, handle);
 }
 
 /**
@@ -589,25 +864,25 @@ void channel_release(struct channel *chan, struct lttng_ust_shm_handle *handle,
  * They should release their handle at that point. 
  */
 void channel_destroy(struct channel *chan, struct lttng_ust_shm_handle *handle,
-		int shadow)
+		int consumer)
 {
-	if (shadow) {
-		channel_release(chan, handle, shadow);
-		return;
+	if (consumer) {
+		/*
+		 * Note: the consumer takes care of finalizing and
+		 * switching the buffers.
+		 */
+		channel_unregister_notifiers(chan, handle);
+		/*
+		 * The consumer prints errors.
+		 */
+		channel_print_errors(chan, handle);
 	}
 
-	channel_unregister_notifiers(chan, handle);
-
-	/*
-	 * Note: the consumer takes care of finalizing and switching the
-	 * buffers.
-	 */
-
 	/*
 	 * sessiond/consumer are keeping a reference on the shm file
 	 * descriptor directly. No need to refcount.
 	 */
-	channel_release(chan, handle, shadow);
+	channel_release(chan, handle);
 	return;
 }
 
@@ -615,36 +890,88 @@ struct lttng_ust_lib_ring_buffer *channel_get_ring_buffer(
 					const struct lttng_ust_lib_ring_buffer_config *config,
 					struct channel *chan, int cpu,
 					struct lttng_ust_shm_handle *handle,
-					int **shm_fd, int **wait_fd,
-					uint64_t **memory_map_size)
+					int *shm_fd, int *wait_fd,
+					int *wakeup_fd,
+					uint64_t *memory_map_size)
 {
 	struct shm_ref *ref;
 
 	if (config->alloc == RING_BUFFER_ALLOC_GLOBAL) {
-		ref = &chan->backend.buf[0].shmp._ref;
-		shm_get_object_data(handle, ref, shm_fd, wait_fd,
-			memory_map_size);
-		return shmp(handle, chan->backend.buf[0].shmp);
+		cpu = 0;
 	} else {
 		if (cpu >= num_possible_cpus())
 			return NULL;
-		ref = &chan->backend.buf[cpu].shmp._ref;
-		shm_get_object_data(handle, ref, shm_fd, wait_fd,
-			memory_map_size);
-		return shmp(handle, chan->backend.buf[cpu].shmp);
 	}
+	ref = &chan->backend.buf[cpu].shmp._ref;
+	*shm_fd = shm_get_shm_fd(handle, ref);
+	*wait_fd = shm_get_wait_fd(handle, ref);
+	*wakeup_fd = shm_get_wakeup_fd(handle, ref);
+	if (shm_get_shm_size(handle, ref, memory_map_size))
+		return NULL;
+	return shmp(handle, chan->backend.buf[cpu].shmp);
 }
 
-int lib_ring_buffer_open_read(struct lttng_ust_lib_ring_buffer *buf,
-			      struct lttng_ust_shm_handle *handle,
-			      int shadow)
+int ring_buffer_channel_close_wait_fd(const struct lttng_ust_lib_ring_buffer_config *config,
+			struct channel *chan,
+			struct lttng_ust_shm_handle *handle)
 {
-	if (shadow) {
-		if (uatomic_cmpxchg(&buf->active_shadow_readers, 0, 1) != 0)
-			return -EBUSY;
-		cmm_smp_mb();
-		return 0;
+	struct shm_ref *ref;
+
+	ref = &handle->chan._ref;
+	return shm_close_wait_fd(handle, ref);
+}
+
+int ring_buffer_channel_close_wakeup_fd(const struct lttng_ust_lib_ring_buffer_config *config,
+			struct channel *chan,
+			struct lttng_ust_shm_handle *handle)
+{
+	struct shm_ref *ref;
+
+	ref = &handle->chan._ref;
+	return shm_close_wakeup_fd(handle, ref);
+}
+
+int ring_buffer_stream_close_wait_fd(const struct lttng_ust_lib_ring_buffer_config *config,
+			struct channel *chan,
+			struct lttng_ust_shm_handle *handle,
+			int cpu)
+{
+	struct shm_ref *ref;
+
+	if (config->alloc == RING_BUFFER_ALLOC_GLOBAL) {
+		cpu = 0;
+	} else {
+		if (cpu >= num_possible_cpus())
+			return -EINVAL;
 	}
+	ref = &chan->backend.buf[cpu].shmp._ref;
+	return shm_close_wait_fd(handle, ref);
+}
+
+int ring_buffer_stream_close_wakeup_fd(const struct lttng_ust_lib_ring_buffer_config *config,
+			struct channel *chan,
+			struct lttng_ust_shm_handle *handle,
+			int cpu)
+{
+	struct shm_ref *ref;
+	int ret;
+
+	if (config->alloc == RING_BUFFER_ALLOC_GLOBAL) {
+		cpu = 0;
+	} else {
+		if (cpu >= num_possible_cpus())
+			return -EINVAL;
+	}
+	ref = &chan->backend.buf[cpu].shmp._ref;
+	pthread_mutex_lock(&wakeup_fd_mutex);
+	ret = shm_close_wakeup_fd(handle, ref);
+	pthread_mutex_unlock(&wakeup_fd_mutex);
+	return ret;
+}
+
+int lib_ring_buffer_open_read(struct lttng_ust_lib_ring_buffer *buf,
+			      struct lttng_ust_shm_handle *handle)
+{
 	if (uatomic_cmpxchg(&buf->active_readers, 0, 1) != 0)
 		return -EBUSY;
 	cmm_smp_mb();
@@ -652,17 +979,10 @@ int lib_ring_buffer_open_read(struct lttng_ust_lib_ring_buffer *buf,
 }
 
 void lib_ring_buffer_release_read(struct lttng_ust_lib_ring_buffer *buf,
-				  struct lttng_ust_shm_handle *handle,
-				  int shadow)
+				  struct lttng_ust_shm_handle *handle)
 {
 	struct channel *chan = shmp(handle, buf->backend.chan);
 
-	if (shadow) {
-		CHAN_WARN_ON(chan, uatomic_read(&buf->active_shadow_readers) != 1);
-		cmm_smp_mb();
-		uatomic_dec(&buf->active_shadow_readers);
-		return;
-	}
 	CHAN_WARN_ON(chan, uatomic_read(&buf->active_readers) != 1);
 	cmm_smp_mb();
 	uatomic_dec(&buf->active_readers);
@@ -728,7 +1048,7 @@ nodata:
 }
 
 /**
- * lib_ring_buffer_put_snapshot - move consumed counter forward
+ * lib_ring_buffer_move_consumer - move consumed counter forward
  * @buf: ring buffer
  * @consumed_new: new consumed count value
  */
@@ -740,8 +1060,7 @@ void lib_ring_buffer_move_consumer(struct lttng_ust_lib_ring_buffer *buf,
 	struct channel *chan = shmp(handle, bufb->chan);
 	unsigned long consumed;
 
-	CHAN_WARN_ON(chan, uatomic_read(&buf->active_readers) != 1
-			&& uatomic_read(&buf->active_shadow_readers) != 1);
+	CHAN_WARN_ON(chan, uatomic_read(&buf->active_readers) != 1);
 
 	/*
 	 * Only push the consumed value forward.
@@ -864,8 +1183,7 @@ void lib_ring_buffer_put_subbuf(struct lttng_ust_lib_ring_buffer *buf,
 	const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
 	unsigned long read_sb_bindex, consumed_idx, consumed;
 
-	CHAN_WARN_ON(chan, uatomic_read(&buf->active_readers) != 1
-			&& uatomic_read(&buf->active_shadow_readers) != 1);
+	CHAN_WARN_ON(chan, uatomic_read(&buf->active_readers) != 1);
 
 	if (!buf->get_subbuf) {
 		/*
@@ -975,29 +1293,36 @@ void lib_ring_buffer_print_buffer_errors(struct lttng_ust_lib_ring_buffer *buf,
 
 static
 void lib_ring_buffer_print_errors(struct channel *chan,
-				  struct lttng_ust_lib_ring_buffer *buf, int cpu,
-				  struct lttng_ust_shm_handle *handle)
+				struct lttng_ust_lib_ring_buffer *buf, int cpu,
+				struct lttng_ust_shm_handle *handle)
 {
 	const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
 	void *priv = channel_get_private(chan);
 
-	DBG("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))
-		DBG("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));
-
+	if (!strcmp(chan->backend.name, "relay-metadata-mmap")) {
+		DBG("ring buffer %s: %lu records written, "
+			"%lu records overrun\n",
+			chan->backend.name,
+			v_read(config, &buf->records_count),
+			v_read(config, &buf->records_overrun));
+	} else {
+		DBG("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))
+			DBG("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));
+	}
 	lib_ring_buffer_print_buffer_errors(buf, chan, priv, cpu, handle);
 }
 
@@ -1187,6 +1512,19 @@ int lib_ring_buffer_try_switch_slow(enum switch_mode mode,
 	 */
 	if (mode == SWITCH_FLUSH || off > 0) {
 		if (caa_unlikely(off == 0)) {
+			/*
+			 * A final flush that encounters an empty
+			 * sub-buffer cannot switch buffer if a
+			 * reader is located within this sub-buffer.
+			 * Anyway, the purpose of final flushing of a
+			 * sub-buffer at offset 0 is to handle the case
+			 * of entirely empty stream.
+			 */
+			if (caa_unlikely(subbuf_trunc(offsets->begin, chan)
+					 - subbuf_trunc((unsigned long)
+					     uatomic_read(&buf->consumed), chan)
+					>= chan->backend.buf_size))
+				return -1;
 			/*
 			 * The client does not save any header information.
 			 * Don't switch empty subbuffer on finalize, because it
@@ -1507,3 +1845,28 @@ int lib_ring_buffer_reserve_slow(struct lttng_ust_lib_ring_buffer_ctx *ctx)
 	ctx->buf_offset = offsets.begin + offsets.pre_header_padding;
 	return 0;
 }
+
+/*
+ * Force a read (imply TLS fixup for dlopen) of TLS variables.
+ */
+void lttng_fixup_ringbuffer_tls(void)
+{
+	asm volatile ("" : : "m" (URCU_TLS(lib_ring_buffer_nesting)));
+}
+
+void lib_ringbuffer_signal_init(void)
+{
+	sigset_t mask;
+	int ret;
+
+	/*
+	 * Block signal for entire process, so only our thread processes
+	 * it.
+	 */
+	rb_setmask(&mask);
+	ret = pthread_sigmask(SIG_BLOCK, &mask, NULL);
+	if (ret) {
+		errno = ret;
+		PERROR("pthread_sigmask");
+	}
+}