[lttng-modules.git] / ltt-ring-buffer-client.h

/*
 * ltt-ring-buffer-client.h
 *
 * Copyright (C) 2010 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
 *
 * LTTng lib ring buffer client template.
 *
 * Dual LGPL v2.1/GPL v2 license.
 */

#include <linux/module.h>
#include <linux/types.h>
#include "wrapper/vmalloc.h"	/* for wrapper_vmalloc_sync_all() */
#include "wrapper/trace-clock.h"
#include "ltt-events.h"
#include "ltt-tracer.h"

/*
 * Keep the natural field alignment for _each field_ within this structure if
 * you ever add/remove a field from this header. Packed attribute is not used
 * because gcc generates poor code on at least powerpc and mips. Don't ever
 * let gcc add padding between the structure elements.
 */
struct packet_header {
	uint32_t magic;			/*
					 * Trace magic number.
					 * contains endianness information.
					 */
	uint8_t trace_uuid[16];
	uint32_t stream_id;
	uint64_t timestamp_begin;	/* Cycle count at subbuffer start */
	uint64_t timestamp_end;	/* Cycle count at subbuffer end */
	uint32_t content_size;		/* Size of data in subbuffer */
	uint32_t packet_size;		/* Subbuffer size (include padding) */
	uint32_t events_lost;		/*
					 * Events lost in this subbuffer since
					 * the beginning of the trace.
					 * (may overflow)
					 */
#if 0
	uint64_t start_time_sec;	/* NTP-corrected start time */
	uint64_t start_time_usec;
	uint64_t start_freq;		/*
					 * Frequency at trace start,
				 * used all along the trace.
					 */
	uint32_t freq_scale;		/* Frequency scaling (divisor) */
#endif //0
	uint8_t header_end[0];		/* End of header */
};


static inline notrace u64 lib_ring_buffer_clock_read(struct channel *chan)
{
	return trace_clock_read64();
}

/*
 * record_header_size - Calculate the header size and padding necessary.
 * @config: ring buffer instance configuration
 * @chan: channel
 * @offset: offset in the write buffer
 * @data_size: size of the payload
 * @pre_header_padding: padding to add before the header (output)
 * @rflags: reservation flags
 * @ctx: reservation context
 *
 * Returns the event header size (including padding).
 *
 * Important note :
 * The event header must be 32-bits. The total offset calculated here :
 *
 * Alignment of header struct on 32 bits (min arch size, header size)
 * + sizeof(header struct)  (32-bits)
 * + (opt) u16 (ext. event id)
 * + (opt) u16 (event_size)
 *             (if event_size == LTT_MAX_SMALL_SIZE, has ext. event size)
 * + (opt) u32 (ext. event size)
 * + (opt) u64 full TSC (aligned on min(64-bits, arch size))
 *
 * The payload must itself determine its own alignment from the biggest type it
 * contains.
 */
static __inline__
unsigned char record_header_size(const struct lib_ring_buffer_config *config,
				 struct channel *chan, size_t offset,
				 size_t data_size, size_t *pre_header_padding,
				 unsigned int rflags,
				 struct lib_ring_buffer_ctx *ctx)
{
	size_t orig_offset = offset;
	size_t padding;

	BUILD_BUG_ON(sizeof(struct event_header) != sizeof(u32));

	padding = lib_ring_buffer_align(offset,
					sizeof(struct event_header));
	offset += padding;
	offset += sizeof(struct event_header);

	if (unlikely(rflags)) {
		switch (rflags) {
		case LTT_RFLAG_ID_SIZE_TSC:
			offset += sizeof(u16) + sizeof(u16);
			if (data_size >= LTT_MAX_SMALL_SIZE)
				offset += sizeof(u32);
			offset += lib_ring_buffer_align(offset, sizeof(u64));
			offset += sizeof(u64);
			break;
		case LTT_RFLAG_ID_SIZE:
			offset += sizeof(u16) + sizeof(u16);
			if (data_size >= LTT_MAX_SMALL_SIZE)
				offset += sizeof(u32);
			break;
		case LTT_RFLAG_ID:
			offset += sizeof(u16);
			break;
		}
	}

	*pre_header_padding = padding;
	return offset - orig_offset;
}

#include "wrapper/ringbuffer/api.h"

extern
void ltt_write_event_header_slow(const struct lib_ring_buffer_config *config,
				 struct lib_ring_buffer_ctx *ctx,
				 u16 eID, u32 event_size);

/*
 * ltt_write_event_header
 *
 * Writes the event header to the offset (already aligned on 32-bits).
 *
 * @config: ring buffer instance configuration
 * @ctx: reservation context
 * @eID : event ID
 * @event_size : size of the event, excluding the event header.
 */
static __inline__
void ltt_write_event_header(const struct lib_ring_buffer_config *config,
			    struct lib_ring_buffer_ctx *ctx,
			    u16 eID, u32 event_size)
{
	struct event_header header;

	if (unlikely(ctx->rflags))
		goto slow_path;

	header.id_time = eID << LTT_TSC_BITS;
	header.id_time |= (u32)ctx->tsc & LTT_TSC_MASK;
	lib_ring_buffer_write(config, ctx, &header, sizeof(header));

slow_path:
	ltt_write_event_header_slow(config, ctx, eID, event_size);
}

void ltt_write_event_header_slow(const struct lib_ring_buffer_config *config,
				   struct lib_ring_buffer_ctx *ctx,
				   u16 eID, u32 event_size)
{
	struct event_header header;
	u16 small_size;

	switch (ctx->rflags) {
	case LTT_RFLAG_ID_SIZE_TSC:
		header.id_time = 29 << LTT_TSC_BITS;
		break;
	case LTT_RFLAG_ID_SIZE:
		header.id_time = 30 << LTT_TSC_BITS;
		break;
	case LTT_RFLAG_ID:
		header.id_time = 31 << LTT_TSC_BITS;
		break;
	default:
		WARN_ON_ONCE(1);
		header.id_time = 0;
	}

	header.id_time |= (u32)ctx->tsc & LTT_TSC_MASK;
	lib_ring_buffer_write(config, ctx, &header, sizeof(header));

	switch (ctx->rflags) {
	case LTT_RFLAG_ID_SIZE_TSC:
		small_size = (u16)min_t(u32, event_size, LTT_MAX_SMALL_SIZE);
		lib_ring_buffer_write(config, ctx, &eID, sizeof(u16));
		lib_ring_buffer_write(config, ctx, &small_size, sizeof(u16));
		if (small_size == LTT_MAX_SMALL_SIZE)
			lib_ring_buffer_write(config, ctx, &event_size,
					      sizeof(u32));
		lib_ring_buffer_align_ctx(ctx, sizeof(u64));
		lib_ring_buffer_write(config, ctx, &ctx->tsc, sizeof(u64));
		break;
	case LTT_RFLAG_ID_SIZE:
		small_size = (u16)min_t(u32, event_size, LTT_MAX_SMALL_SIZE);
		lib_ring_buffer_write(config, ctx, &eID, sizeof(u16));
		lib_ring_buffer_write(config, ctx, &small_size, sizeof(u16));
		if (small_size == LTT_MAX_SMALL_SIZE)
			lib_ring_buffer_write(config, ctx, &event_size,
					      sizeof(u32));
		break;
	case LTT_RFLAG_ID:
		lib_ring_buffer_write(config, ctx, &eID, sizeof(u16));
		break;
	}
}

static const struct lib_ring_buffer_config client_config;

static u64 client_ring_buffer_clock_read(struct channel *chan)
{
	return lib_ring_buffer_clock_read(chan);
}

static
size_t client_record_header_size(const struct lib_ring_buffer_config *config,
				 struct channel *chan, size_t offset,
				 size_t data_size,
				 size_t *pre_header_padding,
				 unsigned int rflags,
				 struct lib_ring_buffer_ctx *ctx)
{
	return record_header_size(config, chan, offset, data_size,
				  pre_header_padding, rflags, ctx);
}

/**
 * client_packet_header_size - called on buffer-switch to a new sub-buffer
 *
 * Return header size without padding after the structure. Don't use packed
 * structure because gcc generates inefficient code on some architectures
 * (powerpc, mips..)
 */
static size_t client_packet_header_size(void)
{
	return offsetof(struct packet_header, header_end);
}

static void client_buffer_begin(struct lib_ring_buffer *buf, u64 tsc,
				unsigned int subbuf_idx)
{
	struct channel *chan = buf->backend.chan;
	struct packet_header *header =
		(struct packet_header *)
			lib_ring_buffer_offset_address(&buf->backend,
				subbuf_idx * chan->backend.subbuf_size);
	struct ltt_session *session = channel_get_private(chan);

	header->magic = CTF_MAGIC_NUMBER;
	memcpy(header->trace_uuid, session->uuid.b, sizeof(session->uuid));
	header->timestamp_begin = tsc;
	header->timestamp_end = 0;
	header->content_size = 0xFFFFFFFF; /* for debugging */
	header->packet_size = 0xFFFFFFFF;
	header->events_lost = 0;
#if 0
	header->start_time_sec = ltt_chan->session->start_time.tv_sec;
	header->start_time_usec = ltt_chan->session->start_time.tv_usec;
	header->start_freq = ltt_chan->session->start_freq;
	header->freq_scale = ltt_chan->session->freq_scale;
#endif //0
}

/*
 * offset is assumed to never be 0 here : never deliver a completely empty
 * subbuffer. data_size is between 1 and subbuf_size.
 */
static void client_buffer_end(struct lib_ring_buffer *buf, u64 tsc,
			      unsigned int subbuf_idx, unsigned long data_size)
{
	struct channel *chan = buf->backend.chan;
	struct packet_header *header =
		(struct packet_header *)
			lib_ring_buffer_offset_address(&buf->backend,
				subbuf_idx * chan->backend.subbuf_size);
	unsigned long records_lost = 0;

	header->timestamp_end = tsc;
	header->content_size = data_size;
	header->packet_size = PAGE_ALIGN(data_size);
	records_lost += lib_ring_buffer_get_records_lost_full(&client_config, buf);
	records_lost += lib_ring_buffer_get_records_lost_wrap(&client_config, buf);
	records_lost += lib_ring_buffer_get_records_lost_big(&client_config, buf);
	header->events_lost = records_lost;
}

static int client_buffer_create(struct lib_ring_buffer *buf, void *priv,
				int cpu, const char *name)
{
	return 0;
}

static void client_buffer_finalize(struct lib_ring_buffer *buf, void *priv, int cpu)
{
}

static const struct lib_ring_buffer_config client_config = {
	.cb.ring_buffer_clock_read = client_ring_buffer_clock_read,
	.cb.record_header_size = client_record_header_size,
	.cb.subbuffer_header_size = client_packet_header_size,
	.cb.buffer_begin = client_buffer_begin,
	.cb.buffer_end = client_buffer_end,
	.cb.buffer_create = client_buffer_create,
	.cb.buffer_finalize = client_buffer_finalize,

	.tsc_bits = 32,
	.alloc = RING_BUFFER_ALLOC_PER_CPU,
	.sync = RING_BUFFER_SYNC_PER_CPU,
	.mode = RING_BUFFER_MODE_TEMPLATE,
	.backend = RING_BUFFER_PAGE,
	.output = RING_BUFFER_SPLICE,
	.oops = RING_BUFFER_OOPS_CONSISTENCY,
	.ipi = RING_BUFFER_IPI_BARRIER,
	.wakeup = RING_BUFFER_WAKEUP_BY_TIMER,
};

static
struct channel *_channel_create(const char *name,
				struct ltt_session *session, void *buf_addr,
				size_t subbuf_size, size_t num_subbuf,
				unsigned int switch_timer_interval,
				unsigned int read_timer_interval)
{
	return channel_create(&client_config, name, session, buf_addr,
			      subbuf_size, num_subbuf, switch_timer_interval,
			      read_timer_interval);
}

static
void ltt_channel_destroy(struct channel *chan)
{
	channel_destroy(chan);
}

static
struct lib_ring_buffer *ltt_buffer_read_open(struct channel *chan)
{
	struct lib_ring_buffer *buf;
	int cpu;

	for_each_channel_cpu(cpu, chan) {
		buf = channel_get_ring_buffer(&client_config, chan, cpu);
		if (!lib_ring_buffer_open_read(buf))
			return buf;
	}
	return NULL;
}

static
void ltt_buffer_read_close(struct lib_ring_buffer *buf)
{
	lib_ring_buffer_release_read(buf);
	
}

static
int ltt_event_reserve(struct lib_ring_buffer_ctx *ctx)
{
	int ret, cpu;

	cpu = lib_ring_buffer_get_cpu(&client_config);
	if (cpu < 0)
		return -EPERM;
	ctx->cpu = cpu;

	ret = lib_ring_buffer_reserve(&client_config, ctx);
	if (ret)
		goto put;
	return ret;

put:
	lib_ring_buffer_put_cpu(&client_config);
	return ret;
}

static
void ltt_event_commit(struct lib_ring_buffer_ctx *ctx)
{
	lib_ring_buffer_commit(&client_config, ctx);
	lib_ring_buffer_put_cpu(&client_config);
}

static
void ltt_event_write(struct lib_ring_buffer_ctx *ctx, const void *src,
		     size_t len)
{
	lib_ring_buffer_write(&client_config, ctx, src, len);
}

static
wait_queue_head_t *ltt_get_reader_wait_queue(struct ltt_channel *chan)
{
	return &chan->chan->read_wait;
}

static struct ltt_transport ltt_relay_transport = {
	.name = "relay-" RING_BUFFER_MODE_TEMPLATE_STRING,
	.owner = THIS_MODULE,
	.ops = {
		.channel_create = _channel_create,
		.channel_destroy = ltt_channel_destroy,
		.buffer_read_open = ltt_buffer_read_open,
		.buffer_read_close = ltt_buffer_read_close,
		.event_reserve = ltt_event_reserve,
		.event_commit = ltt_event_commit,
		.event_write = ltt_event_write,
		.get_reader_wait_queue = ltt_get_reader_wait_queue,
	},
};

static int __init ltt_ring_buffer_client_init(void)
{
	/*
	 * This vmalloc sync all also takes care of the lib ring buffer
	 * vmalloc'd module pages when it is built as a module into LTTng.
	 */
	wrapper_vmalloc_sync_all();
	printk(KERN_INFO "LTT : ltt ring buffer client init\n");
	ltt_transport_register(&ltt_relay_transport);
	return 0;
}

module_init(ltt_ring_buffer_client_init);

static void __exit ltt_ring_buffer_client_exit(void)
{
	printk(KERN_INFO "LTT : ltt ring buffer client exit\n");
	ltt_transport_unregister(&ltt_relay_transport);
}

module_exit(ltt_ring_buffer_client_exit);

MODULE_LICENSE("GPL and additional rights");
MODULE_AUTHOR("Mathieu Desnoyers");
MODULE_DESCRIPTION("LTTng ring buffer " RING_BUFFER_MODE_TEMPLATE_STRING
		   " client");
Commit	Line	Data
7514523f	1	/*
3d084699	2	* ltt-ring-buffer-client.h
7514523f MD	3	*
	4	* Copyright (C) 2010 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
	5	*
3d084699	6	* LTTng lib ring buffer client template.
7514523f MD	7	*
	8	* Dual LGPL v2.1/GPL v2 license.
	9	*/
	10
	11	#include <linux/module.h>
c0e31d2e	12	#include <linux/types.h>
b13f3ebe	13	#include "wrapper/vmalloc.h" /* for wrapper_vmalloc_sync_all() */
f3bc08c5	14	#include "wrapper/trace-clock.h"
c0e31d2e	15	#include "ltt-events.h"
7514523f MD	16	#include "ltt-tracer.h"
7514523f MD	17
d793d5e1 MD	18	/*
	19	* Keep the natural field alignment for _each field_ within this structure if
	20	* you ever add/remove a field from this header. Packed attribute is not used
	21	* because gcc generates poor code on at least powerpc and mips. Don't ever
	22	* let gcc add padding between the structure elements.
	23	*/
	24	struct packet_header {
	25	uint32_t magic; /*
	26	* Trace magic number.
	27	* contains endianness information.
	28	*/
	29	uint8_t trace_uuid[16];
	30	uint32_t stream_id;
	31	uint64_t timestamp_begin; /* Cycle count at subbuffer start */
	32	uint64_t timestamp_end; /* Cycle count at subbuffer end */
	33	uint32_t content_size; /* Size of data in subbuffer */
	34	uint32_t packet_size; /* Subbuffer size (include padding) */
	35	uint32_t events_lost; /*
	36	* Events lost in this subbuffer since
	37	* the beginning of the trace.
	38	* (may overflow)
	39	*/
	40	#if 0
	41	uint64_t start_time_sec; /* NTP-corrected start time */
	42	uint64_t start_time_usec;
	43	uint64_t start_freq; /*
	44	* Frequency at trace start,
	45	* used all along the trace.
	46	*/
	47	uint32_t freq_scale; /* Frequency scaling (divisor) */
	48	#endif //0
	49	uint8_t header_end[0]; /* End of header */
	50	};
	51
	52
881833e3 MD	53	static inline notrace u64 lib_ring_buffer_clock_read(struct channel *chan)
	54	{
	55	return trace_clock_read64();
	56	}
	57
	58	/*
	59	* record_header_size - Calculate the header size and padding necessary.
	60	* @config: ring buffer instance configuration
	61	* @chan: channel
	62	* @offset: offset in the write buffer
	63	* @data_size: size of the payload
	64	* @pre_header_padding: padding to add before the header (output)
	65	* @rflags: reservation flags
	66	* @ctx: reservation context
	67	*
	68	* Returns the event header size (including padding).
	69	*
	70	* Important note :
	71	* The event header must be 32-bits. The total offset calculated here :
	72	*
	73	* Alignment of header struct on 32 bits (min arch size, header size)
	74	* + sizeof(header struct) (32-bits)
	75	* + (opt) u16 (ext. event id)
	76	* + (opt) u16 (event_size)
	77	* (if event_size == LTT_MAX_SMALL_SIZE, has ext. event size)
	78	* + (opt) u32 (ext. event size)
	79	* + (opt) u64 full TSC (aligned on min(64-bits, arch size))
	80	*
	81	* The payload must itself determine its own alignment from the biggest type it
	82	* contains.
	83	*/
	84	static __inline__
	85	unsigned char record_header_size(const struct lib_ring_buffer_config *config,
	86	struct channel *chan, size_t offset,
	87	size_t data_size, size_t *pre_header_padding,
	88	unsigned int rflags,
	89	struct lib_ring_buffer_ctx *ctx)
	90	{
	91	size_t orig_offset = offset;
	92	size_t padding;
	93
	94	BUILD_BUG_ON(sizeof(struct event_header) != sizeof(u32));
	95
	96	padding = lib_ring_buffer_align(offset,
	97	sizeof(struct event_header));
	98	offset += padding;
	99	offset += sizeof(struct event_header);
	100
	101	if (unlikely(rflags)) {
	102	switch (rflags) {
	103	case LTT_RFLAG_ID_SIZE_TSC:
	104	offset += sizeof(u16) + sizeof(u16);
	105	if (data_size >= LTT_MAX_SMALL_SIZE)
	106	offset += sizeof(u32);
	107	offset += lib_ring_buffer_align(offset, sizeof(u64));
	108	offset += sizeof(u64);
	109	break;
	110	case LTT_RFLAG_ID_SIZE:
	111	offset += sizeof(u16) + sizeof(u16);
	112	if (data_size >= LTT_MAX_SMALL_SIZE)
	113	offset += sizeof(u32);
	114	break;
	115	case LTT_RFLAG_ID:
	116	offset += sizeof(u16);
117	break;
118	}
119	}
120
121	*pre_header_padding = padding;
122	return offset - orig_offset;
123	}
124
125	#include "wrapper/ringbuffer/api.h"
126
127	extern
128	void ltt_write_event_header_slow(const struct lib_ring_buffer_config *config,
129	struct lib_ring_buffer_ctx *ctx,
130	u16 eID, u32 event_size);
131
132	/*
133	* ltt_write_event_header
134	*
135	* Writes the event header to the offset (already aligned on 32-bits).
136	*
137	* @config: ring buffer instance configuration
138	* @ctx: reservation context
139	* @eID : event ID
140	* @event_size : size of the event, excluding the event header.
141	*/
142	static __inline__
143	void ltt_write_event_header(const struct lib_ring_buffer_config *config,
144	struct lib_ring_buffer_ctx *ctx,
145	u16 eID, u32 event_size)
146	{
147	struct event_header header;
148
149	if (unlikely(ctx->rflags))
150	goto slow_path;
151
152	header.id_time = eID << LTT_TSC_BITS;
153	header.id_time \|= (u32)ctx->tsc & LTT_TSC_MASK;
154	lib_ring_buffer_write(config, ctx, &header, sizeof(header));
155
156	slow_path:
157	ltt_write_event_header_slow(config, ctx, eID, event_size);
158	}
159
881833e3 MD	160	void ltt_write_event_header_slow(const struct lib_ring_buffer_config *config,
	161	struct lib_ring_buffer_ctx *ctx,
	162	u16 eID, u32 event_size)
	163	{
	164	struct event_header header;
	165	u16 small_size;
	166
	167	switch (ctx->rflags) {
	168	case LTT_RFLAG_ID_SIZE_TSC:
	169	header.id_time = 29 << LTT_TSC_BITS;
	170	break;
	171	case LTT_RFLAG_ID_SIZE:
	172	header.id_time = 30 << LTT_TSC_BITS;
	173	break;
	174	case LTT_RFLAG_ID:
	175	header.id_time = 31 << LTT_TSC_BITS;
	176	break;
	177	default:
	178	WARN_ON_ONCE(1);
	179	header.id_time = 0;
	180	}
	181
	182	header.id_time \|= (u32)ctx->tsc & LTT_TSC_MASK;
	183	lib_ring_buffer_write(config, ctx, &header, sizeof(header));
	184
	185	switch (ctx->rflags) {
	186	case LTT_RFLAG_ID_SIZE_TSC:
	187	small_size = (u16)min_t(u32, event_size, LTT_MAX_SMALL_SIZE);
	188	lib_ring_buffer_write(config, ctx, &eID, sizeof(u16));
	189	lib_ring_buffer_write(config, ctx, &small_size, sizeof(u16));
	190	if (small_size == LTT_MAX_SMALL_SIZE)
	191	lib_ring_buffer_write(config, ctx, &event_size,
	192	sizeof(u32));
	193	lib_ring_buffer_align_ctx(ctx, sizeof(u64));
	194	lib_ring_buffer_write(config, ctx, &ctx->tsc, sizeof(u64));
	195	break;
	196	case LTT_RFLAG_ID_SIZE:
	197	small_size = (u16)min_t(u32, event_size, LTT_MAX_SMALL_SIZE);
	198	lib_ring_buffer_write(config, ctx, &eID, sizeof(u16));
	199	lib_ring_buffer_write(config, ctx, &small_size, sizeof(u16));
	200	if (small_size == LTT_MAX_SMALL_SIZE)
	201	lib_ring_buffer_write(config, ctx, &event_size,
	202	sizeof(u32));
	203	break;
	204	case LTT_RFLAG_ID:
	205	lib_ring_buffer_write(config, ctx, &eID, sizeof(u16));
	206	break;
	207	}
	208	}
	209
7514523f MD	210	static const struct lib_ring_buffer_config client_config;
	211
	212	static u64 client_ring_buffer_clock_read(struct channel *chan)
	213	{
	214	return lib_ring_buffer_clock_read(chan);
	215	}
	216
1e2015dc	217	static
7514523f MD	218	size_t client_record_header_size(const struct lib_ring_buffer_config *config,
	219	struct channel *chan, size_t offset,
	220	size_t data_size,
	221	size_t *pre_header_padding,
	222	unsigned int rflags,
	223	struct lib_ring_buffer_ctx *ctx)
	224	{
	225	return record_header_size(config, chan, offset, data_size,
	226	pre_header_padding, rflags, ctx);
	227	}
	228
	229	/**
1c25284c	230	* client_packet_header_size - called on buffer-switch to a new sub-buffer
7514523f MD	231	*
	232	* Return header size without padding after the structure. Don't use packed
	233	* structure because gcc generates inefficient code on some architectures
	234	* (powerpc, mips..)
	235	*/
1c25284c	236	static size_t client_packet_header_size(void)
7514523f	237	{
1c25284c	238	return offsetof(struct packet_header, header_end);
7514523f MD	239	}
	240
	241	static void client_buffer_begin(struct lib_ring_buffer *buf, u64 tsc,
	242	unsigned int subbuf_idx)
	243	{
	244	struct channel *chan = buf->backend.chan;
1c25284c MD	245	struct packet_header *header =
1c25284c MD	246	(struct packet_header *)
7514523f MD	247	lib_ring_buffer_offset_address(&buf->backend,
7514523f MD	248	subbuf_idx * chan->backend.subbuf_size);
d793d5e1	249	struct ltt_session *session = channel_get_private(chan);
7514523f	250
d793d5e1 MD	251	header->magic = CTF_MAGIC_NUMBER;
d793d5e1 MD	252	memcpy(header->trace_uuid, session->uuid.b, sizeof(session->uuid));
1c25284c	253	header->timestamp_begin = tsc;
d793d5e1	254	header->timestamp_end = 0;
1c25284c	255	header->content_size = 0xFFFFFFFF; /* for debugging */
d793d5e1 MD	256	header->packet_size = 0xFFFFFFFF;
	257	header->events_lost = 0;
	258	#if 0
	259	header->start_time_sec = ltt_chan->session->start_time.tv_sec;
	260	header->start_time_usec = ltt_chan->session->start_time.tv_usec;
	261	header->start_freq = ltt_chan->session->start_freq;
	262	header->freq_scale = ltt_chan->session->freq_scale;
	263	#endif //0
7514523f MD	264	}
	265
	266	/*
	267	* offset is assumed to never be 0 here : never deliver a completely empty
	268	* subbuffer. data_size is between 1 and subbuf_size.
	269	*/
	270	static void client_buffer_end(struct lib_ring_buffer *buf, u64 tsc,
	271	unsigned int subbuf_idx, unsigned long data_size)
	272	{
	273	struct channel *chan = buf->backend.chan;
1c25284c MD	274	struct packet_header *header =
1c25284c MD	275	(struct packet_header *)
7514523f MD	276	lib_ring_buffer_offset_address(&buf->backend,
	277	subbuf_idx * chan->backend.subbuf_size);
	278	unsigned long records_lost = 0;
	279
d793d5e1	280	header->timestamp_end = tsc;
1c25284c MD	281	header->content_size = data_size;
1c25284c MD	282	header->packet_size = PAGE_ALIGN(data_size);
7514523f MD	283	records_lost += lib_ring_buffer_get_records_lost_full(&client_config, buf);
	284	records_lost += lib_ring_buffer_get_records_lost_wrap(&client_config, buf);
	285	records_lost += lib_ring_buffer_get_records_lost_big(&client_config, buf);
	286	header->events_lost = records_lost;
7514523f MD	287	}
	288
	289	static int client_buffer_create(struct lib_ring_buffer buf, void priv,
	290	int cpu, const char *name)
	291	{
1c25284c	292	return 0;
7514523f MD	293	}
	294
	295	static void client_buffer_finalize(struct lib_ring_buffer buf, void priv, int cpu)
	296	{
7514523f MD	297	}
	298
	299	static const struct lib_ring_buffer_config client_config = {
	300	.cb.ring_buffer_clock_read = client_ring_buffer_clock_read,
	301	.cb.record_header_size = client_record_header_size,
1c25284c	302	.cb.subbuffer_header_size = client_packet_header_size,
7514523f MD	303	.cb.buffer_begin = client_buffer_begin,
	304	.cb.buffer_end = client_buffer_end,
	305	.cb.buffer_create = client_buffer_create,
	306	.cb.buffer_finalize = client_buffer_finalize,
	307
	308	.tsc_bits = 32,
	309	.alloc = RING_BUFFER_ALLOC_PER_CPU,
	310	.sync = RING_BUFFER_SYNC_PER_CPU,
3d084699	311	.mode = RING_BUFFER_MODE_TEMPLATE,
7514523f MD	312	.backend = RING_BUFFER_PAGE,
	313	.output = RING_BUFFER_SPLICE,
	314	.oops = RING_BUFFER_OOPS_CONSISTENCY,
	315	.ipi = RING_BUFFER_IPI_BARRIER,
	316	.wakeup = RING_BUFFER_WAKEUP_BY_TIMER,
	317	};
	318
1e2015dc	319	static
1c25284c MD	320	struct channel _channel_create(const char name,
	321	struct ltt_session session, void buf_addr,
	322	size_t subbuf_size, size_t num_subbuf,
	323	unsigned int switch_timer_interval,
	324	unsigned int read_timer_interval)
7514523f	325	{
1c25284c	326	return channel_create(&client_config, name, session, buf_addr,
7514523f MD	327	subbuf_size, num_subbuf, switch_timer_interval,
7514523f MD	328	read_timer_interval);
7514523f MD	329	}
7514523f MD	330
1e2015dc	331	static
7514523f MD	332	void ltt_channel_destroy(struct channel *chan)
7514523f MD	333	{
7514523f	334	channel_destroy(chan);
7514523f MD	335	}
7514523f MD	336
ad1c05e1 MD	337	static
	338	struct lib_ring_buffer ltt_buffer_read_open(struct channel chan)
	339	{
	340	struct lib_ring_buffer *buf;
	341	int cpu;
	342
1c25284c MD	343	for_each_channel_cpu(cpu, chan) {
1c25284c MD	344	buf = channel_get_ring_buffer(&client_config, chan, cpu);
ad1c05e1 MD	345	if (!lib_ring_buffer_open_read(buf))
	346	return buf;
	347	}
	348	return NULL;
	349	}
	350
	351	static
1c25284c	352	void ltt_buffer_read_close(struct lib_ring_buffer *buf)
ad1c05e1 MD	353	{
ad1c05e1 MD	354	lib_ring_buffer_release_read(buf);
1c25284c MD	355
	356	}
	357
c099397a	358	static
1c25284c MD	359	int ltt_event_reserve(struct lib_ring_buffer_ctx *ctx)
	360	{
	361	int ret, cpu;
	362
	363	cpu = lib_ring_buffer_get_cpu(&client_config);
	364	if (cpu < 0)
	365	return -EPERM;
	366	ctx->cpu = cpu;
	367
	368	ret = lib_ring_buffer_reserve(&client_config, ctx);
	369	if (ret)
	370	goto put;
	371	return ret;
	372
	373	put:
	374	lib_ring_buffer_put_cpu(&client_config);
	375	return ret;
ad1c05e1 MD	376	}
ad1c05e1 MD	377
c099397a	378	static
1c25284c MD	379	void ltt_event_commit(struct lib_ring_buffer_ctx *ctx)
	380	{
	381	lib_ring_buffer_commit(&client_config, ctx);
	382	lib_ring_buffer_put_cpu(&client_config);
	383	}
	384
c099397a	385	static
e763dbf5 MD	386	void ltt_event_write(struct lib_ring_buffer_ctx ctx, const void src,
	387	size_t len)
	388	{
	389	lib_ring_buffer_write(&client_config, ctx, src, len);
	390	}
1c25284c	391
c099397a MD	392	static
	393	wait_queue_head_t ltt_get_reader_wait_queue(struct ltt_channel chan)
	394	{
	395	return &chan->chan->read_wait;
	396	}
	397
7514523f	398	static struct ltt_transport ltt_relay_transport = {
3d084699	399	.name = "relay-" RING_BUFFER_MODE_TEMPLATE_STRING,
7514523f MD	400	.owner = THIS_MODULE,
7514523f MD	401	.ops = {
1c25284c MD	402	.channel_create = _channel_create,
1c25284c MD	403	.channel_destroy = ltt_channel_destroy,
ad1c05e1 MD	404	.buffer_read_open = ltt_buffer_read_open,
ad1c05e1 MD	405	.buffer_read_close = ltt_buffer_read_close,
1c25284c MD	406	.event_reserve = ltt_event_reserve,
1c25284c MD	407	.event_commit = ltt_event_commit,
e763dbf5	408	.event_write = ltt_event_write,
c099397a	409	.get_reader_wait_queue = ltt_get_reader_wait_queue,
7514523f MD	410	},
	411	};
	412
3d084699	413	static int __init ltt_ring_buffer_client_init(void)
7514523f	414	{
a509e133 MD	415	/*
	416	* This vmalloc sync all also takes care of the lib ring buffer
	417	* vmalloc'd module pages when it is built as a module into LTTng.
	418	*/
6d2a620c	419	wrapper_vmalloc_sync_all();
7514523f MD	420	printk(KERN_INFO "LTT : ltt ring buffer client init\n");
	421	ltt_transport_register(&ltt_relay_transport);
	422	return 0;
	423	}
	424
1c25284c MD	425	module_init(ltt_ring_buffer_client_init);
1c25284c MD	426
3d084699	427	static void __exit ltt_ring_buffer_client_exit(void)
7514523f MD	428	{
	429	printk(KERN_INFO "LTT : ltt ring buffer client exit\n");
	430	ltt_transport_unregister(&ltt_relay_transport);
	431	}
	432
1c25284c MD	433	module_exit(ltt_ring_buffer_client_exit);
1c25284c MD	434
7514523f MD	435	MODULE_LICENSE("GPL and additional rights");
7514523f MD	436	MODULE_AUTHOR("Mathieu Desnoyers");
3d084699 MD	437	MODULE_DESCRIPTION("LTTng ring buffer " RING_BUFFER_MODE_TEMPLATE_STRING
3d084699 MD	438	" client");