X-Git-Url: http://git.liburcu.org/?a=blobdiff_plain;f=libringbuffer%2Fring_buffer_frontend.c;h=7762da7291ae5d1f215ee0c47039bd8c84f78d28;hb=8c90a710949a214d87084ff67fc85f284bc93ef2;hp=970fde90a2a79bd0d8638f2eefc168b07c7e8499;hpb=f645cfa794fc481dff6cb30ea56e93f825b3d423;p=lttng-ust.git diff --git a/libringbuffer/ring_buffer_frontend.c b/libringbuffer/ring_buffer_frontend.c index 970fde90..7762da72 100644 --- a/libringbuffer/ring_buffer_frontend.c +++ b/libringbuffer/ring_buffer_frontend.c @@ -1,7 +1,22 @@ /* * ring_buffer_frontend.c * - * (C) Copyright 2005-2010 - Mathieu Desnoyers + * Copyright (C) 2005-2012 Mathieu Desnoyers + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; only + * version 2.1 of the License. + * + * This library is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + * * * Ring buffer wait-free buffer synchronization. Producer-consumer and flight * recorder (overwrite) modes. See thesis: @@ -34,8 +49,6 @@ * - splice one subbuffer worth of data to a pipe * - splice the data from pipe to disk/network * - put_subbuf - * - * Dual LGPL v2.1/GPL v2 license. */ #define _GNU_SOURCE @@ -45,6 +58,7 @@ #include #include #include +#include #include #include "smp.h" @@ -54,6 +68,7 @@ #include "frontend.h" #include "shm.h" #include "tlsfixup.h" +#include "../liblttng-ust/compat.h" /* For ENODATA */ #ifndef max #define max(a, b) ((a) > (b) ? (a) : (b)) @@ -88,7 +103,7 @@ struct switch_offsets { switch_old_end:1; }; -__thread unsigned int lib_ring_buffer_nesting; +DEFINE_URCU_TLS(unsigned int, lib_ring_buffer_nesting); /* * TODO: this is unused. Errors are saved within the ring buffer. @@ -241,7 +256,7 @@ static void switch_buffer_timer(unsigned long data) /* * Only flush buffers periodically if readers are active. */ - if (uatomic_read(&buf->active_readers) || uatomic_read(&buf->active_shadow_readers)) + if (uatomic_read(&buf->active_readers)) lib_ring_buffer_switch_slow(buf, SWITCH_ACTIVE, handle); //TODO timers @@ -299,7 +314,7 @@ static void read_buffer_timer(unsigned long data) CHAN_WARN_ON(chan, !buf->backend.allocated); - if (uatomic_read(&buf->active_readers) || uatomic_read(&buf->active_shadow_readers)) + if (uatomic_read(&buf->active_readers)) && lib_ring_buffer_poll_deliver(config, buf, chan)) { //TODO //wake_up_interruptible(&buf->read_wait); @@ -387,11 +402,9 @@ static void channel_unregister_notifiers(struct channel *chan, //channel_backend_unregister_notifiers(&chan->backend); } -static void channel_free(struct channel *chan, struct lttng_ust_shm_handle *handle, - int shadow) +static void channel_free(struct channel *chan, struct lttng_ust_shm_handle *handle) { - if (!shadow) - channel_backend_free(&chan->backend, handle); + channel_backend_free(&chan->backend, handle); /* chan is freed by shm teardown */ shm_object_table_destroy(handle->table); free(handle); @@ -425,15 +438,19 @@ struct lttng_ust_shm_handle *channel_create(const struct lttng_ust_lib_ring_buff void *priv_data_init, void *buf_addr, size_t subbuf_size, size_t num_subbuf, unsigned int switch_timer_interval, - unsigned int read_timer_interval, - int **shm_fd, int **wait_fd, uint64_t **memory_map_size) + unsigned int read_timer_interval) { int ret, cpu; size_t shmsize, chansize; struct channel *chan; struct lttng_ust_shm_handle *handle; struct shm_object *shmobj; - struct shm_ref *ref; + unsigned int nr_streams; + + if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) + nr_streams = num_possible_cpus(); + else + nr_streams = 1; if (lib_ring_buffer_check_config(config, switch_timer_interval, read_timer_interval)) @@ -451,15 +468,14 @@ struct lttng_ust_shm_handle *channel_create(const struct lttng_ust_lib_ring_buff /* Calculate the shm allocation layout */ shmsize = sizeof(struct channel); shmsize += offset_align(shmsize, __alignof__(struct lttng_ust_lib_ring_buffer_shmp)); - if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) - shmsize += sizeof(struct lttng_ust_lib_ring_buffer_shmp) * num_possible_cpus(); - else - shmsize += sizeof(struct lttng_ust_lib_ring_buffer_shmp); + shmsize += sizeof(struct lttng_ust_lib_ring_buffer_shmp) * nr_streams; chansize = shmsize; - shmsize += offset_align(shmsize, priv_data_align); + if (priv_data_align) + shmsize += offset_align(shmsize, priv_data_align); shmsize += priv_data_size; - shmobj = shm_object_table_append(handle->table, shmsize); + /* Allocate normal memory for channel (not shared) */ + shmobj = shm_object_table_alloc(handle->table, shmsize, SHM_OBJECT_MEM); if (!shmobj) goto error_append; /* struct channel is at object 0, offset 0 (hardcoded) */ @@ -469,6 +485,7 @@ struct lttng_ust_shm_handle *channel_create(const struct lttng_ust_lib_ring_buff chan = shmp(handle, handle->chan); if (!chan) goto error_append; + chan->nr_streams = nr_streams; /* space for private data */ if (priv_data_size) { @@ -483,7 +500,8 @@ struct lttng_ust_shm_handle *channel_create(const struct lttng_ust_lib_ring_buff memcpy(*priv_data, priv_data_init, priv_data_size); } else { chan->priv_data_offset = -1; - *priv_data = NULL; + if (priv_data) + *priv_data = NULL; } ret = channel_backend_init(&chan->backend, name, config, @@ -516,8 +534,6 @@ struct lttng_ust_shm_handle *channel_create(const struct lttng_ust_lib_ring_buff lib_ring_buffer_start_switch_timer(buf, handle); lib_ring_buffer_start_read_timer(buf, handle); } - ref = &handle->chan._ref; - shm_get_object_data(handle, ref, shm_fd, wait_fd, memory_map_size); return handle; error_backend_init: @@ -528,7 +544,7 @@ error_table_alloc: return NULL; } -struct lttng_ust_shm_handle *channel_handle_create(int shm_fd, int wait_fd, +struct lttng_ust_shm_handle *channel_handle_create(void *data, uint64_t memory_map_size) { struct lttng_ust_shm_handle *handle; @@ -543,8 +559,8 @@ struct lttng_ust_shm_handle *channel_handle_create(int shm_fd, int wait_fd, if (!handle->table) goto error_table_alloc; /* Add channel object */ - object = shm_object_table_append_shadow(handle->table, - shm_fd, wait_fd, memory_map_size); + object = shm_object_table_append_mem(handle->table, data, + memory_map_size); if (!object) goto error_table_object; /* struct channel is at object 0, offset 0 (hardcoded) */ @@ -560,23 +576,30 @@ error_table_alloc: } int channel_handle_add_stream(struct lttng_ust_shm_handle *handle, - int shm_fd, int wait_fd, uint64_t memory_map_size) + int shm_fd, int wakeup_fd, uint32_t stream_nr, + uint64_t memory_map_size) { struct shm_object *object; /* Add stream object */ - object = shm_object_table_append_shadow(handle->table, - shm_fd, wait_fd, memory_map_size); + object = shm_object_table_append_shm(handle->table, + shm_fd, wakeup_fd, stream_nr, + memory_map_size); if (!object) - return -1; + return -EINVAL; return 0; } +unsigned int channel_handle_get_nr_streams(struct lttng_ust_shm_handle *handle) +{ + assert(handle->table); + return handle->table->allocated_len - 1; +} + static -void channel_release(struct channel *chan, struct lttng_ust_shm_handle *handle, - int shadow) +void channel_release(struct channel *chan, struct lttng_ust_shm_handle *handle) { - channel_free(chan, handle, shadow); + channel_free(chan, handle); } /** @@ -590,25 +613,21 @@ void channel_release(struct channel *chan, struct lttng_ust_shm_handle *handle, * They should release their handle at that point. */ void channel_destroy(struct channel *chan, struct lttng_ust_shm_handle *handle, - int shadow) + int consumer) { - if (shadow) { - channel_release(chan, handle, shadow); - return; + if (consumer) { + /* + * Note: the consumer takes care of finalizing and + * switching the buffers. + */ + channel_unregister_notifiers(chan, handle); } - channel_unregister_notifiers(chan, handle); - - /* - * Note: the consumer takes care of finalizing and switching the - * buffers. - */ - /* * sessiond/consumer are keeping a reference on the shm file * descriptor directly. No need to refcount. */ - channel_release(chan, handle, shadow); + channel_release(chan, handle); return; } @@ -616,36 +635,64 @@ struct lttng_ust_lib_ring_buffer *channel_get_ring_buffer( const struct lttng_ust_lib_ring_buffer_config *config, struct channel *chan, int cpu, struct lttng_ust_shm_handle *handle, - int **shm_fd, int **wait_fd, - uint64_t **memory_map_size) + int *shm_fd, int *wait_fd, + int *wakeup_fd, + uint64_t *memory_map_size) { struct shm_ref *ref; if (config->alloc == RING_BUFFER_ALLOC_GLOBAL) { - ref = &chan->backend.buf[0].shmp._ref; - shm_get_object_data(handle, ref, shm_fd, wait_fd, - memory_map_size); - return shmp(handle, chan->backend.buf[0].shmp); + cpu = 0; } else { if (cpu >= num_possible_cpus()) return NULL; - ref = &chan->backend.buf[cpu].shmp._ref; - shm_get_object_data(handle, ref, shm_fd, wait_fd, - memory_map_size); - return shmp(handle, chan->backend.buf[cpu].shmp); } + ref = &chan->backend.buf[cpu].shmp._ref; + *shm_fd = shm_get_shm_fd(handle, ref); + *wait_fd = shm_get_wait_fd(handle, ref); + *wakeup_fd = shm_get_wakeup_fd(handle, ref); + if (shm_get_shm_size(handle, ref, memory_map_size)) + return NULL; + return shmp(handle, chan->backend.buf[cpu].shmp); } -int lib_ring_buffer_open_read(struct lttng_ust_lib_ring_buffer *buf, - struct lttng_ust_shm_handle *handle, - int shadow) +int ring_buffer_close_wait_fd(const struct lttng_ust_lib_ring_buffer_config *config, + struct channel *chan, + struct lttng_ust_shm_handle *handle, + int cpu) { - if (shadow) { - if (uatomic_cmpxchg(&buf->active_shadow_readers, 0, 1) != 0) - return -EBUSY; - cmm_smp_mb(); - return 0; + struct shm_ref *ref; + + if (config->alloc == RING_BUFFER_ALLOC_GLOBAL) { + cpu = 0; + } else { + if (cpu >= num_possible_cpus()) + return -EINVAL; } + ref = &chan->backend.buf[cpu].shmp._ref; + return shm_close_wait_fd(handle, ref); +} + +int ring_buffer_close_wakeup_fd(const struct lttng_ust_lib_ring_buffer_config *config, + struct channel *chan, + struct lttng_ust_shm_handle *handle, + int cpu) +{ + struct shm_ref *ref; + + if (config->alloc == RING_BUFFER_ALLOC_GLOBAL) { + cpu = 0; + } else { + if (cpu >= num_possible_cpus()) + return -EINVAL; + } + ref = &chan->backend.buf[cpu].shmp._ref; + return shm_close_wakeup_fd(handle, ref); +} + +int lib_ring_buffer_open_read(struct lttng_ust_lib_ring_buffer *buf, + struct lttng_ust_shm_handle *handle) +{ if (uatomic_cmpxchg(&buf->active_readers, 0, 1) != 0) return -EBUSY; cmm_smp_mb(); @@ -653,17 +700,10 @@ int lib_ring_buffer_open_read(struct lttng_ust_lib_ring_buffer *buf, } void lib_ring_buffer_release_read(struct lttng_ust_lib_ring_buffer *buf, - struct lttng_ust_shm_handle *handle, - int shadow) + struct lttng_ust_shm_handle *handle) { struct channel *chan = shmp(handle, buf->backend.chan); - if (shadow) { - CHAN_WARN_ON(chan, uatomic_read(&buf->active_shadow_readers) != 1); - cmm_smp_mb(); - uatomic_dec(&buf->active_shadow_readers); - return; - } CHAN_WARN_ON(chan, uatomic_read(&buf->active_readers) != 1); cmm_smp_mb(); uatomic_dec(&buf->active_readers); @@ -741,8 +781,7 @@ void lib_ring_buffer_move_consumer(struct lttng_ust_lib_ring_buffer *buf, struct channel *chan = shmp(handle, bufb->chan); unsigned long consumed; - CHAN_WARN_ON(chan, uatomic_read(&buf->active_readers) != 1 - && uatomic_read(&buf->active_shadow_readers) != 1); + CHAN_WARN_ON(chan, uatomic_read(&buf->active_readers) != 1); /* * Only push the consumed value forward. @@ -865,8 +904,7 @@ void lib_ring_buffer_put_subbuf(struct lttng_ust_lib_ring_buffer *buf, const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config; unsigned long read_sb_bindex, consumed_idx, consumed; - CHAN_WARN_ON(chan, uatomic_read(&buf->active_readers) != 1 - && uatomic_read(&buf->active_shadow_readers) != 1); + CHAN_WARN_ON(chan, uatomic_read(&buf->active_readers) != 1); if (!buf->get_subbuf) { /* @@ -982,23 +1020,30 @@ void lib_ring_buffer_print_errors(struct channel *chan, const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config; void *priv = channel_get_private(chan); - DBG("ring buffer %s, cpu %d: %lu records written, " - "%lu records overrun\n", - chan->backend.name, cpu, - v_read(config, &buf->records_count), - v_read(config, &buf->records_overrun)); - - if (v_read(config, &buf->records_lost_full) - || v_read(config, &buf->records_lost_wrap) - || v_read(config, &buf->records_lost_big)) - DBG("ring buffer %s, cpu %d: records were lost. Caused by:\n" - " [ %lu buffer full, %lu nest buffer wrap-around, " - "%lu event too big ]\n", - chan->backend.name, cpu, - v_read(config, &buf->records_lost_full), - v_read(config, &buf->records_lost_wrap), - v_read(config, &buf->records_lost_big)); - + if (!strcmp(chan->backend.name, "relay-metadata-mmap")) { + DBG("ring buffer %s: %lu records written, " + "%lu records overrun\n", + chan->backend.name, + v_read(config, &buf->records_count), + v_read(config, &buf->records_overrun)); + } else { + DBG("ring buffer %s, cpu %d: %lu records written, " + "%lu records overrun\n", + chan->backend.name, cpu, + v_read(config, &buf->records_count), + v_read(config, &buf->records_overrun)); + + if (v_read(config, &buf->records_lost_full) + || v_read(config, &buf->records_lost_wrap) + || v_read(config, &buf->records_lost_big)) + DBG("ring buffer %s, cpu %d: records were lost. Caused by:\n" + " [ %lu buffer full, %lu nest buffer wrap-around, " + "%lu event too big ]\n", + chan->backend.name, cpu, + v_read(config, &buf->records_lost_full), + v_read(config, &buf->records_lost_wrap), + v_read(config, &buf->records_lost_big)); + } lib_ring_buffer_print_buffer_errors(buf, chan, priv, cpu, handle); } @@ -1514,5 +1559,5 @@ int lib_ring_buffer_reserve_slow(struct lttng_ust_lib_ring_buffer_ctx *ctx) */ void lttng_fixup_ringbuffer_tls(void) { - asm volatile ("" : : "m" (lib_ring_buffer_nesting)); + asm volatile ("" : : "m" (URCU_TLS(lib_ring_buffer_nesting))); }