[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 lib_ring_buffer_config client_config;

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

static inline
size_t record_header_size(const struct lib_ring_buffer_config *config,
				 struct 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 channel *chan)
{
	return 0;
}

static
size_t client_record_header_size(const struct lib_ring_buffer_config *config,
				 struct 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 lib_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 lib_ring_buffer *buf, u64 tsc,
			      unsigned int subbuf_idx, unsigned long data_size)
{
}

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 int client_timestamp_begin(const struct lib_ring_buffer_config *config,
		struct lib_ring_buffer *buf, uint64_t *timestamp_begin)
{
	return -ENOSYS;
}

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

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

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

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

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

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

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

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

static void client_record_get(const struct lib_ring_buffer_config *config,
	struct channel *chan, struct lib_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 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,
	.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 channel *chan)
{
	channel_destroy(chan);
}

static
struct 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 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 lib_ring_buffer *lttng_buffer_read_open(struct channel *chan)
{
	struct lib_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 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 (!atomic_long_read(&buf->active_readers))
			return 1;
	}
	return 0;
}

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

static
void lttng_write_event_notifier_header(const struct lib_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 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)
{
	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)
{
	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 channel *chan)
{
	unsigned long o_begin;
	struct lib_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 channel *chan, int cpu)
{
	struct lib_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 channel *chan)
{
	return &chan->hp_wait;
}

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

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