[lttng-modules.git] / lib / ringbuffer / frontend.h

#ifndef _LINUX_RING_BUFFER_FRONTEND_H
#define _LINUX_RING_BUFFER_FRONTEND_H

/*
 * linux/ringbuffer/frontend.h
 *
 * (C) Copyright 2005-2010 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
 *
 * Ring Buffer Library Synchronization Header (API).
 *
 * Author:
 *	Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
 *
 * See ring_buffer_frontend.c for more information on wait-free algorithms.
 *
 * Dual LGPL v2.1/GPL v2 license.
 */

#include <linux/pipe_fs_i.h>
#include <linux/rcupdate.h>
#include <linux/cpumask.h>
#include <linux/module.h>
#include <linux/bitops.h>
#include <linux/splice.h>
#include <linux/string.h>
#include <linux/timer.h>
#include <linux/sched.h>
#include <linux/cache.h>
#include <linux/time.h>
#include <linux/slab.h>
#include <linux/init.h>
#include <linux/stat.h>
#include <linux/cpu.h>
#include <linux/fs.h>

#include <asm/atomic.h>
#include <asm/local.h>

/* Internal helpers */
#include "../../wrapper/ringbuffer/frontend_internal.h"

/* Buffer creation/removal and setup operations */

/*
 * switch_timer_interval is the time interval (in us) to fill sub-buffers with
 * padding to let readers get those sub-buffers.  Used for live streaming.
 *
 * read_timer_interval is the time interval (in us) to wake up pending readers.
 *
 * buf_addr is a pointer the the beginning of the preallocated buffer contiguous
 * address mapping. It is used only by RING_BUFFER_STATIC configuration. It can
 * be set to NULL for other backends.
 */

extern
struct channel *channel_create(const struct lib_ring_buffer_config *config,
			       const char *name, void *priv,
			       void *buf_addr,
			       size_t subbuf_size, size_t num_subbuf,
			       unsigned int switch_timer_interval,
			       unsigned int read_timer_interval);

/*
 * channel_destroy returns the private data pointer. It finalizes all channel's
 * buffers, waits for readers to release all references, and destroys the
 * channel.
 */
extern
void *channel_destroy(struct channel *chan);


/* Buffer read operations */

/*
 * Iteration on channel cpumask needs to issue a read barrier to match the write
 * barrier in cpu hotplug. It orders the cpumask read before read of per-cpu
 * buffer data. The per-cpu buffer is never removed by cpu hotplug; teardown is
 * only performed at channel destruction.
 */
#define for_each_channel_cpu(cpu, chan)					\
	for ((cpu) = -1;						\
		({ (cpu) = cpumask_next(cpu, (chan)->backend.cpumask);	\
		   smp_read_barrier_depends(); (cpu) < nr_cpu_ids; });)

extern struct lib_ring_buffer *channel_get_ring_buffer(
				const struct lib_ring_buffer_config *config,
				struct channel *chan, int cpu);
extern int lib_ring_buffer_open_read(struct lib_ring_buffer *buf);
extern void lib_ring_buffer_release_read(struct lib_ring_buffer *buf);

/*
 * Read sequence: snapshot, many get_subbuf/put_subbuf, move_consumer.
 */
extern int lib_ring_buffer_snapshot(struct lib_ring_buffer *buf,
				    unsigned long *consumed,
				    unsigned long *produced);
extern void lib_ring_buffer_move_consumer(struct lib_ring_buffer *buf,
					  unsigned long consumed_new);

extern int lib_ring_buffer_get_subbuf(struct lib_ring_buffer *buf,
				      unsigned long consumed);
extern void lib_ring_buffer_put_subbuf(struct lib_ring_buffer *buf);

/*
 * lib_ring_buffer_get_next_subbuf/lib_ring_buffer_put_next_subbuf are helpers
 * to read sub-buffers sequentially.
 */
static inline int lib_ring_buffer_get_next_subbuf(struct lib_ring_buffer *buf)
{
	int ret;

	ret = lib_ring_buffer_snapshot(buf, &buf->cons_snapshot,
				       &buf->prod_snapshot);
	if (ret)
		return ret;
	ret = lib_ring_buffer_get_subbuf(buf, buf->cons_snapshot);
	return ret;
}

static inline void lib_ring_buffer_put_next_subbuf(struct lib_ring_buffer *buf)
{
	lib_ring_buffer_put_subbuf(buf);
	lib_ring_buffer_move_consumer(buf, subbuf_align(buf->cons_snapshot,
						    buf->backend.chan));
}

extern void channel_reset(struct channel *chan);
extern void lib_ring_buffer_reset(struct lib_ring_buffer *buf);

static inline
unsigned long lib_ring_buffer_get_offset(const struct lib_ring_buffer_config *config,
					 struct lib_ring_buffer *buf)
{
	return v_read(config, &buf->offset);
}

static inline
unsigned long lib_ring_buffer_get_consumed(const struct lib_ring_buffer_config *config,
					   struct lib_ring_buffer *buf)
{
	return atomic_long_read(&buf->consumed);
}

/*
 * Must call lib_ring_buffer_is_finalized before reading counters (memory
 * ordering enforced with respect to trace teardown).
 */
static inline
int lib_ring_buffer_is_finalized(const struct lib_ring_buffer_config *config,
				 struct lib_ring_buffer *buf)
{
	int finalized = ACCESS_ONCE(buf->finalized);
	/*
	 * Read finalized before counters.
	 */
	smp_rmb();
	return finalized;
}

static inline
int lib_ring_buffer_channel_is_finalized(const struct channel *chan)
{
	return chan->finalized;
}

static inline
unsigned long lib_ring_buffer_get_read_data_size(
				const struct lib_ring_buffer_config *config,
				struct lib_ring_buffer *buf)
{
	return subbuffer_get_read_data_size(config, &buf->backend);
}

static inline
unsigned long lib_ring_buffer_get_records_count(
				const struct lib_ring_buffer_config *config,
				struct lib_ring_buffer *buf)
{
	return v_read(config, &buf->records_count);
}

static inline
unsigned long lib_ring_buffer_get_records_overrun(
				const struct lib_ring_buffer_config *config,
				struct lib_ring_buffer *buf)
{
	return v_read(config, &buf->records_overrun);
}

static inline
unsigned long lib_ring_buffer_get_records_lost_full(
				const struct lib_ring_buffer_config *config,
				struct lib_ring_buffer *buf)
{
	return v_read(config, &buf->records_lost_full);
}

static inline
unsigned long lib_ring_buffer_get_records_lost_wrap(
				const struct lib_ring_buffer_config *config,
				struct lib_ring_buffer *buf)
{
	return v_read(config, &buf->records_lost_wrap);
}

static inline
unsigned long lib_ring_buffer_get_records_lost_big(
				const struct lib_ring_buffer_config *config,
				struct lib_ring_buffer *buf)
{
	return v_read(config, &buf->records_lost_big);
}

static inline
unsigned long lib_ring_buffer_get_records_read(
				const struct lib_ring_buffer_config *config,
				struct lib_ring_buffer *buf)
{
	return v_read(config, &buf->backend.records_read);
}

#endif /* _LINUX_RING_BUFFER_FRONTEND_H */
Commit	Line	Data
f3bc08c5 MD	1	#ifndef _LINUX_RING_BUFFER_FRONTEND_H
	2	#define _LINUX_RING_BUFFER_FRONTEND_H
	3
	4	/*
	5	* linux/ringbuffer/frontend.h
	6	*
	7	* (C) Copyright 2005-2010 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
	8	*
	9	* Ring Buffer Library Synchronization Header (API).
	10	*
	11	* Author:
	12	* Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
	13	*
	14	* See ring_buffer_frontend.c for more information on wait-free algorithms.
	15	*
	16	* Dual LGPL v2.1/GPL v2 license.
	17	*/
	18
	19	#include <linux/pipe_fs_i.h>
	20	#include <linux/rcupdate.h>
	21	#include <linux/cpumask.h>
	22	#include <linux/module.h>
	23	#include <linux/bitops.h>
	24	#include <linux/splice.h>
	25	#include <linux/string.h>
	26	#include <linux/timer.h>
	27	#include <linux/sched.h>
	28	#include <linux/cache.h>
	29	#include <linux/time.h>
	30	#include <linux/slab.h>
	31	#include <linux/init.h>
	32	#include <linux/stat.h>
	33	#include <linux/cpu.h>
	34	#include <linux/fs.h>
	35
	36	#include <asm/atomic.h>
	37	#include <asm/local.h>
	38
	39	/* Internal helpers */
	40	#include "../../wrapper/ringbuffer/frontend_internal.h"
	41
	42	/* Buffer creation/removal and setup operations */
	43
	44	/*
	45	* switch_timer_interval is the time interval (in us) to fill sub-buffers with
	46	* padding to let readers get those sub-buffers. Used for live streaming.
	47	*
	48	* read_timer_interval is the time interval (in us) to wake up pending readers.
	49	*
	50	* buf_addr is a pointer the the beginning of the preallocated buffer contiguous
	51	* address mapping. It is used only by RING_BUFFER_STATIC configuration. It can
	52	* be set to NULL for other backends.
	53	*/
	54
	55	extern
	56	struct channel channel_create(const struct lib_ring_buffer_config config,
	57	const char name, void priv,
	58	void *buf_addr,
	59	size_t subbuf_size, size_t num_subbuf,
	60	unsigned int switch_timer_interval,
	61	unsigned int read_timer_interval);
	62
	63	/*
	64	* channel_destroy returns the private data pointer. It finalizes all channel's
65	* buffers, waits for readers to release all references, and destroys the
66	* channel.
67	*/
68	extern
69	void channel_destroy(struct channel chan);
70
71
72	/* Buffer read operations */
73
74	/*
75	* Iteration on channel cpumask needs to issue a read barrier to match the write
76	* barrier in cpu hotplug. It orders the cpumask read before read of per-cpu
77	* buffer data. The per-cpu buffer is never removed by cpu hotplug; teardown is
78	* only performed at channel destruction.
79	*/
80	#define for_each_channel_cpu(cpu, chan) \
81	for ((cpu) = -1; \
82	({ (cpu) = cpumask_next(cpu, (chan)->backend.cpumask); \
83	smp_read_barrier_depends(); (cpu) < nr_cpu_ids; });)
84
85	extern struct lib_ring_buffer *channel_get_ring_buffer(
86	const struct lib_ring_buffer_config *config,
87	struct channel *chan, int cpu);
88	extern int lib_ring_buffer_open_read(struct lib_ring_buffer *buf);
89	extern void lib_ring_buffer_release_read(struct lib_ring_buffer *buf);
90
91	/*
92	* Read sequence: snapshot, many get_subbuf/put_subbuf, move_consumer.
93	*/
94	extern int lib_ring_buffer_snapshot(struct lib_ring_buffer *buf,
95	unsigned long *consumed,
96	unsigned long *produced);
97	extern void lib_ring_buffer_move_consumer(struct lib_ring_buffer *buf,
98	unsigned long consumed_new);
99
100	extern int lib_ring_buffer_get_subbuf(struct lib_ring_buffer *buf,
101	unsigned long consumed);
102	extern void lib_ring_buffer_put_subbuf(struct lib_ring_buffer *buf);
103
104	/*
105	* lib_ring_buffer_get_next_subbuf/lib_ring_buffer_put_next_subbuf are helpers
106	* to read sub-buffers sequentially.
107	*/
108	static inline int lib_ring_buffer_get_next_subbuf(struct lib_ring_buffer *buf)
109	{
110	int ret;
111
112	ret = lib_ring_buffer_snapshot(buf, &buf->cons_snapshot,
113	&buf->prod_snapshot);
114	if (ret)
115	return ret;
116	ret = lib_ring_buffer_get_subbuf(buf, buf->cons_snapshot);
117	return ret;
118	}
119
120	static inline void lib_ring_buffer_put_next_subbuf(struct lib_ring_buffer *buf)
121	{
122	lib_ring_buffer_put_subbuf(buf);
123	lib_ring_buffer_move_consumer(buf, subbuf_align(buf->cons_snapshot,
124	buf->backend.chan));
125	}
126
127	extern void channel_reset(struct channel *chan);
128	extern void lib_ring_buffer_reset(struct lib_ring_buffer *buf);
129
130	static inline
131	unsigned long lib_ring_buffer_get_offset(const struct lib_ring_buffer_config *config,
132	struct lib_ring_buffer *buf)
133	{
134	return v_read(config, &buf->offset);
135	}
136
137	static inline
138	unsigned long lib_ring_buffer_get_consumed(const struct lib_ring_buffer_config *config,
139	struct lib_ring_buffer *buf)
140	{
141	return atomic_long_read(&buf->consumed);
142	}
143
144	/*
145	* Must call lib_ring_buffer_is_finalized before reading counters (memory
146	* ordering enforced with respect to trace teardown).
147	*/
148	static inline
149	int lib_ring_buffer_is_finalized(const struct lib_ring_buffer_config *config,
150	struct lib_ring_buffer *buf)
151	{
152	int finalized = ACCESS_ONCE(buf->finalized);
153	/*
154	* Read finalized before counters.
155	*/
156	smp_rmb();
157	return finalized;
158	}
159
24cedcfe MD	160	static inline
	161	int lib_ring_buffer_channel_is_finalized(const struct channel *chan)
	162	{
	163	return chan->finalized;
	164	}
	165
f3bc08c5 MD	166	static inline
	167	unsigned long lib_ring_buffer_get_read_data_size(
	168	const struct lib_ring_buffer_config *config,
	169	struct lib_ring_buffer *buf)
	170	{
	171	return subbuffer_get_read_data_size(config, &buf->backend);
	172	}
	173
	174	static inline
	175	unsigned long lib_ring_buffer_get_records_count(
	176	const struct lib_ring_buffer_config *config,
	177	struct lib_ring_buffer *buf)
	178	{
	179	return v_read(config, &buf->records_count);
	180	}
	181
	182	static inline
	183	unsigned long lib_ring_buffer_get_records_overrun(
	184	const struct lib_ring_buffer_config *config,
	185	struct lib_ring_buffer *buf)
	186	{
	187	return v_read(config, &buf->records_overrun);
	188	}
	189
	190	static inline
	191	unsigned long lib_ring_buffer_get_records_lost_full(
	192	const struct lib_ring_buffer_config *config,
	193	struct lib_ring_buffer *buf)
	194	{
	195	return v_read(config, &buf->records_lost_full);
	196	}
	197
	198	static inline
	199	unsigned long lib_ring_buffer_get_records_lost_wrap(
	200	const struct lib_ring_buffer_config *config,
	201	struct lib_ring_buffer *buf)
	202	{
	203	return v_read(config, &buf->records_lost_wrap);
	204	}
	205
	206	static inline
	207	unsigned long lib_ring_buffer_get_records_lost_big(
	208	const struct lib_ring_buffer_config *config,
	209	struct lib_ring_buffer *buf)
	210	{
	211	return v_read(config, &buf->records_lost_big);
	212	}
	213
	214	static inline
	215	unsigned long lib_ring_buffer_get_records_read(
	216	const struct lib_ring_buffer_config *config,
	217	struct lib_ring_buffer *buf)
	218	{
	219	return v_read(config, &buf->backend.records_read);
	220	}
	221
f3bc08c5	222	#endif /* _LINUX_RING_BUFFER_FRONTEND_H */