[linux-block.git] / fs / relayfs / relay.c

/*
 * Public API and common code for RelayFS.
 *
 * See Documentation/filesystems/relayfs.txt for an overview of relayfs.
 *
 * Copyright (C) 2002-2005 - Tom Zanussi (zanussi@us.ibm.com), IBM Corp
 * Copyright (C) 1999-2005 - Karim Yaghmour (karim@opersys.com)
 *
 * This file is released under the GPL.
 */

#include <linux/errno.h>
#include <linux/stddef.h>
#include <linux/slab.h>
#include <linux/module.h>
#include <linux/string.h>
#include <linux/relayfs_fs.h>
#include "relay.h"
#include "buffers.h"

/**
 *	relay_buf_empty - boolean, is the channel buffer empty?
 *	@buf: channel buffer
 *
 *	Returns 1 if the buffer is empty, 0 otherwise.
 */
int relay_buf_empty(struct rchan_buf *buf)
{
	return (buf->subbufs_produced - buf->subbufs_consumed) ? 0 : 1;
}

/**
 *	relay_buf_full - boolean, is the channel buffer full?
 *	@buf: channel buffer
 *
 *	Returns 1 if the buffer is full, 0 otherwise.
 */
int relay_buf_full(struct rchan_buf *buf)
{
	size_t ready = buf->subbufs_produced - buf->subbufs_consumed;
	return (ready >= buf->chan->n_subbufs) ? 1 : 0;
}

/*
 * High-level relayfs kernel API and associated functions.
 */

/*
 * rchan_callback implementations defining default channel behavior.  Used
 * in place of corresponding NULL values in client callback struct.
 */

/*
 * subbuf_start() default callback.  Does nothing.
 */
static int subbuf_start_default_callback (struct rchan_buf *buf,
					  void *subbuf,
					  void *prev_subbuf,
					  size_t prev_padding)
{
	if (relay_buf_full(buf))
		return 0;

	return 1;
}

/*
 * buf_mapped() default callback.  Does nothing.
 */
static void buf_mapped_default_callback(struct rchan_buf *buf,
					struct file *filp)
{
}

/*
 * buf_unmapped() default callback.  Does nothing.
 */
static void buf_unmapped_default_callback(struct rchan_buf *buf,
					  struct file *filp)
{
}

/*
 * create_buf_file_create() default callback.  Creates file to represent buf.
 */
static struct dentry *create_buf_file_default_callback(const char *filename,
						       struct dentry *parent,
						       int mode,
						       struct rchan_buf *buf,
						       int *is_global)
{
	return relayfs_create_file(filename, parent, mode,
				   &relay_file_operations, buf);
}

/*
 * remove_buf_file() default callback.  Removes file representing relay buffer.
 */
static int remove_buf_file_default_callback(struct dentry *dentry)
{
	return relayfs_remove(dentry);
}

/* relay channel default callbacks */
static struct rchan_callbacks default_channel_callbacks = {
	.subbuf_start = subbuf_start_default_callback,
	.buf_mapped = buf_mapped_default_callback,
	.buf_unmapped = buf_unmapped_default_callback,
	.create_buf_file = create_buf_file_default_callback,
	.remove_buf_file = remove_buf_file_default_callback,
};

/**
 *	wakeup_readers - wake up readers waiting on a channel
 *	@private: the channel buffer
 *
 *	This is the work function used to defer reader waking.  The
 *	reason waking is deferred is that calling directly from write
 *	causes problems if you're writing from say the scheduler.
 */
static void wakeup_readers(void *private)
{
	struct rchan_buf *buf = private;
	wake_up_interruptible(&buf->read_wait);
}

/**
 *	__relay_reset - reset a channel buffer
 *	@buf: the channel buffer
 *	@init: 1 if this is a first-time initialization
 *
 *	See relay_reset for description of effect.
 */
static inline void __relay_reset(struct rchan_buf *buf, unsigned int init)
{
	size_t i;

	if (init) {
		init_waitqueue_head(&buf->read_wait);
		kref_init(&buf->kref);
		INIT_WORK(&buf->wake_readers, NULL, NULL);
	} else {
		cancel_delayed_work(&buf->wake_readers);
		flush_scheduled_work();
	}

	buf->subbufs_produced = 0;
	buf->subbufs_consumed = 0;
	buf->bytes_consumed = 0;
	buf->finalized = 0;
	buf->data = buf->start;
	buf->offset = 0;

	for (i = 0; i < buf->chan->n_subbufs; i++)
		buf->padding[i] = 0;

	buf->chan->cb->subbuf_start(buf, buf->data, NULL, 0);
}

/**
 *	relay_reset - reset the channel
 *	@chan: the channel
 *
 *	This has the effect of erasing all data from all channel buffers
 *	and restarting the channel in its initial state.  The buffers
 *	are not freed, so any mappings are still in effect.
 *
 *	NOTE: Care should be taken that the channel isn't actually
 *	being used by anything when this call is made.
 */
void relay_reset(struct rchan *chan)
{
	unsigned int i;
	struct rchan_buf *prev = NULL;

	if (!chan)
		return;

	for (i = 0; i < NR_CPUS; i++) {
		if (!chan->buf[i] || chan->buf[i] == prev)
			break;
		__relay_reset(chan->buf[i], 0);
		prev = chan->buf[i];
	}
}

/**
 *	relay_open_buf - create a new channel buffer in relayfs
 *
 *	Internal - used by relay_open().
 */
static struct rchan_buf *relay_open_buf(struct rchan *chan,
					const char *filename,
					struct dentry *parent,
					int *is_global)
{
	struct rchan_buf *buf;
	struct dentry *dentry;

	if (*is_global)
		return chan->buf[0];

 	buf = relay_create_buf(chan);
 	if (!buf)
 		return NULL;

	/* Create file in fs */
 	dentry = chan->cb->create_buf_file(filename, parent, S_IRUSR,
 					   buf, is_global);
 	if (!dentry) {
 		relay_destroy_buf(buf);
		return NULL;
 	}

	buf->dentry = dentry;
	__relay_reset(buf, 1);

	return buf;
}

/**
 *	relay_close_buf - close a channel buffer
 *	@buf: channel buffer
 *
 *	Marks the buffer finalized and restores the default callbacks.
 *	The channel buffer and channel buffer data structure are then freed
 *	automatically when the last reference is given up.
 */
static inline void relay_close_buf(struct rchan_buf *buf)
{
	buf->finalized = 1;
	buf->chan->cb = &default_channel_callbacks;
	cancel_delayed_work(&buf->wake_readers);
	flush_scheduled_work();
	kref_put(&buf->kref, relay_remove_buf);
}

static inline void setup_callbacks(struct rchan *chan,
				   struct rchan_callbacks *cb)
{
	if (!cb) {
		chan->cb = &default_channel_callbacks;
		return;
	}

	if (!cb->subbuf_start)
		cb->subbuf_start = subbuf_start_default_callback;
	if (!cb->buf_mapped)
		cb->buf_mapped = buf_mapped_default_callback;
	if (!cb->buf_unmapped)
		cb->buf_unmapped = buf_unmapped_default_callback;
	if (!cb->create_buf_file)
		cb->create_buf_file = create_buf_file_default_callback;
	if (!cb->remove_buf_file)
		cb->remove_buf_file = remove_buf_file_default_callback;
	chan->cb = cb;
}

/**
 *	relay_open - create a new relayfs channel
 *	@base_filename: base name of files to create
 *	@parent: dentry of parent directory, NULL for root directory
 *	@subbuf_size: size of sub-buffers
 *	@n_subbufs: number of sub-buffers
 *	@cb: client callback functions
 *
 *	Returns channel pointer if successful, NULL otherwise.
 *
 *	Creates a channel buffer for each cpu using the sizes and
 *	attributes specified.  The created channel buffer files
 *	will be named base_filename0...base_filenameN-1.  File
 *	permissions will be S_IRUSR.
 */
struct rchan *relay_open(const char *base_filename,
			 struct dentry *parent,
			 size_t subbuf_size,
			 size_t n_subbufs,
			 struct rchan_callbacks *cb)
{
	unsigned int i;
	struct rchan *chan;
	char *tmpname;
	int is_global = 0;

	if (!base_filename)
		return NULL;

	if (!(subbuf_size && n_subbufs))
		return NULL;

	chan = kcalloc(1, sizeof(struct rchan), GFP_KERNEL);
	if (!chan)
		return NULL;

	chan->version = RELAYFS_CHANNEL_VERSION;
	chan->n_subbufs = n_subbufs;
	chan->subbuf_size = subbuf_size;
	chan->alloc_size = FIX_SIZE(subbuf_size * n_subbufs);
	setup_callbacks(chan, cb);
	kref_init(&chan->kref);

	tmpname = kmalloc(NAME_MAX + 1, GFP_KERNEL);
	if (!tmpname)
		goto free_chan;

	for_each_online_cpu(i) {
		sprintf(tmpname, "%s%d", base_filename, i);
		chan->buf[i] = relay_open_buf(chan, tmpname, parent,
					      &is_global);
		chan->buf[i]->cpu = i;
		if (!chan->buf[i])
			goto free_bufs;
	}

	kfree(tmpname);
	return chan;

free_bufs:
	for (i = 0; i < NR_CPUS; i++) {
		if (!chan->buf[i])
			break;
		relay_close_buf(chan->buf[i]);
		if (is_global)
			break;
	}
	kfree(tmpname);

free_chan:
	kref_put(&chan->kref, relay_destroy_channel);
	return NULL;
}

/**
 *	relay_switch_subbuf - switch to a new sub-buffer
 *	@buf: channel buffer
 *	@length: size of current event
 *
 *	Returns either the length passed in or 0 if full.

 *	Performs sub-buffer-switch tasks such as invoking callbacks,
 *	updating padding counts, waking up readers, etc.
 */
size_t relay_switch_subbuf(struct rchan_buf *buf, size_t length)
{
	void *old, *new;
	size_t old_subbuf, new_subbuf;

	if (unlikely(length > buf->chan->subbuf_size))
		goto toobig;

	if (buf->offset != buf->chan->subbuf_size + 1) {
		buf->prev_padding = buf->chan->subbuf_size - buf->offset;
		old_subbuf = buf->subbufs_produced % buf->chan->n_subbufs;
		buf->padding[old_subbuf] = buf->prev_padding;
		buf->subbufs_produced++;
		if (waitqueue_active(&buf->read_wait)) {
			PREPARE_WORK(&buf->wake_readers, wakeup_readers, buf);
			schedule_delayed_work(&buf->wake_readers, 1);
		}
	}

	old = buf->data;
	new_subbuf = buf->subbufs_produced % buf->chan->n_subbufs;
	new = buf->start + new_subbuf * buf->chan->subbuf_size;
	buf->offset = 0;
	if (!buf->chan->cb->subbuf_start(buf, new, old, buf->prev_padding)) {
		buf->offset = buf->chan->subbuf_size + 1;
		return 0;
	}
	buf->data = new;
	buf->padding[new_subbuf] = 0;

	if (unlikely(length + buf->offset > buf->chan->subbuf_size))
		goto toobig;

	return length;

toobig:
	buf->chan->last_toobig = length;
	return 0;
}

/**
 *	relay_subbufs_consumed - update the buffer's sub-buffers-consumed count
 *	@chan: the channel
 *	@cpu: the cpu associated with the channel buffer to update
 *	@subbufs_consumed: number of sub-buffers to add to current buf's count
 *
 *	Adds to the channel buffer's consumed sub-buffer count.
 *	subbufs_consumed should be the number of sub-buffers newly consumed,
 *	not the total consumed.
 *
 *	NOTE: kernel clients don't need to call this function if the channel
 *	mode is 'overwrite'.
 */
void relay_subbufs_consumed(struct rchan *chan,
			    unsigned int cpu,
			    size_t subbufs_consumed)
{
	struct rchan_buf *buf;

	if (!chan)
		return;

	if (cpu >= NR_CPUS || !chan->buf[cpu])
		return;

	buf = chan->buf[cpu];
	buf->subbufs_consumed += subbufs_consumed;
	if (buf->subbufs_consumed > buf->subbufs_produced)
		buf->subbufs_consumed = buf->subbufs_produced;
}

/**
 *	relay_destroy_channel - free the channel struct
 *
 *	Should only be called from kref_put().
 */
void relay_destroy_channel(struct kref *kref)
{
	struct rchan *chan = container_of(kref, struct rchan, kref);
	kfree(chan);
}

/**
 *	relay_close - close the channel
 *	@chan: the channel
 *
 *	Closes all channel buffers and frees the channel.
 */
void relay_close(struct rchan *chan)
{
	unsigned int i;
	struct rchan_buf *prev = NULL;

	if (!chan)
		return;

	for (i = 0; i < NR_CPUS; i++) {
		if (!chan->buf[i] || chan->buf[i] == prev)
			break;
		relay_close_buf(chan->buf[i]);
		prev = chan->buf[i];
	}

	if (chan->last_toobig)
		printk(KERN_WARNING "relayfs: one or more items not logged "
		       "[item size (%Zd) > sub-buffer size (%Zd)]\n",
		       chan->last_toobig, chan->subbuf_size);

	kref_put(&chan->kref, relay_destroy_channel);
}

/**
 *	relay_flush - close the channel
 *	@chan: the channel
 *
 *	Flushes all channel buffers i.e. forces buffer switch.
 */
void relay_flush(struct rchan *chan)
{
	unsigned int i;
	struct rchan_buf *prev = NULL;

	if (!chan)
		return;

	for (i = 0; i < NR_CPUS; i++) {
		if (!chan->buf[i] || chan->buf[i] == prev)
			break;
		relay_switch_subbuf(chan->buf[i], 0);
		prev = chan->buf[i];
	}
}

EXPORT_SYMBOL_GPL(relay_open);
EXPORT_SYMBOL_GPL(relay_close);
EXPORT_SYMBOL_GPL(relay_flush);
EXPORT_SYMBOL_GPL(relay_reset);
EXPORT_SYMBOL_GPL(relay_subbufs_consumed);
EXPORT_SYMBOL_GPL(relay_switch_subbuf);
EXPORT_SYMBOL_GPL(relay_buf_full);
Commit	Line	Data
e82894f8 TZ	1	/*
	2	* Public API and common code for RelayFS.
	3	*
	4	* See Documentation/filesystems/relayfs.txt for an overview of relayfs.
	5	*
	6	* Copyright (C) 2002-2005 - Tom Zanussi (zanussi@us.ibm.com), IBM Corp
	7	* Copyright (C) 1999-2005 - Karim Yaghmour (karim@opersys.com)
	8	*
	9	* This file is released under the GPL.
	10	*/
	11
	12	#include <linux/errno.h>
	13	#include <linux/stddef.h>
	14	#include <linux/slab.h>
	15	#include <linux/module.h>
	16	#include <linux/string.h>
	17	#include <linux/relayfs_fs.h>
	18	#include "relay.h"
	19	#include "buffers.h"
	20
	21	/**
	22	* relay_buf_empty - boolean, is the channel buffer empty?
	23	* @buf: channel buffer
	24	*
	25	* Returns 1 if the buffer is empty, 0 otherwise.
	26	*/
	27	int relay_buf_empty(struct rchan_buf *buf)
	28	{
	29	return (buf->subbufs_produced - buf->subbufs_consumed) ? 0 : 1;
	30	}
	31
	32	/**
	33	* relay_buf_full - boolean, is the channel buffer full?
	34	* @buf: channel buffer
	35	*
	36	* Returns 1 if the buffer is full, 0 otherwise.
	37	*/
	38	int relay_buf_full(struct rchan_buf *buf)
	39	{
	40	size_t ready = buf->subbufs_produced - buf->subbufs_consumed;
	41	return (ready >= buf->chan->n_subbufs) ? 1 : 0;
	42	}
	43
	44	/*
	45	* High-level relayfs kernel API and associated functions.
	46	*/
	47
	48	/*
	49	* rchan_callback implementations defining default channel behavior. Used
	50	* in place of corresponding NULL values in client callback struct.
	51	*/
	52
	53	/*
	54	* subbuf_start() default callback. Does nothing.
	55	*/
	56	static int subbuf_start_default_callback (struct rchan_buf *buf,
	57	void *subbuf,
	58	void *prev_subbuf,
	59	size_t prev_padding)
	60	{
	61	if (relay_buf_full(buf))
	62	return 0;
	63
	64	return 1;
65	}
66
67	/*
68	* buf_mapped() default callback. Does nothing.
69	*/
70	static void buf_mapped_default_callback(struct rchan_buf *buf,
71	struct file *filp)
72	{
73	}
74
75	/*
76	* buf_unmapped() default callback. Does nothing.
77	*/
78	static void buf_unmapped_default_callback(struct rchan_buf *buf,
79	struct file *filp)
80	{
81	}
82
08c541a7 TZ	83	/*
	84	* create_buf_file_create() default callback. Creates file to represent buf.
	85	*/
	86	static struct dentry create_buf_file_default_callback(const char filename,
	87	struct dentry *parent,
	88	int mode,
e6c08367 TZ	89	struct rchan_buf *buf,
e6c08367 TZ	90	int *is_global)
08c541a7 TZ	91	{
08c541a7 TZ	92	return relayfs_create_file(filename, parent, mode,
761da5c8	93	&relay_file_operations, buf);
08c541a7 TZ	94	}
	95
	96	/*
	97	* remove_buf_file() default callback. Removes file representing relay buffer.
	98	*/
	99	static int remove_buf_file_default_callback(struct dentry *dentry)
	100	{
	101	return relayfs_remove(dentry);
	102	}
	103
e82894f8 TZ	104	/* relay channel default callbacks */
	105	static struct rchan_callbacks default_channel_callbacks = {
	106	.subbuf_start = subbuf_start_default_callback,
	107	.buf_mapped = buf_mapped_default_callback,
	108	.buf_unmapped = buf_unmapped_default_callback,
08c541a7 TZ	109	.create_buf_file = create_buf_file_default_callback,
08c541a7 TZ	110	.remove_buf_file = remove_buf_file_default_callback,
e82894f8 TZ	111	};
	112
	113	/**
	114	* wakeup_readers - wake up readers waiting on a channel
	115	* @private: the channel buffer
	116	*
	117	* This is the work function used to defer reader waking. The
	118	* reason waking is deferred is that calling directly from write
	119	* causes problems if you're writing from say the scheduler.
	120	*/
	121	static void wakeup_readers(void *private)
	122	{
	123	struct rchan_buf *buf = private;
	124	wake_up_interruptible(&buf->read_wait);
	125	}
	126
	127	/**
	128	* __relay_reset - reset a channel buffer
	129	* @buf: the channel buffer
	130	* @init: 1 if this is a first-time initialization
	131	*
	132	* See relay_reset for description of effect.
	133	*/
	134	static inline void __relay_reset(struct rchan_buf *buf, unsigned int init)
	135	{
	136	size_t i;
	137
	138	if (init) {
	139	init_waitqueue_head(&buf->read_wait);
	140	kref_init(&buf->kref);
	141	INIT_WORK(&buf->wake_readers, NULL, NULL);
	142	} else {
	143	cancel_delayed_work(&buf->wake_readers);
	144	flush_scheduled_work();
	145	}
	146
	147	buf->subbufs_produced = 0;
	148	buf->subbufs_consumed = 0;
	149	buf->bytes_consumed = 0;
	150	buf->finalized = 0;
	151	buf->data = buf->start;
	152	buf->offset = 0;
	153
	154	for (i = 0; i < buf->chan->n_subbufs; i++)
	155	buf->padding[i] = 0;
	156
	157	buf->chan->cb->subbuf_start(buf, buf->data, NULL, 0);
	158	}
	159
	160	/**
	161	* relay_reset - reset the channel
	162	* @chan: the channel
	163	*
	164	* This has the effect of erasing all data from all channel buffers
	165	* and restarting the channel in its initial state. The buffers
	166	* are not freed, so any mappings are still in effect.
	167	*
	168	* NOTE: Care should be taken that the channel isn't actually
	169	* being used by anything when this call is made.
	170	*/
	171	void relay_reset(struct rchan *chan)
	172	{
	173	unsigned int i;
e6c08367	174	struct rchan_buf *prev = NULL;
e82894f8 TZ	175
	176	if (!chan)
	177	return;
	178
	179	for (i = 0; i < NR_CPUS; i++) {
e6c08367 TZ	180	if (!chan->buf[i] \|\| chan->buf[i] == prev)
e6c08367 TZ	181	break;
e82894f8	182	__relay_reset(chan->buf[i], 0);
e6c08367	183	prev = chan->buf[i];
e82894f8 TZ	184	}
	185	}
	186
	187	/**
	188	* relay_open_buf - create a new channel buffer in relayfs
	189	*
	190	* Internal - used by relay_open().
	191	*/
	192	static struct rchan_buf relay_open_buf(struct rchan chan,
	193	const char *filename,
e6c08367 TZ	194	struct dentry *parent,
e6c08367 TZ	195	int *is_global)
e82894f8 TZ	196	{
	197	struct rchan_buf *buf;
	198	struct dentry *dentry;
	199
e6c08367 TZ	200	if (*is_global)
	201	return chan->buf[0];
	202
6625b861 TZ	203	buf = relay_create_buf(chan);
	204	if (!buf)
	205	return NULL;
	206
e82894f8	207	/* Create file in fs */
08c541a7	208	dentry = chan->cb->create_buf_file(filename, parent, S_IRUSR,
e6c08367	209	buf, is_global);
6625b861 TZ	210	if (!dentry) {
6625b861 TZ	211	relay_destroy_buf(buf);
e82894f8	212	return NULL;
6625b861	213	}
e82894f8	214
e82894f8 TZ	215	buf->dentry = dentry;
	216	__relay_reset(buf, 1);
	217
	218	return buf;
	219	}
	220
	221	/**
	222	* relay_close_buf - close a channel buffer
	223	* @buf: channel buffer
	224	*
	225	* Marks the buffer finalized and restores the default callbacks.
	226	* The channel buffer and channel buffer data structure are then freed
	227	* automatically when the last reference is given up.
	228	*/
	229	static inline void relay_close_buf(struct rchan_buf *buf)
	230	{
	231	buf->finalized = 1;
	232	buf->chan->cb = &default_channel_callbacks;
	233	cancel_delayed_work(&buf->wake_readers);
	234	flush_scheduled_work();
	235	kref_put(&buf->kref, relay_remove_buf);
	236	}
	237
	238	static inline void setup_callbacks(struct rchan *chan,
	239	struct rchan_callbacks *cb)
	240	{
	241	if (!cb) {
	242	chan->cb = &default_channel_callbacks;
	243	return;
	244	}
	245
	246	if (!cb->subbuf_start)
	247	cb->subbuf_start = subbuf_start_default_callback;
	248	if (!cb->buf_mapped)
	249	cb->buf_mapped = buf_mapped_default_callback;
	250	if (!cb->buf_unmapped)
	251	cb->buf_unmapped = buf_unmapped_default_callback;
08c541a7 TZ	252	if (!cb->create_buf_file)
	253	cb->create_buf_file = create_buf_file_default_callback;
	254	if (!cb->remove_buf_file)
	255	cb->remove_buf_file = remove_buf_file_default_callback;
e82894f8 TZ	256	chan->cb = cb;
	257	}
	258
	259	/**
	260	* relay_open - create a new relayfs channel
	261	* @base_filename: base name of files to create
	262	* @parent: dentry of parent directory, NULL for root directory
	263	* @subbuf_size: size of sub-buffers
	264	* @n_subbufs: number of sub-buffers
	265	* @cb: client callback functions
	266	*
	267	* Returns channel pointer if successful, NULL otherwise.
	268	*
	269	* Creates a channel buffer for each cpu using the sizes and
	270	* attributes specified. The created channel buffer files
	271	* will be named base_filename0...base_filenameN-1. File
	272	* permissions will be S_IRUSR.
	273	*/
	274	struct rchan relay_open(const char base_filename,
	275	struct dentry *parent,
	276	size_t subbuf_size,
	277	size_t n_subbufs,
	278	struct rchan_callbacks *cb)
	279	{
	280	unsigned int i;
	281	struct rchan *chan;
	282	char *tmpname;
e6c08367	283	int is_global = 0;
e82894f8 TZ	284
	285	if (!base_filename)
	286	return NULL;
	287
	288	if (!(subbuf_size && n_subbufs))
	289	return NULL;
	290
	291	chan = kcalloc(1, sizeof(struct rchan), GFP_KERNEL);
	292	if (!chan)
	293	return NULL;
	294
	295	chan->version = RELAYFS_CHANNEL_VERSION;
	296	chan->n_subbufs = n_subbufs;
	297	chan->subbuf_size = subbuf_size;
	298	chan->alloc_size = FIX_SIZE(subbuf_size * n_subbufs);
	299	setup_callbacks(chan, cb);
	300	kref_init(&chan->kref);
	301
	302	tmpname = kmalloc(NAME_MAX + 1, GFP_KERNEL);
	303	if (!tmpname)
	304	goto free_chan;
	305
	306	for_each_online_cpu(i) {
	307	sprintf(tmpname, "%s%d", base_filename, i);
e6c08367 TZ	308	chan->buf[i] = relay_open_buf(chan, tmpname, parent,
e6c08367 TZ	309	&is_global);
e82894f8 TZ	310	chan->buf[i]->cpu = i;
	311	if (!chan->buf[i])
	312	goto free_bufs;
	313	}
	314
	315	kfree(tmpname);
	316	return chan;
	317
	318	free_bufs:
	319	for (i = 0; i < NR_CPUS; i++) {
	320	if (!chan->buf[i])
	321	break;
	322	relay_close_buf(chan->buf[i]);
e6c08367 TZ	323	if (is_global)
e6c08367 TZ	324	break;
e82894f8 TZ	325	}
	326	kfree(tmpname);
	327
	328	free_chan:
	329	kref_put(&chan->kref, relay_destroy_channel);
	330	return NULL;
	331	}
	332
	333	/**
	334	* relay_switch_subbuf - switch to a new sub-buffer
	335	* @buf: channel buffer
	336	* @length: size of current event
	337	*
	338	* Returns either the length passed in or 0 if full.
	339
	340	* Performs sub-buffer-switch tasks such as invoking callbacks,
	341	* updating padding counts, waking up readers, etc.
	342	*/
	343	size_t relay_switch_subbuf(struct rchan_buf *buf, size_t length)
	344	{
	345	void old, new;
	346	size_t old_subbuf, new_subbuf;
	347
	348	if (unlikely(length > buf->chan->subbuf_size))
	349	goto toobig;
	350
	351	if (buf->offset != buf->chan->subbuf_size + 1) {
	352	buf->prev_padding = buf->chan->subbuf_size - buf->offset;
	353	old_subbuf = buf->subbufs_produced % buf->chan->n_subbufs;
	354	buf->padding[old_subbuf] = buf->prev_padding;
	355	buf->subbufs_produced++;
	356	if (waitqueue_active(&buf->read_wait)) {
	357	PREPARE_WORK(&buf->wake_readers, wakeup_readers, buf);
	358	schedule_delayed_work(&buf->wake_readers, 1);
	359	}
	360	}
	361
	362	old = buf->data;
	363	new_subbuf = buf->subbufs_produced % buf->chan->n_subbufs;
	364	new = buf->start + new_subbuf * buf->chan->subbuf_size;
	365	buf->offset = 0;
	366	if (!buf->chan->cb->subbuf_start(buf, new, old, buf->prev_padding)) {
	367	buf->offset = buf->chan->subbuf_size + 1;
	368	return 0;
	369	}
	370	buf->data = new;
	371	buf->padding[new_subbuf] = 0;
	372
	373	if (unlikely(length + buf->offset > buf->chan->subbuf_size))
	374	goto toobig;
	375
	376	return length;
	377
	378	toobig:
fd30fc32	379	buf->chan->last_toobig = length;
e82894f8 TZ	380	return 0;
	381	}
	382
	383	/**
	384	* relay_subbufs_consumed - update the buffer's sub-buffers-consumed count
	385	* @chan: the channel
	386	* @cpu: the cpu associated with the channel buffer to update
	387	* @subbufs_consumed: number of sub-buffers to add to current buf's count
	388	*
	389	* Adds to the channel buffer's consumed sub-buffer count.
	390	* subbufs_consumed should be the number of sub-buffers newly consumed,
	391	* not the total consumed.
	392	*
	393	* NOTE: kernel clients don't need to call this function if the channel
	394	* mode is 'overwrite'.
	395	*/
	396	void relay_subbufs_consumed(struct rchan *chan,
	397	unsigned int cpu,
	398	size_t subbufs_consumed)
	399	{
	400	struct rchan_buf *buf;
	401
	402	if (!chan)
	403	return;
	404
	405	if (cpu >= NR_CPUS \|\| !chan->buf[cpu])
	406	return;
	407
	408	buf = chan->buf[cpu];
	409	buf->subbufs_consumed += subbufs_consumed;
	410	if (buf->subbufs_consumed > buf->subbufs_produced)
	411	buf->subbufs_consumed = buf->subbufs_produced;
	412	}
	413
	414	/**
	415	* relay_destroy_channel - free the channel struct
	416	*
	417	* Should only be called from kref_put().
	418	*/
	419	void relay_destroy_channel(struct kref *kref)
	420	{
	421	struct rchan *chan = container_of(kref, struct rchan, kref);
	422	kfree(chan);
	423	}
	424
	425	/**
	426	* relay_close - close the channel
	427	* @chan: the channel
	428	*
	429	* Closes all channel buffers and frees the channel.
	430	*/
	431	void relay_close(struct rchan *chan)
	432	{
	433	unsigned int i;
e6c08367	434	struct rchan_buf *prev = NULL;
e82894f8 TZ	435
	436	if (!chan)
	437	return;
	438
	439	for (i = 0; i < NR_CPUS; i++) {
e6c08367 TZ	440	if (!chan->buf[i] \|\| chan->buf[i] == prev)
e6c08367 TZ	441	break;
e82894f8	442	relay_close_buf(chan->buf[i]);
e6c08367	443	prev = chan->buf[i];
e82894f8 TZ	444	}
e82894f8 TZ	445
fd30fc32 TZ	446	if (chan->last_toobig)
	447	printk(KERN_WARNING "relayfs: one or more items not logged "
	448	"[item size (%Zd) > sub-buffer size (%Zd)]\n",
	449	chan->last_toobig, chan->subbuf_size);
	450
e82894f8 TZ	451	kref_put(&chan->kref, relay_destroy_channel);
	452	}
	453
	454	/**
	455	* relay_flush - close the channel
	456	* @chan: the channel
	457	*
	458	* Flushes all channel buffers i.e. forces buffer switch.
	459	*/
	460	void relay_flush(struct rchan *chan)
	461	{
	462	unsigned int i;
e6c08367	463	struct rchan_buf *prev = NULL;
e82894f8 TZ	464
	465	if (!chan)
	466	return;
	467
	468	for (i = 0; i < NR_CPUS; i++) {
e6c08367 TZ	469	if (!chan->buf[i] \|\| chan->buf[i] == prev)
e6c08367 TZ	470	break;
e82894f8	471	relay_switch_subbuf(chan->buf[i], 0);
e6c08367	472	prev = chan->buf[i];
e82894f8 TZ	473	}
	474	}
	475
	476	EXPORT_SYMBOL_GPL(relay_open);
	477	EXPORT_SYMBOL_GPL(relay_close);
	478	EXPORT_SYMBOL_GPL(relay_flush);
	479	EXPORT_SYMBOL_GPL(relay_reset);
	480	EXPORT_SYMBOL_GPL(relay_subbufs_consumed);
	481	EXPORT_SYMBOL_GPL(relay_switch_subbuf);
	482	EXPORT_SYMBOL_GPL(relay_buf_full);