X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Flib%2Fringbuffer%2Fring_buffer_frontend.c;h=df0705cbd4245f1f2221a00ed41e9884f7c1a5a6;hb=c4d6de426e681f6760a883fe47f2b6c52075b021;hp=d3dfcb1b1e4d9cfbf283bd47548118462014f71c;hpb=5a15f70c5211ff4a398171a6971586e2948eb419;p=lttng-modules.git diff --git a/src/lib/ringbuffer/ring_buffer_frontend.c b/src/lib/ringbuffer/ring_buffer_frontend.c index d3dfcb1b..df0705cb 100644 --- a/src/lib/ringbuffer/ring_buffer_frontend.c +++ b/src/lib/ringbuffer/ring_buffer_frontend.c @@ -37,9 +37,11 @@ * - put_subbuf */ +#include #include #include #include +#include #include #include @@ -47,9 +49,8 @@ #include #include #include -#include +#include #include -#include #include #include @@ -79,16 +80,16 @@ DEFINE_PER_CPU(unsigned int, lib_ring_buffer_nesting); EXPORT_PER_CPU_SYMBOL(lib_ring_buffer_nesting); static -void lib_ring_buffer_print_errors(struct channel *chan, - struct lib_ring_buffer *buf, int cpu); +void lib_ring_buffer_print_errors(struct lttng_kernel_ring_buffer_channel *chan, + struct lttng_kernel_ring_buffer *buf, int cpu); static -void _lib_ring_buffer_switch_remote(struct lib_ring_buffer *buf, +void _lib_ring_buffer_switch_remote(struct lttng_kernel_ring_buffer *buf, enum switch_mode mode); static -int lib_ring_buffer_poll_deliver(const struct lib_ring_buffer_config *config, - struct lib_ring_buffer *buf, - struct channel *chan) +int lib_ring_buffer_poll_deliver(const struct lttng_kernel_ring_buffer_config *config, + struct lttng_kernel_ring_buffer *buf, + struct lttng_kernel_ring_buffer_channel *chan) { unsigned long consumed_old, consumed_idx, commit_count, write_offset; @@ -129,9 +130,11 @@ int lib_ring_buffer_poll_deliver(const struct lib_ring_buffer_config *config, /* * Must be called under cpu hotplug protection. */ -void lib_ring_buffer_free(struct lib_ring_buffer *buf) +void lib_ring_buffer_free(struct lttng_kernel_ring_buffer *buf) { - struct channel *chan = buf->backend.chan; + struct lttng_kernel_ring_buffer_channel *chan = buf->backend.chan; + + irq_work_sync(&buf->wakeup_pending); lib_ring_buffer_print_errors(chan, buf, buf->backend.cpu); lttng_kvfree(buf->commit_hot); @@ -150,10 +153,10 @@ void lib_ring_buffer_free(struct lib_ring_buffer *buf) * should not be using the iterator concurrently with reset. The previous * current iterator record is reset. */ -void lib_ring_buffer_reset(struct lib_ring_buffer *buf) +void lib_ring_buffer_reset(struct lttng_kernel_ring_buffer *buf) { - struct channel *chan = buf->backend.chan; - const struct lib_ring_buffer_config *config = &chan->backend.config; + struct lttng_kernel_ring_buffer_channel *chan = buf->backend.chan; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; unsigned int i; /* @@ -170,7 +173,7 @@ void lib_ring_buffer_reset(struct lib_ring_buffer *buf) } atomic_long_set(&buf->consumed, 0); atomic_set(&buf->record_disabled, 0); - v_set(config, &buf->last_tsc, 0); + v_set(config, &buf->last_timestamp, 0); lib_ring_buffer_backend_reset(&buf->backend); /* Don't reset number of active readers */ v_set(config, &buf->records_lost_full, 0); @@ -191,7 +194,7 @@ EXPORT_SYMBOL_GPL(lib_ring_buffer_reset); * be using the iterator concurrently with reset. The previous current iterator * record is reset. */ -void channel_reset(struct channel *chan) +void channel_reset(struct lttng_kernel_ring_buffer_channel *chan) { /* * Reset iterators first. Will put the subbuffer if held for reading. @@ -206,17 +209,30 @@ void channel_reset(struct channel *chan) } EXPORT_SYMBOL_GPL(channel_reset); +static void lib_ring_buffer_pending_wakeup_buf(struct irq_work *entry) +{ + struct lttng_kernel_ring_buffer *buf = container_of(entry, struct lttng_kernel_ring_buffer, + wakeup_pending); + wake_up_interruptible(&buf->read_wait); +} + +static void lib_ring_buffer_pending_wakeup_chan(struct irq_work *entry) +{ + struct lttng_kernel_ring_buffer_channel *chan = container_of(entry, struct lttng_kernel_ring_buffer_channel, wakeup_pending); + wake_up_interruptible(&chan->read_wait); +} + /* * Must be called under cpu hotplug protection. */ -int lib_ring_buffer_create(struct lib_ring_buffer *buf, +int lib_ring_buffer_create(struct lttng_kernel_ring_buffer *buf, struct channel_backend *chanb, int cpu) { - const struct lib_ring_buffer_config *config = &chanb->config; - struct channel *chan = container_of(chanb, struct channel, backend); + const struct lttng_kernel_ring_buffer_config *config = &chanb->config; + struct lttng_kernel_ring_buffer_channel *chan = container_of(chanb, struct lttng_kernel_ring_buffer_channel, backend); void *priv = chanb->priv; size_t subbuf_header_size; - u64 tsc; + u64 timestamp; int ret; /* Test for cpu hotplug */ @@ -268,6 +284,7 @@ int lib_ring_buffer_create(struct lib_ring_buffer *buf, init_waitqueue_head(&buf->read_wait); init_waitqueue_head(&buf->write_wait); + init_irq_work(&buf->wakeup_pending, lib_ring_buffer_pending_wakeup_buf); raw_spin_lock_init(&buf->raw_tick_nohz_spinlock); /* @@ -277,8 +294,8 @@ int lib_ring_buffer_create(struct lib_ring_buffer *buf, subbuf_header_size = config->cb.subbuffer_header_size(); v_set(config, &buf->offset, subbuf_header_size); subbuffer_id_clear_noref(config, &buf->backend.buf_wsb[0].id); - tsc = config->cb.ring_buffer_clock_read(buf->backend.chan); - config->cb.buffer_begin(buf, tsc, 0); + timestamp = config->cb.ring_buffer_clock_read(buf->backend.chan); + config->cb.buffer_begin(buf, timestamp, 0); v_add(config, subbuf_header_size, &buf->commit_hot[0].cc); if (config->cb.buffer_create) { @@ -317,9 +334,9 @@ free_chanbuf: static void switch_buffer_timer(LTTNG_TIMER_FUNC_ARG_TYPE t) { - struct lib_ring_buffer *buf = lttng_from_timer(buf, t, switch_timer); - struct channel *chan = buf->backend.chan; - const struct lib_ring_buffer_config *config = &chan->backend.config; + struct lttng_kernel_ring_buffer *buf = lttng_from_timer(buf, t, switch_timer); + struct lttng_kernel_ring_buffer_channel *chan = buf->backend.chan; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; /* * Only flush buffers periodically if readers are active. @@ -338,10 +355,10 @@ static void switch_buffer_timer(LTTNG_TIMER_FUNC_ARG_TYPE t) /* * Called with ring_buffer_nohz_lock held for per-cpu buffers. */ -static void lib_ring_buffer_start_switch_timer(struct lib_ring_buffer *buf) +static void lib_ring_buffer_start_switch_timer(struct lttng_kernel_ring_buffer *buf) { - struct channel *chan = buf->backend.chan; - const struct lib_ring_buffer_config *config = &chan->backend.config; + struct lttng_kernel_ring_buffer_channel *chan = buf->backend.chan; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; unsigned int flags = 0; if (!chan->switch_timer_interval || buf->switch_timer_enabled) @@ -364,9 +381,9 @@ static void lib_ring_buffer_start_switch_timer(struct lib_ring_buffer *buf) /* * Called with ring_buffer_nohz_lock held for per-cpu buffers. */ -static void lib_ring_buffer_stop_switch_timer(struct lib_ring_buffer *buf) +static void lib_ring_buffer_stop_switch_timer(struct lttng_kernel_ring_buffer *buf) { - struct channel *chan = buf->backend.chan; + struct lttng_kernel_ring_buffer_channel *chan = buf->backend.chan; if (!chan->switch_timer_interval || !buf->switch_timer_enabled) return; @@ -380,9 +397,9 @@ static void lib_ring_buffer_stop_switch_timer(struct lib_ring_buffer *buf) */ static void read_buffer_timer(LTTNG_TIMER_FUNC_ARG_TYPE t) { - struct lib_ring_buffer *buf = lttng_from_timer(buf, t, read_timer); - struct channel *chan = buf->backend.chan; - const struct lib_ring_buffer_config *config = &chan->backend.config; + struct lttng_kernel_ring_buffer *buf = lttng_from_timer(buf, t, read_timer); + struct lttng_kernel_ring_buffer_channel *chan = buf->backend.chan; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; CHAN_WARN_ON(chan, !buf->backend.allocated); @@ -403,10 +420,10 @@ static void read_buffer_timer(LTTNG_TIMER_FUNC_ARG_TYPE t) /* * Called with ring_buffer_nohz_lock held for per-cpu buffers. */ -static void lib_ring_buffer_start_read_timer(struct lib_ring_buffer *buf) +static void lib_ring_buffer_start_read_timer(struct lttng_kernel_ring_buffer *buf) { - struct channel *chan = buf->backend.chan; - const struct lib_ring_buffer_config *config = &chan->backend.config; + struct lttng_kernel_ring_buffer_channel *chan = buf->backend.chan; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; unsigned int flags = 0; if (config->wakeup != RING_BUFFER_WAKEUP_BY_TIMER @@ -431,10 +448,10 @@ static void lib_ring_buffer_start_read_timer(struct lib_ring_buffer *buf) /* * Called with ring_buffer_nohz_lock held for per-cpu buffers. */ -static void lib_ring_buffer_stop_read_timer(struct lib_ring_buffer *buf) +static void lib_ring_buffer_stop_read_timer(struct lttng_kernel_ring_buffer *buf) { - struct channel *chan = buf->backend.chan; - const struct lib_ring_buffer_config *config = &chan->backend.config; + struct lttng_kernel_ring_buffer_channel *chan = buf->backend.chan; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; if (config->wakeup != RING_BUFFER_WAKEUP_BY_TIMER || !chan->read_timer_interval @@ -453,7 +470,7 @@ 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)) +#if (LTTNG_LINUX_VERSION_CODE >= LTTNG_KERNEL_VERSION(4,10,0)) enum cpuhp_state lttng_rb_hp_prepare; enum cpuhp_state lttng_rb_hp_online; @@ -473,10 +490,10 @@ EXPORT_SYMBOL_GPL(lttng_rb_set_hp_online); int lttng_cpuhp_rb_frontend_dead(unsigned int cpu, struct lttng_cpuhp_node *node) { - struct channel *chan = container_of(node, struct channel, + struct lttng_kernel_ring_buffer_channel *chan = container_of(node, struct lttng_kernel_ring_buffer_channel, cpuhp_prepare); - struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, cpu); - const struct lib_ring_buffer_config *config = &chan->backend.config; + struct lttng_kernel_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, cpu); + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; CHAN_WARN_ON(chan, config->alloc == RING_BUFFER_ALLOC_GLOBAL); @@ -494,10 +511,10 @@ EXPORT_SYMBOL_GPL(lttng_cpuhp_rb_frontend_dead); int lttng_cpuhp_rb_frontend_online(unsigned int cpu, struct lttng_cpuhp_node *node) { - struct channel *chan = container_of(node, struct channel, + struct lttng_kernel_ring_buffer_channel *chan = container_of(node, struct lttng_kernel_ring_buffer_channel, cpuhp_online); - struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, cpu); - const struct lib_ring_buffer_config *config = &chan->backend.config; + struct lttng_kernel_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, cpu); + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; CHAN_WARN_ON(chan, config->alloc == RING_BUFFER_ALLOC_GLOBAL); @@ -511,10 +528,10 @@ EXPORT_SYMBOL_GPL(lttng_cpuhp_rb_frontend_online); int lttng_cpuhp_rb_frontend_offline(unsigned int cpu, struct lttng_cpuhp_node *node) { - struct channel *chan = container_of(node, struct channel, + struct lttng_kernel_ring_buffer_channel *chan = container_of(node, struct lttng_kernel_ring_buffer_channel, cpuhp_online); - struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, cpu); - const struct lib_ring_buffer_config *config = &chan->backend.config; + struct lttng_kernel_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, cpu); + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; CHAN_WARN_ON(chan, config->alloc == RING_BUFFER_ALLOC_GLOBAL); @@ -524,7 +541,7 @@ 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)) */ +#else /* #if (LTTNG_LINUX_VERSION_CODE >= LTTNG_KERNEL_VERSION(4,10,0)) */ #ifdef CONFIG_HOTPLUG_CPU @@ -542,10 +559,10 @@ int lib_ring_buffer_cpu_hp_callback(struct notifier_block *nb, void *hcpu) { unsigned int cpu = (unsigned long)hcpu; - struct channel *chan = container_of(nb, struct channel, + struct lttng_kernel_ring_buffer_channel *chan = container_of(nb, struct lttng_kernel_ring_buffer_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; + struct lttng_kernel_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, cpu); + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; if (!chan->cpu_hp_enable) return NOTIFY_DONE; @@ -586,7 +603,7 @@ int lib_ring_buffer_cpu_hp_callback(struct notifier_block *nb, #endif -#endif /* #else #if (LINUX_VERSION_CODE >= KERNEL_VERSION(4,10,0)) */ +#endif /* #else #if (LTTNG_LINUX_VERSION_CODE >= LTTNG_KERNEL_VERSION(4,10,0)) */ #if defined(CONFIG_NO_HZ) && defined(CONFIG_LIB_RING_BUFFER) /* @@ -601,10 +618,10 @@ static int notrace ring_buffer_tick_nohz_callback(struct notifier_block *nb, unsigned long val, void *data) { - struct channel *chan = container_of(nb, struct channel, + struct lttng_kernel_ring_buffer_channel *chan = container_of(nb, struct lttng_kernel_ring_buffer_channel, tick_nohz_notifier); - const struct lib_ring_buffer_config *config = &chan->backend.config; - struct lib_ring_buffer *buf; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; + struct lttng_kernel_ring_buffer *buf; int cpu = smp_processor_id(); if (config->alloc != RING_BUFFER_ALLOC_PER_CPU) { @@ -635,16 +652,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; } @@ -673,9 +690,9 @@ void notrace lib_ring_buffer_tick_nohz_restart(void) /* * Holds CPU hotplug. */ -static void channel_unregister_notifiers(struct channel *chan) +static void channel_unregister_notifiers(struct lttng_kernel_ring_buffer_channel *chan) { - const struct lib_ring_buffer_config *config = &chan->backend.config; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; channel_iterator_unregister_notifiers(chan); if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) { @@ -692,7 +709,7 @@ static void channel_unregister_notifiers(struct channel *chan) * concurrency. */ #endif /* CONFIG_NO_HZ */ -#if (LINUX_VERSION_CODE >= KERNEL_VERSION(4,10,0)) +#if (LTTNG_LINUX_VERSION_CODE >= LTTNG_KERNEL_VERSION(4,10,0)) { int ret; @@ -703,33 +720,33 @@ 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)) */ +#else /* #if (LTTNG_LINUX_VERSION_CODE >= LTTNG_KERNEL_VERSION(4,10,0)) */ { int cpu; #ifdef CONFIG_HOTPLUG_CPU - get_online_cpus(); + lttng_cpus_read_lock(); chan->cpu_hp_enable = 0; for_each_online_cpu(cpu) { - struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, + struct lttng_kernel_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(); + lttng_cpus_read_unlock(); unregister_cpu_notifier(&chan->cpu_hp_notifier); #else for_each_possible_cpu(cpu) { - struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, + struct lttng_kernel_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)) */ +#endif /* #else #if (LTTNG_LINUX_VERSION_CODE >= LTTNG_KERNEL_VERSION(4,10,0)) */ } else { - struct lib_ring_buffer *buf = chan->backend.buf; + struct lttng_kernel_ring_buffer *buf = chan->backend.buf; lib_ring_buffer_stop_switch_timer(buf); lib_ring_buffer_stop_read_timer(buf); @@ -737,7 +754,7 @@ static void channel_unregister_notifiers(struct channel *chan) channel_backend_unregister_notifiers(&chan->backend); } -static void lib_ring_buffer_set_quiescent(struct lib_ring_buffer *buf) +static void lib_ring_buffer_set_quiescent(struct lttng_kernel_ring_buffer *buf) { if (!buf->quiescent) { buf->quiescent = true; @@ -745,56 +762,56 @@ static void lib_ring_buffer_set_quiescent(struct lib_ring_buffer *buf) } } -static void lib_ring_buffer_clear_quiescent(struct lib_ring_buffer *buf) +static void lib_ring_buffer_clear_quiescent(struct lttng_kernel_ring_buffer *buf) { buf->quiescent = false; } -void lib_ring_buffer_set_quiescent_channel(struct channel *chan) +void lib_ring_buffer_set_quiescent_channel(struct lttng_kernel_ring_buffer_channel *chan) { int cpu; - const struct lib_ring_buffer_config *config = &chan->backend.config; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) { - get_online_cpus(); + lttng_cpus_read_lock(); for_each_channel_cpu(cpu, chan) { - struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, + struct lttng_kernel_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, cpu); lib_ring_buffer_set_quiescent(buf); } - put_online_cpus(); + lttng_cpus_read_unlock(); } else { - struct lib_ring_buffer *buf = chan->backend.buf; + struct lttng_kernel_ring_buffer *buf = chan->backend.buf; lib_ring_buffer_set_quiescent(buf); } } EXPORT_SYMBOL_GPL(lib_ring_buffer_set_quiescent_channel); -void lib_ring_buffer_clear_quiescent_channel(struct channel *chan) +void lib_ring_buffer_clear_quiescent_channel(struct lttng_kernel_ring_buffer_channel *chan) { int cpu; - const struct lib_ring_buffer_config *config = &chan->backend.config; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) { - get_online_cpus(); + lttng_cpus_read_lock(); for_each_channel_cpu(cpu, chan) { - struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, + struct lttng_kernel_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, cpu); lib_ring_buffer_clear_quiescent(buf); } - put_online_cpus(); + lttng_cpus_read_unlock(); } else { - struct lib_ring_buffer *buf = chan->backend.buf; + struct lttng_kernel_ring_buffer *buf = chan->backend.buf; lib_ring_buffer_clear_quiescent(buf); } } EXPORT_SYMBOL_GPL(lib_ring_buffer_clear_quiescent_channel); -static void channel_free(struct channel *chan) +static void channel_free(struct lttng_kernel_ring_buffer_channel *chan) { if (chan->backend.release_priv_ops) { chan->backend.release_priv_ops(chan->backend.priv_ops); @@ -822,20 +839,20 @@ static void channel_free(struct channel *chan) * Holds cpu hotplug. * Returns NULL on failure. */ -struct channel *channel_create(const struct lib_ring_buffer_config *config, +struct lttng_kernel_ring_buffer_channel *channel_create(const struct lttng_kernel_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) { int ret; - struct channel *chan; + struct lttng_kernel_ring_buffer_channel *chan; if (lib_ring_buffer_check_config(config, switch_timer_interval, read_timer_interval)) return NULL; - chan = kzalloc(sizeof(struct channel), GFP_KERNEL); + chan = kzalloc(sizeof(struct lttng_kernel_ring_buffer_channel), GFP_KERNEL); if (!chan) return NULL; @@ -854,9 +871,10 @@ struct channel *channel_create(const struct lib_ring_buffer_config *config, kref_init(&chan->ref); init_waitqueue_head(&chan->read_wait); init_waitqueue_head(&chan->hp_wait); + init_irq_work(&chan->wakeup_pending, lib_ring_buffer_pending_wakeup_chan); if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) { -#if (LINUX_VERSION_CODE >= KERNEL_VERSION(4,10,0)) +#if (LTTNG_LINUX_VERSION_CODE >= LTTNG_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); @@ -868,7 +886,7 @@ 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)) */ +#else /* #if (LTTNG_LINUX_VERSION_CODE >= LTTNG_KERNEL_VERSION(4,10,0)) */ { int cpu; /* @@ -882,9 +900,9 @@ struct channel *channel_create(const struct lib_ring_buffer_config *config, chan->cpu_hp_notifier.priority = 6; register_cpu_notifier(&chan->cpu_hp_notifier); - get_online_cpus(); + lttng_cpus_read_lock(); for_each_online_cpu(cpu) { - struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, + struct lttng_kernel_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); @@ -892,10 +910,10 @@ struct channel *channel_create(const struct lib_ring_buffer_config *config, spin_unlock(&per_cpu(ring_buffer_nohz_lock, cpu)); } chan->cpu_hp_enable = 1; - put_online_cpus(); + lttng_cpus_read_unlock(); #else for_each_possible_cpu(cpu) { - struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, + struct lttng_kernel_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); @@ -904,7 +922,7 @@ struct channel *channel_create(const struct lib_ring_buffer_config *config, } #endif } -#endif /* #else #if (LINUX_VERSION_CODE >= KERNEL_VERSION(4,10,0)) */ +#endif /* #else #if (LTTNG_LINUX_VERSION_CODE >= LTTNG_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. */ @@ -916,7 +934,7 @@ struct channel *channel_create(const struct lib_ring_buffer_config *config, #endif /* defined(CONFIG_NO_HZ) && defined(CONFIG_LIB_RING_BUFFER) */ } else { - struct lib_ring_buffer *buf = chan->backend.buf; + struct lttng_kernel_ring_buffer *buf = chan->backend.buf; lib_ring_buffer_start_switch_timer(buf); lib_ring_buffer_start_read_timer(buf); @@ -924,13 +942,13 @@ struct channel *channel_create(const struct lib_ring_buffer_config *config, return chan; -#if (LINUX_VERSION_CODE >= KERNEL_VERSION(4,10,0)) +#if (LTTNG_LINUX_VERSION_CODE >= LTTNG_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)) */ +#endif /* #if (LTTNG_LINUX_VERSION_CODE >= LTTNG_KERNEL_VERSION(4,10,0)) */ error_free_backend: channel_backend_free(&chan->backend); error: @@ -942,7 +960,7 @@ EXPORT_SYMBOL_GPL(channel_create); static void channel_release(struct kref *kref) { - struct channel *chan = container_of(kref, struct channel, ref); + struct lttng_kernel_ring_buffer_channel *chan = container_of(kref, struct lttng_kernel_ring_buffer_channel, ref); channel_free(chan); } @@ -957,12 +975,14 @@ void channel_release(struct kref *kref) * They should release their handle at that point. Returns the private * data pointer. */ -void *channel_destroy(struct channel *chan) +void *channel_destroy(struct lttng_kernel_ring_buffer_channel *chan) { int cpu; - const struct lib_ring_buffer_config *config = &chan->backend.config; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; void *priv; + irq_work_sync(&chan->wakeup_pending); + channel_unregister_notifiers(chan); if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) { @@ -971,7 +991,7 @@ void *channel_destroy(struct channel *chan) * unregistered. */ for_each_channel_cpu(cpu, chan) { - struct lib_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, + struct lttng_kernel_ring_buffer *buf = per_cpu_ptr(chan->backend.buf, cpu); if (config->cb.buffer_finalize) @@ -986,7 +1006,7 @@ void *channel_destroy(struct channel *chan) wake_up_interruptible(&buf->read_wait); } } else { - struct lib_ring_buffer *buf = chan->backend.buf; + struct lttng_kernel_ring_buffer *buf = chan->backend.buf; if (config->cb.buffer_finalize) config->cb.buffer_finalize(buf, chan->backend.priv, -1); @@ -1006,9 +1026,9 @@ void *channel_destroy(struct channel *chan) } EXPORT_SYMBOL_GPL(channel_destroy); -struct lib_ring_buffer *channel_get_ring_buffer( - const struct lib_ring_buffer_config *config, - struct channel *chan, int cpu) +struct lttng_kernel_ring_buffer *channel_get_ring_buffer( + const struct lttng_kernel_ring_buffer_config *config, + struct lttng_kernel_ring_buffer_channel *chan, int cpu) { if (config->alloc == RING_BUFFER_ALLOC_GLOBAL) return chan->backend.buf; @@ -1017,9 +1037,9 @@ struct lib_ring_buffer *channel_get_ring_buffer( } EXPORT_SYMBOL_GPL(channel_get_ring_buffer); -int lib_ring_buffer_open_read(struct lib_ring_buffer *buf) +int lib_ring_buffer_open_read(struct lttng_kernel_ring_buffer *buf) { - struct channel *chan = buf->backend.chan; + struct lttng_kernel_ring_buffer_channel *chan = buf->backend.chan; if (!atomic_long_add_unless(&buf->active_readers, 1, 1)) return -EBUSY; @@ -1027,17 +1047,17 @@ int lib_ring_buffer_open_read(struct lib_ring_buffer *buf) atomic_long_dec(&buf->active_readers); return -EOVERFLOW; } - lttng_smp_mb__after_atomic(); + smp_mb__after_atomic(); return 0; } EXPORT_SYMBOL_GPL(lib_ring_buffer_open_read); -void lib_ring_buffer_release_read(struct lib_ring_buffer *buf) +void lib_ring_buffer_release_read(struct lttng_kernel_ring_buffer *buf) { - struct channel *chan = buf->backend.chan; + struct lttng_kernel_ring_buffer_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); } @@ -1065,16 +1085,16 @@ static void remote_mb(void *info) * Busy-loop trying to get data if the tick_nohz sequence lock is held. */ -int lib_ring_buffer_snapshot(struct lib_ring_buffer *buf, +int lib_ring_buffer_snapshot(struct lttng_kernel_ring_buffer *buf, unsigned long *consumed, unsigned long *produced) { - struct channel *chan = buf->backend.chan; - const struct lib_ring_buffer_config *config = &chan->backend.config; + struct lttng_kernel_ring_buffer_channel *chan = buf->backend.chan; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; unsigned long consumed_cur, write_offset; int finalized; retry: - finalized = READ_ONCE(buf->finalized); + finalized = LTTNG_READ_ONCE(buf->finalized); /* * Read finalized before counters. */ @@ -1128,11 +1148,11 @@ EXPORT_SYMBOL_GPL(lib_ring_buffer_snapshot); * This function is meant to provide information on the exact producer and * consumer positions without regard for the "snapshot" feature. */ -int lib_ring_buffer_snapshot_sample_positions(struct lib_ring_buffer *buf, +int lib_ring_buffer_snapshot_sample_positions(struct lttng_kernel_ring_buffer *buf, unsigned long *consumed, unsigned long *produced) { - struct channel *chan = buf->backend.chan; - const struct lib_ring_buffer_config *config = &chan->backend.config; + struct lttng_kernel_ring_buffer_channel *chan = buf->backend.chan; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; smp_rmb(); *consumed = atomic_long_read(&buf->consumed); @@ -1155,11 +1175,11 @@ int lib_ring_buffer_snapshot_sample_positions(struct lib_ring_buffer *buf, * @buf: ring buffer * @consumed_new: new consumed count value */ -void lib_ring_buffer_move_consumer(struct lib_ring_buffer *buf, +void lib_ring_buffer_move_consumer(struct lttng_kernel_ring_buffer *buf, unsigned long consumed_new) { - struct lib_ring_buffer_backend *bufb = &buf->backend; - struct channel *chan = bufb->chan; + struct lttng_kernel_ring_buffer_backend *bufb = &buf->backend; + struct lttng_kernel_ring_buffer_channel *chan = bufb->chan; unsigned long consumed; CHAN_WARN_ON(chan, atomic_long_read(&buf->active_readers) != 1); @@ -1180,16 +1200,26 @@ EXPORT_SYMBOL_GPL(lib_ring_buffer_move_consumer); #if ARCH_IMPLEMENTS_FLUSH_DCACHE_PAGE static void lib_ring_buffer_flush_read_subbuf_dcache( - const struct lib_ring_buffer_config *config, - struct channel *chan, - struct lib_ring_buffer *buf) + const struct lttng_kernel_ring_buffer_config *config, + struct lttng_kernel_ring_buffer_channel *chan, + struct lttng_kernel_ring_buffer *buf) { - struct lib_ring_buffer_backend_pages *pages; + struct lttng_kernel_ring_buffer_backend_pages *pages; unsigned long sb_bindex, id, i, nr_pages; if (config->output != RING_BUFFER_MMAP) return; +#ifdef cpu_dcache_is_aliasing + /* + * Some architectures implement flush_dcache_page() but don't + * actually have aliasing dcache. cpu_dcache_is_aliasing() was + * introduced in kernel v6.9 to query this more precisely. + */ + if (!cpu_dcache_is_aliasing()) + return; +#endif + /* * Architectures with caches aliased on virtual addresses may * use different cache lines for the linear mapping vs @@ -1204,7 +1234,7 @@ static void lib_ring_buffer_flush_read_subbuf_dcache( pages = buf->backend.array[sb_bindex]; nr_pages = buf->backend.num_pages_per_subbuf; for (i = 0; i < nr_pages; i++) { - struct lib_ring_buffer_backend_page *backend_page; + struct lttng_kernel_ring_buffer_backend_page *backend_page; backend_page = &pages->p[i]; flush_dcache_page(pfn_to_page(backend_page->pfn)); @@ -1212,9 +1242,9 @@ static void lib_ring_buffer_flush_read_subbuf_dcache( } #else static void lib_ring_buffer_flush_read_subbuf_dcache( - const struct lib_ring_buffer_config *config, - struct channel *chan, - struct lib_ring_buffer *buf) + const struct lttng_kernel_ring_buffer_config *config, + struct lttng_kernel_ring_buffer_channel *chan, + struct lttng_kernel_ring_buffer *buf) { } #endif @@ -1228,11 +1258,11 @@ static void lib_ring_buffer_flush_read_subbuf_dcache( * data to read at consumed position, or 0 if the get operation succeeds. * Busy-loop trying to get data if the tick_nohz sequence lock is held. */ -int lib_ring_buffer_get_subbuf(struct lib_ring_buffer *buf, +int lib_ring_buffer_get_subbuf(struct lttng_kernel_ring_buffer *buf, unsigned long consumed) { - struct channel *chan = buf->backend.chan; - const struct lib_ring_buffer_config *config = &chan->backend.config; + struct lttng_kernel_ring_buffer_channel *chan = buf->backend.chan; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; unsigned long consumed_cur, consumed_idx, commit_count, write_offset; int ret; int finalized; @@ -1245,7 +1275,7 @@ int lib_ring_buffer_get_subbuf(struct lib_ring_buffer *buf, return -EBUSY; } retry: - finalized = READ_ONCE(buf->finalized); + finalized = LTTNG_READ_ONCE(buf->finalized); /* * Read finalized before counters. */ @@ -1383,11 +1413,11 @@ EXPORT_SYMBOL_GPL(lib_ring_buffer_get_subbuf); * lib_ring_buffer_put_subbuf - release exclusive subbuffer access * @buf: ring buffer */ -void lib_ring_buffer_put_subbuf(struct lib_ring_buffer *buf) +void lib_ring_buffer_put_subbuf(struct lttng_kernel_ring_buffer *buf) { - struct lib_ring_buffer_backend *bufb = &buf->backend; - struct channel *chan = bufb->chan; - const struct lib_ring_buffer_config *config = &chan->backend.config; + struct lttng_kernel_ring_buffer_backend *bufb = &buf->backend; + struct lttng_kernel_ring_buffer_channel *chan = bufb->chan; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; unsigned long read_sb_bindex, consumed_idx, consumed; CHAN_WARN_ON(chan, atomic_long_read(&buf->active_readers) != 1); @@ -1441,12 +1471,12 @@ EXPORT_SYMBOL_GPL(lib_ring_buffer_put_subbuf); * position and the writer position. (inclusive) */ static -void lib_ring_buffer_print_subbuffer_errors(struct lib_ring_buffer *buf, - struct channel *chan, +void lib_ring_buffer_print_subbuffer_errors(struct lttng_kernel_ring_buffer *buf, + struct lttng_kernel_ring_buffer_channel *chan, unsigned long cons_offset, int cpu) { - const struct lib_ring_buffer_config *config = &chan->backend.config; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; unsigned long cons_idx, commit_count, commit_count_sb; cons_idx = subbuf_index(cons_offset, chan); @@ -1468,11 +1498,11 @@ void lib_ring_buffer_print_subbuffer_errors(struct lib_ring_buffer *buf, } static -void lib_ring_buffer_print_buffer_errors(struct lib_ring_buffer *buf, - struct channel *chan, +void lib_ring_buffer_print_buffer_errors(struct lttng_kernel_ring_buffer *buf, + struct lttng_kernel_ring_buffer_channel *chan, void *priv, int cpu) { - const struct lib_ring_buffer_config *config = &chan->backend.config; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; unsigned long write_offset, cons_offset; /* @@ -1500,11 +1530,11 @@ void lib_ring_buffer_print_buffer_errors(struct lib_ring_buffer *buf, #ifdef LTTNG_RING_BUFFER_COUNT_EVENTS static -void lib_ring_buffer_print_records_count(struct channel *chan, - struct lib_ring_buffer *buf, +void lib_ring_buffer_print_records_count(struct lttng_kernel_ring_buffer_channel *chan, + struct lttng_kernel_ring_buffer *buf, int cpu) { - const struct lib_ring_buffer_config *config = &chan->backend.config; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; if (!strcmp(chan->backend.name, "relay-metadata")) { printk(KERN_DEBUG "LTTng: ring buffer %s: %lu records written, " @@ -1522,18 +1552,18 @@ void lib_ring_buffer_print_records_count(struct channel *chan, } #else static -void lib_ring_buffer_print_records_count(struct channel *chan, - struct lib_ring_buffer *buf, +void lib_ring_buffer_print_records_count(struct lttng_kernel_ring_buffer_channel *chan, + struct lttng_kernel_ring_buffer *buf, int cpu) { } #endif static -void lib_ring_buffer_print_errors(struct channel *chan, - struct lib_ring_buffer *buf, int cpu) +void lib_ring_buffer_print_errors(struct lttng_kernel_ring_buffer_channel *chan, + struct lttng_kernel_ring_buffer *buf, int cpu) { - const struct lib_ring_buffer_config *config = &chan->backend.config; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; void *priv = chan->backend.priv; lib_ring_buffer_print_records_count(chan, buf, cpu); @@ -1559,17 +1589,17 @@ void lib_ring_buffer_print_errors(struct channel *chan, * Only executed when the buffer is finalized, in SWITCH_FLUSH. */ static -void lib_ring_buffer_switch_old_start(struct lib_ring_buffer *buf, - struct channel *chan, +void lib_ring_buffer_switch_old_start(struct lttng_kernel_ring_buffer *buf, + struct lttng_kernel_ring_buffer_channel *chan, struct switch_offsets *offsets, - u64 tsc) + const struct lttng_kernel_ring_buffer_ctx *ctx) { - const struct lib_ring_buffer_config *config = &chan->backend.config; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; unsigned long oldidx = subbuf_index(offsets->old, chan); unsigned long commit_count; struct commit_counters_hot *cc_hot; - config->cb.buffer_begin(buf, tsc, oldidx); + config->cb.buffer_begin(buf, ctx->priv.timestamp, oldidx); /* * Order all writes to buffer before the commit count update that will @@ -1589,7 +1619,7 @@ void lib_ring_buffer_switch_old_start(struct lib_ring_buffer *buf, commit_count = v_read(config, &cc_hot->cc); /* Check if the written buffer has to be delivered */ lib_ring_buffer_check_deliver(config, buf, chan, offsets->old, - commit_count, oldidx, tsc); + commit_count, oldidx, ctx); lib_ring_buffer_write_commit_counter(config, buf, chan, offsets->old + config->cb.subbuffer_header_size(), commit_count, cc_hot); @@ -1604,12 +1634,12 @@ void lib_ring_buffer_switch_old_start(struct lib_ring_buffer *buf, * subbuffer. */ static -void lib_ring_buffer_switch_old_end(struct lib_ring_buffer *buf, - struct channel *chan, +void lib_ring_buffer_switch_old_end(struct lttng_kernel_ring_buffer *buf, + struct lttng_kernel_ring_buffer_channel *chan, struct switch_offsets *offsets, - u64 tsc) + const struct lttng_kernel_ring_buffer_ctx *ctx) { - const struct lib_ring_buffer_config *config = &chan->backend.config; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; unsigned long oldidx = subbuf_index(offsets->old - 1, chan); unsigned long commit_count, padding_size, data_size; struct commit_counters_hot *cc_hot; @@ -1628,7 +1658,7 @@ void lib_ring_buffer_switch_old_end(struct lib_ring_buffer *buf, * postponed until the commit counter is incremented for the * current space reservation. */ - *ts_end = tsc; + *ts_end = ctx->priv.timestamp; /* * Order all writes to buffer and store to ts_end before the commit @@ -1647,7 +1677,7 @@ void lib_ring_buffer_switch_old_end(struct lib_ring_buffer *buf, v_add(config, padding_size, &cc_hot->cc); commit_count = v_read(config, &cc_hot->cc); lib_ring_buffer_check_deliver(config, buf, chan, offsets->old - 1, - commit_count, oldidx, tsc); + commit_count, oldidx, ctx); lib_ring_buffer_write_commit_counter(config, buf, chan, offsets->old + padding_size, commit_count, cc_hot); @@ -1661,17 +1691,17 @@ void lib_ring_buffer_switch_old_end(struct lib_ring_buffer *buf, * that this code is executed before the deliver of this sub-buffer. */ static -void lib_ring_buffer_switch_new_start(struct lib_ring_buffer *buf, - struct channel *chan, +void lib_ring_buffer_switch_new_start(struct lttng_kernel_ring_buffer *buf, + struct lttng_kernel_ring_buffer_channel *chan, struct switch_offsets *offsets, - u64 tsc) + const struct lttng_kernel_ring_buffer_ctx *ctx) { - const struct lib_ring_buffer_config *config = &chan->backend.config; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; unsigned long beginidx = subbuf_index(offsets->begin, chan); unsigned long commit_count; struct commit_counters_hot *cc_hot; - config->cb.buffer_begin(buf, tsc, beginidx); + config->cb.buffer_begin(buf, ctx->priv.timestamp, beginidx); /* * Order all writes to buffer before the commit count update that will @@ -1691,7 +1721,7 @@ void lib_ring_buffer_switch_new_start(struct lib_ring_buffer *buf, commit_count = v_read(config, &cc_hot->cc); /* Check if the written buffer has to be delivered */ lib_ring_buffer_check_deliver(config, buf, chan, offsets->begin, - commit_count, beginidx, tsc); + commit_count, beginidx, ctx); lib_ring_buffer_write_commit_counter(config, buf, chan, offsets->begin + config->cb.subbuffer_header_size(), commit_count, cc_hot); @@ -1706,12 +1736,12 @@ void lib_ring_buffer_switch_new_start(struct lib_ring_buffer *buf, * we are currently doing the space reservation. */ static -void lib_ring_buffer_switch_new_end(struct lib_ring_buffer *buf, - struct channel *chan, +void lib_ring_buffer_switch_new_end(struct lttng_kernel_ring_buffer *buf, + struct lttng_kernel_ring_buffer_channel *chan, struct switch_offsets *offsets, - u64 tsc) + const struct lttng_kernel_ring_buffer_ctx *ctx) { - const struct lib_ring_buffer_config *config = &chan->backend.config; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; unsigned long endidx, data_size; u64 *ts_end; @@ -1727,7 +1757,7 @@ void lib_ring_buffer_switch_new_end(struct lib_ring_buffer *buf, * postponed until the commit counter is incremented for the * current space reservation. */ - *ts_end = tsc; + *ts_end = ctx->priv.timestamp; } /* @@ -1737,12 +1767,12 @@ void lib_ring_buffer_switch_new_end(struct lib_ring_buffer *buf, */ static int lib_ring_buffer_try_switch_slow(enum switch_mode mode, - struct lib_ring_buffer *buf, - struct channel *chan, + struct lttng_kernel_ring_buffer *buf, + struct lttng_kernel_ring_buffer_channel *chan, struct switch_offsets *offsets, - u64 *tsc) + struct lttng_kernel_ring_buffer_ctx *ctx) { - const struct lib_ring_buffer_config *config = &chan->backend.config; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; unsigned long off, reserve_commit_diff; offsets->begin = v_read(config, &buf->offset); @@ -1750,7 +1780,7 @@ int lib_ring_buffer_try_switch_slow(enum switch_mode mode, offsets->switch_old_start = 0; off = subbuf_offset(offsets->begin, chan); - *tsc = config->cb.ring_buffer_clock_read(chan); + ctx->priv.timestamp = config->cb.ring_buffer_clock_read(chan); /* * Ensure we flush the header of an empty subbuffer when doing the @@ -1832,6 +1862,13 @@ int lib_ring_buffer_try_switch_slow(enum switch_mode mode, offsets->begin = subbuf_align(offsets->begin, chan); /* Note: old points to the next subbuf at offset 0 */ offsets->end = offsets->begin; + /* + * Populate the records lost counters prior to performing a + * sub-buffer switch. + */ + ctx->priv.records_lost_full = v_read(config, &buf->records_lost_full); + ctx->priv.records_lost_wrap = v_read(config, &buf->records_lost_wrap); + ctx->priv.records_lost_big = v_read(config, &buf->records_lost_big); return 0; } @@ -1843,13 +1880,13 @@ int lib_ring_buffer_try_switch_slow(enum switch_mode mode, * operations, this function must be called from the CPU which owns the buffer * for a ACTIVE flush. */ -void lib_ring_buffer_switch_slow(struct lib_ring_buffer *buf, enum switch_mode mode) +void lib_ring_buffer_switch_slow(struct lttng_kernel_ring_buffer *buf, enum switch_mode mode) { - struct channel *chan = buf->backend.chan; - const struct lib_ring_buffer_config *config = &chan->backend.config; + struct lttng_kernel_ring_buffer_channel *chan = buf->backend.chan; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; + struct lttng_kernel_ring_buffer_ctx ctx; struct switch_offsets offsets; unsigned long oldidx; - u64 tsc; offsets.size = 0; @@ -1858,18 +1895,18 @@ void lib_ring_buffer_switch_slow(struct lib_ring_buffer *buf, enum switch_mode m */ do { if (lib_ring_buffer_try_switch_slow(mode, buf, chan, &offsets, - &tsc)) + &ctx)) return; /* Switch not needed */ } while (v_cmpxchg(config, &buf->offset, offsets.old, offsets.end) != offsets.old); /* - * Atomically update last_tsc. This update races against concurrent - * atomic updates, but the race will always cause supplementary full TSC - * records, never the opposite (missing a full TSC record when it would - * be needed). + * Atomically update last_timestamp. This update races against concurrent + * atomic updates, but the race will always cause supplementary + * full timestamp records, never the opposite (missing a full + * timestamp record when it would be needed). */ - save_last_tsc(config, buf, tsc); + save_last_timestamp(config, buf, ctx.priv.timestamp); /* * Push the reader if necessary @@ -1883,35 +1920,35 @@ void lib_ring_buffer_switch_slow(struct lib_ring_buffer *buf, enum switch_mode m * May need to populate header start on SWITCH_FLUSH. */ if (offsets.switch_old_start) { - lib_ring_buffer_switch_old_start(buf, chan, &offsets, tsc); + lib_ring_buffer_switch_old_start(buf, chan, &offsets, &ctx); offsets.old += config->cb.subbuffer_header_size(); } /* * Switch old subbuffer. */ - lib_ring_buffer_switch_old_end(buf, chan, &offsets, tsc); + lib_ring_buffer_switch_old_end(buf, chan, &offsets, &ctx); } EXPORT_SYMBOL_GPL(lib_ring_buffer_switch_slow); struct switch_param { - struct lib_ring_buffer *buf; + struct lttng_kernel_ring_buffer *buf; enum switch_mode mode; }; static void remote_switch(void *info) { struct switch_param *param = info; - struct lib_ring_buffer *buf = param->buf; + struct lttng_kernel_ring_buffer *buf = param->buf; lib_ring_buffer_switch_slow(buf, param->mode); } -static void _lib_ring_buffer_switch_remote(struct lib_ring_buffer *buf, +static void _lib_ring_buffer_switch_remote(struct lttng_kernel_ring_buffer *buf, enum switch_mode mode) { - struct channel *chan = buf->backend.chan; - const struct lib_ring_buffer_config *config = &chan->backend.config; + struct lttng_kernel_ring_buffer_channel *chan = buf->backend.chan; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; int ret; struct switch_param param; @@ -1944,23 +1981,23 @@ static void _lib_ring_buffer_switch_remote(struct lib_ring_buffer *buf, } /* Switch sub-buffer if current sub-buffer is non-empty. */ -void lib_ring_buffer_switch_remote(struct lib_ring_buffer *buf) +void lib_ring_buffer_switch_remote(struct lttng_kernel_ring_buffer *buf) { _lib_ring_buffer_switch_remote(buf, SWITCH_ACTIVE); } EXPORT_SYMBOL_GPL(lib_ring_buffer_switch_remote); /* Switch sub-buffer even if current sub-buffer is empty. */ -void lib_ring_buffer_switch_remote_empty(struct lib_ring_buffer *buf) +void lib_ring_buffer_switch_remote_empty(struct lttng_kernel_ring_buffer *buf) { _lib_ring_buffer_switch_remote(buf, SWITCH_FLUSH); } EXPORT_SYMBOL_GPL(lib_ring_buffer_switch_remote_empty); -void lib_ring_buffer_clear(struct lib_ring_buffer *buf) +void lib_ring_buffer_clear(struct lttng_kernel_ring_buffer *buf) { - struct lib_ring_buffer_backend *bufb = &buf->backend; - struct channel *chan = bufb->chan; + struct lttng_kernel_ring_buffer_backend *bufb = &buf->backend; + struct lttng_kernel_ring_buffer_channel *chan = bufb->chan; lib_ring_buffer_switch_remote(buf); lib_ring_buffer_clear_reader(buf, chan); @@ -1975,13 +2012,13 @@ EXPORT_SYMBOL_GPL(lib_ring_buffer_clear); * -EIO if data cannot be written into the buffer for any other reason. */ static -int lib_ring_buffer_try_reserve_slow(struct lib_ring_buffer *buf, - struct channel *chan, +int lib_ring_buffer_try_reserve_slow(struct lttng_kernel_ring_buffer *buf, + struct lttng_kernel_ring_buffer_channel *chan, struct switch_offsets *offsets, - struct lib_ring_buffer_ctx *ctx, + struct lttng_kernel_ring_buffer_ctx *ctx, void *client_ctx) { - const struct lib_ring_buffer_config *config = &chan->backend.config; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; unsigned long reserve_commit_diff, offset_cmp; retry: @@ -1992,14 +2029,14 @@ retry: offsets->switch_old_end = 0; offsets->pre_header_padding = 0; - ctx->tsc = config->cb.ring_buffer_clock_read(chan); - if ((int64_t) ctx->tsc == -EIO) + ctx->priv.timestamp = config->cb.ring_buffer_clock_read(chan); + if ((int64_t) ctx->priv.timestamp == -EIO) return -EIO; - if (last_tsc_overflow(config, buf, ctx->tsc)) - ctx->rflags |= RING_BUFFER_RFLAG_FULL_TSC; + if (last_timestamp_overflow(config, buf, ctx->priv.timestamp)) + ctx->priv.rflags |= RING_BUFFER_RFLAG_FULL_TIMESTAMP; - if (unlikely(subbuf_offset(offsets->begin, ctx->chan) == 0)) { + if (unlikely(subbuf_offset(offsets->begin, ctx->priv.chan) == 0)) { offsets->switch_new_start = 1; /* For offsets->begin */ } else { offsets->size = config->cb.record_header_size(config, chan, @@ -2123,12 +2160,21 @@ retry: */ offsets->switch_new_end = 1; /* For offsets->begin */ } + /* + * Populate the records lost counters when the space reservation + * may cause a sub-buffer switch. + */ + if (offsets->switch_new_end || offsets->switch_old_end) { + ctx->priv.records_lost_full = v_read(config, &buf->records_lost_full); + ctx->priv.records_lost_wrap = v_read(config, &buf->records_lost_wrap); + ctx->priv.records_lost_big = v_read(config, &buf->records_lost_big); + } return 0; } -static struct lib_ring_buffer *get_current_buf(struct channel *chan, int cpu) +static struct lttng_kernel_ring_buffer *get_current_buf(struct lttng_kernel_ring_buffer_channel *chan, int cpu) { - const struct lib_ring_buffer_config *config = &chan->backend.config; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) return per_cpu_ptr(chan->backend.buf, cpu); @@ -2136,10 +2182,10 @@ static struct lib_ring_buffer *get_current_buf(struct channel *chan, int cpu) return chan->backend.buf; } -void lib_ring_buffer_lost_event_too_big(struct channel *chan) +void lib_ring_buffer_lost_event_too_big(struct lttng_kernel_ring_buffer_channel *chan) { - const struct lib_ring_buffer_config *config = &chan->backend.config; - struct lib_ring_buffer *buf = get_current_buf(chan, smp_processor_id()); + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; + struct lttng_kernel_ring_buffer *buf = get_current_buf(chan, smp_processor_id()); v_inc(config, &buf->records_lost_big); } @@ -2153,16 +2199,16 @@ EXPORT_SYMBOL_GPL(lib_ring_buffer_lost_event_too_big); * -EIO for other errors, else returns 0. * It will take care of sub-buffer switching. */ -int lib_ring_buffer_reserve_slow(struct lib_ring_buffer_ctx *ctx, +int lib_ring_buffer_reserve_slow(struct lttng_kernel_ring_buffer_ctx *ctx, void *client_ctx) { - struct channel *chan = ctx->chan; - const struct lib_ring_buffer_config *config = &chan->backend.config; - struct lib_ring_buffer *buf; + struct lttng_kernel_ring_buffer_channel *chan = ctx->priv.chan; + const struct lttng_kernel_ring_buffer_config *config = &chan->backend.config; + struct lttng_kernel_ring_buffer *buf; struct switch_offsets offsets; int ret; - ctx->buf = buf = get_current_buf(chan, ctx->cpu); + ctx->priv.buf = buf = get_current_buf(chan, ctx->priv.reserve_cpu); offsets.size = 0; do { @@ -2175,12 +2221,12 @@ int lib_ring_buffer_reserve_slow(struct lib_ring_buffer_ctx *ctx, != offsets.old)); /* - * Atomically update last_tsc. This update races against concurrent - * atomic updates, but the race will always cause supplementary full TSC - * records, never the opposite (missing a full TSC record when it would - * be needed). + * Atomically update last_timestamp. This update races against concurrent + * atomic updates, but the race will always cause supplementary + * full timestamp records, never the opposite (missing a full + * timestamp record when it would be needed). */ - save_last_tsc(config, buf, ctx->tsc); + save_last_timestamp(config, buf, ctx->priv.timestamp); /* * Push the reader if necessary @@ -2199,28 +2245,28 @@ int lib_ring_buffer_reserve_slow(struct lib_ring_buffer_ctx *ctx, if (unlikely(offsets.switch_old_end)) { lib_ring_buffer_clear_noref(config, &buf->backend, subbuf_index(offsets.old - 1, chan)); - lib_ring_buffer_switch_old_end(buf, chan, &offsets, ctx->tsc); + lib_ring_buffer_switch_old_end(buf, chan, &offsets, ctx); } /* * Populate new subbuffer. */ if (unlikely(offsets.switch_new_start)) - lib_ring_buffer_switch_new_start(buf, chan, &offsets, ctx->tsc); + lib_ring_buffer_switch_new_start(buf, chan, &offsets, ctx); if (unlikely(offsets.switch_new_end)) - lib_ring_buffer_switch_new_end(buf, chan, &offsets, ctx->tsc); + lib_ring_buffer_switch_new_end(buf, chan, &offsets, ctx); - ctx->slot_size = offsets.size; - ctx->pre_offset = offsets.begin; - ctx->buf_offset = offsets.begin + offsets.pre_header_padding; + ctx->priv.slot_size = offsets.size; + ctx->priv.pre_offset = offsets.begin; + ctx->priv.buf_offset = offsets.begin + offsets.pre_header_padding; return 0; } EXPORT_SYMBOL_GPL(lib_ring_buffer_reserve_slow); static -void lib_ring_buffer_vmcore_check_deliver(const struct lib_ring_buffer_config *config, - struct lib_ring_buffer *buf, +void lib_ring_buffer_vmcore_check_deliver(const struct lttng_kernel_ring_buffer_config *config, + struct lttng_kernel_ring_buffer *buf, unsigned long commit_count, unsigned long idx) { @@ -2234,8 +2280,8 @@ void lib_ring_buffer_vmcore_check_deliver(const struct lib_ring_buffer_config *c */ #ifdef LTTNG_RING_BUFFER_COUNT_EVENTS static -void deliver_count_events(const struct lib_ring_buffer_config *config, - struct lib_ring_buffer *buf, +void deliver_count_events(const struct lttng_kernel_ring_buffer_config *config, + struct lttng_kernel_ring_buffer *buf, unsigned long idx) { v_add(config, subbuffer_get_records_count(config, @@ -2247,21 +2293,21 @@ void deliver_count_events(const struct lib_ring_buffer_config *config, } #else /* LTTNG_RING_BUFFER_COUNT_EVENTS */ static -void deliver_count_events(const struct lib_ring_buffer_config *config, - struct lib_ring_buffer *buf, +void deliver_count_events(const struct lttng_kernel_ring_buffer_config *config, + struct lttng_kernel_ring_buffer *buf, unsigned long idx) { } #endif /* #else LTTNG_RING_BUFFER_COUNT_EVENTS */ -void lib_ring_buffer_check_deliver_slow(const struct lib_ring_buffer_config *config, - struct lib_ring_buffer *buf, - struct channel *chan, +void lib_ring_buffer_check_deliver_slow(const struct lttng_kernel_ring_buffer_config *config, + struct lttng_kernel_ring_buffer *buf, + struct lttng_kernel_ring_buffer_channel *chan, unsigned long offset, unsigned long commit_count, unsigned long idx, - u64 tsc) + const struct lttng_kernel_ring_buffer_ctx *ctx) { unsigned long old_commit_count = commit_count - chan->backend.subbuf_size; @@ -2321,7 +2367,7 @@ void lib_ring_buffer_check_deliver_slow(const struct lib_ring_buffer_config *con config->cb.buffer_end(buf, *ts_end, idx, lib_ring_buffer_get_data_size(config, buf, - idx)); + idx), ctx); /* * Increment the packet counter while we have exclusive @@ -2356,19 +2402,21 @@ void lib_ring_buffer_check_deliver_slow(const struct lib_ring_buffer_config *con commit_count, idx); /* - * RING_BUFFER_WAKEUP_BY_WRITER wakeup is not lock-free. + * RING_BUFFER_WAKEUP_BY_WRITER uses an irq_work to issue + * the wakeups. */ if (config->wakeup == RING_BUFFER_WAKEUP_BY_WRITER && atomic_long_read(&buf->active_readers) && lib_ring_buffer_poll_deliver(config, buf, chan)) { - wake_up_interruptible(&buf->read_wait); - wake_up_interruptible(&chan->read_wait); + irq_work_queue(&buf->wakeup_pending); + irq_work_queue(&chan->wakeup_pending); } } } EXPORT_SYMBOL_GPL(lib_ring_buffer_check_deliver_slow); +static int __init init_lib_ring_buffer_frontend(void) { int cpu; @@ -2380,6 +2428,7 @@ int __init init_lib_ring_buffer_frontend(void) module_init(init_lib_ring_buffer_frontend); +static void __exit exit_lib_ring_buffer_frontend(void) { }