[lttng-modules.git] / ltt-relay-lockless.c

/*
 * ltt/ltt-relay-lockless.c
 *
 * (C) Copyright 2005-2008 - Mathieu Desnoyers (mathieu.desnoyers@polymtl.ca)
 *
 * LTTng lockless buffer space management (reader/writer).
 *
 * Author:
 *	Mathieu Desnoyers (mathieu.desnoyers@polymtl.ca)
 *
 * Inspired from LTT :
 *  Karim Yaghmour (karim@opersys.com)
 *  Tom Zanussi (zanussi@us.ibm.com)
 *  Bob Wisniewski (bob@watson.ibm.com)
 * And from K42 :
 *  Bob Wisniewski (bob@watson.ibm.com)
 *
 * Changelog:
 *  08/10/08, Cleanup.
 *  19/10/05, Complete lockless mechanism.
 *  27/05/05, Modular redesign and rewrite.
 *
 * Userspace reader semantic :
 * while (poll fd != POLLHUP) {
 *   - ioctl RELAY_GET_SUBBUF_SIZE
 *   while (1) {
 *     - ioctl GET_SUBBUF
 *     - splice 1 subbuffer worth of data to a pipe
 *     - splice the data from pipe to disk/network
 *     - ioctl PUT_SUBBUF, check error value
 *       if err val < 0, previous subbuffer was corrupted.
 *   }
 * }
 *
 * Dual LGPL v2.1/GPL v2 license.
 */

#include <linux/time.h>
#include <linux/module.h>
#include <linux/string.h>
#include <linux/slab.h>
#include <linux/init.h>
#include <linux/rcupdate.h>
#include <linux/timer.h>
#include <linux/sched.h>
#include <linux/bitops.h>
#include <linux/smp_lock.h>
#include <linux/stat.h>
#include <linux/cpu.h>
#include <linux/idle.h>
#include <linux/delay.h>
#include <linux/notifier.h>
#include <asm/atomic.h>
#include <asm/local.h>

#include "ltt-tracer.h"
#include "ltt-relay.h"
#include "ltt-relay-lockless.h"

#if 0
#define printk_dbg(fmt, args...) printk(fmt, args)
#else
#define printk_dbg(fmt, args...)
#endif

struct ltt_reserve_switch_offsets {
	long begin, end, old;
	long begin_switch, end_switch_current, end_switch_old;
	size_t before_hdr_pad, size;
};

static
void ltt_force_switch(struct ltt_chanbuf *buf, enum force_switch_mode mode);

static
void ltt_relay_print_buffer_errors(struct ltt_chan *chan, unsigned int cpu);

static const struct file_operations ltt_file_operations;

static
void ltt_buffer_begin(struct ltt_chanbuf *buf, u64 tsc, unsigned int subbuf_idx)
{
	struct ltt_chan *chan = container_of(buf->a.chan, struct ltt_chan, a);
	struct ltt_subbuffer_header *header =
		(struct ltt_subbuffer_header *)
			ltt_relay_offset_address(&buf->a,
				subbuf_idx * chan->a.sb_size);

	header->cycle_count_begin = tsc;
	header->data_size = 0xFFFFFFFF; /* for debugging */
	ltt_write_trace_header(chan->a.trace, header);
}

/*
 * offset is assumed to never be 0 here : never deliver a completely empty
 * subbuffer. The lost size is between 0 and subbuf_size-1.
 */
static
void ltt_buffer_end(struct ltt_chanbuf *buf, u64 tsc, unsigned int offset,
		    unsigned int subbuf_idx)
{
	struct ltt_chan *chan = container_of(buf->a.chan, struct ltt_chan, a);
	struct ltt_subbuffer_header *header =
		(struct ltt_subbuffer_header *)
			ltt_relay_offset_address(&buf->a,
				subbuf_idx * chan->a.sb_size);
	u32 data_size = SUBBUF_OFFSET(offset - 1, chan) + 1;

	header->data_size = data_size;
	header->sb_size = PAGE_ALIGN(data_size);
	header->cycle_count_end = tsc;
	header->events_lost = local_read(&buf->events_lost);
	header->subbuf_corrupt = local_read(&buf->corrupted_subbuffers);
}

/*
 * Must be called under trace lock or cpu hotplug protection.
 */
void ltt_chanbuf_free(struct ltt_chanbuf *buf)
{
	struct ltt_chan *chan = container_of(buf->a.chan, struct ltt_chan, a);

	ltt_relay_print_buffer_errors(chan, buf->a.cpu);
#ifdef LTT_VMCORE
	kfree(buf->commit_seq);
#endif
	kfree(buf->commit_count);

	ltt_chanbuf_alloc_free(&buf->a);
}

/*
 * Must be called under trace lock or cpu hotplug protection.
 */
int ltt_chanbuf_create(struct ltt_chanbuf *buf, struct ltt_chan_alloc *chana,
		       int cpu)
{
	struct ltt_chan *chan = container_of(chana, struct ltt_chan, a);
	struct ltt_trace *trace = chana->trace;
	unsigned int j, n_sb;
	int ret;

	/* Test for cpu hotplug */
	if (buf->a.allocated)
		return 0;

	ret = ltt_chanbuf_alloc_create(&buf->a, &chan->a, cpu);
	if (ret)
		return ret;

	buf->commit_count =
		kzalloc_node(ALIGN(sizeof(*buf->commit_count) * chan->a.n_sb,
				   1 << INTERNODE_CACHE_SHIFT),
			GFP_KERNEL, cpu_to_node(cpu));
	if (!buf->commit_count) {
		ret = -ENOMEM;
		goto free_chanbuf;
	}

#ifdef LTT_VMCORE
	buf->commit_seq =
		kzalloc_node(ALIGN(sizeof(*buf->commit_seq) * chan->a.n_sb,
				   1 << INTERNODE_CACHE_SHIFT),
			GFP_KERNEL, cpu_to_node(cpu));
	if (!buf->commit_seq) {
		kfree(buf->commit_count);
		ret = -ENOMEM;
		goto free_commit;
	}
#endif

	local_set(&buf->offset, ltt_sb_header_size());
	atomic_long_set(&buf->consumed, 0);
	atomic_long_set(&buf->active_readers, 0);
	n_sb = chan->a.n_sb;
	for (j = 0; j < n_sb; j++) {
		local_set(&buf->commit_count[j].cc, 0);
		local_set(&buf->commit_count[j].cc_sb, 0);
		local_set(&buf->commit_count[j].events, 0);
	}
	init_waitqueue_head(&buf->write_wait);
	init_waitqueue_head(&buf->read_wait);
	spin_lock_init(&buf->full_lock);

	RCHAN_SB_CLEAR_NOREF(buf->a.buf_wsb[0].pages);
	ltt_buffer_begin(buf, trace->start_tsc, 0);
	/* atomic_add made on local variable on data that belongs to
	 * various CPUs : ok because tracing not started (for this cpu). */
	local_add(ltt_sb_header_size(), &buf->commit_count[0].cc);

	local_set(&buf->events_lost, 0);
	local_set(&buf->corrupted_subbuffers, 0);
	buf->finalized = 0;

	ret = ltt_chanbuf_create_file(chan->a.filename, chan->a.parent,
				      S_IRUSR, buf);
	if (ret)
		goto free_init;

	/*
	 * Ensure the buffer is ready before setting it to allocated.
	 * Used for cpu hotplug vs async wakeup.
	 */
	smp_wmb();
	buf->a.allocated = 1;

	return 0;

	/* Error handling */
free_init:
#ifdef LTT_VMCORE
	kfree(buf->commit_seq);
free_commit:
#endif
	kfree(buf->commit_count);
free_chanbuf:
	ltt_chanbuf_alloc_free(&buf->a);
	return ret;
}

void ltt_chan_remove_files(struct ltt_chan *chan)
{
	ltt_ascii_remove(chan);
	ltt_chan_alloc_remove_files(&chan->a);
}
EXPORT_SYMBOL_GPL(ltt_chan_remove_files);


void ltt_chan_free(struct kref *kref)
{
	struct ltt_chan *chan = container_of(kref, struct ltt_chan, a.kref);

	ltt_chan_alloc_free(&chan->a);
}
EXPORT_SYMBOL_GPL(ltt_chan_free);

/**
 * ltt_chan_create - Create channel.
 */
int ltt_chan_create(const char *base_filename,
		    struct ltt_chan *chan, struct dentry *parent,
		    size_t sb_size, size_t n_sb,
		    int overwrite, struct ltt_trace *trace)
{
	int ret;

	chan->overwrite = overwrite;

	ret = ltt_chan_alloc_init(&chan->a, trace, base_filename, parent,
				  sb_size, n_sb, overwrite, overwrite);
	if (ret)
		goto error;

	chan->commit_count_mask = (~0UL >> chan->a.n_sb_order);

	ret = ltt_ascii_create(chan);
	if (ret)
		goto error_chan_alloc_free;

	return ret;

error_chan_alloc_free:
	ltt_chan_alloc_free(&chan->a);
error:
	return ret;
}
EXPORT_SYMBOL_GPL(ltt_chan_create);

int ltt_chanbuf_open_read(struct ltt_chanbuf *buf)
{
	kref_get(&buf->a.chan->kref);
	if (!atomic_long_add_unless(&buf->active_readers, 1, 1)) {
		kref_put(&buf->a.chan->kref, ltt_chan_free);
		return -EBUSY;
	}

	return 0;
}
EXPORT_SYMBOL_GPL(ltt_chanbuf_open_read);

void ltt_chanbuf_release_read(struct ltt_chanbuf *buf)
{
	//ltt_relay_destroy_buffer(&buf->a.chan->a, buf->a.cpu);
	WARN_ON(atomic_long_read(&buf->active_readers) != 1);
	atomic_long_dec(&buf->active_readers);
	kref_put(&buf->a.chan->kref, ltt_chan_free);
}
EXPORT_SYMBOL_GPL(ltt_chanbuf_release_read);

/*
 * Wake writers :
 *
 * This must be done after the trace is removed from the RCU list so that there
 * are no stalled writers.
 */
static void ltt_relay_wake_writers(struct ltt_chanbuf *buf)
{

	if (waitqueue_active(&buf->write_wait))
		wake_up_interruptible(&buf->write_wait);
}

/*
 * This function should not be called from NMI interrupt context
 */
static void ltt_buf_unfull(struct ltt_chanbuf *buf)
{
	ltt_relay_wake_writers(buf);
}

/*
 * Promote compiler barrier to a smp_mb().
 * For the specific LTTng case, this IPI call should be removed if the
 * architecture does not reorder writes.  This should eventually be provided by
 * a separate architecture-specific infrastructure.
 */
static void remote_mb(void *info)
{
	smp_mb();
}

int ltt_chanbuf_get_subbuf(struct ltt_chanbuf *buf, unsigned long *consumed)
{
	struct ltt_chan *chan = container_of(buf->a.chan, struct ltt_chan, a);
	long consumed_old, consumed_idx, commit_count, write_offset;
	int ret;

	consumed_old = atomic_long_read(&buf->consumed);
	consumed_idx = SUBBUF_INDEX(consumed_old, chan);
	commit_count = local_read(&buf->commit_count[consumed_idx].cc_sb);
	/*
	 * Make sure we read the commit count before reading the buffer
	 * data and the write offset. Correct consumed offset ordering
	 * wrt commit count is insured by the use of cmpxchg to update
	 * the consumed offset.
	 * smp_call_function_single can fail if the remote CPU is offline,
	 * this is OK because then there is no wmb to execute there.
	 * If our thread is executing on the same CPU as the on the buffers
	 * belongs to, we don't have to synchronize it at all. If we are
	 * migrated, the scheduler will take care of the memory barriers.
	 * Normally, smp_call_function_single() should ensure program order when
	 * executing the remote function, which implies that it surrounds the
	 * function execution with :
	 * smp_mb()
	 * send IPI
	 * csd_lock_wait
	 *                recv IPI
	 *                smp_mb()
	 *                exec. function
	 *                smp_mb()
	 *                csd unlock
	 * smp_mb()
	 *
	 * However, smp_call_function_single() does not seem to clearly execute
	 * such barriers. It depends on spinlock semantic to provide the barrier
	 * before executing the IPI and, when busy-looping, csd_lock_wait only
	 * executes smp_mb() when it has to wait for the other CPU.
	 *
	 * I don't trust this code. Therefore, let's add the smp_mb() sequence
	 * required ourself, even if duplicated. It has no performance impact
	 * anyway.
	 *
	 * smp_mb() is needed because smp_rmb() and smp_wmb() only order read vs
	 * read and write vs write. They do not ensure core synchronization. We
	 * really have to ensure total order between the 3 barriers running on
	 * the 2 CPUs.
	 */
#ifdef LTT_NO_IPI_BARRIER
	/*
	 * Local rmb to match the remote wmb to read the commit count before the
	 * buffer data and the write offset.
	 */
	smp_rmb();
#else
	if (raw_smp_processor_id() != buf->a.cpu) {
		smp_mb();	/* Total order with IPI handler smp_mb() */
		smp_call_function_single(buf->a.cpu, remote_mb, NULL, 1);
		smp_mb();	/* Total order with IPI handler smp_mb() */
	}
#endif
	write_offset = local_read(&buf->offset);
	/*
	 * Check that the subbuffer we are trying to consume has been
	 * already fully committed.
	 */
	if (((commit_count - chan->a.sb_size)
	     & chan->commit_count_mask)
	    - (BUFFER_TRUNC(consumed_old, chan)
	       >> chan->a.n_sb_order)
	    != 0) {
		return -EAGAIN;
	}
	/*
	 * Check that we are not about to read the same subbuffer in
	 * which the writer head is.
	 */
	if ((SUBBUF_TRUNC(write_offset, chan)
	   - SUBBUF_TRUNC(consumed_old, chan))
	   == 0) {
		return -EAGAIN;
	}

	ret = update_read_sb_index(&buf->a, &chan->a, consumed_idx);
	if (ret)
		return ret;

	*consumed = consumed_old;
	return 0;
}
EXPORT_SYMBOL_GPL(ltt_chanbuf_get_subbuf);

int ltt_chanbuf_put_subbuf(struct ltt_chanbuf *buf, unsigned long consumed)
{
	struct ltt_chan *chan = container_of(buf->a.chan, struct ltt_chan, a);
	long consumed_new, consumed_old;

	WARN_ON(atomic_long_read(&buf->active_readers) != 1);

	consumed_old = consumed;
	consumed_new = SUBBUF_ALIGN(consumed_old, chan);
	WARN_ON_ONCE(RCHAN_SB_IS_NOREF(buf->a.buf_rsb.pages));
	RCHAN_SB_SET_NOREF(buf->a.buf_rsb.pages);

	spin_lock(&buf->full_lock);
	if (atomic_long_cmpxchg(&buf->consumed, consumed_old, consumed_new)
	    != consumed_old) {
		/* We have been pushed by the writer. */
		spin_unlock(&buf->full_lock);
		/*
		 * We exchanged the subbuffer pages. No corruption possible
		 * even if the writer did push us. No more -EIO possible.
		 */
		return 0;
	} else {
		/* tell the client that buffer is now unfull */
		int index;
		long data;
		index = SUBBUF_INDEX(consumed_old, chan);
		data = BUFFER_OFFSET(consumed_old, chan);
		ltt_buf_unfull(buf);
		spin_unlock(&buf->full_lock);
	}
	return 0;
}
EXPORT_SYMBOL_GPL(ltt_chanbuf_put_subbuf);

static void switch_buffer(unsigned long data)
{
	struct ltt_chanbuf *buf = (struct ltt_chanbuf *)data;
	struct ltt_chan *chan = container_of(buf->a.chan, struct ltt_chan, a);

	/*
	 * Only flush buffers periodically if readers are active.
	 */
	if (atomic_long_read(&buf->active_readers))
		ltt_force_switch(buf, FORCE_ACTIVE);

	mod_timer_pinned(&buf->switch_timer,
			 jiffies + chan->switch_timer_interval);
}

static void ltt_chanbuf_start_switch_timer(struct ltt_chanbuf *buf)
{
	struct ltt_chan *chan = container_of(buf->a.chan, struct ltt_chan, a);

	if (!chan->switch_timer_interval)
		return;

	init_timer_deferrable(&buf->switch_timer);
	buf->switch_timer.function = switch_buffer;
	buf->switch_timer.expires = jiffies + chan->switch_timer_interval;
	buf->switch_timer.data = (unsigned long)buf;
	add_timer_on(&buf->switch_timer, buf->a.cpu);
}

/*
 * called with ltt traces lock held.
 */
void ltt_chan_start_switch_timer(struct ltt_chan *chan)
{
	int cpu;

	if (!chan->switch_timer_interval)
		return;

	for_each_online_cpu(cpu) {
		struct ltt_chanbuf *buf;

		buf = per_cpu_ptr(chan->a.buf, cpu);
		ltt_chanbuf_start_switch_timer(buf);
	}
}

static void ltt_chanbuf_stop_switch_timer(struct ltt_chanbuf *buf)
{
	struct ltt_chan *chan = container_of(buf->a.chan, struct ltt_chan, a);

	if (!chan->switch_timer_interval)
		return;

	del_timer_sync(&buf->switch_timer);
}

/*
 * called with ltt traces lock held.
 */
void ltt_chan_stop_switch_timer(struct ltt_chan *chan)
{
	int cpu;

	if (!chan->switch_timer_interval)
		return;

	for_each_online_cpu(cpu) {
		struct ltt_chanbuf *buf;

		buf = per_cpu_ptr(chan->a.buf, cpu);
		ltt_chanbuf_stop_switch_timer(buf);
	}
}

static void ltt_chanbuf_idle_switch(struct ltt_chanbuf *buf)
{
	struct ltt_chan *chan = container_of(buf->a.chan, struct ltt_chan, a);

	if (chan->switch_timer_interval)
		ltt_force_switch(buf, FORCE_ACTIVE);
}

/*
 * ltt_chanbuf_switch is called from a remote CPU to ensure that the buffers of
 * a cpu which went down are flushed. Note that if we execute concurrently
 * with trace allocation, a buffer might appear be unallocated (because it
 * detects that the target CPU is offline).
 */
static void ltt_chanbuf_switch(struct ltt_chanbuf *buf)
{
	if (buf->a.allocated)
		ltt_force_switch(buf, FORCE_ACTIVE);
}

/**
 *	ltt_chanbuf_hotcpu_callback - CPU hotplug callback
 *	@nb: notifier block
 *	@action: hotplug action to take
 *	@hcpu: CPU number
 *
 *	Returns the success/failure of the operation. (%NOTIFY_OK, %NOTIFY_BAD)
 */
static
int ltt_chanbuf_hotcpu_callback(struct notifier_block *nb,
					  unsigned long action,
					  void *hcpu)
{
	unsigned int cpu = (unsigned long)hcpu;

	switch (action) {
	case CPU_DOWN_FAILED:
	case CPU_DOWN_FAILED_FROZEN:
	case CPU_ONLINE:
	case CPU_ONLINE_FROZEN:
		/*
		 * CPU hotplug lock protects trace lock from this callback.
		 */
		ltt_chan_for_each_channel(ltt_chanbuf_start_switch_timer, cpu);
		return NOTIFY_OK;

	case CPU_DOWN_PREPARE:
	case CPU_DOWN_PREPARE_FROZEN:
		/*
		 * Performs an IPI to delete the timer locally on the target
		 * CPU.	CPU hotplug lock protects trace lock from this
		 * callback.
		 */
		ltt_chan_for_each_channel(ltt_chanbuf_stop_switch_timer, cpu);
		return NOTIFY_OK;

	case CPU_DEAD:
	case CPU_DEAD_FROZEN:
		/*
		 * Performing a buffer switch on a remote CPU. Performed by
		 * the CPU responsible for doing the hotunplug after the target
		 * CPU stopped running completely. Ensures that all data
		 * from that remote CPU is flushed. CPU hotplug lock protects
		 * trace lock from this callback.
		 */
		ltt_chan_for_each_channel(ltt_chanbuf_switch, cpu);
		return NOTIFY_OK;

	default:
		return NOTIFY_DONE;
	}
}

static int pm_idle_entry_callback(struct notifier_block *self,
				  unsigned long val, void *data)
{
	if (val == IDLE_START) {
		rcu_read_lock_sched_notrace();
		ltt_chan_for_each_channel(ltt_chanbuf_idle_switch,
					  smp_processor_id());
		rcu_read_unlock_sched_notrace();
	}
	return 0;
}

struct notifier_block pm_idle_entry_notifier = {
	.notifier_call = pm_idle_entry_callback,
	.priority = ~0U,	/* smallest prio, run after tracing events */
};

static
void ltt_relay_print_written(struct ltt_chan *chan, long cons_off,
			     unsigned int cpu)
{
	struct ltt_chanbuf *buf = per_cpu_ptr(chan->a.buf, cpu);
	long cons_idx, events_count;

	cons_idx = SUBBUF_INDEX(cons_off, chan);
	events_count = local_read(&buf->commit_count[cons_idx].events);

	if (events_count)
		printk(KERN_INFO
			"LTT: %lu events written in channel %s "
			"(cpu %u, index %lu)\n",
			events_count, chan->a.filename, cpu, cons_idx);
}

static
void ltt_relay_print_subbuffer_errors(struct ltt_chanbuf *buf,
				      struct ltt_chan *chan, long cons_off,
				      unsigned int cpu)
{
	long cons_idx, commit_count, commit_count_sb, write_offset;

	cons_idx = SUBBUF_INDEX(cons_off, chan);
	commit_count = local_read(&buf->commit_count[cons_idx].cc);
	commit_count_sb = local_read(&buf->commit_count[cons_idx].cc_sb);
	/*
	 * No need to order commit_count and write_offset reads because we
	 * execute after trace is stopped when there are no readers left.
	 */
	write_offset = local_read(&buf->offset);
	printk(KERN_WARNING
	       "LTT : unread channel %s offset is %ld "
	       "and cons_off : %ld (cpu %u)\n",
	       chan->a.filename, write_offset, cons_off, cpu);
	/* Check each sub-buffer for non filled commit count */
	if (((commit_count - chan->a.sb_size) & chan->commit_count_mask)
	    - (BUFFER_TRUNC(cons_off, chan) >> chan->a.n_sb_order)
	    != 0)
		printk(KERN_ALERT
		       "LTT : %s : subbuffer %lu has non filled "
		       "commit count [cc, cc_sb] [%lu,%lu].\n",
		       chan->a.filename, cons_idx, commit_count,
		       commit_count_sb);
	printk(KERN_ALERT "LTT : %s : commit count : %lu, subbuf size %lu\n",
	       chan->a.filename, commit_count, chan->a.sb_size);
}

static
void ltt_relay_print_errors(struct ltt_chanbuf *buf, struct ltt_chan *chan,
			    struct ltt_trace *trace, int cpu)
{
	long cons_off;

	/*
	 * Can be called in the error path of allocation when
	 * trans_channel_data is not yet set.
	 */
	if (!chan)
		return;
	for (cons_off = 0; cons_off < chan->a.buf_size;
	     cons_off = SUBBUF_ALIGN(cons_off, chan))
		ltt_relay_print_written(chan, cons_off, cpu);
	for (cons_off = atomic_long_read(&buf->consumed);
			(SUBBUF_TRUNC(local_read(&buf->offset), chan)
			 - cons_off) > 0;
			cons_off = SUBBUF_ALIGN(cons_off, chan))
		ltt_relay_print_subbuffer_errors(buf, chan, cons_off, cpu);
}

static
void ltt_relay_print_buffer_errors(struct ltt_chan *chan, unsigned int cpu)
{
	struct ltt_trace *trace = chan->a.trace;
	struct ltt_chanbuf *buf = per_cpu_ptr(chan->a.buf, cpu);

	if (local_read(&buf->events_lost))
		printk(KERN_ALERT
		       "LTT : %s : %ld events lost "
		       "in %s channel (cpu %u).\n",
		       chan->a.filename, local_read(&buf->events_lost),
		       chan->a.filename, cpu);
	if (local_read(&buf->corrupted_subbuffers))
		printk(KERN_ALERT
		       "LTT : %s : %ld corrupted subbuffers "
		       "in %s channel (cpu %u).\n",
		       chan->a.filename,
		       local_read(&buf->corrupted_subbuffers),
		       chan->a.filename, cpu);

	ltt_relay_print_errors(buf, chan, trace, cpu);
}

static void ltt_relay_remove_dirs(struct ltt_trace *trace)
{
	ltt_ascii_remove_dir(trace);
	debugfs_remove(trace->dentry.trace_root);
}

static int ltt_relay_create_dirs(struct ltt_trace *new_trace)
{
	struct dentry *ltt_root_dentry;
	int ret;

	ltt_root_dentry = get_ltt_root();
	if (!ltt_root_dentry)
		return ENOENT;

	new_trace->dentry.trace_root = debugfs_create_dir(new_trace->trace_name,
							  ltt_root_dentry);
	put_ltt_root();
	if (new_trace->dentry.trace_root == NULL) {
		printk(KERN_ERR "LTT : Trace directory name %s already taken\n",
		       new_trace->trace_name);
		return EEXIST;
	}
	ret = ltt_ascii_create_dir(new_trace);
	if (ret)
		printk(KERN_WARNING "LTT : Unable to create ascii output file "
				    "for trace %s\n", new_trace->trace_name);

	return 0;
}

/*
 * LTTng channel flush function.
 *
 * Must be called when no tracing is active in the channel, because of
 * accesses across CPUs.
 */
static notrace void ltt_relay_buffer_flush(struct ltt_chanbuf *buf)
{
	buf->finalized = 1;
	ltt_force_switch(buf, FORCE_FLUSH);
}

static void ltt_relay_async_wakeup_chan(struct ltt_chan *chan)
{
	unsigned int i;

	for_each_possible_cpu(i) {
		struct ltt_chanbuf *buf;

		buf = per_cpu_ptr(chan->a.buf, i);
		if (!buf->a.allocated)
			continue;
		/*
		 * Ensure the buffer has been allocated before reading its
		 * content. Sync cpu hotplug vs async wakeup.
		 */
		smp_rmb();
		if (ltt_poll_deliver(buf, chan))
			wake_up_interruptible(&buf->read_wait);
	}
}

static void ltt_relay_finish_buffer(struct ltt_chan *chan, unsigned int cpu)
{
	struct ltt_chanbuf *buf = per_cpu_ptr(chan->a.buf, cpu);

	if (buf->a.allocated) {
		ltt_relay_buffer_flush(buf);
		ltt_relay_wake_writers(buf);
	}
}


static void ltt_relay_finish_channel(struct ltt_chan *chan)
{
	unsigned int i;

	for_each_possible_cpu(i)
		ltt_relay_finish_buffer(chan, i);
}

/*
 * This is called with preemption disabled when user space has requested
 * blocking mode.  If one of the active traces has free space below a
 * specific threshold value, we reenable preemption and block.
 */
static
int ltt_relay_user_blocking(struct ltt_trace *trace, unsigned int chan_index,
			    size_t data_size, struct user_dbg_data *dbg)
{
	struct ltt_chanbuf *buf;
	struct ltt_chan *chan;
	int cpu;
	DECLARE_WAITQUEUE(wait, current);

	chan = &trace->channels[chan_index];
	cpu = smp_processor_id();
	buf = per_cpu_ptr(chan->a.buf, cpu);

	/*
	 * Check if data is too big for the channel : do not
	 * block for it.
	 */
	if (LTT_RESERVE_CRITICAL + data_size > chan->a.sb_size)
		return 0;

	/*
	 * If free space too low, we block. We restart from the
	 * beginning after we resume (cpu id may have changed
	 * while preemption is active).
	 */
	spin_lock(&buf->full_lock);
	if (!chan->overwrite) {
		dbg->write = local_read(&buf->offset);
		dbg->read = atomic_long_read(&buf->consumed);
		dbg->avail_size = dbg->write + LTT_RESERVE_CRITICAL + data_size
				  - SUBBUF_TRUNC(dbg->read, chan);
		if (dbg->avail_size > chan->a.buf_size) {
			__set_current_state(TASK_INTERRUPTIBLE);
			add_wait_queue(&buf->write_wait, &wait);
			spin_unlock(&buf->full_lock);
			preempt_enable();
			schedule();
			__set_current_state(TASK_RUNNING);
			remove_wait_queue(&buf->write_wait, &wait);
			if (signal_pending(current))
				return -ERESTARTSYS;
			preempt_disable();
			return 1;
		}
	}
	spin_unlock(&buf->full_lock);
	return 0;
}

static
void ltt_relay_print_user_errors(struct ltt_trace *trace,
				 unsigned int chan_index, size_t data_size,
				 struct user_dbg_data *dbg, int cpu)
{
	struct ltt_chanbuf *buf;
	struct ltt_chan *chan;

	chan = &trace->channels[chan_index];
	buf = per_cpu_ptr(chan->a.buf, cpu);

	printk(KERN_ERR "Error in LTT usertrace : "
	       "buffer full : event lost in blocking "
	       "mode. Increase LTT_RESERVE_CRITICAL.\n");
	printk(KERN_ERR "LTT nesting level is %u.\n",
	       per_cpu(ltt_nesting, cpu));
	printk(KERN_ERR "LTT available size %lu.\n",
	       dbg->avail_size);
	printk(KERN_ERR "available write : %lu, read : %lu\n",
	       dbg->write, dbg->read);

	dbg->write = local_read(&buf->offset);
	dbg->read = atomic_long_read(&buf->consumed);

	printk(KERN_ERR "LTT current size %lu.\n",
		dbg->write + LTT_RESERVE_CRITICAL + data_size
		- SUBBUF_TRUNC(dbg->read, chan));
	printk(KERN_ERR "current write : %lu, read : %lu\n",
			dbg->write, dbg->read);
}

/*
 * ltt_reserve_switch_old_subbuf: switch old subbuffer
 *
 * Concurrency safe because we are the last and only thread to alter this
 * sub-buffer. As long as it is not delivered and read, no other thread can
 * alter the offset, alter the reserve_count or call the
 * client_buffer_end_callback on this sub-buffer.
 *
 * The only remaining threads could be the ones with pending commits. They will
 * have to do the deliver themselves.  Not concurrency safe in overwrite mode.
 * We detect corrupted subbuffers with commit and reserve counts. We keep a
 * corrupted sub-buffers count and push the readers across these sub-buffers.
 *
 * Not concurrency safe if a writer is stalled in a subbuffer and another writer
 * switches in, finding out it's corrupted.  The result will be than the old
 * (uncommited) subbuffer will be declared corrupted, and that the new subbuffer
 * will be declared corrupted too because of the commit count adjustment.
 *
 * Note : offset_old should never be 0 here.
 */
static
void ltt_reserve_switch_old_subbuf(struct ltt_chanbuf *buf,
				   struct ltt_chan *chan,
				   struct ltt_reserve_switch_offsets *offsets,
				   u64 *tsc)
{
	long oldidx = SUBBUF_INDEX(offsets->old - 1, chan);
	long commit_count, padding_size;

	padding_size = chan->a.sb_size
			- (SUBBUF_OFFSET(offsets->old - 1, chan) + 1);
	ltt_buffer_end(buf, *tsc, offsets->old, oldidx);

	/*
	 * Must write slot data before incrementing commit count.
	 * This compiler barrier is upgraded into a smp_wmb() by the IPI
	 * sent by get_subbuf() when it does its smp_rmb().
	 */
	barrier();
	local_add(padding_size, &buf->commit_count[oldidx].cc);
	commit_count = local_read(&buf->commit_count[oldidx].cc);
	ltt_check_deliver(buf, chan, offsets->old - 1, commit_count, oldidx);
	ltt_write_commit_counter(buf, chan, oldidx, offsets->old, commit_count,
				 padding_size);
}

/*
 * ltt_reserve_switch_new_subbuf: Populate new subbuffer.
 *
 * This code can be executed unordered : writers may already have written to the
 * sub-buffer before this code gets executed, caution.  The commit makes sure
 * that this code is executed before the deliver of this sub-buffer.
 */
static
void ltt_reserve_switch_new_subbuf(struct ltt_chanbuf *buf,
				   struct ltt_chan *chan,
				   struct ltt_reserve_switch_offsets *offsets,
				   u64 *tsc)
{
	long beginidx = SUBBUF_INDEX(offsets->begin, chan);
	long commit_count;

	ltt_buffer_begin(buf, *tsc, beginidx);

	/*
	 * Must write slot data before incrementing commit count.
	 * This compiler barrier is upgraded into a smp_wmb() by the IPI
	 * sent by get_subbuf() when it does its smp_rmb().
	 */
	barrier();
	local_add(ltt_sb_header_size(), &buf->commit_count[beginidx].cc);
	commit_count = local_read(&buf->commit_count[beginidx].cc);
	/* Check if the written buffer has to be delivered */
	ltt_check_deliver(buf, chan, offsets->begin, commit_count, beginidx);
	ltt_write_commit_counter(buf, chan, beginidx, offsets->begin,
				 commit_count, ltt_sb_header_size());
}


/*
 * ltt_reserve_end_switch_current: finish switching current subbuffer
 *
 * Concurrency safe because we are the last and only thread to alter this
 * sub-buffer. As long as it is not delivered and read, no other thread can
 * alter the offset, alter the reserve_count or call the
 * client_buffer_end_callback on this sub-buffer.
 *
 * The only remaining threads could be the ones with pending commits. They will
 * have to do the deliver themselves.  Not concurrency safe in overwrite mode.
 * We detect corrupted subbuffers with commit and reserve counts. We keep a
 * corrupted sub-buffers count and push the readers across these sub-buffers.
 *
 * Not concurrency safe if a writer is stalled in a subbuffer and another writer
 * switches in, finding out it's corrupted.  The result will be than the old
 * (uncommited) subbuffer will be declared corrupted, and that the new subbuffer
 * will be declared corrupted too because of the commit count adjustment.
 */
static
void ltt_reserve_end_switch_current(struct ltt_chanbuf *buf,
				    struct ltt_chan *chan,
				    struct ltt_reserve_switch_offsets *offsets,
				    u64 *tsc)
{
	long endidx = SUBBUF_INDEX(offsets->end - 1, chan);
	long commit_count, padding_size;

	padding_size = chan->a.sb_size
			- (SUBBUF_OFFSET(offsets->end - 1, chan) + 1);

	ltt_buffer_end(buf, *tsc, offsets->end, endidx);

	/*
	 * Must write slot data before incrementing commit count.
	 * This compiler barrier is upgraded into a smp_wmb() by the IPI
	 * sent by get_subbuf() when it does its smp_rmb().
	 */
	barrier();
	local_add(padding_size, &buf->commit_count[endidx].cc);
	commit_count = local_read(&buf->commit_count[endidx].cc);
	ltt_check_deliver(buf, chan, offsets->end - 1, commit_count, endidx);
	ltt_write_commit_counter(buf, chan, endidx, offsets->end, commit_count,
				 padding_size);
}

/*
 * Returns :
 * 0 if ok
 * !0 if execution must be aborted.
 */
static
int ltt_relay_try_switch_slow(enum force_switch_mode mode,
			      struct ltt_chanbuf *buf, struct ltt_chan *chan,
			      struct ltt_reserve_switch_offsets *offsets,
			      u64 *tsc)
{
	long sb_index;
	long reserve_commit_diff;
	long off;

	offsets->begin = local_read(&buf->offset);
	offsets->old = offsets->begin;
	offsets->begin_switch = 0;
	offsets->end_switch_old = 0;

	*tsc = trace_clock_read64();

	off = SUBBUF_OFFSET(offsets->begin, chan);
	if ((mode != FORCE_ACTIVE && off > 0) || off > ltt_sb_header_size()) {
		offsets->begin = SUBBUF_ALIGN(offsets->begin, chan);
		offsets->end_switch_old = 1;
	} else {
		/* we do not have to switch : buffer is empty */
		return -1;
	}
	if (mode == FORCE_ACTIVE)
		offsets->begin += ltt_sb_header_size();
	/*
	 * Always begin_switch in FORCE_ACTIVE mode.
	 * Test new buffer integrity
	 */
	sb_index = SUBBUF_INDEX(offsets->begin, chan);
	reserve_commit_diff =
		(BUFFER_TRUNC(offsets->begin, chan)
		 >> chan->a.n_sb_order)
		- (local_read(&buf->commit_count[sb_index].cc_sb)
			& chan->commit_count_mask);
	if (reserve_commit_diff == 0) {
		/* Next buffer not corrupted. */
		if (mode == FORCE_ACTIVE
		    && !chan->overwrite
		    && offsets->begin - atomic_long_read(&buf->consumed)
		       >= chan->a.buf_size) {
			/*
			 * We do not overwrite non consumed buffers and we are
			 * full : ignore switch while tracing is active.
			 */
			return -1;
		}
	} else {
		/*
		 * Next subbuffer corrupted. Force pushing reader even in normal
		 * mode
		 */
	}
	offsets->end = offsets->begin;
	return 0;
}

/*
 * Force a sub-buffer switch for a per-cpu buffer. This operation is
 * completely reentrant : can be called while tracing is active with
 * absolutely no lock held.
 *
 * Note, however, that as a local_cmpxchg is used for some atomic
 * operations, this function must be called from the CPU which owns the buffer
 * for a ACTIVE flush.
 */
void ltt_force_switch_lockless_slow(struct ltt_chanbuf *buf,
				    enum force_switch_mode mode)
{
	struct ltt_chan *chan = container_of(buf->a.chan, struct ltt_chan, a);
	struct ltt_reserve_switch_offsets offsets;
	u64 tsc;

	offsets.size = 0;

	/*
	 * Perform retryable operations.
	 */
	do {
		if (ltt_relay_try_switch_slow(mode, buf, chan, &offsets, &tsc))
			return;
	} while (local_cmpxchg(&buf->offset, offsets.old, offsets.end)
		 != offsets.old);

	/*
	 * Atomically update last_tsc. This update races against concurrent
	 * atomic updates, but the race will always cause supplementary full TSC
	 * events, never the opposite (missing a full TSC event when it would be
	 * needed).
	 */
	save_last_tsc(buf, tsc);

	/*
	 * Push the reader if necessary
	 */
	if (mode == FORCE_ACTIVE) {
		ltt_reserve_push_reader(buf, chan, offsets.end - 1);
		ltt_clear_noref_flag(&buf->a, SUBBUF_INDEX(offsets.end - 1,
							   chan));
	}

	/*
	 * Switch old subbuffer if needed.
	 */
	if (offsets.end_switch_old) {
		ltt_clear_noref_flag(&buf->a, SUBBUF_INDEX(offsets.old - 1,
							   chan));
		ltt_reserve_switch_old_subbuf(buf, chan, &offsets, &tsc);
	}

	/*
	 * Populate new subbuffer.
	 */
	if (mode == FORCE_ACTIVE)
		ltt_reserve_switch_new_subbuf(buf, chan, &offsets, &tsc);
}
EXPORT_SYMBOL_GPL(ltt_force_switch_lockless_slow);

/*
 * Returns :
 * 0 if ok
 * !0 if execution must be aborted.
 */
static
int ltt_relay_try_reserve_slow(struct ltt_chanbuf *buf, struct ltt_chan *chan,
			       struct ltt_reserve_switch_offsets *offsets,
			       size_t data_size, u64 *tsc, unsigned int *rflags,
			       int largest_align)
{
	long reserve_commit_diff;

	offsets->begin = local_read(&buf->offset);
	offsets->old = offsets->begin;
	offsets->begin_switch = 0;
	offsets->end_switch_current = 0;
	offsets->end_switch_old = 0;

	*tsc = trace_clock_read64();
	if (last_tsc_overflow(buf, *tsc))
		*rflags = LTT_RFLAG_ID_SIZE_TSC;

	if (unlikely(SUBBUF_OFFSET(offsets->begin, chan) == 0)) {
		offsets->begin_switch = 1;		/* For offsets->begin */
	} else {
		offsets->size = ltt_get_header_size(chan, offsets->begin,
						    data_size,
						    &offsets->before_hdr_pad,
						    *rflags);
		offsets->size += ltt_align(offsets->begin + offsets->size,
					   largest_align)
				 + data_size;
		if (unlikely((SUBBUF_OFFSET(offsets->begin, chan) +
			     offsets->size) > chan->a.sb_size)) {
			offsets->end_switch_old = 1;	/* For offsets->old */
			offsets->begin_switch = 1;	/* For offsets->begin */
		}
	}
	if (unlikely(offsets->begin_switch)) {
		long sb_index;

		/*
		 * We are typically not filling the previous buffer completely.
		 */
		if (likely(offsets->end_switch_old))
			offsets->begin = SUBBUF_ALIGN(offsets->begin, chan);
		offsets->begin = offsets->begin + ltt_sb_header_size();
		/* Test new buffer integrity */
		sb_index = SUBBUF_INDEX(offsets->begin, chan);
		reserve_commit_diff =
		  (BUFFER_TRUNC(offsets->begin, chan)
		   >> chan->a.n_sb_order)
		  - (local_read(&buf->commit_count[sb_index].cc_sb)
				& chan->commit_count_mask);
		if (likely(reserve_commit_diff == 0)) {
			/* Next buffer not corrupted. */
			if (unlikely(!chan->overwrite &&
				(SUBBUF_TRUNC(offsets->begin, chan)
				 - SUBBUF_TRUNC(atomic_long_read(&buf->consumed),
						chan))
				>= chan->a.buf_size)) {
				/*
				 * We do not overwrite non consumed buffers
				 * and we are full : event is lost.
				 */
				local_inc(&buf->events_lost);
				return -1;
			} else {
				/*
				 * next buffer not corrupted, we are either in
				 * overwrite mode or the buffer is not full.
				 * It's safe to write in this new subbuffer.
				 */
			}
		} else {
			/*
			 * Next subbuffer corrupted. Drop event in normal and
			 * overwrite mode. Caused by either a writer OOPS or
			 * too many nested writes over a reserve/commit pair.
			 */
			local_inc(&buf->events_lost);
			return -1;
		}
		offsets->size = ltt_get_header_size(chan, offsets->begin,
						    data_size,
						    &offsets->before_hdr_pad,
						    *rflags);
		offsets->size += ltt_align(offsets->begin + offsets->size,
					   largest_align)
				 + data_size;
		if (unlikely((SUBBUF_OFFSET(offsets->begin, chan)
			     + offsets->size) > chan->a.sb_size)) {
			/*
			 * Event too big for subbuffers, report error, don't
			 * complete the sub-buffer switch.
			 */
			local_inc(&buf->events_lost);
			return -1;
		} else {
			/*
			 * We just made a successful buffer switch and the event
			 * fits in the new subbuffer. Let's write.
			 */
		}
	} else {
		/*
		 * Event fits in the current buffer and we are not on a switch
		 * boundary. It's safe to write.
		 */
	}
	offsets->end = offsets->begin + offsets->size;

	if (unlikely((SUBBUF_OFFSET(offsets->end, chan)) == 0)) {
		/*
		 * The offset_end will fall at the very beginning of the next
		 * subbuffer.
		 */
		offsets->end_switch_current = 1;	/* For offsets->begin */
	}
	return 0;
}

/**
 * ltt_relay_reserve_slot_lockless_slow - Atomic slot reservation in a buffer.
 * @trace: the trace structure to log to.
 * @ltt_channel: channel structure
 * @transport_data: data structure specific to ltt relay
 * @data_size: size of the variable length data to log.
 * @slot_size: pointer to total size of the slot (out)
 * @buf_offset : pointer to reserved buffer offset (out)
 * @tsc: pointer to the tsc at the slot reservation (out)
 * @cpu: cpuid
 *
 * Return : -ENOSPC if not enough space, else returns 0.
 * It will take care of sub-buffer switching.
 */
int ltt_reserve_slot_lockless_slow(struct ltt_chan *chan,
				   struct ltt_trace *trace, size_t data_size,
				   int largest_align, int cpu,
				   struct ltt_chanbuf **ret_buf,
				   size_t *slot_size, long *buf_offset,
				   u64 *tsc, unsigned int *rflags)
{
	struct ltt_chanbuf *buf = *ret_buf = per_cpu_ptr(chan->a.buf, cpu);
	struct ltt_reserve_switch_offsets offsets;

	offsets.size = 0;

	do {
		if (unlikely(ltt_relay_try_reserve_slow(buf, chan, &offsets,
							data_size, tsc, rflags,
							largest_align)))
			return -ENOSPC;
	} while (unlikely(local_cmpxchg(&buf->offset, offsets.old, offsets.end)
			  != offsets.old));

	/*
	 * Atomically update last_tsc. This update races against concurrent
	 * atomic updates, but the race will always cause supplementary full TSC
	 * events, never the opposite (missing a full TSC event when it would be
	 * needed).
	 */
	save_last_tsc(buf, *tsc);

	/*
	 * Push the reader if necessary
	 */
	ltt_reserve_push_reader(buf, chan, offsets.end - 1);

	/*
	 * Clear noref flag for this subbuffer.
	 */
	ltt_clear_noref_flag(&buf->a, SUBBUF_INDEX(offsets.end - 1, chan));

	/*
	 * Switch old subbuffer if needed.
	 */
	if (unlikely(offsets.end_switch_old)) {
		ltt_clear_noref_flag(&buf->a, SUBBUF_INDEX(offsets.old - 1,
							  chan));
		ltt_reserve_switch_old_subbuf(buf, chan, &offsets, tsc);
	}

	/*
	 * Populate new subbuffer.
	 */
	if (unlikely(offsets.begin_switch))
		ltt_reserve_switch_new_subbuf(buf, chan, &offsets, tsc);

	if (unlikely(offsets.end_switch_current))
		ltt_reserve_end_switch_current(buf, chan, &offsets, tsc);

	*slot_size = offsets.size;
	*buf_offset = offsets.begin + offsets.before_hdr_pad;
	return 0;
}
EXPORT_SYMBOL_GPL(ltt_reserve_slot_lockless_slow);

static struct ltt_transport ltt_relay_transport = {
	.name = "relay",
	.owner = THIS_MODULE,
	.ops = {
		.create_dirs = ltt_relay_create_dirs,
		.remove_dirs = ltt_relay_remove_dirs,
		.create_channel = ltt_chan_create,
		.finish_channel = ltt_relay_finish_channel,
		.remove_channel = ltt_chan_free,
		.remove_channel_files = ltt_chan_remove_files,
		.wakeup_channel = ltt_relay_async_wakeup_chan,
		.user_blocking = ltt_relay_user_blocking,
		.user_errors = ltt_relay_print_user_errors,
		.start_switch_timer = ltt_chan_start_switch_timer,
		.stop_switch_timer = ltt_chan_stop_switch_timer,
	},
};

static struct notifier_block fn_ltt_chanbuf_hotcpu_callback = {
	.notifier_call = ltt_chanbuf_hotcpu_callback,
	.priority = 6,
};

int __init ltt_relay_init(void)
{
	printk(KERN_INFO "LTT : ltt-relay init\n");

	ltt_transport_register(&ltt_relay_transport);
	register_cpu_notifier(&fn_ltt_chanbuf_hotcpu_callback);
	register_idle_notifier(&pm_idle_entry_notifier);

	return 0;
}

void __exit ltt_relay_exit(void)
{
	printk(KERN_INFO "LTT : ltt-relay exit\n");

	unregister_idle_notifier(&pm_idle_entry_notifier);
	unregister_cpu_notifier(&fn_ltt_chanbuf_hotcpu_callback);
	ltt_transport_unregister(&ltt_relay_transport);
}

MODULE_LICENSE("GPL and additional rights");
MODULE_AUTHOR("Mathieu Desnoyers");
MODULE_DESCRIPTION("Linux Trace Toolkit Next Generation Lockless Relay");