X-Git-Url: http://git.liburcu.org/?a=blobdiff_plain;f=libringbuffer%2Fring_buffer_frontend.c;h=3cba68d4e7ef354b9d9ddd842378459fa975a4d1;hb=a9ff648cc;hp=6f94040e9b186ca9a2788c3875d9d453bddde8c6;hpb=74d81a6cca2cd4a7718bba9368f382f9f2fbba84;p=lttng-ust.git

diff --git a/libringbuffer/ring_buffer_frontend.c b/libringbuffer/ring_buffer_frontend.c
index 6f94040e..3cba68d4 100644
--- a/libringbuffer/ring_buffer_frontend.c
+++ b/libringbuffer/ring_buffer_frontend.c
@@ -55,9 +55,14 @@
 #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 <poll.h>
 #include <helper.h>
 
 #include "smp.h"
@@ -76,6 +81,18 @@
 /* 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
+#define LTTNG_UST_RING_BUFFER_GET_RETRY		10
+#define LTTNG_UST_RING_BUFFER_RETRY_DELAY_MS	10
+
+/*
+ * Non-static to ensure the compiler does not optimize away the xor.
+ */
+uint8_t lttng_crash_magic_xor[] = RB_CRASH_DUMP_ABI_MAGIC_XOR;
+
 /*
  * Use POSIX SHM: shm_open(3) and shm_unlink(3).
  * close(2) to close the fd returned by shm_open.
@@ -102,17 +119,38 @@ 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 lock within the structure.
+ */
+struct timer_signal_data {
+	pthread_t tid;	/* thread id managing signals */
+	int setup_done;
+	int qs_done;
+	pthread_mutex_t lock;
+};
+
+static struct timer_signal_data timer_signal = {
+	.tid = 0,
+	.setup_done = 0,
+	.qs_done = 0,
+	.lock = PTHREAD_MUTEX_INITIALIZER,
+};
 
 /**
  * lib_ring_buffer_reset - Reset ring buffer to initial values.
@@ -175,6 +213,95 @@ void channel_reset(struct channel *chan)
 	/* Don't reset reader reference count */
 }
 
+static
+void init_crash_abi(const struct lttng_ust_lib_ring_buffer_config *config,
+		struct lttng_crash_abi *crash_abi,
+		struct lttng_ust_lib_ring_buffer *buf,
+		struct channel_backend *chanb,
+		struct shm_object *shmobj,
+		struct lttng_ust_shm_handle *handle)
+{
+	int i;
+
+	for (i = 0; i < RB_CRASH_DUMP_ABI_MAGIC_LEN; i++)
+		crash_abi->magic[i] = lttng_crash_magic_xor[i] ^ 0xFF;
+	crash_abi->mmap_length = shmobj->memory_map_size;
+	crash_abi->endian = RB_CRASH_ENDIAN;
+	crash_abi->major = RB_CRASH_DUMP_ABI_MAJOR;
+	crash_abi->minor = RB_CRASH_DUMP_ABI_MINOR;
+	crash_abi->word_size = sizeof(unsigned long);
+	crash_abi->layout_type = LTTNG_CRASH_TYPE_UST;
+
+	/* Offset of fields */
+	crash_abi->offset.prod_offset =
+		(uint32_t) ((char *) &buf->offset - (char *) buf);
+	crash_abi->offset.consumed_offset =
+		(uint32_t) ((char *) &buf->consumed - (char *) buf);
+	crash_abi->offset.commit_hot_array =
+		(uint32_t) ((char *) shmp(handle, buf->commit_hot) - (char *) buf);
+	crash_abi->offset.commit_hot_seq =
+		offsetof(struct commit_counters_hot, seq);
+	crash_abi->offset.buf_wsb_array =
+		(uint32_t) ((char *) shmp(handle, buf->backend.buf_wsb) - (char *) buf);
+	crash_abi->offset.buf_wsb_id =
+		offsetof(struct lttng_ust_lib_ring_buffer_backend_subbuffer, id);
+	crash_abi->offset.sb_array =
+		(uint32_t) ((char *) shmp(handle, buf->backend.array) - (char *) buf);
+	crash_abi->offset.sb_array_shmp_offset =
+		offsetof(struct lttng_ust_lib_ring_buffer_backend_pages_shmp,
+			shmp._ref.offset);
+	crash_abi->offset.sb_backend_p_offset =
+		offsetof(struct lttng_ust_lib_ring_buffer_backend_pages,
+			p._ref.offset);
+
+	/* Field length */
+	crash_abi->length.prod_offset = sizeof(buf->offset);
+	crash_abi->length.consumed_offset = sizeof(buf->consumed);
+	crash_abi->length.commit_hot_seq =
+		sizeof(((struct commit_counters_hot *) NULL)->seq);
+	crash_abi->length.buf_wsb_id =
+		sizeof(((struct lttng_ust_lib_ring_buffer_backend_subbuffer *) NULL)->id);
+	crash_abi->length.sb_array_shmp_offset =
+		sizeof(((struct lttng_ust_lib_ring_buffer_backend_pages_shmp *) NULL)->shmp._ref.offset);
+	crash_abi->length.sb_backend_p_offset =
+		sizeof(((struct lttng_ust_lib_ring_buffer_backend_pages *) NULL)->p._ref.offset);
+
+	/* Array stride */
+	crash_abi->stride.commit_hot_array =
+		sizeof(struct commit_counters_hot);
+	crash_abi->stride.buf_wsb_array =
+		sizeof(struct lttng_ust_lib_ring_buffer_backend_subbuffer);
+	crash_abi->stride.sb_array =
+		sizeof(struct lttng_ust_lib_ring_buffer_backend_pages_shmp);
+
+	/* Buffer constants */
+	crash_abi->buf_size = chanb->buf_size;
+	crash_abi->subbuf_size = chanb->subbuf_size;
+	crash_abi->num_subbuf = chanb->num_subbuf;
+	crash_abi->mode = (uint32_t) chanb->config.mode;
+
+	if (config->cb.content_size_field) {
+		size_t offset, length;
+
+		config->cb.content_size_field(config, &offset, &length);
+		crash_abi->offset.content_size = offset;
+		crash_abi->length.content_size = length;
+	} else {
+		crash_abi->offset.content_size = 0;
+		crash_abi->length.content_size = 0;
+	}
+	if (config->cb.packet_size_field) {
+		size_t offset, length;
+
+		config->cb.packet_size_field(config, &offset, &length);
+		crash_abi->offset.packet_size = offset;
+		crash_abi->length.packet_size = length;
+	} else {
+		crash_abi->offset.packet_size = 0;
+		crash_abi->length.packet_size = 0;
+	}
+}
+
 /*
  * Must be called under cpu hotplug protection.
  */
@@ -194,18 +321,12 @@ int lib_ring_buffer_create(struct lttng_ust_lib_ring_buffer *buf,
 	if (buf->backend.allocated)
 		return 0;
 
-	ret = lib_ring_buffer_backend_create(&buf->backend, &chan->backend,
-			cpu, handle, shmobj);
-	if (ret)
-		return ret;
-
 	align_shm(shmobj, __alignof__(struct commit_counters_hot));
 	set_shmp(buf->commit_hot,
 		 zalloc_shm(shmobj,
 			sizeof(struct commit_counters_hot) * chan->backend.num_subbuf));
 	if (!shmp(handle, buf->commit_hot)) {
-		ret = -ENOMEM;
-		goto free_chanbuf;
+		return -ENOMEM;
 	}
 
 	align_shm(shmobj, __alignof__(struct commit_counters_cold));
@@ -217,6 +338,12 @@ int lib_ring_buffer_create(struct lttng_ust_lib_ring_buffer *buf,
 		goto free_commit;
 	}
 
+	ret = lib_ring_buffer_backend_create(&buf->backend, &chan->backend,
+			cpu, handle, shmobj);
+	if (ret) {
+		goto free_init;
+	}
+
 	/*
 	 * Write the subbuffer header for first subbuffer so we know the total
 	 * duration of data gathering.
@@ -227,12 +354,16 @@ int lib_ring_buffer_create(struct lttng_ust_lib_ring_buffer *buf,
 	tsc = config->cb.ring_buffer_clock_read(shmp(handle, buf->backend.chan));
 	config->cb.buffer_begin(buf, tsc, 0, handle);
 	v_add(config, subbuf_header_size, &shmp_index(handle, buf->commit_hot, 0)->cc);
+	v_add(config, subbuf_header_size, &shmp_index(handle, buf->commit_hot, 0)->seq);
 
 	if (config->cb.buffer_create) {
 		ret = config->cb.buffer_create(buf, priv, cpu, chanb->name, handle);
 		if (ret)
-			goto free_init;
+			goto free_chanbuf;
 	}
+
+	init_crash_abi(config, &buf->crash_abi, buf, chanb, shmobj, handle);
+
 	buf->backend.allocated = 1;
 	return 0;
 
@@ -245,163 +376,380 @@ 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))
-		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;
 
-	if (!chan->switch_timer_interval || buf->switch_timer_enabled)
-		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;
+	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_stop_switch_timer(struct lttng_ust_lib_ring_buffer *buf,
-			   struct lttng_ust_shm_handle *handle)
+static
+void lib_ring_buffer_channel_read_timer(int sig, siginfo_t *si, void *uc)
 {
-	struct channel *chan = shmp(handle, buf->backend.chan);
+	struct channel *chan;
 
-	if (!chan->switch_timer_interval || !buf->switch_timer_enabled)
-		return;
+	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;
 
-	//TODO
-	//del_timer_sync(&buf->switch_timer);
-	buf->switch_timer_enabled = 0;
+	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;
 }
 
-#if 0
 /*
- * Polling timer to check the channels for data.
+ * Ensure only a single thread listens on the timer signal.
  */
-static void read_buffer_timer(unsigned long data)
+static
+void lib_ring_buffer_setup_timer_thread(void)
 {
-	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;
+	pthread_t thread;
+	int ret;
 
-	CHAN_WARN_ON(chan, !buf->backend.allocated);
+	pthread_mutex_lock(&timer_signal.lock);
+	if (timer_signal.setup_done)
+		goto end;
 
-	if (uatomic_read(&buf->active_readers))
-	    && lib_ring_buffer_poll_deliver(config, buf, chan)) {
-		//TODO
-		//wake_up_interruptible(&buf->read_wait);
-		//wake_up_interruptible(&chan->read_wait);
+	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;
+end:
+	pthread_mutex_unlock(&timer_signal.lock);
+}
+
+/*
+ * Wait for signal-handling thread quiescent state.
+ */
+static
+void lib_ring_buffer_wait_signal_thread_qs(unsigned int signr)
+{
+	sigset_t pending_set;
+	int ret;
 
-	//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);
+	/*
+	 * We need to be the only thread interacting with the thread
+	 * that manages signals for teardown synchronization.
+	 */
+	pthread_mutex_lock(&timer_signal.lock);
+
+	/*
+	 * 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, signr))
+			break;
+		caa_cpu_relax();
+	}
+
+	/*
+	 * 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();
+
+	pthread_mutex_unlock(&timer_signal.lock);
 }
-#endif //0
 
-static void lib_ring_buffer_start_read_timer(struct lttng_ust_lib_ring_buffer *buf,
-			   struct lttng_ust_shm_handle *handle)
+static
+void lib_ring_buffer_channel_switch_timer_start(struct channel *chan)
+{
+	struct sigevent sev;
+	struct itimerspec its;
+	int ret;
+
+	if (!chan->switch_timer_interval || chan->switch_timer_enabled)
+		return;
+
+	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");
+	}
+}
+
+static
+void lib_ring_buffer_channel_switch_timer_stop(struct channel *chan)
+{
+	int ret;
+
+	if (!chan->switch_timer_interval || !chan->switch_timer_enabled)
+		return;
+
+	ret = timer_delete(chan->switch_timer);
+	if (ret == -1) {
+		PERROR("timer_delete");
+	}
+
+	lib_ring_buffer_wait_signal_thread_qs(LTTNG_UST_RB_SIG_FLUSH);
+
+	chan->switch_timer = 0;
+	chan->switch_timer_enabled = 0;
+}
+
+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)
+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;
+	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);
-	}
-	buf->read_timer_enabled = 0;
+	lib_ring_buffer_channel_do_read(chan);
+
+	lib_ring_buffer_wait_signal_thread_qs(LTTNG_UST_RB_SIG_READ);
+
+	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)
+static void channel_free(struct channel *chan,
+		struct lttng_ust_shm_handle *handle)
 {
 	channel_backend_free(&chan->backend, handle);
 	/* chan is freed by shm teardown */
@@ -425,6 +773,7 @@ static void channel_free(struct channel *chan, struct lttng_ust_shm_handle *hand
  *                         padding to let readers get those sub-buffers.
  *                         Used for live streaming.
  * @read_timer_interval: Time interval (in us) to wake up pending readers.
+ * @shm_path: Shared memory files path.
  *
  * Holds cpu hotplug.
  * Returns NULL on failure.
@@ -437,9 +786,10 @@ 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)
+		   unsigned int read_timer_interval,
+		   const char *shm_path)
 {
-	int ret, cpu;
+	int ret;
 	size_t shmsize, chansize;
 	struct channel *chan;
 	struct lttng_ust_shm_handle *handle;
@@ -474,7 +824,8 @@ struct lttng_ust_shm_handle *channel_create(const struct lttng_ust_lib_ring_buff
 	shmsize += priv_data_size;
 
 	/* Allocate normal memory for channel (not shared) */
-	shmobj = shm_object_table_alloc(handle->table, shmsize, SHM_OBJECT_MEM);
+	shmobj = shm_object_table_alloc(handle->table, shmsize, SHM_OBJECT_MEM,
+			NULL);
 	if (!shmobj)
 		goto error_append;
 	/* struct channel is at object 0, offset 0 (hardcoded) */
@@ -504,35 +855,19 @@ struct lttng_ust_shm_handle *channel_create(const struct lttng_ust_lib_ring_buff
 	}
 
 	ret = channel_backend_init(&chan->backend, name, config,
-				   subbuf_size, num_subbuf, handle);
+				   subbuf_size, num_subbuf, handle,
+				   shm_path);
 	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);
-	}
 	return handle;
 
 error_backend_init:
@@ -544,7 +879,8 @@ error_table_alloc:
 }
 
 struct lttng_ust_shm_handle *channel_handle_create(void *data,
-					uint64_t memory_map_size)
+					uint64_t memory_map_size,
+					int wakeup_fd)
 {
 	struct lttng_ust_shm_handle *handle;
 	struct shm_object *object;
@@ -559,7 +895,7 @@ struct lttng_ust_shm_handle *channel_handle_create(void *data,
 		goto error_table_alloc;
 	/* Add channel object */
 	object = shm_object_table_append_mem(handle->table, data,
-			memory_map_size);
+			memory_map_size, wakeup_fd);
 	if (!object)
 		goto error_table_object;
 	/* struct channel is at object 0, offset 0 (hardcoded) */
@@ -620,6 +956,10 @@ void channel_destroy(struct channel *chan, struct lttng_ust_shm_handle *handle,
 		 * switching the buffers.
 		 */
 		channel_unregister_notifiers(chan, handle);
+		/*
+		 * The consumer prints errors.
+		 */
+		channel_print_errors(chan, handle);
 	}
 
 	/*
@@ -655,7 +995,27 @@ struct lttng_ust_lib_ring_buffer *channel_get_ring_buffer(
 	return shmp(handle, chan->backend.buf[cpu].shmp);
 }
 
-int ring_buffer_close_wait_fd(const struct lttng_ust_lib_ring_buffer_config *config,
+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)
+{
+	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)
@@ -672,12 +1032,13 @@ int ring_buffer_close_wait_fd(const struct lttng_ust_lib_ring_buffer_config *con
 	return shm_close_wait_fd(handle, ref);
 }
 
-int ring_buffer_close_wakeup_fd(const struct lttng_ust_lib_ring_buffer_config *config,
+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;
@@ -686,7 +1047,10 @@ int ring_buffer_close_wakeup_fd(const struct lttng_ust_lib_ring_buffer_config *c
 			return -EINVAL;
 	}
 	ref = &chan->backend.buf[cpu].shmp._ref;
-	return shm_close_wakeup_fd(handle, 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,
@@ -768,7 +1132,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
  */
@@ -808,8 +1172,7 @@ int lib_ring_buffer_get_subbuf(struct lttng_ust_lib_ring_buffer *buf,
 	struct channel *chan = shmp(handle, buf->backend.chan);
 	const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
 	unsigned long consumed_cur, consumed_idx, commit_count, write_offset;
-	int ret;
-	int finalized;
+	int ret, finalized, nr_retry = LTTNG_UST_RING_BUFFER_GET_RETRY;
 
 retry:
 	finalized = CMM_ACCESS_ONCE(buf->finalized);
@@ -844,20 +1207,72 @@ retry:
 
 	/*
 	 * Check that the subbuffer we are trying to consume has been
-	 * already fully committed.
+	 * already fully committed. There are a few causes that can make
+	 * this unavailability situation occur:
+	 *
+	 * Temporary (short-term) situation:
+	 * - Application is running on a different CPU, between reserve
+	 *   and commit ring buffer operations,
+	 * - Application is preempted between reserve and commit ring
+	 *   buffer operations,
+	 *
+	 * Long-term situation:
+	 * - Application is stopped (SIGSTOP) between reserve and commit
+	 *   ring buffer operations. Could eventually be resumed by
+	 *   SIGCONT.
+	 * - Application is killed (SIGTERM, SIGINT, SIGKILL) between
+	 *   reserve and commit ring buffer operation.
+	 *
+	 * From a consumer perspective, handling short-term
+	 * unavailability situations is performed by retrying a few
+	 * times after a delay. Handling long-term unavailability
+	 * situations is handled by failing to get the sub-buffer.
+	 *
+	 * In all of those situations, if the application is taking a
+	 * long time to perform its commit after ring buffer space
+	 * reservation, we can end up in a situation where the producer
+	 * will fill the ring buffer and try to write into the same
+	 * sub-buffer again (which has a missing commit). This is
+	 * handled by the producer in the sub-buffer switch handling
+	 * code of the reserve routine by detecting unbalanced
+	 * reserve/commit counters and discarding all further events
+	 * until the situation is resolved in those situations. Two
+	 * scenarios can occur:
+	 *
+	 * 1) The application causing the reserve/commit counters to be
+	 *    unbalanced has been terminated. In this situation, all
+	 *    further events will be discarded in the buffers, and no
+	 *    further buffer data will be readable by the consumer
+	 *    daemon. Tearing down the UST tracing session and starting
+	 *    anew is a work-around for those situations. Note that this
+	 *    only affects per-UID tracing. In per-PID tracing, the
+	 *    application vanishes with the termination, and therefore
+	 *    no more data needs to be written to the buffers.
+	 * 2) The application causing the unbalance has been delayed for
+	 *    a long time, but will eventually try to increment the
+	 *    commit counter after eventually writing to the sub-buffer.
+	 *    This situation can cause events to be discarded until the
+	 *    application resumes its operations.
 	 */
 	if (((commit_count - chan->backend.subbuf_size)
 	     & chan->commit_count_mask)
-	    - (buf_trunc(consumed_cur, chan)
+	    - (buf_trunc(consumed, chan)
 	       >> chan->backend.num_subbuf_order)
-	    != 0)
-		goto nodata;
+	    != 0) {
+		if (nr_retry-- > 0) {
+			if (nr_retry <= (LTTNG_UST_RING_BUFFER_GET_RETRY >> 1))
+				(void) poll(NULL, 0, LTTNG_UST_RING_BUFFER_RETRY_DELAY_MS);
+			goto retry;
+		} else {
+			goto nodata;
+		}
+	}
 
 	/*
 	 * Check that we are not about to read the same subbuffer in
 	 * which the writer head is.
 	 */
-	if (subbuf_trunc(write_offset, chan) - subbuf_trunc(consumed_cur, chan)
+	if (subbuf_trunc(write_offset, chan) - subbuf_trunc(consumed, chan)
 	    == 0)
 		goto nodata;
 
@@ -867,12 +1282,23 @@ retry:
 	 * the writer is getting access to a subbuffer we were trying to get
 	 * access to. Also checks that the "consumed" buffer count we are
 	 * looking for matches the one contained in the subbuffer id.
+	 *
+	 * The short-lived race window described here can be affected by
+	 * application signals and preemption, thus requiring to bound
+	 * the loop to a maximum number of retry.
 	 */
 	ret = update_read_sb_index(config, &buf->backend, &chan->backend,
 				   consumed_idx, buf_trunc_val(consumed, chan),
 				   handle);
-	if (ret)
-		goto retry;
+	if (ret) {
+		if (nr_retry-- > 0) {
+			if (nr_retry <= (LTTNG_UST_RING_BUFFER_GET_RETRY >> 1))
+				(void) poll(NULL, 0, LTTNG_UST_RING_BUFFER_RETRY_DELAY_MS);
+			goto retry;
+		} else {
+			goto nodata;
+		}
+	}
 	subbuffer_id_clear_noref(config, &buf->backend.buf_rsb.id);
 
 	buf->get_subbuf_consumed = consumed;
@@ -1013,8 +1439,8 @@ 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);
@@ -1074,11 +1500,10 @@ void lib_ring_buffer_switch_old_start(struct lttng_ust_lib_ring_buffer *buf,
 	commit_count = v_read(config, &shmp_index(handle, buf->commit_hot, oldidx)->cc);
 	/* Check if the written buffer has to be delivered */
 	lib_ring_buffer_check_deliver(config, buf, chan, offsets->old,
-				      commit_count, oldidx, handle);
+				      commit_count, oldidx, handle, tsc);
 	lib_ring_buffer_write_commit_counter(config, buf, chan, oldidx,
-					     offsets->old, commit_count,
-					     config->cb.subbuffer_header_size(),
-					     handle);
+			offsets->old + config->cb.subbuffer_header_size(),
+			commit_count, handle);
 }
 
 /*
@@ -1113,10 +1538,9 @@ void lib_ring_buffer_switch_old_end(struct lttng_ust_lib_ring_buffer *buf,
 	v_add(config, padding_size, &shmp_index(handle, buf->commit_hot, oldidx)->cc);
 	commit_count = v_read(config, &shmp_index(handle, buf->commit_hot, oldidx)->cc);
 	lib_ring_buffer_check_deliver(config, buf, chan, offsets->old - 1,
-				      commit_count, oldidx, handle);
+				      commit_count, oldidx, handle, tsc);
 	lib_ring_buffer_write_commit_counter(config, buf, chan, oldidx,
-					     offsets->old, commit_count,
-					     padding_size, handle);
+			offsets->old + padding_size, commit_count, handle);
 }
 
 /*
@@ -1149,18 +1573,19 @@ void lib_ring_buffer_switch_new_start(struct lttng_ust_lib_ring_buffer *buf,
 	commit_count = v_read(config, &shmp_index(handle, buf->commit_hot, beginidx)->cc);
 	/* Check if the written buffer has to be delivered */
 	lib_ring_buffer_check_deliver(config, buf, chan, offsets->begin,
-				      commit_count, beginidx, handle);
+				      commit_count, beginidx, handle, tsc);
 	lib_ring_buffer_write_commit_counter(config, buf, chan, beginidx,
-					     offsets->begin, commit_count,
-					     config->cb.subbuffer_header_size(),
-					     handle);
+			offsets->begin + config->cb.subbuffer_header_size(),
+			commit_count, handle);
 }
 
 /*
  * lib_ring_buffer_switch_new_end: finish switching current subbuffer
  *
- * The only remaining threads could be the ones with pending commits. They will
- * have to do the deliver themselves.
+ * Calls subbuffer_set_data_size() to set the data size of the current
+ * sub-buffer. We do not need to perform check_deliver nor commit here,
+ * since this task will be done by the "commit" of the event for which
+ * we are currently doing the space reservation.
  */
 static
 void lib_ring_buffer_switch_new_end(struct lttng_ust_lib_ring_buffer *buf,
@@ -1170,26 +1595,12 @@ void lib_ring_buffer_switch_new_end(struct lttng_ust_lib_ring_buffer *buf,
 				    struct lttng_ust_shm_handle *handle)
 {
 	const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
-	unsigned long endidx = subbuf_index(offsets->end - 1, chan);
-	unsigned long commit_count, padding_size, data_size;
+	unsigned long endidx, data_size;
 
+	endidx = subbuf_index(offsets->end - 1, chan);
 	data_size = subbuf_offset(offsets->end - 1, chan) + 1;
-	padding_size = chan->backend.subbuf_size - data_size;
 	subbuffer_set_data_size(config, &buf->backend, endidx, data_size,
 				handle);
-
-	/*
-	 * Order all writes to buffer before the commit count update that will
-	 * determine that the subbuffer is full.
-	 */
-	cmm_smp_wmb();
-	v_add(config, padding_size, &shmp_index(handle, buf->commit_hot, endidx)->cc);
-	commit_count = v_read(config, &shmp_index(handle, buf->commit_hot, endidx)->cc);
-	lib_ring_buffer_check_deliver(config, buf, chan, offsets->end - 1,
-				  commit_count, endidx, handle);
-	lib_ring_buffer_write_commit_counter(config, buf, chan, endidx,
-					     offsets->end, commit_count,
-					     padding_size, handle);
 }
 
 /*
@@ -1202,10 +1613,11 @@ int lib_ring_buffer_try_switch_slow(enum switch_mode mode,
 				    struct lttng_ust_lib_ring_buffer *buf,
 				    struct channel *chan,
 				    struct switch_offsets *offsets,
-				    uint64_t *tsc)
+				    uint64_t *tsc,
+				    struct lttng_ust_shm_handle *handle)
 {
 	const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
-	unsigned long off;
+	unsigned long off, reserve_commit_diff;
 
 	offsets->begin = v_read(config, &buf->offset);
 	offsets->old = offsets->begin;
@@ -1230,23 +1642,69 @@ int lib_ring_buffer_try_switch_slow(enum switch_mode mode,
 	 * timestamps) are visible to the reader. This is required for
 	 * quiescence guarantees for the fusion merge.
 	 */
-	if (mode == SWITCH_FLUSH || off > 0) {
-		if (caa_unlikely(off == 0)) {
-			/*
-			 * The client does not save any header information.
-			 * Don't switch empty subbuffer on finalize, because it
-			 * is invalid to deliver a completely empty subbuffer.
-			 */
-			if (!config->cb.subbuffer_header_size())
+	if (mode != SWITCH_FLUSH && !off)
+		return -1;	/* we do not have to switch : buffer is empty */
+
+	if (caa_unlikely(off == 0)) {
+		unsigned long sb_index, commit_count;
+
+		/*
+		 * We are performing a SWITCH_FLUSH. At this stage, there are no
+		 * concurrent writes into the buffer.
+		 *
+		 * The client does not save any header information.  Don't
+		 * switch empty subbuffer on finalize, because it is invalid to
+		 * deliver a completely empty subbuffer.
+		 */
+		if (!config->cb.subbuffer_header_size())
+			return -1;
+
+		/* Test new buffer integrity */
+		sb_index = subbuf_index(offsets->begin, chan);
+		commit_count = v_read(config,
+				&shmp_index(handle, buf->commit_cold,
+					sb_index)->cc_sb);
+		reserve_commit_diff =
+		  (buf_trunc(offsets->begin, chan)
+		   >> chan->backend.num_subbuf_order)
+		  - (commit_count & chan->commit_count_mask);
+		if (caa_likely(reserve_commit_diff == 0)) {
+			/* Next subbuffer not being written to. */
+			if (caa_unlikely(config->mode != RING_BUFFER_OVERWRITE &&
+				subbuf_trunc(offsets->begin, chan)
+				 - subbuf_trunc((unsigned long)
+				     uatomic_read(&buf->consumed), chan)
+				>= chan->backend.buf_size)) {
+				/*
+				 * We do not overwrite non consumed buffers
+				 * and we are full : don't switch.
+				 */
 				return -1;
+			} else {
+				/*
+				 * Next subbuffer not being written to, and we
+				 * are either in overwrite mode or the buffer is
+				 * not full. It's safe to write in this new
+				 * subbuffer.
+				 */
+			}
+		} else {
 			/*
-			 * Need to write the subbuffer start header on finalize.
+			 * Next subbuffer reserve offset does not match the
+			 * commit offset. Don't perform switch in
+			 * producer-consumer and overwrite mode.  Caused by
+			 * either a writer OOPS or too many nested writes over a
+			 * reserve/commit pair.
 			 */
-			offsets->switch_old_start = 1;
+			return -1;
 		}
-		offsets->begin = subbuf_align(offsets->begin, chan);
-	} else
-		return -1;	/* we do not have to switch : buffer is empty */
+
+		/*
+		 * Need to write the subbuffer start header on finalize.
+		 */
+		offsets->switch_old_start = 1;
+	}
+	offsets->begin = subbuf_align(offsets->begin, chan);
 	/* Note: old points to the next subbuf at offset 0 */
 	offsets->end = offsets->begin;
 	return 0;
@@ -1276,7 +1734,7 @@ void lib_ring_buffer_switch_slow(struct lttng_ust_lib_ring_buffer *buf, enum swi
 	 */
 	do {
 		if (lib_ring_buffer_try_switch_slow(mode, buf, chan, &offsets,
-						    &tsc))
+						    &tsc, handle))
 			return;	/* Switch not needed */
 	} while (v_cmpxchg(config, &buf->offset, offsets.old, offsets.end)
 		 != offsets.old);
@@ -1326,9 +1784,10 @@ int lib_ring_buffer_try_reserve_slow(struct lttng_ust_lib_ring_buffer *buf,
 {
 	const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
 	struct lttng_ust_shm_handle *handle = ctx->handle;
-	unsigned long reserve_commit_diff;
+	unsigned long reserve_commit_diff, offset_cmp;
 
-	offsets->begin = v_read(config, &buf->offset);
+retry:
+	offsets->begin = offset_cmp = v_read(config, &buf->offset);
 	offsets->old = offsets->begin;
 	offsets->switch_new_start = 0;
 	offsets->switch_new_end = 0;
@@ -1360,7 +1819,7 @@ int lib_ring_buffer_try_reserve_slow(struct lttng_ust_lib_ring_buffer *buf,
 		}
 	}
 	if (caa_unlikely(offsets->switch_new_start)) {
-		unsigned long sb_index;
+		unsigned long sb_index, commit_count;
 
 		/*
 		 * We are typically not filling the previous buffer completely.
@@ -1371,12 +1830,32 @@ int lib_ring_buffer_try_reserve_slow(struct lttng_ust_lib_ring_buffer *buf,
 				 + config->cb.subbuffer_header_size();
 		/* Test new buffer integrity */
 		sb_index = subbuf_index(offsets->begin, chan);
+		/*
+		 * Read buf->offset before buf->commit_cold[sb_index].cc_sb.
+		 * lib_ring_buffer_check_deliver() has the matching
+		 * memory barriers required around commit_cold cc_sb
+		 * updates to ensure reserve and commit counter updates
+		 * are not seen reordered when updated by another CPU.
+		 */
+		cmm_smp_rmb();
+		commit_count = v_read(config,
+				&shmp_index(handle, buf->commit_cold,
+					sb_index)->cc_sb);
+		/* Read buf->commit_cold[sb_index].cc_sb before buf->offset. */
+		cmm_smp_rmb();
+		if (caa_unlikely(offset_cmp != v_read(config, &buf->offset))) {
+			/*
+			 * The reserve counter have been concurrently updated
+			 * while we read the commit counter. This means the
+			 * commit counter we read might not match buf->offset
+			 * due to concurrent update. We therefore need to retry.
+			 */
+			goto retry;
+		}
 		reserve_commit_diff =
 		  (buf_trunc(offsets->begin, chan)
 		   >> chan->backend.num_subbuf_order)
-		  - ((unsigned long) v_read(config,
-					    &shmp_index(handle, buf->commit_cold, sb_index)->cc_sb)
-		     & chan->commit_count_mask);
+		  - (commit_count & chan->commit_count_mask);
 		if (caa_likely(reserve_commit_diff == 0)) {
 			/* Next subbuffer not being written to. */
 			if (caa_unlikely(config->mode != RING_BUFFER_OVERWRITE &&
@@ -1411,7 +1890,8 @@ int lib_ring_buffer_try_reserve_slow(struct lttng_ust_lib_ring_buffer *buf,
 
 			/*
 			 * Next subbuffer reserve offset does not match the
-			 * commit offset. Drop record in producer-consumer and
+			 * commit offset, and this did not involve update to the
+			 * reserve counter. Drop record in producer-consumer and
 			 * overwrite mode. Caused by either a writer OOPS or too
 			 * many nested writes over a reserve/commit pair.
 			 */
@@ -1558,5 +2038,22 @@ int lib_ring_buffer_reserve_slow(struct lttng_ust_lib_ring_buffer_ctx *ctx)
  */
 void lttng_fixup_ringbuffer_tls(void)
 {
-	asm volatile ("" : : "m" (lib_ring_buffer_nesting));
+	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");
+	}
 }