X-Git-Url: http://git.liburcu.org/?a=blobdiff_plain;f=lib%2Fringbuffer%2Fring_buffer_frontend.c;h=8ffb702e027a692ee88f6ab7581aafcff85d1f03;hb=5760f3f4e64af16249a4f208e680c7dcffb4f990;hp=0d8279bb293d07331b36848f54e4315e063cf7e7;hpb=a8f2d0c75c9cc179fc9e7f7ca17ea3b3b3b5af41;p=lttng-modules.git diff --git a/lib/ringbuffer/ring_buffer_frontend.c b/lib/ringbuffer/ring_buffer_frontend.c index 0d8279bb..8ffb702e 100644 --- a/lib/ringbuffer/ring_buffer_frontend.c +++ b/lib/ringbuffer/ring_buffer_frontend.c @@ -1,23 +1,9 @@ -/* +/* SPDX-License-Identifier: (GPL-2.0-only OR LGPL-2.1-only) + * * ring_buffer_frontend.c * * 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: * @@ -54,18 +40,16 @@ #include #include #include +#include +#include +#include #include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include +#include +#include +#include +#include +#include /* * Internal structure representing offsets to use at a sub-buffer switch. @@ -148,8 +132,9 @@ void lib_ring_buffer_free(struct lib_ring_buffer *buf) struct channel *chan = buf->backend.chan; lib_ring_buffer_print_errors(chan, buf, buf->backend.cpu); - lttng_kvfree(buf->commit_hot); - lttng_kvfree(buf->commit_cold); + kvfree(buf->commit_hot); + kvfree(buf->commit_cold); + kvfree(buf->ts_end); lib_ring_buffer_backend_free(&buf->backend); } @@ -179,6 +164,7 @@ void lib_ring_buffer_reset(struct lib_ring_buffer *buf) v_set(config, &buf->commit_hot[i].cc, 0); v_set(config, &buf->commit_hot[i].seq, 0); v_set(config, &buf->commit_cold[i].cc_sb, 0); + buf->ts_end[i] = 0; } atomic_long_set(&buf->consumed, 0); atomic_set(&buf->record_disabled, 0); @@ -246,7 +232,7 @@ int lib_ring_buffer_create(struct lib_ring_buffer *buf, return ret; buf->commit_hot = - lttng_kvzalloc_node(ALIGN(sizeof(*buf->commit_hot) + kvzalloc_node(ALIGN(sizeof(*buf->commit_hot) * chan->backend.num_subbuf, 1 << INTERNODE_CACHE_SHIFT), GFP_KERNEL | __GFP_NOWARN, @@ -257,7 +243,7 @@ int lib_ring_buffer_create(struct lib_ring_buffer *buf, } buf->commit_cold = - lttng_kvzalloc_node(ALIGN(sizeof(*buf->commit_cold) + kvzalloc_node(ALIGN(sizeof(*buf->commit_cold) * chan->backend.num_subbuf, 1 << INTERNODE_CACHE_SHIFT), GFP_KERNEL | __GFP_NOWARN, @@ -267,6 +253,17 @@ int lib_ring_buffer_create(struct lib_ring_buffer *buf, goto free_commit; } + buf->ts_end = + kvzalloc_node(ALIGN(sizeof(*buf->ts_end) + * chan->backend.num_subbuf, + 1 << INTERNODE_CACHE_SHIFT), + GFP_KERNEL | __GFP_NOWARN, + cpu_to_node(max(cpu, 0))); + if (!buf->ts_end) { + ret = -ENOMEM; + goto free_commit_cold; + } + init_waitqueue_head(&buf->read_wait); init_waitqueue_head(&buf->write_wait); raw_spin_lock_init(&buf->raw_tick_nohz_spinlock); @@ -301,24 +298,24 @@ int lib_ring_buffer_create(struct lib_ring_buffer *buf, chan->backend.cpumask)); cpumask_set_cpu(cpu, chan->backend.cpumask); } - return 0; /* Error handling */ free_init: - lttng_kvfree(buf->commit_cold); + kvfree(buf->ts_end); +free_commit_cold: + kvfree(buf->commit_cold); free_commit: - lttng_kvfree(buf->commit_hot); + kvfree(buf->commit_hot); free_chanbuf: lib_ring_buffer_backend_free(&buf->backend); return ret; } -static void switch_buffer_timer(LTTNG_TIMER_FUNC_ARG_TYPE t) +static void switch_buffer_timer(struct timer_list *t) { - struct lib_ring_buffer *buf = lttng_from_timer(buf, t, switch_timer); + struct lib_ring_buffer *buf = from_timer(buf, t, switch_timer); struct channel *chan = buf->backend.chan; - const struct lib_ring_buffer_config *config = &chan->backend.config; /* * Only flush buffers periodically if readers are active. @@ -326,12 +323,8 @@ static void switch_buffer_timer(LTTNG_TIMER_FUNC_ARG_TYPE t) if (atomic_long_read(&buf->active_readers)) lib_ring_buffer_switch_slow(buf, SWITCH_ACTIVE); - if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) - lttng_mod_timer_pinned(&buf->switch_timer, - jiffies + chan->switch_timer_interval); - else - mod_timer(&buf->switch_timer, - jiffies + chan->switch_timer_interval); + mod_timer(&buf->switch_timer, + jiffies + chan->switch_timer_interval); } /* @@ -347,9 +340,9 @@ static void lib_ring_buffer_start_switch_timer(struct lib_ring_buffer *buf) return; if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) - flags = LTTNG_TIMER_PINNED; + flags = TIMER_PINNED; - lttng_timer_setup(&buf->switch_timer, switch_buffer_timer, flags, buf); + timer_setup(&buf->switch_timer, switch_buffer_timer, flags); buf->switch_timer.expires = jiffies + chan->switch_timer_interval; if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) @@ -377,9 +370,9 @@ static void lib_ring_buffer_stop_switch_timer(struct lib_ring_buffer *buf) /* * Polling timer to check the channels for data. */ -static void read_buffer_timer(LTTNG_TIMER_FUNC_ARG_TYPE t) +static void read_buffer_timer(struct timer_list *t) { - struct lib_ring_buffer *buf = lttng_from_timer(buf, t, read_timer); + struct lib_ring_buffer *buf = from_timer(buf, t, read_timer); struct channel *chan = buf->backend.chan; const struct lib_ring_buffer_config *config = &chan->backend.config; @@ -391,12 +384,8 @@ static void read_buffer_timer(LTTNG_TIMER_FUNC_ARG_TYPE t) wake_up_interruptible(&chan->read_wait); } - if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) - lttng_mod_timer_pinned(&buf->read_timer, - jiffies + chan->read_timer_interval); - else - mod_timer(&buf->read_timer, - jiffies + chan->read_timer_interval); + mod_timer(&buf->read_timer, + jiffies + chan->read_timer_interval); } /* @@ -406,7 +395,7 @@ static void lib_ring_buffer_start_read_timer(struct lib_ring_buffer *buf) { struct channel *chan = buf->backend.chan; const struct lib_ring_buffer_config *config = &chan->backend.config; - unsigned int flags; + unsigned int flags = 0; if (config->wakeup != RING_BUFFER_WAKEUP_BY_TIMER || !chan->read_timer_interval @@ -414,9 +403,9 @@ static void lib_ring_buffer_start_read_timer(struct lib_ring_buffer *buf) return; if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) - flags = LTTNG_TIMER_PINNED; + flags = TIMER_PINNED; - lttng_timer_setup(&buf->read_timer, read_buffer_timer, flags, buf); + timer_setup(&buf->read_timer, read_buffer_timer, flags); buf->read_timer.expires = jiffies + chan->read_timer_interval; if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) @@ -452,8 +441,6 @@ static void lib_ring_buffer_stop_read_timer(struct lib_ring_buffer *buf) buf->read_timer_enabled = 0; } -#if (LINUX_VERSION_CODE >= KERNEL_VERSION(4,10,0)) - enum cpuhp_state lttng_rb_hp_prepare; enum cpuhp_state lttng_rb_hp_online; @@ -523,70 +510,6 @@ int lttng_cpuhp_rb_frontend_offline(unsigned int cpu, } EXPORT_SYMBOL_GPL(lttng_cpuhp_rb_frontend_offline); -#else /* #if (LINUX_VERSION_CODE >= KERNEL_VERSION(4,10,0)) */ - -#ifdef CONFIG_HOTPLUG_CPU - -/** - * lib_ring_buffer_cpu_hp_callback - CPU hotplug callback - * @nb: notifier block - * @action: hotplug action to take - * @hcpu: CPU number - * - * Returns the success/failure of the operation. (%NOTIFY_OK, %NOTIFY_BAD) - */ -static -int lib_ring_buffer_cpu_hp_callback(struct notifier_block *nb, - unsigned long action, - void *hcpu) -{ - unsigned int cpu = (unsigned long)hcpu; - struct channel *chan = container_of(nb, struct channel, - cpu_hp_notifier); - struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, cpu); - const struct lib_ring_buffer_config *config = &chan->backend.config; - - if (!chan->cpu_hp_enable) - return NOTIFY_DONE; - - CHAN_WARN_ON(chan, config->alloc == RING_BUFFER_ALLOC_GLOBAL); - - switch (action) { - case CPU_DOWN_FAILED: - case CPU_DOWN_FAILED_FROZEN: - case CPU_ONLINE: - case CPU_ONLINE_FROZEN: - wake_up_interruptible(&chan->hp_wait); - lib_ring_buffer_start_switch_timer(buf); - lib_ring_buffer_start_read_timer(buf); - return NOTIFY_OK; - - case CPU_DOWN_PREPARE: - case CPU_DOWN_PREPARE_FROZEN: - lib_ring_buffer_stop_switch_timer(buf); - lib_ring_buffer_stop_read_timer(buf); - return NOTIFY_OK; - - case CPU_DEAD: - case CPU_DEAD_FROZEN: - /* - * Performing a buffer switch on a remote CPU. Performed by - * the CPU responsible for doing the hotunplug after the target - * CPU stopped running completely. Ensures that all data - * from that remote CPU is flushed. - */ - lib_ring_buffer_switch_slow(buf, SWITCH_ACTIVE); - return NOTIFY_OK; - - default: - return NOTIFY_DONE; - } -} - -#endif - -#endif /* #else #if (LINUX_VERSION_CODE >= KERNEL_VERSION(4,10,0)) */ - #if defined(CONFIG_NO_HZ) && defined(CONFIG_LIB_RING_BUFFER) /* * For per-cpu buffers, call the reader wakeups before switching the buffer, so @@ -634,16 +557,16 @@ static int notrace ring_buffer_tick_nohz_callback(struct notifier_block *nb, raw_spin_unlock(&buf->raw_tick_nohz_spinlock); break; case TICK_NOHZ_STOP: - spin_lock(lttng_this_cpu_ptr(&ring_buffer_nohz_lock)); + spin_lock(this_cpu_ptr(&ring_buffer_nohz_lock)); lib_ring_buffer_stop_switch_timer(buf); lib_ring_buffer_stop_read_timer(buf); - spin_unlock(lttng_this_cpu_ptr(&ring_buffer_nohz_lock)); + spin_unlock(this_cpu_ptr(&ring_buffer_nohz_lock)); break; case TICK_NOHZ_RESTART: - spin_lock(lttng_this_cpu_ptr(&ring_buffer_nohz_lock)); + spin_lock(this_cpu_ptr(&ring_buffer_nohz_lock)); lib_ring_buffer_start_read_timer(buf); lib_ring_buffer_start_switch_timer(buf); - spin_unlock(lttng_this_cpu_ptr(&ring_buffer_nohz_lock)); + spin_unlock(this_cpu_ptr(&ring_buffer_nohz_lock)); break; } @@ -691,7 +614,6 @@ static void channel_unregister_notifiers(struct channel *chan) * concurrency. */ #endif /* CONFIG_NO_HZ */ -#if (LINUX_VERSION_CODE >= KERNEL_VERSION(4,10,0)) { int ret; @@ -702,31 +624,6 @@ static void channel_unregister_notifiers(struct channel *chan) &chan->cpuhp_prepare.node); WARN_ON(ret); } -#else /* #if (LINUX_VERSION_CODE >= KERNEL_VERSION(4,10,0)) */ - { - int cpu; - -#ifdef CONFIG_HOTPLUG_CPU - get_online_cpus(); - chan->cpu_hp_enable = 0; - for_each_online_cpu(cpu) { - struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, - cpu); - lib_ring_buffer_stop_switch_timer(buf); - lib_ring_buffer_stop_read_timer(buf); - } - put_online_cpus(); - unregister_cpu_notifier(&chan->cpu_hp_notifier); -#else - for_each_possible_cpu(cpu) { - struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, - cpu); - lib_ring_buffer_stop_switch_timer(buf); - lib_ring_buffer_stop_read_timer(buf); - } -#endif - } -#endif /* #else #if (LINUX_VERSION_CODE >= KERNEL_VERSION(4,10,0)) */ } else { struct lib_ring_buffer *buf = chan->backend.buf; @@ -855,7 +752,6 @@ struct channel *channel_create(const struct lib_ring_buffer_config *config, init_waitqueue_head(&chan->hp_wait); if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) { -#if (LINUX_VERSION_CODE >= KERNEL_VERSION(4,10,0)) chan->cpuhp_prepare.component = LTTNG_RING_BUFFER_FRONTEND; ret = cpuhp_state_add_instance_nocalls(lttng_rb_hp_prepare, &chan->cpuhp_prepare.node); @@ -867,43 +763,6 @@ struct channel *channel_create(const struct lib_ring_buffer_config *config, &chan->cpuhp_online.node); if (ret) goto cpuhp_online_error; -#else /* #if (LINUX_VERSION_CODE >= KERNEL_VERSION(4,10,0)) */ - { - int cpu; - /* - * In case of non-hotplug cpu, if the ring-buffer is allocated - * in early initcall, it will not be notified of secondary cpus. - * In that off case, we need to allocate for all possible cpus. - */ -#ifdef CONFIG_HOTPLUG_CPU - chan->cpu_hp_notifier.notifier_call = - lib_ring_buffer_cpu_hp_callback; - chan->cpu_hp_notifier.priority = 6; - register_cpu_notifier(&chan->cpu_hp_notifier); - - get_online_cpus(); - for_each_online_cpu(cpu) { - struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, - cpu); - spin_lock(&per_cpu(ring_buffer_nohz_lock, cpu)); - lib_ring_buffer_start_switch_timer(buf); - lib_ring_buffer_start_read_timer(buf); - spin_unlock(&per_cpu(ring_buffer_nohz_lock, cpu)); - } - chan->cpu_hp_enable = 1; - put_online_cpus(); -#else - for_each_possible_cpu(cpu) { - struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, - cpu); - spin_lock(&per_cpu(ring_buffer_nohz_lock, cpu)); - lib_ring_buffer_start_switch_timer(buf); - lib_ring_buffer_start_read_timer(buf); - spin_unlock(&per_cpu(ring_buffer_nohz_lock, cpu)); - } -#endif - } -#endif /* #else #if (LINUX_VERSION_CODE >= KERNEL_VERSION(4,10,0)) */ #if defined(CONFIG_NO_HZ) && defined(CONFIG_LIB_RING_BUFFER) /* Only benefit from NO_HZ idle with per-cpu buffers for now. */ @@ -923,13 +782,11 @@ struct channel *channel_create(const struct lib_ring_buffer_config *config, return chan; -#if (LINUX_VERSION_CODE >= KERNEL_VERSION(4,10,0)) cpuhp_online_error: ret = cpuhp_state_remove_instance_nocalls(lttng_rb_hp_prepare, &chan->cpuhp_prepare.node); WARN_ON(ret); cpuhp_prepare_error: -#endif /* #if (LINUX_VERSION_CODE >= KERNEL_VERSION(4,10,0)) */ error_free_backend: channel_backend_free(&chan->backend); error: @@ -977,8 +834,6 @@ void *channel_destroy(struct channel *chan) config->cb.buffer_finalize(buf, chan->backend.priv, cpu); - if (buf->backend.allocated) - lib_ring_buffer_set_quiescent(buf); /* * Perform flush before writing to finalized. */ @@ -991,8 +846,6 @@ void *channel_destroy(struct channel *chan) if (config->cb.buffer_finalize) config->cb.buffer_finalize(buf, chan->backend.priv, -1); - if (buf->backend.allocated) - lib_ring_buffer_set_quiescent(buf); /* * Perform flush before writing to finalized. */ @@ -1026,11 +879,8 @@ int lib_ring_buffer_open_read(struct lib_ring_buffer *buf) if (!atomic_long_add_unless(&buf->active_readers, 1, 1)) return -EBUSY; - if (!lttng_kref_get(&chan->ref)) { - atomic_long_dec(&buf->active_readers); - return -EOVERFLOW; - } - lttng_smp_mb__after_atomic(); + kref_get(&chan->ref); + smp_mb__after_atomic(); return 0; } EXPORT_SYMBOL_GPL(lib_ring_buffer_open_read); @@ -1040,7 +890,7 @@ void lib_ring_buffer_release_read(struct lib_ring_buffer *buf) struct channel *chan = buf->backend.chan; CHAN_WARN_ON(chan, atomic_long_read(&buf->active_readers) != 1); - lttng_smp_mb__before_atomic(); + smp_mb__before_atomic(); atomic_long_dec(&buf->active_readers); kref_put(&chan->ref, channel_release); } @@ -1501,12 +1351,13 @@ void lib_ring_buffer_print_buffer_errors(struct lib_ring_buffer *buf, cpu); } +#ifdef LTTNG_RING_BUFFER_COUNT_EVENTS static -void lib_ring_buffer_print_errors(struct channel *chan, - struct lib_ring_buffer *buf, int cpu) +void lib_ring_buffer_print_records_count(struct channel *chan, + struct lib_ring_buffer *buf, + int cpu) { const struct lib_ring_buffer_config *config = &chan->backend.config; - void *priv = chan->backend.priv; if (!strcmp(chan->backend.name, "relay-metadata")) { printk(KERN_DEBUG "ring buffer %s: %lu records written, " @@ -1520,7 +1371,26 @@ void lib_ring_buffer_print_errors(struct channel *chan, chan->backend.name, cpu, v_read(config, &buf->records_count), v_read(config, &buf->records_overrun)); + } +} +#else +static +void lib_ring_buffer_print_records_count(struct channel *chan, + struct lib_ring_buffer *buf, + int cpu) +{ +} +#endif +static +void lib_ring_buffer_print_errors(struct channel *chan, + struct lib_ring_buffer *buf, int cpu) +{ + const struct lib_ring_buffer_config *config = &chan->backend.config; + void *priv = chan->backend.priv; + + lib_ring_buffer_print_records_count(chan, buf, cpu); + if (strcmp(chan->backend.name, "relay-metadata")) { if (v_read(config, &buf->records_lost_full) || v_read(config, &buf->records_lost_wrap) || v_read(config, &buf->records_lost_big)) @@ -1596,14 +1466,26 @@ void lib_ring_buffer_switch_old_end(struct lib_ring_buffer *buf, unsigned long oldidx = subbuf_index(offsets->old - 1, chan); unsigned long commit_count, padding_size, data_size; struct commit_counters_hot *cc_hot; + u64 *ts_end; data_size = subbuf_offset(offsets->old - 1, chan) + 1; padding_size = chan->backend.subbuf_size - data_size; subbuffer_set_data_size(config, &buf->backend, oldidx, data_size); + ts_end = &buf->ts_end[oldidx]; /* - * Order all writes to buffer before the commit count update that will - * determine that the subbuffer is full. + * This is the last space reservation in that sub-buffer before + * it gets delivered. This provides exclusive access to write to + * this sub-buffer's ts_end. There are also no concurrent + * readers of that ts_end because delivery of that sub-buffer is + * postponed until the commit counter is incremented for the + * current space reservation. + */ + *ts_end = tsc; + + /* + * Order all writes to buffer and store to ts_end before the commit + * count update that will determine that the subbuffer is full. */ if (config->ipi == RING_BUFFER_IPI_BARRIER) { /* @@ -1684,10 +1566,21 @@ void lib_ring_buffer_switch_new_end(struct lib_ring_buffer *buf, { const struct lib_ring_buffer_config *config = &chan->backend.config; unsigned long endidx, data_size; + u64 *ts_end; endidx = subbuf_index(offsets->end - 1, chan); data_size = subbuf_offset(offsets->end - 1, chan) + 1; subbuffer_set_data_size(config, &buf->backend, endidx, data_size); + ts_end = &buf->ts_end[endidx]; + /* + * This is the last space reservation in that sub-buffer before + * it gets delivered. This provides exclusive access to write to + * this sub-buffer's ts_end. There are also no concurrent + * readers of that ts_end because delivery of that sub-buffer is + * postponed until the commit counter is incremented for the + * current space reservation. + */ + *ts_end = tsc; } /* @@ -1917,6 +1810,16 @@ void lib_ring_buffer_switch_remote_empty(struct lib_ring_buffer *buf) } EXPORT_SYMBOL_GPL(lib_ring_buffer_switch_remote_empty); +void lib_ring_buffer_clear(struct lib_ring_buffer *buf) +{ + struct lib_ring_buffer_backend *bufb = &buf->backend; + struct channel *chan = bufb->chan; + + lib_ring_buffer_switch_remote(buf); + lib_ring_buffer_clear_reader(buf, chan); +} +EXPORT_SYMBOL_GPL(lib_ring_buffer_clear); + /* * Returns : * 0 if ok @@ -2251,14 +2154,24 @@ void lib_ring_buffer_check_deliver_slow(const struct lib_ring_buffer_config *con if (likely(v_cmpxchg(config, &buf->commit_cold[idx].cc_sb, old_commit_count, old_commit_count + 1) == old_commit_count)) { + u64 *ts_end; + /* * Start of exclusive subbuffer access. We are * guaranteed to be the last writer in this subbuffer * and any other writer trying to access this subbuffer * in this state is required to drop records. + * + * We can read the ts_end for the current sub-buffer + * which has been saved by the very last space + * reservation for the current sub-buffer. + * + * Order increment of commit counter before reading ts_end. */ + smp_mb(); + ts_end = &buf->ts_end[idx]; deliver_count_events(config, buf, idx); - config->cb.buffer_end(buf, tsc, idx, + config->cb.buffer_end(buf, *ts_end, idx, lib_ring_buffer_get_data_size(config, buf, idx));