[lttng-ust.git] / libringbuffer / backend.h

/*
 * SPDX-License-Identifier: LGPL-2.1-only
 *
 * Copyright (C) 2011-2012 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
 *
 * Ring buffer backend (API).
 *
 * Credits to Steven Rostedt for proposing to use an extra-subbuffer owned by
 * the reader in flight recorder mode.
 */

#ifndef _LTTNG_RING_BUFFER_BACKEND_H
#define _LTTNG_RING_BUFFER_BACKEND_H

#include <stddef.h>
#include <unistd.h>

#include "ust-helper.h"

/* Internal helpers */
#include "backend_internal.h"
#include "frontend_internal.h"

/* Ring buffer backend API */

/* Ring buffer backend access (read/write) */

LTTNG_HIDDEN
extern size_t lib_ring_buffer_read(struct lttng_ust_lib_ring_buffer_backend *bufb,
				   size_t offset, void *dest, size_t len,
				   struct lttng_ust_shm_handle *handle);

LTTNG_HIDDEN
extern int lib_ring_buffer_read_cstr(struct lttng_ust_lib_ring_buffer_backend *bufb,
				     size_t offset, void *dest, size_t len,
				     struct lttng_ust_shm_handle *handle);

/*
 * Return the address where a given offset is located.
 * Should be used to get the current subbuffer header pointer. Given we know
 * it's never on a page boundary, it's safe to write directly to this address,
 * as long as the write is never bigger than a page size.
 */
LTTNG_HIDDEN
extern void *
lib_ring_buffer_offset_address(struct lttng_ust_lib_ring_buffer_backend *bufb,
			       size_t offset,
			       struct lttng_ust_shm_handle *handle);
LTTNG_HIDDEN
extern void *
lib_ring_buffer_read_offset_address(struct lttng_ust_lib_ring_buffer_backend *bufb,
				    size_t offset,
				    struct lttng_ust_shm_handle *handle);

/**
 * lib_ring_buffer_write - write data to a buffer backend
 * @config : ring buffer instance configuration
 * @ctx: ring buffer context. (input arguments only)
 * @src : source pointer to copy from
 * @len : length of data to copy
 *
 * This function copies "len" bytes of data from a source pointer to a buffer
 * backend, at the current context offset. This is more or less a buffer
 * backend-specific memcpy() operation. Calls the slow path (_ring_buffer_write)
 * if copy is crossing a page boundary.
 */
static inline __attribute__((always_inline))
void lib_ring_buffer_write(const struct lttng_ust_lib_ring_buffer_config *config,
			   struct lttng_ust_lib_ring_buffer_ctx *ctx,
			   const void *src, size_t len)
{
	struct channel_backend *chanb = &ctx->chan->backend;
	struct lttng_ust_shm_handle *handle = ctx->handle;
	size_t offset = ctx->buf_offset;
	struct lttng_ust_lib_ring_buffer_backend_pages *backend_pages;
	void *p;

	if (caa_unlikely(!len))
		return;
	/*
	 * Underlying layer should never ask for writes across
	 * subbuffers.
	 */
	CHAN_WARN_ON(chanb, (offset & (chanb->buf_size - 1)) + len > chanb->buf_size);
	backend_pages = lib_ring_buffer_get_backend_pages_from_ctx(config, ctx);
	if (caa_unlikely(!backend_pages)) {
		if (lib_ring_buffer_backend_get_pages(config, ctx, &backend_pages))
			return;
	}
	p = shmp_index(handle, backend_pages->p, offset & (chanb->subbuf_size - 1));
	if (caa_unlikely(!p))
		return;
	lib_ring_buffer_do_copy(config, p, src, len);
	ctx->buf_offset += len;
}

/*
 * Copy up to @len string bytes from @src to @dest. Stop whenever a NULL
 * terminating character is found in @src. Returns the number of bytes
 * copied. Does *not* terminate @dest with NULL terminating character.
 */
static inline __attribute__((always_inline))
size_t lib_ring_buffer_do_strcpy(const struct lttng_ust_lib_ring_buffer_config *config,
		char *dest, const char *src, size_t len)
{
	size_t count;

	for (count = 0; count < len; count++) {
		char c;

		/*
		 * Only read source character once, in case it is
		 * modified concurrently.
		 */
		c = CMM_LOAD_SHARED(src[count]);
		if (!c)
			break;
		lib_ring_buffer_do_copy(config, &dest[count], &c, 1);
	}
	return count;
}

/**
 * lib_ring_buffer_strcpy - write string data to a buffer backend
 * @config : ring buffer instance configuration
 * @ctx: ring buffer context. (input arguments only)
 * @src : source pointer to copy from
 * @len : length of data to copy
 * @pad : character to use for padding
 *
 * This function copies @len - 1 bytes of string data from a source
 * pointer to a buffer backend, followed by a terminating '\0'
 * character, at the current context offset. This is more or less a
 * buffer backend-specific strncpy() operation. If a terminating '\0'
 * character is found in @src before @len - 1 characters are copied, pad
 * the buffer with @pad characters (e.g. '#').
 */
static inline __attribute__((always_inline))
void lib_ring_buffer_strcpy(const struct lttng_ust_lib_ring_buffer_config *config,
			   struct lttng_ust_lib_ring_buffer_ctx *ctx,
			   const char *src, size_t len, int pad)
{
	struct channel_backend *chanb = &ctx->chan->backend;
	struct lttng_ust_shm_handle *handle = ctx->handle;
	size_t count;
	size_t offset = ctx->buf_offset;
	struct lttng_ust_lib_ring_buffer_backend_pages *backend_pages;
	void *p;

	if (caa_unlikely(!len))
		return;
	/*
	 * Underlying layer should never ask for writes across
	 * subbuffers.
	 */
	CHAN_WARN_ON(chanb, (offset & (chanb->buf_size - 1)) + len > chanb->buf_size);
	backend_pages = lib_ring_buffer_get_backend_pages_from_ctx(config, ctx);
	if (caa_unlikely(!backend_pages)) {
		if (lib_ring_buffer_backend_get_pages(config, ctx, &backend_pages))
			return;
	}
	p = shmp_index(handle, backend_pages->p, offset & (chanb->subbuf_size - 1));
	if (caa_unlikely(!p))
		return;

	count = lib_ring_buffer_do_strcpy(config, p, src, len - 1);
	offset += count;
	/* Padding */
	if (caa_unlikely(count < len - 1)) {
		size_t pad_len = len - 1 - count;

		p = shmp_index(handle, backend_pages->p, offset & (chanb->subbuf_size - 1));
		if (caa_unlikely(!p))
			return;
		lib_ring_buffer_do_memset(p, pad, pad_len);
		offset += pad_len;
	}
	/* Final '\0' */
	p = shmp_index(handle, backend_pages->p, offset & (chanb->subbuf_size - 1));
	if (caa_unlikely(!p))
		return;
	lib_ring_buffer_do_memset(p, '\0', 1);
	ctx->buf_offset += len;
}

/*
 * This accessor counts the number of unread records in a buffer.
 * It only provides a consistent value if no reads not writes are performed
 * concurrently.
 */
static inline
unsigned long lib_ring_buffer_get_records_unread(
				const struct lttng_ust_lib_ring_buffer_config *config,
				struct lttng_ust_lib_ring_buffer *buf,
				struct lttng_ust_shm_handle *handle)
{
	struct lttng_ust_lib_ring_buffer_backend *bufb = &buf->backend;
	unsigned long records_unread = 0, sb_bindex;
	unsigned int i;
	struct channel *chan;

	chan = shmp(handle, bufb->chan);
	if (!chan)
		return 0;
	for (i = 0; i < chan->backend.num_subbuf; i++) {
		struct lttng_ust_lib_ring_buffer_backend_subbuffer *wsb;
		struct lttng_ust_lib_ring_buffer_backend_pages_shmp *rpages;
		struct lttng_ust_lib_ring_buffer_backend_pages *backend_pages;

		wsb = shmp_index(handle, bufb->buf_wsb, i);
		if (!wsb)
			return 0;
		sb_bindex = subbuffer_id_get_index(config, wsb->id);
		rpages = shmp_index(handle, bufb->array, sb_bindex);
		if (!rpages)
			return 0;
		backend_pages = shmp(handle, rpages->shmp);
		if (!backend_pages)
			return 0;
		records_unread += v_read(config, &backend_pages->records_unread);
	}
	if (config->mode == RING_BUFFER_OVERWRITE) {
		struct lttng_ust_lib_ring_buffer_backend_pages_shmp *rpages;
		struct lttng_ust_lib_ring_buffer_backend_pages *backend_pages;

		sb_bindex = subbuffer_id_get_index(config, bufb->buf_rsb.id);
		rpages = shmp_index(handle, bufb->array, sb_bindex);
		if (!rpages)
			return 0;
		backend_pages = shmp(handle, rpages->shmp);
		if (!backend_pages)
			return 0;
		records_unread += v_read(config, &backend_pages->records_unread);
	}
	return records_unread;
}

#endif /* _LTTNG_RING_BUFFER_BACKEND_H */
Commit	Line	Data
852c2936	1	/*
c0c0989a	2	* SPDX-License-Identifier: LGPL-2.1-only
852c2936	3	*
e92f3e28 MD	4	* Copyright (C) 2011-2012 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
e92f3e28 MD	5	*
c0c0989a	6	* Ring buffer backend (API).
852c2936 MD	7	*
	8	* Credits to Steven Rostedt for proposing to use an extra-subbuffer owned by
	9	* the reader in flight recorder mode.
	10	*/
	11
c0c0989a MJ	12	#ifndef _LTTNG_RING_BUFFER_BACKEND_H
	13	#define _LTTNG_RING_BUFFER_BACKEND_H
	14
b4051ad8	15	#include <stddef.h>
14641deb MD	16	#include <unistd.h>
14641deb MD	17
071dec43 MJ	18	#include "ust-helper.h"
071dec43 MJ	19
852c2936	20	/* Internal helpers */
4931a13e MD	21	#include "backend_internal.h"
4931a13e MD	22	#include "frontend_internal.h"
852c2936 MD	23
	24	/* Ring buffer backend API */
	25
	26	/* Ring buffer backend access (read/write) */
	27
071dec43	28	LTTNG_HIDDEN
4cfec15c	29	extern size_t lib_ring_buffer_read(struct lttng_ust_lib_ring_buffer_backend *bufb,
1d498196	30	size_t offset, void *dest, size_t len,
38fae1d3	31	struct lttng_ust_shm_handle *handle);
852c2936	32
071dec43	33	LTTNG_HIDDEN
4cfec15c	34	extern int lib_ring_buffer_read_cstr(struct lttng_ust_lib_ring_buffer_backend *bufb,
1d498196	35	size_t offset, void *dest, size_t len,
38fae1d3	36	struct lttng_ust_shm_handle *handle);
852c2936	37
852c2936 MD	38	/*
	39	* Return the address where a given offset is located.
	40	* Should be used to get the current subbuffer header pointer. Given we know
	41	* it's never on a page boundary, it's safe to write directly to this address,
	42	* as long as the write is never bigger than a page size.
	43	*/
071dec43	44	LTTNG_HIDDEN
852c2936	45	extern void *
4cfec15c	46	lib_ring_buffer_offset_address(struct lttng_ust_lib_ring_buffer_backend *bufb,
1d498196	47	size_t offset,
38fae1d3	48	struct lttng_ust_shm_handle *handle);
071dec43	49	LTTNG_HIDDEN
852c2936	50	extern void *
4cfec15c	51	lib_ring_buffer_read_offset_address(struct lttng_ust_lib_ring_buffer_backend *bufb,
1d498196	52	size_t offset,
38fae1d3	53	struct lttng_ust_shm_handle *handle);
852c2936 MD	54
	55	/**
	56	* lib_ring_buffer_write - write data to a buffer backend
	57	* @config : ring buffer instance configuration
	58	* @ctx: ring buffer context. (input arguments only)
	59	* @src : source pointer to copy from
	60	* @len : length of data to copy
	61	*
	62	* This function copies "len" bytes of data from a source pointer to a buffer
	63	* backend, at the current context offset. This is more or less a buffer
	64	* backend-specific memcpy() operation. Calls the slow path (_ring_buffer_write)
	65	* if copy is crossing a page boundary.
	66	*/
00d0f8eb	67	static inline __attribute__((always_inline))
4cfec15c MD	68	void lib_ring_buffer_write(const struct lttng_ust_lib_ring_buffer_config *config,
4cfec15c MD	69	struct lttng_ust_lib_ring_buffer_ctx *ctx,
852c2936 MD	70	const void *src, size_t len)
852c2936 MD	71	{
852c2936	72	struct channel_backend *chanb = &ctx->chan->backend;
38fae1d3	73	struct lttng_ust_shm_handle *handle = ctx->handle;
852c2936	74	size_t offset = ctx->buf_offset;
15500a1b MD	75	struct lttng_ust_lib_ring_buffer_backend_pages *backend_pages;
15500a1b MD	76	void *p;
852c2936	77
0bf3c920 MD	78	if (caa_unlikely(!len))
0bf3c920 MD	79	return;
a6352fd4 MD	80	/*
	81	* Underlying layer should never ask for writes across
	82	* subbuffers.
	83	*/
a3492932 MD	84	CHAN_WARN_ON(chanb, (offset & (chanb->buf_size - 1)) + len > chanb->buf_size);
	85	backend_pages = lib_ring_buffer_get_backend_pages_from_ctx(config, ctx);
	86	if (caa_unlikely(!backend_pages)) {
	87	if (lib_ring_buffer_backend_get_pages(config, ctx, &backend_pages))
	88	return;
	89	}
15500a1b MD	90	p = shmp_index(handle, backend_pages->p, offset & (chanb->subbuf_size - 1));
	91	if (caa_unlikely(!p))
	92	return;
	93	lib_ring_buffer_do_copy(config, p, src, len);
852c2936 MD	94	ctx->buf_offset += len;
	95	}
	96
a44c74d9 MD	97	/*
	98	* Copy up to @len string bytes from @src to @dest. Stop whenever a NULL
	99	* terminating character is found in @src. Returns the number of bytes
	100	* copied. Does not terminate @dest with NULL terminating character.
	101	*/
00d0f8eb	102	static inline __attribute__((always_inline))
a44c74d9 MD	103	size_t lib_ring_buffer_do_strcpy(const struct lttng_ust_lib_ring_buffer_config *config,
	104	char dest, const char src, size_t len)
	105	{
	106	size_t count;
	107
	108	for (count = 0; count < len; count++) {
	109	char c;
	110
	111	/*
	112	* Only read source character once, in case it is
	113	* modified concurrently.
	114	*/
	115	c = CMM_LOAD_SHARED(src[count]);
	116	if (!c)
	117	break;
	118	lib_ring_buffer_do_copy(config, &dest[count], &c, 1);
	119	}
	120	return count;
	121	}
	122
	123	/**
	124	* lib_ring_buffer_strcpy - write string data to a buffer backend
	125	* @config : ring buffer instance configuration
	126	* @ctx: ring buffer context. (input arguments only)
	127	* @src : source pointer to copy from
	128	* @len : length of data to copy
	129	* @pad : character to use for padding
	130	*
	131	* This function copies @len - 1 bytes of string data from a source
	132	* pointer to a buffer backend, followed by a terminating '\0'
	133	* character, at the current context offset. This is more or less a
	134	* buffer backend-specific strncpy() operation. If a terminating '\0'
	135	* character is found in @src before @len - 1 characters are copied, pad
	136	* the buffer with @pad characters (e.g. '#').
	137	*/
00d0f8eb	138	static inline __attribute__((always_inline))
a44c74d9 MD	139	void lib_ring_buffer_strcpy(const struct lttng_ust_lib_ring_buffer_config *config,
	140	struct lttng_ust_lib_ring_buffer_ctx *ctx,
	141	const char *src, size_t len, int pad)
	142	{
a44c74d9 MD	143	struct channel_backend *chanb = &ctx->chan->backend;
a44c74d9 MD	144	struct lttng_ust_shm_handle *handle = ctx->handle;
a3492932	145	size_t count;
a44c74d9	146	size_t offset = ctx->buf_offset;
a3492932 MD	147	struct lttng_ust_lib_ring_buffer_backend_pages *backend_pages;
a3492932 MD	148	void *p;
a44c74d9 MD	149
	150	if (caa_unlikely(!len))
	151	return;
a44c74d9 MD	152	/*
	153	* Underlying layer should never ask for writes across
	154	* subbuffers.
	155	*/
a3492932 MD	156	CHAN_WARN_ON(chanb, (offset & (chanb->buf_size - 1)) + len > chanb->buf_size);
	157	backend_pages = lib_ring_buffer_get_backend_pages_from_ctx(config, ctx);
	158	if (caa_unlikely(!backend_pages)) {
	159	if (lib_ring_buffer_backend_get_pages(config, ctx, &backend_pages))
	160	return;
	161	}
	162	p = shmp_index(handle, backend_pages->p, offset & (chanb->subbuf_size - 1));
	163	if (caa_unlikely(!p))
	164	return;
	165
	166	count = lib_ring_buffer_do_strcpy(config, p, src, len - 1);
a44c74d9 MD	167	offset += count;
	168	/* Padding */
	169	if (caa_unlikely(count < len - 1)) {
	170	size_t pad_len = len - 1 - count;
	171
a3492932 MD	172	p = shmp_index(handle, backend_pages->p, offset & (chanb->subbuf_size - 1));
	173	if (caa_unlikely(!p))
	174	return;
	175	lib_ring_buffer_do_memset(p, pad, pad_len);
a44c74d9 MD	176	offset += pad_len;
	177	}
	178	/* Final '\0' */
a3492932 MD	179	p = shmp_index(handle, backend_pages->p, offset & (chanb->subbuf_size - 1));
	180	if (caa_unlikely(!p))
	181	return;
	182	lib_ring_buffer_do_memset(p, '\0', 1);
a44c74d9 MD	183	ctx->buf_offset += len;
	184	}
	185
852c2936 MD	186	/*
	187	* This accessor counts the number of unread records in a buffer.
	188	* It only provides a consistent value if no reads not writes are performed
	189	* concurrently.
	190	*/
	191	static inline
	192	unsigned long lib_ring_buffer_get_records_unread(
4cfec15c MD	193	const struct lttng_ust_lib_ring_buffer_config *config,
4cfec15c MD	194	struct lttng_ust_lib_ring_buffer *buf,
38fae1d3	195	struct lttng_ust_shm_handle *handle)
852c2936	196	{
4cfec15c	197	struct lttng_ust_lib_ring_buffer_backend *bufb = &buf->backend;
15500a1b	198	unsigned long records_unread = 0, sb_bindex;
852c2936	199	unsigned int i;
15500a1b	200	struct channel *chan;
852c2936	201
15500a1b MD	202	chan = shmp(handle, bufb->chan);
	203	if (!chan)
	204	return 0;
	205	for (i = 0; i < chan->backend.num_subbuf; i++) {
	206	struct lttng_ust_lib_ring_buffer_backend_subbuffer *wsb;
	207	struct lttng_ust_lib_ring_buffer_backend_pages_shmp *rpages;
	208	struct lttng_ust_lib_ring_buffer_backend_pages *backend_pages;
	209
	210	wsb = shmp_index(handle, bufb->buf_wsb, i);
	211	if (!wsb)
	212	return 0;
	213	sb_bindex = subbuffer_id_get_index(config, wsb->id);
	214	rpages = shmp_index(handle, bufb->array, sb_bindex);
	215	if (!rpages)
	216	return 0;
	217	backend_pages = shmp(handle, rpages->shmp);
	218	if (!backend_pages)
	219	return 0;
	220	records_unread += v_read(config, &backend_pages->records_unread);
852c2936 MD	221	}
852c2936 MD	222	if (config->mode == RING_BUFFER_OVERWRITE) {
15500a1b MD	223	struct lttng_ust_lib_ring_buffer_backend_pages_shmp *rpages;
	224	struct lttng_ust_lib_ring_buffer_backend_pages *backend_pages;
	225
	226	sb_bindex = subbuffer_id_get_index(config, bufb->buf_rsb.id);
	227	rpages = shmp_index(handle, bufb->array, sb_bindex);
	228	if (!rpages)
	229	return 0;
	230	backend_pages = shmp(handle, rpages->shmp);
	231	if (!backend_pages)
	232	return 0;
	233	records_unread += v_read(config, &backend_pages->records_unread);
852c2936 MD	234	}
	235	return records_unread;
	236	}
	237
e92f3e28	238	#endif /* _LTTNG_RING_BUFFER_BACKEND_H */