X-Git-Url: http://git.liburcu.org/?a=blobdiff_plain;f=libringbuffer%2Fring_buffer_backend.c;h=a0ef74461bd12606b9d3bd6eaa14ac2ae9f600ab;hb=ab12eb57cf3fdd2c25df16a3d1e770a2642b93e8;hp=40fae8017aed823401ae85a328be5bb8833a8231;hpb=2657d1baecb6b33ead9c9caa7c8ec5bf423d1fde;p=lttng-ust.git

diff --git a/libringbuffer/ring_buffer_backend.c b/libringbuffer/ring_buffer_backend.c
index 40fae801..a0ef7446 100644
--- a/libringbuffer/ring_buffer_backend.c
+++ b/libringbuffer/ring_buffer_backend.c
@@ -19,6 +19,7 @@
  */
 
 #define _GNU_SOURCE
+#define _LGPL_SOURCE
 #include <unistd.h>
 #include <urcu/arch.h>
 #include <limits.h>
@@ -46,12 +47,16 @@ int lib_ring_buffer_backend_allocate(const struct lttng_ust_lib_ring_buffer_conf
 				     struct lttng_ust_shm_handle *handle,
 				     struct shm_object *shmobj)
 {
-	struct channel_backend *chanb = &shmp(handle, bufb->chan)->backend;
+	struct channel_backend *chanb;
 	unsigned long subbuf_size, mmap_offset = 0;
 	unsigned long num_subbuf_alloc;
 	unsigned long i;
 	long page_size;
 
+	chanb = &shmp(handle, bufb->chan)->backend;
+	if (!chanb)
+		return -EINVAL;
+
 	subbuf_size = chanb->subbuf_size;
 	num_subbuf_alloc = num_subbuf;
 
@@ -97,8 +102,14 @@ int lib_ring_buffer_backend_allocate(const struct lttng_ust_lib_ring_buffer_conf
 	if (caa_unlikely(!shmp(handle, bufb->buf_wsb)))
 		goto free_array;
 
-	for (i = 0; i < num_subbuf; i++)
-		shmp_index(handle, bufb->buf_wsb, i)->id = subbuffer_id(config, 0, 1, i);
+	for (i = 0; i < num_subbuf; i++) {
+		struct lttng_ust_lib_ring_buffer_backend_subbuffer *sb;
+
+		sb = shmp_index(handle, bufb->buf_wsb, i);
+		if (!sb)
+			goto free_array;
+		sb->id = subbuffer_id(config, 0, 1, i);
+	}
 
 	/* Assign read-side subbuffer table */
 	if (extra_reader_sb)
@@ -107,23 +118,40 @@ int lib_ring_buffer_backend_allocate(const struct lttng_ust_lib_ring_buffer_conf
 	else
 		bufb->buf_rsb.id = subbuffer_id(config, 0, 1, 0);
 
+	/* Allocate subbuffer packet counter table */
+	align_shm(shmobj, __alignof__(struct lttng_ust_lib_ring_buffer_backend_counts));
+	set_shmp(bufb->buf_cnt, zalloc_shm(shmobj,
+				sizeof(struct lttng_ust_lib_ring_buffer_backend_counts)
+				* num_subbuf));
+	if (caa_unlikely(!shmp(handle, bufb->buf_cnt)))
+		goto free_wsb;
+
 	/* Assign pages to page index */
 	for (i = 0; i < num_subbuf_alloc; i++) {
+		struct lttng_ust_lib_ring_buffer_backend_pages_shmp *sbp;
+		struct lttng_ust_lib_ring_buffer_backend_pages *pages;
 		struct shm_ref ref;
 
 		ref.index = bufb->memory_map._ref.index;
 		ref.offset = bufb->memory_map._ref.offset;
 		ref.offset += i * subbuf_size;
 
-		set_shmp(shmp(handle, shmp_index(handle, bufb->array, i)->shmp)->p,
-			 ref);
+		sbp = shmp_index(handle, bufb->array, i);
+		if (!sbp)
+			goto free_array;
+		pages = shmp(handle, sbp->shmp);
+		if (!pages)
+			goto free_array;
+		set_shmp(pages->p, ref);
 		if (config->output == RING_BUFFER_MMAP) {
-			shmp(handle, shmp_index(handle, bufb->array, i)->shmp)->mmap_offset = mmap_offset;
+			pages->mmap_offset = mmap_offset;
 			mmap_offset += subbuf_size;
 		}
 	}
 	return 0;
 
+free_wsb:
+	/* bufb->buf_wsb will be freed by shm teardown */
 free_array:
 	/* bufb->array[i] will be freed by shm teardown */
 memory_map_error:
@@ -152,17 +180,28 @@ int lib_ring_buffer_backend_create(struct lttng_ust_lib_ring_buffer_backend *buf
 void lib_ring_buffer_backend_reset(struct lttng_ust_lib_ring_buffer_backend *bufb,
 				   struct lttng_ust_shm_handle *handle)
 {
-	struct channel_backend *chanb = &shmp(handle, bufb->chan)->backend;
-	const struct lttng_ust_lib_ring_buffer_config *config = &chanb->config;
+	struct channel_backend *chanb;
+	const struct lttng_ust_lib_ring_buffer_config *config;
 	unsigned long num_subbuf_alloc;
 	unsigned int i;
 
+	chanb = &shmp(handle, bufb->chan)->backend;
+	if (!chanb)
+		return;
+	config = &chanb->config;
+
 	num_subbuf_alloc = chanb->num_subbuf;
 	if (chanb->extra_reader_sb)
 		num_subbuf_alloc++;
 
-	for (i = 0; i < chanb->num_subbuf; i++)
-		shmp_index(handle, bufb->buf_wsb, i)->id = subbuffer_id(config, 0, 1, i);
+	for (i = 0; i < chanb->num_subbuf; i++) {
+		struct lttng_ust_lib_ring_buffer_backend_subbuffer *sb;
+
+		sb = shmp_index(handle, bufb->buf_wsb, i);
+		if (!sb)
+			return;
+		sb->id = subbuffer_id(config, 0, 1, i);
+	}
 	if (chanb->extra_reader_sb)
 		bufb->buf_rsb.id = subbuffer_id(config, 0, 1,
 						num_subbuf_alloc - 1);
@@ -170,10 +209,19 @@ void lib_ring_buffer_backend_reset(struct lttng_ust_lib_ring_buffer_backend *buf
 		bufb->buf_rsb.id = subbuffer_id(config, 0, 1, 0);
 
 	for (i = 0; i < num_subbuf_alloc; i++) {
+		struct lttng_ust_lib_ring_buffer_backend_pages_shmp *sbp;
+		struct lttng_ust_lib_ring_buffer_backend_pages *pages;
+
+		sbp = shmp_index(handle, bufb->array, i);
+		if (!sbp)
+			return;
+		pages = shmp(handle, sbp->shmp);
+		if (!pages)
+			return;
 		/* Don't reset mmap_offset */
-		v_set(config, &shmp(handle, shmp_index(handle, bufb->array, i)->shmp)->records_commit, 0);
-		v_set(config, &shmp(handle, shmp_index(handle, bufb->array, i)->shmp)->records_unread, 0);
-		shmp(handle, shmp_index(handle, bufb->array, i)->shmp)->data_size = 0;
+		v_set(config, &pages->records_commit, 0);
+		v_set(config, &pages->records_unread, 0);
+		pages->data_size = 0;
 		/* Don't reset backend page and virt addresses */
 	}
 	/* Don't reset num_pages_per_subbuf, cpu, allocated */
@@ -286,6 +334,8 @@ int channel_backend_init(struct channel_backend *chanb,
 	shmsize += sizeof(struct lttng_ust_lib_ring_buffer_backend_pages) * num_subbuf_alloc;
 	shmsize += offset_align(shmsize, __alignof__(struct lttng_ust_lib_ring_buffer_backend_subbuffer));
 	shmsize += sizeof(struct lttng_ust_lib_ring_buffer_backend_subbuffer) * num_subbuf;
+	shmsize += offset_align(shmsize, __alignof__(struct lttng_ust_lib_ring_buffer_backend_counts));
+	shmsize += sizeof(struct lttng_ust_lib_ring_buffer_backend_counts) * num_subbuf;
 	/* Per-cpu buffer size: control (after backend) */
 	shmsize += offset_align(shmsize, __alignof__(struct commit_counters_hot));
 	shmsize += sizeof(struct commit_counters_hot) * num_subbuf;
@@ -301,7 +351,7 @@ int channel_backend_init(struct channel_backend *chanb,
 			struct shm_object *shmobj;
 
 			shmobj = shm_object_table_alloc(handle->table, shmsize,
-					SHM_OBJECT_SHM, stream_fds[i]);
+					SHM_OBJECT_SHM, stream_fds[i], i);
 			if (!shmobj)
 				goto end;
 			align_shm(shmobj, __alignof__(struct lttng_ust_lib_ring_buffer));
@@ -320,7 +370,7 @@ int channel_backend_init(struct channel_backend *chanb,
 		struct lttng_ust_lib_ring_buffer *buf;
 
 		shmobj = shm_object_table_alloc(handle->table, shmsize,
-					SHM_OBJECT_SHM, stream_fds[0]);
+					SHM_OBJECT_SHM, stream_fds[0], -1);
 		if (!shmobj)
 			goto end;
 		align_shm(shmobj, __alignof__(struct lttng_ust_lib_ring_buffer));
@@ -369,12 +419,18 @@ void channel_backend_free(struct channel_backend *chanb,
 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)
 {
-	struct channel_backend *chanb = &shmp(handle, bufb->chan)->backend;
-	const struct lttng_ust_lib_ring_buffer_config *config = &chanb->config;
+	struct channel_backend *chanb;
+	const struct lttng_ust_lib_ring_buffer_config *config;
 	ssize_t orig_len;
 	struct lttng_ust_lib_ring_buffer_backend_pages_shmp *rpages;
+	struct lttng_ust_lib_ring_buffer_backend_pages *backend_pages;
 	unsigned long sb_bindex, id;
+	void *src;
 
+	chanb = &shmp(handle, bufb->chan)->backend;
+	if (!chanb)
+		return 0;
+	config = &chanb->config;
 	orig_len = len;
 	offset &= chanb->buf_size - 1;
 
@@ -383,6 +439,8 @@ size_t lib_ring_buffer_read(struct lttng_ust_lib_ring_buffer_backend *bufb, size
 	id = bufb->buf_rsb.id;
 	sb_bindex = subbuffer_id_get_index(config, id);
 	rpages = shmp_index(handle, bufb->array, sb_bindex);
+	if (!rpages)
+		return 0;
 	/*
 	 * Underlying layer should never ask for reads across
 	 * subbuffers.
@@ -390,7 +448,13 @@ size_t lib_ring_buffer_read(struct lttng_ust_lib_ring_buffer_backend *bufb, size
 	CHAN_WARN_ON(chanb, offset >= chanb->buf_size);
 	CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
 		     && subbuffer_id_is_noref(config, id));
-	memcpy(dest, shmp_index(handle, shmp(handle, rpages->shmp)->p, offset & (chanb->subbuf_size - 1)), len);
+	backend_pages = shmp(handle, rpages->shmp);
+	if (!backend_pages)
+		return 0;
+	src = shmp_index(handle, backend_pages->p, offset & (chanb->subbuf_size - 1));
+	if (caa_unlikely(!src))
+		return 0;
+	memcpy(dest, src, len);
 	return orig_len;
 }
 
@@ -408,13 +472,18 @@ size_t lib_ring_buffer_read(struct lttng_ust_lib_ring_buffer_backend *bufb, size
 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)
 {
-	struct channel_backend *chanb = &shmp(handle, bufb->chan)->backend;
-	const struct lttng_ust_lib_ring_buffer_config *config = &chanb->config;
+	struct channel_backend *chanb;
+	const struct lttng_ust_lib_ring_buffer_config *config;
 	ssize_t string_len, orig_offset;
 	char *str;
 	struct lttng_ust_lib_ring_buffer_backend_pages_shmp *rpages;
+	struct lttng_ust_lib_ring_buffer_backend_pages *backend_pages;
 	unsigned long sb_bindex, id;
 
+	chanb = &shmp(handle, bufb->chan)->backend;
+	if (!chanb)
+		return -EINVAL;
+	config = &chanb->config;
 	if (caa_unlikely(!len))
 		return -EINVAL;
 	offset &= chanb->buf_size - 1;
@@ -422,6 +491,8 @@ int lib_ring_buffer_read_cstr(struct lttng_ust_lib_ring_buffer_backend *bufb, si
 	id = bufb->buf_rsb.id;
 	sb_bindex = subbuffer_id_get_index(config, id);
 	rpages = shmp_index(handle, bufb->array, sb_bindex);
+	if (!rpages)
+		return -EINVAL;
 	/*
 	 * Underlying layer should never ask for reads across
 	 * subbuffers.
@@ -429,7 +500,12 @@ int lib_ring_buffer_read_cstr(struct lttng_ust_lib_ring_buffer_backend *bufb, si
 	CHAN_WARN_ON(chanb, offset >= chanb->buf_size);
 	CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
 		     && subbuffer_id_is_noref(config, id));
-	str = shmp_index(handle, shmp(handle, rpages->shmp)->p, offset & (chanb->subbuf_size - 1));
+	backend_pages = shmp(handle, rpages->shmp);
+	if (!backend_pages)
+		return -EINVAL;
+	str = shmp_index(handle, backend_pages->p, offset & (chanb->subbuf_size - 1));
+	if (caa_unlikely(!str))
+		return -EINVAL;
 	string_len = strnlen(str, len);
 	if (dest && len) {
 		memcpy(dest, str, string_len);
@@ -454,17 +530,27 @@ void *lib_ring_buffer_read_offset_address(struct lttng_ust_lib_ring_buffer_backe
 					  struct lttng_ust_shm_handle *handle)
 {
 	struct lttng_ust_lib_ring_buffer_backend_pages_shmp *rpages;
-	struct channel_backend *chanb = &shmp(handle, bufb->chan)->backend;
-	const struct lttng_ust_lib_ring_buffer_config *config = &chanb->config;
+	struct lttng_ust_lib_ring_buffer_backend_pages *backend_pages;
+	struct channel_backend *chanb;
+	const struct lttng_ust_lib_ring_buffer_config *config;
 	unsigned long sb_bindex, id;
 
+	chanb = &shmp(handle, bufb->chan)->backend;
+	if (!chanb)
+		return NULL;
+	config = &chanb->config;
 	offset &= chanb->buf_size - 1;
 	id = bufb->buf_rsb.id;
 	sb_bindex = subbuffer_id_get_index(config, id);
 	rpages = shmp_index(handle, bufb->array, sb_bindex);
+	if (!rpages)
+		return NULL;
 	CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
 		     && subbuffer_id_is_noref(config, id));
-	return shmp_index(handle, shmp(handle, rpages->shmp)->p, offset & (chanb->subbuf_size - 1));
+	backend_pages = shmp(handle, rpages->shmp);
+	if (!backend_pages)
+		return NULL;
+	return shmp_index(handle, backend_pages->p, offset & (chanb->subbuf_size - 1));
 }
 
 /**
@@ -483,16 +569,30 @@ void *lib_ring_buffer_offset_address(struct lttng_ust_lib_ring_buffer_backend *b
 {
 	size_t sbidx;
 	struct lttng_ust_lib_ring_buffer_backend_pages_shmp *rpages;
-	struct channel_backend *chanb = &shmp(handle, bufb->chan)->backend;
-	const struct lttng_ust_lib_ring_buffer_config *config = &chanb->config;
+	struct lttng_ust_lib_ring_buffer_backend_pages *backend_pages;
+	struct channel_backend *chanb;
+	const struct lttng_ust_lib_ring_buffer_config *config;
 	unsigned long sb_bindex, id;
+	struct lttng_ust_lib_ring_buffer_backend_subbuffer *sb;
 
+	chanb = &shmp(handle, bufb->chan)->backend;
+	if (!chanb)
+		return NULL;
+	config = &chanb->config;
 	offset &= chanb->buf_size - 1;
 	sbidx = offset >> chanb->subbuf_size_order;
-	id = shmp_index(handle, bufb->buf_wsb, sbidx)->id;
+	sb = shmp_index(handle, bufb->buf_wsb, sbidx);
+	if (!sb)
+		return NULL;
+	id = sb->id;
 	sb_bindex = subbuffer_id_get_index(config, id);
 	rpages = shmp_index(handle, bufb->array, sb_bindex);
+	if (!rpages)
+		return NULL;
 	CHAN_WARN_ON(chanb, config->mode == RING_BUFFER_OVERWRITE
 		     && subbuffer_id_is_noref(config, id));
-	return shmp_index(handle, shmp(handle, rpages->shmp)->p, offset & (chanb->subbuf_size - 1));
+	backend_pages = shmp(handle, rpages->shmp);
+	if (!backend_pages)
+		return NULL;
+	return shmp_index(handle, backend_pages->p, offset & (chanb->subbuf_size - 1));
 }