X-Git-Url: http://git.liburcu.org/?a=blobdiff_plain;f=libringbuffer%2Fbackend_internal.h;h=7bd5c738936084a0ed84fec4fa822a44211812f7;hb=9c995331b51a9b9a8c85457a897fa89e770de111;hp=788cdfb14fe378a6e3277a24c554e205e7eb935e;hpb=38fae1d354502ffb37f471fbd0fef03412606cc0;p=lttng-ust.git diff --git a/libringbuffer/backend_internal.h b/libringbuffer/backend_internal.h index 788cdfb1..7bd5c738 100644 --- a/libringbuffer/backend_internal.h +++ b/libringbuffer/backend_internal.h @@ -1,20 +1,32 @@ -#ifndef _LINUX_RING_BUFFER_BACKEND_INTERNAL_H -#define _LINUX_RING_BUFFER_BACKEND_INTERNAL_H +#ifndef _LTTNG_RING_BUFFER_BACKEND_INTERNAL_H +#define _LTTNG_RING_BUFFER_BACKEND_INTERNAL_H /* - * linux/ringbuffer/backend_internal.h - * - * Copyright (C) 2008-2010 - Mathieu Desnoyers + * libringbuffer/backend_internal.h * * Ring buffer backend (internal helpers). * - * Dual LGPL v2.1/GPL v2 license. + * 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 */ #include #include -#include +#include #include "backend_types.h" #include "frontend_types.h" #include "shm.h" @@ -23,28 +35,29 @@ /* Ring buffer and channel backend create/free */ -int lib_ring_buffer_backend_create(struct lib_ring_buffer_backend *bufb, +int lib_ring_buffer_backend_create(struct lttng_ust_lib_ring_buffer_backend *bufb, struct channel_backend *chan, int cpu, struct lttng_ust_shm_handle *handle, struct shm_object *shmobj); void channel_backend_unregister_notifiers(struct channel_backend *chanb); -void lib_ring_buffer_backend_free(struct lib_ring_buffer_backend *bufb); +void lib_ring_buffer_backend_free(struct lttng_ust_lib_ring_buffer_backend *bufb); int channel_backend_init(struct channel_backend *chanb, const char *name, - const struct lib_ring_buffer_config *config, - void *priv, size_t subbuf_size, - size_t num_subbuf, struct lttng_ust_shm_handle *handle); + const struct lttng_ust_lib_ring_buffer_config *config, + size_t subbuf_size, + size_t num_subbuf, struct lttng_ust_shm_handle *handle, + const int *stream_fds); void channel_backend_free(struct channel_backend *chanb, struct lttng_ust_shm_handle *handle); -void lib_ring_buffer_backend_reset(struct lib_ring_buffer_backend *bufb, +void lib_ring_buffer_backend_reset(struct lttng_ust_lib_ring_buffer_backend *bufb, struct lttng_ust_shm_handle *handle); void channel_backend_reset(struct channel_backend *chanb); int lib_ring_buffer_backend_init(void); void lib_ring_buffer_backend_exit(void); -extern void _lib_ring_buffer_write(struct lib_ring_buffer_backend *bufb, +extern void _lib_ring_buffer_write(struct lttng_ust_lib_ring_buffer_backend *bufb, size_t offset, const void *src, size_t len, ssize_t pagecpy); @@ -86,7 +99,7 @@ extern void _lib_ring_buffer_write(struct lib_ring_buffer_backend *bufb, * mode). */ static inline -unsigned long subbuffer_id(const struct lib_ring_buffer_config *config, +unsigned long subbuffer_id(const struct lttng_ust_lib_ring_buffer_config *config, unsigned long offset, unsigned long noref, unsigned long index) { @@ -103,14 +116,14 @@ unsigned long subbuffer_id(const struct lib_ring_buffer_config *config, * bits are identical, else 0. */ static inline -int subbuffer_id_compare_offset(const struct lib_ring_buffer_config *config, +int subbuffer_id_compare_offset(const struct lttng_ust_lib_ring_buffer_config *config, unsigned long id, unsigned long offset) { return (id & SB_ID_OFFSET_MASK) == (offset << SB_ID_OFFSET_SHIFT); } static inline -unsigned long subbuffer_id_get_index(const struct lib_ring_buffer_config *config, +unsigned long subbuffer_id_get_index(const struct lttng_ust_lib_ring_buffer_config *config, unsigned long id) { if (config->mode == RING_BUFFER_OVERWRITE) @@ -120,7 +133,7 @@ unsigned long subbuffer_id_get_index(const struct lib_ring_buffer_config *config } static inline -unsigned long subbuffer_id_is_noref(const struct lib_ring_buffer_config *config, +unsigned long subbuffer_id_is_noref(const struct lttng_ust_lib_ring_buffer_config *config, unsigned long id) { if (config->mode == RING_BUFFER_OVERWRITE) @@ -134,7 +147,7 @@ unsigned long subbuffer_id_is_noref(const struct lib_ring_buffer_config *config, * needed. */ static inline -void subbuffer_id_set_noref(const struct lib_ring_buffer_config *config, +void subbuffer_id_set_noref(const struct lttng_ust_lib_ring_buffer_config *config, unsigned long *id) { if (config->mode == RING_BUFFER_OVERWRITE) @@ -142,7 +155,7 @@ void subbuffer_id_set_noref(const struct lib_ring_buffer_config *config, } static inline -void subbuffer_id_set_noref_offset(const struct lib_ring_buffer_config *config, +void subbuffer_id_set_noref_offset(const struct lttng_ust_lib_ring_buffer_config *config, unsigned long *id, unsigned long offset) { unsigned long tmp; @@ -159,7 +172,7 @@ void subbuffer_id_set_noref_offset(const struct lib_ring_buffer_config *config, /* No volatile access, since already used locally */ static inline -void subbuffer_id_clear_noref(const struct lib_ring_buffer_config *config, +void subbuffer_id_clear_noref(const struct lttng_ust_lib_ring_buffer_config *config, unsigned long *id) { if (config->mode == RING_BUFFER_OVERWRITE) @@ -174,7 +187,7 @@ void subbuffer_id_clear_noref(const struct lib_ring_buffer_config *config, * -EPERM on failure. */ static inline -int subbuffer_id_check_index(const struct lib_ring_buffer_config *config, +int subbuffer_id_check_index(const struct lttng_ust_lib_ring_buffer_config *config, unsigned long num_subbuf) { if (config->mode == RING_BUFFER_OVERWRITE) @@ -183,9 +196,14 @@ int subbuffer_id_check_index(const struct lib_ring_buffer_config *config, return 0; } +/* + * The ring buffer can count events recorded and overwritten per buffer, + * but it is disabled by default due to its performance overhead. + */ +#ifdef LTTNG_RING_BUFFER_COUNT_EVENTS static inline -void subbuffer_count_record(const struct lib_ring_buffer_config *config, - struct lib_ring_buffer_backend *bufb, +void subbuffer_count_record(const struct lttng_ust_lib_ring_buffer_config *config, + struct lttng_ust_lib_ring_buffer_backend *bufb, unsigned long idx, struct lttng_ust_shm_handle *handle) { unsigned long sb_bindex; @@ -193,14 +211,22 @@ void subbuffer_count_record(const struct lib_ring_buffer_config *config, sb_bindex = subbuffer_id_get_index(config, shmp_index(handle, bufb->buf_wsb, idx)->id); v_inc(config, &shmp(handle, shmp_index(handle, bufb->array, sb_bindex)->shmp)->records_commit); } +#else /* LTTNG_RING_BUFFER_COUNT_EVENTS */ +static inline +void subbuffer_count_record(const struct lttng_ust_lib_ring_buffer_config *config, + struct lttng_ust_lib_ring_buffer_backend *bufb, + unsigned long idx, struct lttng_ust_shm_handle *handle) +{ +} +#endif /* #else LTTNG_RING_BUFFER_COUNT_EVENTS */ /* * Reader has exclusive subbuffer access for record consumption. No need to * perform the decrement atomically. */ static inline -void subbuffer_consume_record(const struct lib_ring_buffer_config *config, - struct lib_ring_buffer_backend *bufb, +void subbuffer_consume_record(const struct lttng_ust_lib_ring_buffer_config *config, + struct lttng_ust_lib_ring_buffer_backend *bufb, struct lttng_ust_shm_handle *handle) { unsigned long sb_bindex; @@ -215,8 +241,8 @@ void subbuffer_consume_record(const struct lib_ring_buffer_config *config, static inline unsigned long subbuffer_get_records_count( - const struct lib_ring_buffer_config *config, - struct lib_ring_buffer_backend *bufb, + const struct lttng_ust_lib_ring_buffer_config *config, + struct lttng_ust_lib_ring_buffer_backend *bufb, unsigned long idx, struct lttng_ust_shm_handle *handle) { @@ -228,18 +254,19 @@ unsigned long subbuffer_get_records_count( /* * Must be executed at subbuffer delivery when the writer has _exclusive_ - * subbuffer access. See ring_buffer_check_deliver() for details. - * ring_buffer_get_records_count() must be called to get the records count - * before this function, because it resets the records_commit count. + * subbuffer access. See lib_ring_buffer_check_deliver() for details. + * lib_ring_buffer_get_records_count() must be called to get the records + * count before this function, because it resets the records_commit + * count. */ static inline unsigned long subbuffer_count_records_overrun( - const struct lib_ring_buffer_config *config, - struct lib_ring_buffer_backend *bufb, + const struct lttng_ust_lib_ring_buffer_config *config, + struct lttng_ust_lib_ring_buffer_backend *bufb, unsigned long idx, struct lttng_ust_shm_handle *handle) { - struct lib_ring_buffer_backend_pages_shmp *pages; + struct lttng_ust_lib_ring_buffer_backend_pages_shmp *pages; unsigned long overruns, sb_bindex; sb_bindex = subbuffer_id_get_index(config, shmp_index(handle, bufb->buf_wsb, idx)->id); @@ -253,13 +280,13 @@ unsigned long subbuffer_count_records_overrun( } static inline -void subbuffer_set_data_size(const struct lib_ring_buffer_config *config, - struct lib_ring_buffer_backend *bufb, +void subbuffer_set_data_size(const struct lttng_ust_lib_ring_buffer_config *config, + struct lttng_ust_lib_ring_buffer_backend *bufb, unsigned long idx, unsigned long data_size, struct lttng_ust_shm_handle *handle) { - struct lib_ring_buffer_backend_pages_shmp *pages; + struct lttng_ust_lib_ring_buffer_backend_pages_shmp *pages; unsigned long sb_bindex; sb_bindex = subbuffer_id_get_index(config, shmp_index(handle, bufb->buf_wsb, idx)->id); @@ -269,11 +296,11 @@ void subbuffer_set_data_size(const struct lib_ring_buffer_config *config, static inline unsigned long subbuffer_get_read_data_size( - const struct lib_ring_buffer_config *config, - struct lib_ring_buffer_backend *bufb, + const struct lttng_ust_lib_ring_buffer_config *config, + struct lttng_ust_lib_ring_buffer_backend *bufb, struct lttng_ust_shm_handle *handle) { - struct lib_ring_buffer_backend_pages_shmp *pages; + struct lttng_ust_lib_ring_buffer_backend_pages_shmp *pages; unsigned long sb_bindex; sb_bindex = subbuffer_id_get_index(config, bufb->buf_rsb.id); @@ -283,12 +310,12 @@ unsigned long subbuffer_get_read_data_size( static inline unsigned long subbuffer_get_data_size( - const struct lib_ring_buffer_config *config, - struct lib_ring_buffer_backend *bufb, + const struct lttng_ust_lib_ring_buffer_config *config, + struct lttng_ust_lib_ring_buffer_backend *bufb, unsigned long idx, struct lttng_ust_shm_handle *handle) { - struct lib_ring_buffer_backend_pages_shmp *pages; + struct lttng_ust_lib_ring_buffer_backend_pages_shmp *pages; unsigned long sb_bindex; sb_bindex = subbuffer_id_get_index(config, shmp_index(handle, bufb->buf_wsb, idx)->id); @@ -296,13 +323,21 @@ unsigned long subbuffer_get_data_size( return shmp(handle, pages->shmp)->data_size; } +static inline +void subbuffer_inc_packet_count(const struct lttng_ust_lib_ring_buffer_config *config, + struct lttng_ust_lib_ring_buffer_backend *bufb, + unsigned long idx, struct lttng_ust_shm_handle *handle) +{ + shmp_index(handle, bufb->buf_cnt, idx)->seq_cnt++; +} + /** * lib_ring_buffer_clear_noref - Clear the noref subbuffer flag, called by * writer. */ static inline -void lib_ring_buffer_clear_noref(const struct lib_ring_buffer_config *config, - struct lib_ring_buffer_backend *bufb, +void lib_ring_buffer_clear_noref(const struct lttng_ust_lib_ring_buffer_config *config, + struct lttng_ust_lib_ring_buffer_backend *bufb, unsigned long idx, struct lttng_ust_shm_handle *handle) { @@ -318,7 +353,7 @@ void lib_ring_buffer_clear_noref(const struct lib_ring_buffer_config *config, id = CMM_ACCESS_ONCE(shmp_index(handle, bufb->buf_wsb, idx)->id); for (;;) { /* This check is called on the fast path for each record. */ - if (likely(!subbuffer_id_is_noref(config, id))) { + if (caa_likely(!subbuffer_id_is_noref(config, id))) { /* * Store after load dependency ordering the writes to * the subbuffer after load and test of the noref flag @@ -330,7 +365,7 @@ void lib_ring_buffer_clear_noref(const struct lib_ring_buffer_config *config, new_id = id; subbuffer_id_clear_noref(config, &new_id); new_id = uatomic_cmpxchg(&shmp_index(handle, bufb->buf_wsb, idx)->id, id, new_id); - if (likely(new_id == id)) + if (caa_likely(new_id == id)) break; id = new_id; } @@ -341,8 +376,8 @@ void lib_ring_buffer_clear_noref(const struct lib_ring_buffer_config *config, * called by writer. */ static inline -void lib_ring_buffer_set_noref_offset(const struct lib_ring_buffer_config *config, - struct lib_ring_buffer_backend *bufb, +void lib_ring_buffer_set_noref_offset(const struct lttng_ust_lib_ring_buffer_config *config, + struct lttng_ust_lib_ring_buffer_backend *bufb, unsigned long idx, unsigned long offset, struct lttng_ust_shm_handle *handle) { @@ -374,8 +409,8 @@ void lib_ring_buffer_set_noref_offset(const struct lib_ring_buffer_config *confi * update_read_sb_index - Read-side subbuffer index update. */ static inline -int update_read_sb_index(const struct lib_ring_buffer_config *config, - struct lib_ring_buffer_backend *bufb, +int update_read_sb_index(const struct lttng_ust_lib_ring_buffer_config *config, + struct lttng_ust_lib_ring_buffer_backend *bufb, struct channel_backend *chanb, unsigned long consumed_idx, unsigned long consumed_count, @@ -391,13 +426,13 @@ int update_read_sb_index(const struct lib_ring_buffer_config *config, * following cmpxchg(). */ old_id = shmp_index(handle, bufb->buf_wsb, consumed_idx)->id; - if (unlikely(!subbuffer_id_is_noref(config, old_id))) + if (caa_unlikely(!subbuffer_id_is_noref(config, old_id))) return -EAGAIN; /* * Make sure the offset count we are expecting matches the one * indicated by the writer. */ - if (unlikely(!subbuffer_id_compare_offset(config, old_id, + if (caa_unlikely(!subbuffer_id_compare_offset(config, old_id, consumed_count))) return -EAGAIN; CHAN_WARN_ON(shmp(handle, bufb->chan), @@ -406,7 +441,7 @@ int update_read_sb_index(const struct lib_ring_buffer_config *config, consumed_count); new_id = uatomic_cmpxchg(&shmp_index(handle, bufb->buf_wsb, consumed_idx)->id, old_id, bufb->buf_rsb.id); - if (unlikely(old_id != new_id)) + if (caa_unlikely(old_id != new_id)) return -EAGAIN; bufb->buf_rsb.id = new_id; } else { @@ -416,6 +451,10 @@ int update_read_sb_index(const struct lib_ring_buffer_config *config, return 0; } +#ifndef inline_memcpy +#define inline_memcpy(dest, src, n) memcpy(dest, src, n) +#endif + /* * Use the architecture-specific memcpy implementation for constant-sized * inputs, but rely on an inline memcpy for length statically unknown. @@ -430,4 +469,57 @@ do { \ inline_memcpy(dest, src, __len); \ } while (0) -#endif /* _LINUX_RING_BUFFER_BACKEND_INTERNAL_H */ +/* + * write len bytes to dest with c + */ +static inline +void lib_ring_buffer_do_memset(char *dest, int c, unsigned long len) +{ + unsigned long i; + + for (i = 0; i < len; i++) + dest[i] = c; +} + +/* arch-agnostic implementation */ + +static inline int lttng_ust_fls(unsigned int x) +{ + int r = 32; + + if (!x) + return 0; + if (!(x & 0xFFFF0000U)) { + x <<= 16; + r -= 16; + } + if (!(x & 0xFF000000U)) { + x <<= 8; + r -= 8; + } + if (!(x & 0xF0000000U)) { + x <<= 4; + r -= 4; + } + if (!(x & 0xC0000000U)) { + x <<= 2; + r -= 2; + } + if (!(x & 0x80000000U)) { + /* No need to bit shift on last operation */ + r -= 1; + } + return r; +} + +static inline int get_count_order(unsigned int count) +{ + int order; + + order = lttng_ust_fls(count) - 1; + if (count & (count - 1)) + order++; + return order; +} + +#endif /* _LTTNG_RING_BUFFER_BACKEND_INTERNAL_H */