X-Git-Url: http://git.lttng.org/?a=blobdiff_plain;f=libringbuffer%2Fring_buffer_frontend.c;h=cf012dfaf2225dfe7a66cf2fb63f4e860c48265d;hb=57773204abc43ed7d4350ac0d641ddbbda1e441e;hp=f242cc07e825d81beb667c4dd401749df98ebaa6;hpb=1fc6c5c9aa62d88f9fa995575006f7a756b1944a;p=lttng-ust.git diff --git a/libringbuffer/ring_buffer_frontend.c b/libringbuffer/ring_buffer_frontend.c index f242cc07..cf012dfa 100644 --- a/libringbuffer/ring_buffer_frontend.c +++ b/libringbuffer/ring_buffer_frontend.c @@ -116,7 +116,7 @@ void lib_ring_buffer_reset(struct lib_ring_buffer *buf, struct shm_handle *handle) { struct channel *chan = shmp(handle, buf->backend.chan); - const struct lib_ring_buffer_config *config = chan->backend.config; + const struct lib_ring_buffer_config *config = &chan->backend.config; unsigned int i; /* @@ -172,7 +172,7 @@ int lib_ring_buffer_create(struct lib_ring_buffer *buf, struct shm_handle *handle, struct shm_object *shmobj) { - const struct lib_ring_buffer_config *config = chanb->config; + const struct lib_ring_buffer_config *config = &chanb->config; struct channel *chan = caa_container_of(chanb, struct channel, backend); void *priv = chanb->priv; unsigned int num_subbuf; @@ -244,12 +244,12 @@ static void switch_buffer_timer(unsigned long data) { struct lib_ring_buffer *buf = (struct lib_ring_buffer *)data; struct channel *chan = shmp(handle, buf->backend.chan); - const struct lib_ring_buffer_config *config = chan->backend.config; + const struct lib_ring_buffer_config *config = &chan->backend.config; /* * Only flush buffers periodically if readers are active. */ - if (uatomic_read(&buf->active_readers)) + if (uatomic_read(&buf->active_readers) || uatomic_read(&buf->active_shadow_readers)) lib_ring_buffer_switch_slow(buf, SWITCH_ACTIVE, handle); //TODO timers @@ -266,7 +266,7 @@ static void lib_ring_buffer_start_switch_timer(struct lib_ring_buffer *buf, struct shm_handle *handle) { struct channel *chan = shmp(handle, buf->backend.chan); - const struct lib_ring_buffer_config *config = chan->backend.config; + //const struct lib_ring_buffer_config *config = &chan->backend.config; if (!chan->switch_timer_interval || buf->switch_timer_enabled) return; @@ -303,11 +303,11 @@ static void read_buffer_timer(unsigned long data) { struct lib_ring_buffer *buf = (struct lib_ring_buffer *)data; struct channel *chan = shmp(handle, buf->backend.chan); - const struct lib_ring_buffer_config *config = chan->backend.config; + const struct lib_ring_buffer_config *config = &chan->backend.config; CHAN_WARN_ON(chan, !buf->backend.allocated); - if (uatomic_read(&buf->active_readers) + if (uatomic_read(&buf->active_readers) || uatomic_read(&buf->active_shadow_readers)) && lib_ring_buffer_poll_deliver(config, buf, chan)) { //TODO //wake_up_interruptible(&buf->read_wait); @@ -328,7 +328,7 @@ static void lib_ring_buffer_start_read_timer(struct lib_ring_buffer *buf, struct shm_handle *handle) { struct channel *chan = shmp(handle, buf->backend.chan); - const struct lib_ring_buffer_config *config = chan->backend.config; + const struct lib_ring_buffer_config *config = &chan->backend.config; if (config->wakeup != RING_BUFFER_WAKEUP_BY_TIMER || !chan->read_timer_interval @@ -352,7 +352,7 @@ static void lib_ring_buffer_stop_read_timer(struct lib_ring_buffer *buf, struct shm_handle *handle) { struct channel *chan = shmp(handle, buf->backend.chan); - const struct lib_ring_buffer_config *config = chan->backend.config; + const struct lib_ring_buffer_config *config = &chan->backend.config; if (config->wakeup != RING_BUFFER_WAKEUP_BY_TIMER || !chan->read_timer_interval @@ -376,7 +376,7 @@ static void lib_ring_buffer_stop_read_timer(struct lib_ring_buffer *buf, static void channel_unregister_notifiers(struct channel *chan, struct shm_handle *handle) { - const struct lib_ring_buffer_config *config = chan->backend.config; + const struct lib_ring_buffer_config *config = &chan->backend.config; int cpu; if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) { @@ -395,11 +395,11 @@ static void channel_unregister_notifiers(struct channel *chan, //channel_backend_unregister_notifiers(&chan->backend); } -static void channel_free(struct channel *chan, struct shm_handle *handle) +static void channel_free(struct channel *chan, struct shm_handle *handle, + int shadow) { - int ret; - - channel_backend_free(&chan->backend, handle); + if (!shadow) + channel_backend_free(&chan->backend, handle); /* chan is freed by shm teardown */ shm_object_table_destroy(handle->table); free(handle); @@ -427,13 +427,15 @@ struct shm_handle *channel_create(const struct lib_ring_buffer_config *config, 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) + unsigned int read_timer_interval, + int *shm_fd, int *wait_fd, uint64_t *memory_map_size) { int ret, cpu; size_t shmsize; struct channel *chan; struct shm_handle *handle; struct shm_object *shmobj; + struct shm_ref *ref; if (lib_ring_buffer_check_config(config, switch_timer_interval, read_timer_interval)) @@ -456,7 +458,12 @@ struct shm_handle *channel_create(const struct lib_ring_buffer_config *config, shmsize += sizeof(struct lib_ring_buffer_shmp); shmobj = shm_object_table_append(handle->table, shmsize); + if (!shmobj) + goto error_append; + /* struct channel is at object 0, offset 0 (hardcoded) */ set_shmp(handle->chan, zalloc_shm(shmobj, sizeof(struct channel))); + assert(handle->chan._ref.index == 0); + assert(handle->chan._ref.offset == 0); chan = shmp(handle, handle->chan); if (!chan) goto error_append; @@ -491,7 +498,8 @@ struct shm_handle *channel_create(const struct lib_ring_buffer_config *config, 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: @@ -502,10 +510,55 @@ error_table_alloc: return NULL; } +struct shm_handle *channel_handle_create(int shm_fd, int wait_fd, + uint64_t memory_map_size) +{ + struct shm_handle *handle; + struct shm_object *object; + + handle = zmalloc(sizeof(struct shm_handle)); + if (!handle) + return NULL; + + /* Allocate table for channel + per-cpu buffers */ + handle->table = shm_object_table_create(1 + num_possible_cpus()); + 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); + if (!object) + goto error_table_object; + /* struct channel is at object 0, offset 0 (hardcoded) */ + handle->chan._ref.index = 0; + handle->chan._ref.offset = 0; + return handle; + +error_table_object: + shm_object_table_destroy(handle->table); +error_table_alloc: + free(handle); + return NULL; +} + +int channel_handle_add_stream(struct shm_handle *handle, + int shm_fd, int wait_fd, 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); + if (!object) + return -1; + return 0; +} + static -void channel_release(struct channel *chan, struct shm_handle *handle) +void channel_release(struct channel *chan, struct shm_handle *handle, + int shadow) { - channel_free(chan, handle); + channel_free(chan, handle, shadow); } /** @@ -519,12 +572,18 @@ void channel_release(struct channel *chan, struct shm_handle *handle) * They should release their handle at that point. Returns the private * data pointer. */ -void *channel_destroy(struct channel *chan, struct shm_handle *handle) +void *channel_destroy(struct channel *chan, struct shm_handle *handle, + int shadow) { - const struct lib_ring_buffer_config *config = chan->backend.config; + const struct lib_ring_buffer_config *config = &chan->backend.config; void *priv; int cpu; + if (shadow) { + channel_release(chan, handle, shadow); + return NULL; + } + channel_unregister_notifiers(chan, handle); if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) { @@ -568,26 +627,44 @@ void *channel_destroy(struct channel *chan, struct shm_handle *handle) * descriptor directly. No need to refcount. */ priv = chan->backend.priv; - channel_release(chan, handle); + channel_release(chan, handle, shadow); return priv; } struct lib_ring_buffer *channel_get_ring_buffer( const struct lib_ring_buffer_config *config, struct channel *chan, int cpu, - struct shm_handle *handle) + struct shm_handle *handle, + int *shm_fd, int *wait_fd, + uint64_t *memory_map_size) { - if (config->alloc == RING_BUFFER_ALLOC_GLOBAL) + 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); - else + } 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); + } } int lib_ring_buffer_open_read(struct lib_ring_buffer *buf, - struct shm_handle *handle) + struct shm_handle *handle, + int shadow) { - struct channel *chan = shmp(handle, buf->backend.chan); - + if (shadow) { + if (uatomic_cmpxchg(&buf->active_shadow_readers, 0, 1) != 0) + return -EBUSY; + cmm_smp_mb(); + return 0; + } if (uatomic_cmpxchg(&buf->active_readers, 0, 1) != 0) return -EBUSY; cmm_smp_mb(); @@ -595,10 +672,17 @@ int lib_ring_buffer_open_read(struct lib_ring_buffer *buf, } void lib_ring_buffer_release_read(struct lib_ring_buffer *buf, - struct shm_handle *handle) + struct shm_handle *handle, + int shadow) { 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); @@ -619,7 +703,7 @@ int lib_ring_buffer_snapshot(struct lib_ring_buffer *buf, struct shm_handle *handle) { struct channel *chan = shmp(handle, buf->backend.chan); - const struct lib_ring_buffer_config *config = chan->backend.config; + const struct lib_ring_buffer_config *config = &chan->backend.config; unsigned long consumed_cur, write_offset; int finalized; @@ -676,7 +760,8 @@ void lib_ring_buffer_move_consumer(struct lib_ring_buffer *buf, struct channel *chan = shmp(handle, bufb->chan); unsigned long consumed; - CHAN_WARN_ON(chan, uatomic_read(&buf->active_readers) != 1); + CHAN_WARN_ON(chan, uatomic_read(&buf->active_readers) != 1 + && uatomic_read(&buf->active_shadow_readers) != 1); /* * Only push the consumed value forward. @@ -702,7 +787,7 @@ int lib_ring_buffer_get_subbuf(struct lib_ring_buffer *buf, struct shm_handle *handle) { struct channel *chan = shmp(handle, buf->backend.chan); - const struct lib_ring_buffer_config *config = chan->backend.config; + const struct lib_ring_buffer_config *config = &chan->backend.config; unsigned long consumed_cur, consumed_idx, commit_count, write_offset; int ret; int finalized; @@ -796,10 +881,11 @@ void lib_ring_buffer_put_subbuf(struct lib_ring_buffer *buf, { struct lib_ring_buffer_backend *bufb = &buf->backend; struct channel *chan = shmp(handle, bufb->chan); - const struct lib_ring_buffer_config *config = chan->backend.config; + const struct 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); + CHAN_WARN_ON(chan, uatomic_read(&buf->active_readers) != 1 + && uatomic_read(&buf->active_shadow_readers) != 1); if (!buf->get_subbuf) { /* @@ -856,7 +942,7 @@ void lib_ring_buffer_print_subbuffer_errors(struct lib_ring_buffer *buf, int cpu, struct shm_handle *handle) { - const struct lib_ring_buffer_config *config = chan->backend.config; + const struct lib_ring_buffer_config *config = &chan->backend.config; unsigned long cons_idx, commit_count, commit_count_sb; cons_idx = subbuf_index(cons_offset, chan); @@ -882,7 +968,7 @@ void lib_ring_buffer_print_buffer_errors(struct lib_ring_buffer *buf, void *priv, int cpu, struct shm_handle *handle) { - const struct lib_ring_buffer_config *config = chan->backend.config; + const struct lib_ring_buffer_config *config = &chan->backend.config; unsigned long write_offset, cons_offset; /* @@ -918,7 +1004,7 @@ void lib_ring_buffer_print_errors(struct channel *chan, struct lib_ring_buffer *buf, int cpu, struct shm_handle *handle) { - const struct lib_ring_buffer_config *config = chan->backend.config; + const struct lib_ring_buffer_config *config = &chan->backend.config; void *priv = chan->backend.priv; ERRMSG("ring buffer %s, cpu %d: %lu records written, " @@ -953,7 +1039,7 @@ void lib_ring_buffer_switch_old_start(struct lib_ring_buffer *buf, u64 tsc, struct shm_handle *handle) { - const struct lib_ring_buffer_config *config = chan->backend.config; + const struct lib_ring_buffer_config *config = &chan->backend.config; unsigned long oldidx = subbuf_index(offsets->old, chan); unsigned long commit_count; @@ -991,7 +1077,7 @@ void lib_ring_buffer_switch_old_end(struct lib_ring_buffer *buf, u64 tsc, struct shm_handle *handle) { - const struct lib_ring_buffer_config *config = chan->backend.config; + const struct lib_ring_buffer_config *config = &chan->backend.config; unsigned long oldidx = subbuf_index(offsets->old - 1, chan); unsigned long commit_count, padding_size, data_size; @@ -1028,7 +1114,7 @@ void lib_ring_buffer_switch_new_start(struct lib_ring_buffer *buf, u64 tsc, struct shm_handle *handle) { - const struct lib_ring_buffer_config *config = chan->backend.config; + const struct lib_ring_buffer_config *config = &chan->backend.config; unsigned long beginidx = subbuf_index(offsets->begin, chan); unsigned long commit_count; @@ -1064,7 +1150,7 @@ void lib_ring_buffer_switch_new_end(struct lib_ring_buffer *buf, u64 tsc, struct shm_handle *handle) { - const struct lib_ring_buffer_config *config = chan->backend.config; + const struct lib_ring_buffer_config *config = &chan->backend.config; unsigned long endidx = subbuf_index(offsets->end - 1, chan); unsigned long commit_count, padding_size, data_size; @@ -1099,7 +1185,7 @@ int lib_ring_buffer_try_switch_slow(enum switch_mode mode, struct switch_offsets *offsets, u64 *tsc) { - const struct lib_ring_buffer_config *config = chan->backend.config; + const struct lib_ring_buffer_config *config = &chan->backend.config; unsigned long off; offsets->begin = v_read(config, &buf->offset); @@ -1159,7 +1245,7 @@ void lib_ring_buffer_switch_slow(struct lib_ring_buffer *buf, enum switch_mode m struct shm_handle *handle) { struct channel *chan = shmp(handle, buf->backend.chan); - const struct lib_ring_buffer_config *config = chan->backend.config; + const struct lib_ring_buffer_config *config = &chan->backend.config; struct switch_offsets offsets; unsigned long oldidx; u64 tsc; @@ -1219,7 +1305,7 @@ int lib_ring_buffer_try_reserve_slow(struct lib_ring_buffer *buf, struct switch_offsets *offsets, struct lib_ring_buffer_ctx *ctx) { - const struct lib_ring_buffer_config *config = chan->backend.config; + const struct lib_ring_buffer_config *config = &chan->backend.config; struct shm_handle *handle = ctx->handle; unsigned long reserve_commit_diff; @@ -1356,7 +1442,7 @@ int lib_ring_buffer_reserve_slow(struct lib_ring_buffer_ctx *ctx) { struct channel *chan = ctx->chan; struct shm_handle *handle = ctx->handle; - const struct lib_ring_buffer_config *config = chan->backend.config; + const struct lib_ring_buffer_config *config = &chan->backend.config; struct lib_ring_buffer *buf; struct switch_offsets offsets; int ret;