[lttng-modules.git] / src / lttng-ring-buffer-event-notifier-client.h

/* SPDX-License-Identifier: (GPL-2.0 or LGPL-2.1)
 *
 * lttng-ring-buffer-event-notifier-client.h
 *
 * LTTng lib ring buffer event notifier client template.
 *
 * Copyright (C) 2010-2020 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
 */

#include <linux/limits.h>
#include <linux/module.h>
#include <linux/types.h>
#include <wrapper/vmalloc.h>	/* for wrapper_vmalloc_sync_mappings() */
#include <lttng/abi.h>
#include <lttng/events.h>
#include <lttng/events-internal.h>
#include <lttng/tracer.h>
#include <wrapper/limits.h>

static struct lttng_transport lttng_relay_transport;

struct event_notifier_packet_header {
	uint8_t  header_end[0];
};

struct event_notifier_record_header {
	uint32_t payload_len;		/* in bytes */
	uint8_t header_end[0];		/* End of header */
};

static const struct lttng_kernel_ring_buffer_config client_config;

static inline
u64 lib_ring_buffer_clock_read(struct lttng_kernel_ring_buffer_channel *chan)
{
	return 0;
}

static inline
size_t record_header_size(const struct lttng_kernel_ring_buffer_config *config,
				 struct lttng_kernel_ring_buffer_channel *chan, size_t offset,
				 size_t *pre_header_padding,
				 struct lttng_kernel_ring_buffer_ctx *ctx,
				 void *client_ctx)
{
	size_t orig_offset = offset;
	size_t padding;

	padding = lib_ring_buffer_align(offset, lttng_alignof(uint32_t));
	offset += padding;

	offset += sizeof(uint32_t);

	*pre_header_padding = padding;

	return offset - orig_offset;
}

#include <ringbuffer/api.h>

static u64 client_ring_buffer_clock_read(struct lttng_kernel_ring_buffer_channel *chan)
{
	return 0;
}

static
size_t client_record_header_size(const struct lttng_kernel_ring_buffer_config *config,
				 struct lttng_kernel_ring_buffer_channel *chan, size_t offset,
				 size_t *pre_header_padding,
				 struct lttng_kernel_ring_buffer_ctx *ctx,
				 void *client_ctx)
{
	return record_header_size(config, chan, offset,
				  pre_header_padding, ctx, client_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 event_notifier_packet_header, header_end);
}

static void client_buffer_begin(struct lttng_kernel_ring_buffer *buf, u64 tsc,
				unsigned int subbuf_idx)
{
}

/*
 * 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 lttng_kernel_ring_buffer *buf, u64 tsc,
			      unsigned int subbuf_idx, unsigned long data_size,
			      const struct lttng_kernel_ring_buffer_ctx *ctx)
{
}

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

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

static int client_timestamp_begin(const struct lttng_kernel_ring_buffer_config *config,
		struct lttng_kernel_ring_buffer *buf, uint64_t *timestamp_begin)
{
	return -ENOSYS;
}

static int client_timestamp_end(const struct lttng_kernel_ring_buffer_config *config,
			struct lttng_kernel_ring_buffer *bufb,
			uint64_t *timestamp_end)
{
	return -ENOSYS;
}

static int client_events_discarded(const struct lttng_kernel_ring_buffer_config *config,
			struct lttng_kernel_ring_buffer *bufb,
			uint64_t *events_discarded)
{
	return -ENOSYS;
}

static int client_current_timestamp(const struct lttng_kernel_ring_buffer_config *config,
		struct lttng_kernel_ring_buffer *bufb,
		uint64_t *ts)
{
	return -ENOSYS;
}

static int client_content_size(const struct lttng_kernel_ring_buffer_config *config,
			struct lttng_kernel_ring_buffer *bufb,
			uint64_t *content_size)
{
	return -ENOSYS;
}

static int client_packet_size(const struct lttng_kernel_ring_buffer_config *config,
			struct lttng_kernel_ring_buffer *bufb,
			uint64_t *packet_size)
{
	return -ENOSYS;
}

static int client_stream_id(const struct lttng_kernel_ring_buffer_config *config,
			struct lttng_kernel_ring_buffer *bufb,
			uint64_t *stream_id)
{
	return -ENOSYS;
}

static int client_sequence_number(const struct lttng_kernel_ring_buffer_config *config,
			struct lttng_kernel_ring_buffer *bufb,
			uint64_t *seq)
{
	return -ENOSYS;
}

static
int client_instance_id(const struct lttng_kernel_ring_buffer_config *config,
		struct lttng_kernel_ring_buffer *bufb,
		uint64_t *id)
{
	return -ENOSYS;
}

static void client_record_get(const struct lttng_kernel_ring_buffer_config *config,
	struct lttng_kernel_ring_buffer_channel *chan, struct lttng_kernel_ring_buffer *buf,
	size_t offset, size_t *header_len,
	size_t *payload_len, u64 *timestamp)
{
	struct event_notifier_record_header header;
	int ret;

	ret = lib_ring_buffer_read(&buf->backend, offset, &header,
			offsetof(struct event_notifier_record_header, header_end));
	CHAN_WARN_ON(chan, ret != offsetof(struct event_notifier_record_header, header_end));
	*header_len = offsetof(struct event_notifier_record_header, header_end);
	*payload_len = header.payload_len;
	*timestamp = 0;
}

static const struct lttng_kernel_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,
	.cb.record_get = client_record_get,

	.tsc_bits = 0,
	.alloc = RING_BUFFER_ALLOC_GLOBAL,
	.sync = RING_BUFFER_SYNC_GLOBAL,
	.mode = RING_BUFFER_MODE_TEMPLATE,
	.backend = RING_BUFFER_PAGE,
	.output = RING_BUFFER_OUTPUT_TEMPLATE,
	.oops = RING_BUFFER_OOPS_CONSISTENCY,
	.ipi = RING_BUFFER_NO_IPI_BARRIER,
	.wakeup = RING_BUFFER_WAKEUP_BY_WRITER,
};

static
void release_priv_ops(void *priv_ops)
{
	module_put(THIS_MODULE);
}

static
void lttng_channel_destroy(struct lttng_kernel_ring_buffer_channel *chan)
{
	channel_destroy(chan);
}

static
struct lttng_kernel_ring_buffer_channel *_channel_create(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)
{
	struct lttng_event_notifier_group *event_notifier_group = priv;
	struct lttng_kernel_ring_buffer_channel *chan;

	chan = channel_create(&client_config, name,
			      event_notifier_group, buf_addr,
			      subbuf_size, num_subbuf, switch_timer_interval,
			      read_timer_interval);
	if (chan) {
		/*
		 * Ensure this module is not unloaded before we finish
		 * using lttng_relay_transport.ops.
		 */
		if (!try_module_get(THIS_MODULE)) {
			printk(KERN_WARNING "LTTng: Can't lock event_notifier transport module.\n");
			goto error;
		}
		chan->backend.priv_ops = &lttng_relay_transport.ops;
		chan->backend.release_priv_ops = release_priv_ops;
	}
	return chan;

error:
	lttng_channel_destroy(chan);
	return NULL;
}

static
struct lttng_kernel_ring_buffer *lttng_buffer_read_open(struct lttng_kernel_ring_buffer_channel *chan)
{
	struct lttng_kernel_ring_buffer *buf;

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

static
int lttng_buffer_has_read_closed_stream(struct lttng_kernel_ring_buffer_channel *chan)
{
	struct lttng_kernel_ring_buffer *buf;
	int cpu;

	for_each_channel_cpu(cpu, chan) {
		buf = channel_get_ring_buffer(&client_config, chan, cpu);
		if (!atomic_long_read(&buf->active_readers))
			return 1;
	}
	return 0;
}

static
void lttng_buffer_read_close(struct lttng_kernel_ring_buffer *buf)
{
	lib_ring_buffer_release_read(buf);
}

static
void lttng_write_event_notifier_header(const struct lttng_kernel_ring_buffer_config *config,
			    struct lttng_kernel_ring_buffer_ctx *ctx)
{
	uint32_t data_size;

	WARN_ON_ONCE(ctx->data_size > U32_MAX);

	data_size = (uint32_t) ctx->data_size;

	lib_ring_buffer_write(config, ctx, &data_size, sizeof(data_size));

	lib_ring_buffer_align_ctx(ctx, ctx->largest_align);
}

static
int lttng_event_reserve(struct lttng_kernel_ring_buffer_ctx *ctx)
{
	struct lttng_kernel_ring_buffer_channel *chan = ctx->client_priv;
	int ret;

	memset(&ctx->priv, 0, sizeof(ctx->priv));
	ctx->priv.chan = chan;

	ret = lib_ring_buffer_reserve(&client_config, ctx, NULL);
	if (ret)
		return ret;
	lib_ring_buffer_backend_get_pages(&client_config, ctx,
			&ctx->priv.backend_pages);

	lttng_write_event_notifier_header(&client_config, ctx);
	return 0;
}

static
void lttng_event_commit(struct lttng_kernel_ring_buffer_ctx *ctx)
{
	lib_ring_buffer_commit(&client_config, ctx);
}

static
void lttng_event_write(struct lttng_kernel_ring_buffer_ctx *ctx, const void *src,
		     size_t len, size_t alignment)
{
	lib_ring_buffer_align_ctx(ctx, alignment);
	lib_ring_buffer_write(&client_config, ctx, src, len);
}

static
void lttng_event_write_from_user(struct lttng_kernel_ring_buffer_ctx *ctx,
			       const void __user *src, size_t len, size_t alignment)
{
	lib_ring_buffer_align_ctx(ctx, alignment);
	lib_ring_buffer_copy_from_user_inatomic(&client_config, ctx, src, len);
}

static
void lttng_event_memset(struct lttng_kernel_ring_buffer_ctx *ctx,
		int c, size_t len)
{
	lib_ring_buffer_memset(&client_config, ctx, c, len);
}

static
void lttng_event_strcpy(struct lttng_kernel_ring_buffer_ctx *ctx, const char *src,
		size_t len)
{
	lib_ring_buffer_strcpy(&client_config, ctx, src, len, '#');
}

static
size_t lttng_packet_avail_size(struct lttng_kernel_ring_buffer_channel *chan)
{
	unsigned long o_begin;
	struct lttng_kernel_ring_buffer *buf;

	buf = chan->backend.buf;	/* Only for global buffer ! */
	o_begin = v_read(&client_config, &buf->offset);
	if (subbuf_offset(o_begin, chan) != 0) {
		return chan->backend.subbuf_size - subbuf_offset(o_begin, chan);
	} else {
		return chan->backend.subbuf_size - subbuf_offset(o_begin, chan)
			- sizeof(struct event_notifier_packet_header);
	}
}

static
wait_queue_head_t *lttng_get_writer_buf_wait_queue(struct lttng_kernel_ring_buffer_channel *chan, int cpu)
{
	struct lttng_kernel_ring_buffer *buf = channel_get_ring_buffer(&client_config,
					chan, cpu);
	return &buf->write_wait;
}

static
wait_queue_head_t *lttng_get_hp_wait_queue(struct lttng_kernel_ring_buffer_channel *chan)
{
	return &chan->hp_wait;
}

static
int lttng_is_finalized(struct lttng_kernel_ring_buffer_channel *chan)
{
	return lib_ring_buffer_channel_is_finalized(chan);
}

static
int lttng_is_disabled(struct lttng_kernel_ring_buffer_channel *chan)
{
	return lib_ring_buffer_channel_is_disabled(chan);
}

static struct lttng_transport lttng_relay_transport = {
	.name = "relay-" RING_BUFFER_MODE_TEMPLATE_STRING,
	.owner = THIS_MODULE,
	.ops = {
		.priv = __LTTNG_COMPOUND_LITERAL(struct lttng_kernel_channel_buffer_ops_private, {
			.pub = &lttng_relay_transport.ops,
			.channel_create = _channel_create,
			.channel_destroy = lttng_channel_destroy,
			.buffer_read_open = lttng_buffer_read_open,
			.buffer_has_read_closed_stream =
				lttng_buffer_has_read_closed_stream,
			.buffer_read_close = lttng_buffer_read_close,
			.packet_avail_size = lttng_packet_avail_size,
			.get_writer_buf_wait_queue = lttng_get_writer_buf_wait_queue,
			.get_hp_wait_queue = lttng_get_hp_wait_queue,
			.is_finalized = lttng_is_finalized,
			.is_disabled = lttng_is_disabled,
			.timestamp_begin = client_timestamp_begin,
			.timestamp_end = client_timestamp_end,
			.events_discarded = client_events_discarded,
			.content_size = client_content_size,
			.packet_size = client_packet_size,
			.stream_id = client_stream_id,
			.current_timestamp = client_current_timestamp,
			.sequence_number = client_sequence_number,
			.instance_id = client_instance_id,
		}),
		.event_reserve = lttng_event_reserve,
		.event_commit = lttng_event_commit,
		.event_write_from_user = lttng_event_write_from_user,
		.event_memset = lttng_event_memset,
		.event_write = lttng_event_write,
		.event_strcpy = lttng_event_strcpy,
	},
};

static int __init lttng_ring_buffer_event_notifier_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_mappings();
	lttng_transport_register(&lttng_relay_transport);
	return 0;
}

module_init(lttng_ring_buffer_event_notifier_client_init);

static void __exit lttng_ring_buffer_event_notifier_client_exit(void)
{
	lttng_transport_unregister(&lttng_relay_transport);
}

module_exit(lttng_ring_buffer_event_notifier_client_exit);

MODULE_LICENSE("GPL and additional rights");
MODULE_AUTHOR("Mathieu Desnoyers <mathieu.desnoyers@efficios.com>");
MODULE_DESCRIPTION("LTTng ring buffer " RING_BUFFER_MODE_TEMPLATE_STRING
		   " client");
MODULE_VERSION(__stringify(LTTNG_MODULES_MAJOR_VERSION) "."
	__stringify(LTTNG_MODULES_MINOR_VERSION) "."
	__stringify(LTTNG_MODULES_PATCHLEVEL_VERSION)
	LTTNG_MODULES_EXTRAVERSION);
Commit	Line	Data
	1	/* SPDX-License-Identifier: (GPL-2.0 or LGPL-2.1)
	2	*
	3	* lttng-ring-buffer-event-notifier-client.h
	4	*
	5	* LTTng lib ring buffer event notifier client template.
	6	*
	7	* Copyright (C) 2010-2020 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
	8	*/
	9
	10	#include <linux/limits.h>
	11	#include <linux/module.h>
	12	#include <linux/types.h>
	13	#include <wrapper/vmalloc.h> /* for wrapper_vmalloc_sync_mappings() */
	14	#include <lttng/abi.h>
	15	#include <lttng/events.h>
	16	#include <lttng/events-internal.h>
	17	#include <lttng/tracer.h>
	18	#include <wrapper/limits.h>
	19
	20	static struct lttng_transport lttng_relay_transport;
	21
	22	struct event_notifier_packet_header {
	23	uint8_t header_end[0];
	24	};
	25
	26	struct event_notifier_record_header {
	27	uint32_t payload_len; /* in bytes */
	28	uint8_t header_end[0]; /* End of header */
	29	};
	30
	31	static const struct lttng_kernel_ring_buffer_config client_config;
	32
	33	static inline
	34	u64 lib_ring_buffer_clock_read(struct lttng_kernel_ring_buffer_channel *chan)
	35	{
	36	return 0;
	37	}
	38
	39	static inline
	40	size_t record_header_size(const struct lttng_kernel_ring_buffer_config *config,
	41	struct lttng_kernel_ring_buffer_channel *chan, size_t offset,
	42	size_t *pre_header_padding,
	43	struct lttng_kernel_ring_buffer_ctx *ctx,
	44	void *client_ctx)
	45	{
	46	size_t orig_offset = offset;
	47	size_t padding;
	48
	49	padding = lib_ring_buffer_align(offset, lttng_alignof(uint32_t));
	50	offset += padding;
	51
	52	offset += sizeof(uint32_t);
	53
	54	*pre_header_padding = padding;
	55
	56	return offset - orig_offset;
	57	}
	58
	59	#include <ringbuffer/api.h>
	60
	61	static u64 client_ring_buffer_clock_read(struct lttng_kernel_ring_buffer_channel *chan)
	62	{
	63	return 0;
	64	}
	65
	66	static
	67	size_t client_record_header_size(const struct lttng_kernel_ring_buffer_config *config,
	68	struct lttng_kernel_ring_buffer_channel *chan, size_t offset,
	69	size_t *pre_header_padding,
	70	struct lttng_kernel_ring_buffer_ctx *ctx,
	71	void *client_ctx)
	72	{
	73	return record_header_size(config, chan, offset,
	74	pre_header_padding, ctx, client_ctx);
	75	}
	76
	77	/**
	78	* client_packet_header_size - called on buffer-switch to a new sub-buffer
	79	*
	80	* Return header size without padding after the structure. Don't use packed
	81	* structure because gcc generates inefficient code on some architectures
	82	* (powerpc, mips..)
	83	*/
	84	static size_t client_packet_header_size(void)
	85	{
	86	return offsetof(struct event_notifier_packet_header, header_end);
	87	}
	88
	89	static void client_buffer_begin(struct lttng_kernel_ring_buffer *buf, u64 tsc,
	90	unsigned int subbuf_idx)
	91	{
	92	}
	93
	94	/*
	95	* offset is assumed to never be 0 here : never deliver a completely empty
	96	* subbuffer. data_size is between 1 and subbuf_size.
	97	*/
	98	static void client_buffer_end(struct lttng_kernel_ring_buffer *buf, u64 tsc,
	99	unsigned int subbuf_idx, unsigned long data_size,
	100	const struct lttng_kernel_ring_buffer_ctx *ctx)
	101	{
	102	}
	103
	104	static int client_buffer_create(struct lttng_kernel_ring_buffer buf, void priv,
	105	int cpu, const char *name)
	106	{
	107	return 0;
	108	}
	109
	110	static void client_buffer_finalize(struct lttng_kernel_ring_buffer buf, void priv, int cpu)
	111	{
	112	}
	113
	114	static int client_timestamp_begin(const struct lttng_kernel_ring_buffer_config *config,
	115	struct lttng_kernel_ring_buffer buf, uint64_t timestamp_begin)
	116	{
	117	return -ENOSYS;
	118	}
	119
	120	static int client_timestamp_end(const struct lttng_kernel_ring_buffer_config *config,
	121	struct lttng_kernel_ring_buffer *bufb,
	122	uint64_t *timestamp_end)
	123	{
	124	return -ENOSYS;
	125	}
	126
	127	static int client_events_discarded(const struct lttng_kernel_ring_buffer_config *config,
	128	struct lttng_kernel_ring_buffer *bufb,
	129	uint64_t *events_discarded)
	130	{
	131	return -ENOSYS;
	132	}
	133
	134	static int client_current_timestamp(const struct lttng_kernel_ring_buffer_config *config,
	135	struct lttng_kernel_ring_buffer *bufb,
	136	uint64_t *ts)
	137	{
	138	return -ENOSYS;
	139	}
	140
	141	static int client_content_size(const struct lttng_kernel_ring_buffer_config *config,
	142	struct lttng_kernel_ring_buffer *bufb,
	143	uint64_t *content_size)
	144	{
	145	return -ENOSYS;
	146	}
	147
	148	static int client_packet_size(const struct lttng_kernel_ring_buffer_config *config,
	149	struct lttng_kernel_ring_buffer *bufb,
	150	uint64_t *packet_size)
	151	{
	152	return -ENOSYS;
	153	}
	154
	155	static int client_stream_id(const struct lttng_kernel_ring_buffer_config *config,
	156	struct lttng_kernel_ring_buffer *bufb,
	157	uint64_t *stream_id)
	158	{
	159	return -ENOSYS;
	160	}
	161
	162	static int client_sequence_number(const struct lttng_kernel_ring_buffer_config *config,
	163	struct lttng_kernel_ring_buffer *bufb,
	164	uint64_t *seq)
	165	{
	166	return -ENOSYS;
	167	}
	168
	169	static
	170	int client_instance_id(const struct lttng_kernel_ring_buffer_config *config,
	171	struct lttng_kernel_ring_buffer *bufb,
	172	uint64_t *id)
	173	{
	174	return -ENOSYS;
	175	}
	176
	177	static void client_record_get(const struct lttng_kernel_ring_buffer_config *config,
	178	struct lttng_kernel_ring_buffer_channel chan, struct lttng_kernel_ring_buffer buf,
	179	size_t offset, size_t *header_len,
	180	size_t payload_len, u64 timestamp)
	181	{
	182	struct event_notifier_record_header header;
	183	int ret;
	184
	185	ret = lib_ring_buffer_read(&buf->backend, offset, &header,
	186	offsetof(struct event_notifier_record_header, header_end));
	187	CHAN_WARN_ON(chan, ret != offsetof(struct event_notifier_record_header, header_end));
	188	*header_len = offsetof(struct event_notifier_record_header, header_end);
	189	*payload_len = header.payload_len;
	190	*timestamp = 0;
	191	}
	192
	193	static const struct lttng_kernel_ring_buffer_config client_config = {
	194	.cb.ring_buffer_clock_read = client_ring_buffer_clock_read,
	195	.cb.record_header_size = client_record_header_size,
	196	.cb.subbuffer_header_size = client_packet_header_size,
	197	.cb.buffer_begin = client_buffer_begin,
	198	.cb.buffer_end = client_buffer_end,
	199	.cb.buffer_create = client_buffer_create,
	200	.cb.buffer_finalize = client_buffer_finalize,
	201	.cb.record_get = client_record_get,
	202
	203	.tsc_bits = 0,
	204	.alloc = RING_BUFFER_ALLOC_GLOBAL,
	205	.sync = RING_BUFFER_SYNC_GLOBAL,
	206	.mode = RING_BUFFER_MODE_TEMPLATE,
	207	.backend = RING_BUFFER_PAGE,
	208	.output = RING_BUFFER_OUTPUT_TEMPLATE,
	209	.oops = RING_BUFFER_OOPS_CONSISTENCY,
	210	.ipi = RING_BUFFER_NO_IPI_BARRIER,
	211	.wakeup = RING_BUFFER_WAKEUP_BY_WRITER,
	212	};
	213
	214	static
	215	void release_priv_ops(void *priv_ops)
	216	{
	217	module_put(THIS_MODULE);
	218	}
	219
	220	static
	221	void lttng_channel_destroy(struct lttng_kernel_ring_buffer_channel *chan)
	222	{
	223	channel_destroy(chan);
	224	}
	225
	226	static
	227	struct lttng_kernel_ring_buffer_channel _channel_create(const char name,
	228	void priv, void buf_addr,
	229	size_t subbuf_size, size_t num_subbuf,
	230	unsigned int switch_timer_interval,
	231	unsigned int read_timer_interval)
	232	{
	233	struct lttng_event_notifier_group *event_notifier_group = priv;
	234	struct lttng_kernel_ring_buffer_channel *chan;
	235
	236	chan = channel_create(&client_config, name,
	237	event_notifier_group, buf_addr,
	238	subbuf_size, num_subbuf, switch_timer_interval,
	239	read_timer_interval);
	240	if (chan) {
	241	/*
	242	* Ensure this module is not unloaded before we finish
	243	* using lttng_relay_transport.ops.
	244	*/
	245	if (!try_module_get(THIS_MODULE)) {
	246	printk(KERN_WARNING "LTTng: Can't lock event_notifier transport module.\n");
	247	goto error;
	248	}
	249	chan->backend.priv_ops = &lttng_relay_transport.ops;
	250	chan->backend.release_priv_ops = release_priv_ops;
	251	}
	252	return chan;
	253
	254	error:
	255	lttng_channel_destroy(chan);
	256	return NULL;
	257	}
	258
	259	static
	260	struct lttng_kernel_ring_buffer lttng_buffer_read_open(struct lttng_kernel_ring_buffer_channel chan)
	261	{
	262	struct lttng_kernel_ring_buffer *buf;
	263
	264	buf = channel_get_ring_buffer(&client_config, chan, 0);
	265	if (!lib_ring_buffer_open_read(buf))
	266	return buf;
	267	return NULL;
	268	}
	269
	270	static
	271	int lttng_buffer_has_read_closed_stream(struct lttng_kernel_ring_buffer_channel *chan)
	272	{
	273	struct lttng_kernel_ring_buffer *buf;
	274	int cpu;
	275
	276	for_each_channel_cpu(cpu, chan) {
	277	buf = channel_get_ring_buffer(&client_config, chan, cpu);
	278	if (!atomic_long_read(&buf->active_readers))
	279	return 1;
	280	}
	281	return 0;
	282	}
	283
	284	static
	285	void lttng_buffer_read_close(struct lttng_kernel_ring_buffer *buf)
	286	{
	287	lib_ring_buffer_release_read(buf);
	288	}
	289
	290	static
	291	void lttng_write_event_notifier_header(const struct lttng_kernel_ring_buffer_config *config,
	292	struct lttng_kernel_ring_buffer_ctx *ctx)
	293	{
	294	uint32_t data_size;
	295
	296	WARN_ON_ONCE(ctx->data_size > U32_MAX);
	297
	298	data_size = (uint32_t) ctx->data_size;
	299
	300	lib_ring_buffer_write(config, ctx, &data_size, sizeof(data_size));
	301
	302	lib_ring_buffer_align_ctx(ctx, ctx->largest_align);
	303	}
	304
	305	static
	306	int lttng_event_reserve(struct lttng_kernel_ring_buffer_ctx *ctx)
	307	{
	308	struct lttng_kernel_ring_buffer_channel *chan = ctx->client_priv;
	309	int ret;
	310
	311	memset(&ctx->priv, 0, sizeof(ctx->priv));
	312	ctx->priv.chan = chan;
	313
	314	ret = lib_ring_buffer_reserve(&client_config, ctx, NULL);
	315	if (ret)
	316	return ret;
	317	lib_ring_buffer_backend_get_pages(&client_config, ctx,
	318	&ctx->priv.backend_pages);
	319
	320	lttng_write_event_notifier_header(&client_config, ctx);
	321	return 0;
	322	}
	323
	324	static
	325	void lttng_event_commit(struct lttng_kernel_ring_buffer_ctx *ctx)
	326	{
	327	lib_ring_buffer_commit(&client_config, ctx);
	328	}
	329
	330	static
	331	void lttng_event_write(struct lttng_kernel_ring_buffer_ctx ctx, const void src,
	332	size_t len, size_t alignment)
	333	{
	334	lib_ring_buffer_align_ctx(ctx, alignment);
	335	lib_ring_buffer_write(&client_config, ctx, src, len);
	336	}
	337
	338	static
	339	void lttng_event_write_from_user(struct lttng_kernel_ring_buffer_ctx *ctx,
	340	const void __user *src, size_t len, size_t alignment)
	341	{
	342	lib_ring_buffer_align_ctx(ctx, alignment);
	343	lib_ring_buffer_copy_from_user_inatomic(&client_config, ctx, src, len);
	344	}
	345
	346	static
	347	void lttng_event_memset(struct lttng_kernel_ring_buffer_ctx *ctx,
	348	int c, size_t len)
	349	{
	350	lib_ring_buffer_memset(&client_config, ctx, c, len);
	351	}
	352
	353	static
	354	void lttng_event_strcpy(struct lttng_kernel_ring_buffer_ctx ctx, const char src,
	355	size_t len)
	356	{
	357	lib_ring_buffer_strcpy(&client_config, ctx, src, len, '#');
	358	}
	359
	360	static
	361	size_t lttng_packet_avail_size(struct lttng_kernel_ring_buffer_channel *chan)
	362	{
	363	unsigned long o_begin;
	364	struct lttng_kernel_ring_buffer *buf;
	365
	366	buf = chan->backend.buf; /* Only for global buffer ! */
	367	o_begin = v_read(&client_config, &buf->offset);
	368	if (subbuf_offset(o_begin, chan) != 0) {
	369	return chan->backend.subbuf_size - subbuf_offset(o_begin, chan);
	370	} else {
	371	return chan->backend.subbuf_size - subbuf_offset(o_begin, chan)
	372	- sizeof(struct event_notifier_packet_header);
	373	}
	374	}
	375
	376	static
	377	wait_queue_head_t lttng_get_writer_buf_wait_queue(struct lttng_kernel_ring_buffer_channel chan, int cpu)
	378	{
	379	struct lttng_kernel_ring_buffer *buf = channel_get_ring_buffer(&client_config,
	380	chan, cpu);
	381	return &buf->write_wait;
	382	}
	383
	384	static
	385	wait_queue_head_t lttng_get_hp_wait_queue(struct lttng_kernel_ring_buffer_channel chan)
	386	{
	387	return &chan->hp_wait;
	388	}
	389
	390	static
	391	int lttng_is_finalized(struct lttng_kernel_ring_buffer_channel *chan)
	392	{
	393	return lib_ring_buffer_channel_is_finalized(chan);
	394	}
	395
	396	static
	397	int lttng_is_disabled(struct lttng_kernel_ring_buffer_channel *chan)
	398	{
	399	return lib_ring_buffer_channel_is_disabled(chan);
	400	}
	401
	402	static struct lttng_transport lttng_relay_transport = {
	403	.name = "relay-" RING_BUFFER_MODE_TEMPLATE_STRING,
	404	.owner = THIS_MODULE,
	405	.ops = {
	406	.priv = __LTTNG_COMPOUND_LITERAL(struct lttng_kernel_channel_buffer_ops_private, {
	407	.pub = &lttng_relay_transport.ops,
	408	.channel_create = _channel_create,
	409	.channel_destroy = lttng_channel_destroy,
	410	.buffer_read_open = lttng_buffer_read_open,
	411	.buffer_has_read_closed_stream =
	412	lttng_buffer_has_read_closed_stream,
	413	.buffer_read_close = lttng_buffer_read_close,
	414	.packet_avail_size = lttng_packet_avail_size,
	415	.get_writer_buf_wait_queue = lttng_get_writer_buf_wait_queue,
	416	.get_hp_wait_queue = lttng_get_hp_wait_queue,
	417	.is_finalized = lttng_is_finalized,
	418	.is_disabled = lttng_is_disabled,
	419	.timestamp_begin = client_timestamp_begin,
	420	.timestamp_end = client_timestamp_end,
	421	.events_discarded = client_events_discarded,
	422	.content_size = client_content_size,
	423	.packet_size = client_packet_size,
	424	.stream_id = client_stream_id,
	425	.current_timestamp = client_current_timestamp,
	426	.sequence_number = client_sequence_number,
	427	.instance_id = client_instance_id,
	428	}),
	429	.event_reserve = lttng_event_reserve,
	430	.event_commit = lttng_event_commit,
	431	.event_write_from_user = lttng_event_write_from_user,
	432	.event_memset = lttng_event_memset,
	433	.event_write = lttng_event_write,
	434	.event_strcpy = lttng_event_strcpy,
	435	},
	436	};
	437
	438	static int __init lttng_ring_buffer_event_notifier_client_init(void)
	439	{
	440	/*
	441	* This vmalloc sync all also takes care of the lib ring buffer
	442	* vmalloc'd module pages when it is built as a module into LTTng.
	443	*/
	444	wrapper_vmalloc_sync_mappings();
	445	lttng_transport_register(&lttng_relay_transport);
	446	return 0;
	447	}
	448
	449	module_init(lttng_ring_buffer_event_notifier_client_init);
	450
	451	static void __exit lttng_ring_buffer_event_notifier_client_exit(void)
	452	{
	453	lttng_transport_unregister(&lttng_relay_transport);
	454	}
	455
	456	module_exit(lttng_ring_buffer_event_notifier_client_exit);
	457
	458	MODULE_LICENSE("GPL and additional rights");
	459	MODULE_AUTHOR("Mathieu Desnoyers <mathieu.desnoyers@efficios.com>");
	460	MODULE_DESCRIPTION("LTTng ring buffer " RING_BUFFER_MODE_TEMPLATE_STRING
	461	" client");
	462	MODULE_VERSION(__stringify(LTTNG_MODULES_MAJOR_VERSION) "."
	463	__stringify(LTTNG_MODULES_MINOR_VERSION) "."
	464	__stringify(LTTNG_MODULES_PATCHLEVEL_VERSION)
	465	LTTNG_MODULES_EXTRAVERSION);