/*
* ring_buffer_frontend.c
*
- * (C) Copyright 2005-2010 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+ * Copyright (C) 2005-2012 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+ *
+ * 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:
* - splice one subbuffer worth of data to a pipe
* - splice the data from pipe to disk/network
* - put_subbuf
- *
- * Dual LGPL v2.1/GPL v2 license.
*/
#include <linux/delay.h>
struct switch_offsets {
unsigned long begin, end, old;
size_t pre_header_padding, size;
- unsigned int switch_new_start:1, switch_new_end:1, switch_old_start:1,
- switch_old_end:1;
+ unsigned int switch_new_start:1, switch_old_start:1, switch_old_end:1;
};
#ifdef CONFIG_NO_HZ
void lib_ring_buffer_reset(struct lib_ring_buffer *buf)
{
struct channel *chan = 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;
/*
int lib_ring_buffer_create(struct lib_ring_buffer *buf,
struct channel_backend *chanb, int cpu)
{
- const struct lib_ring_buffer_config *config = chanb->config;
+ const struct lib_ring_buffer_config *config = &chanb->config;
struct channel *chan = container_of(chanb, struct channel, backend);
void *priv = chanb->priv;
- unsigned int num_subbuf;
size_t subbuf_header_size;
u64 tsc;
int ret;
goto free_commit;
}
- num_subbuf = chan->backend.num_subbuf;
init_waitqueue_head(&buf->read_wait);
+ init_waitqueue_head(&buf->write_wait);
raw_spin_lock_init(&buf->raw_tick_nohz_spinlock);
/*
{
struct lib_ring_buffer *buf = (struct lib_ring_buffer *)data;
struct channel *chan = 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.
static void lib_ring_buffer_start_switch_timer(struct lib_ring_buffer *buf)
{
struct channel *chan = 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;
{
struct lib_ring_buffer *buf = (struct lib_ring_buffer *)data;
struct channel *chan = 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);
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;
+ const struct lib_ring_buffer_config *config = &chan->backend.config;
if (config->wakeup != RING_BUFFER_WAKEUP_BY_TIMER
|| !chan->read_timer_interval
static void lib_ring_buffer_stop_read_timer(struct lib_ring_buffer *buf)
{
struct channel *chan = 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
* Returns the success/failure of the operation. (%NOTIFY_OK, %NOTIFY_BAD)
*/
static
-int __cpuinit lib_ring_buffer_cpu_hp_callback(struct notifier_block *nb,
+int lib_ring_buffer_cpu_hp_callback(struct notifier_block *nb,
unsigned long action,
void *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;
+ const struct lib_ring_buffer_config *config = &chan->backend.config;
if (!chan->cpu_hp_enable)
return NOTIFY_DONE;
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;
{
struct channel *chan = container_of(nb, struct channel,
tick_nohz_notifier);
- 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;
int cpu = smp_processor_id();
*/
static void channel_unregister_notifiers(struct channel *chan)
{
- const struct lib_ring_buffer_config *config = chan->backend.config;
+ const struct lib_ring_buffer_config *config = &chan->backend.config;
int cpu;
channel_iterator_unregister_notifiers(chan);
chan->read_timer_interval = usecs_to_jiffies(read_timer_interval);
kref_init(&chan->ref);
init_waitqueue_head(&chan->read_wait);
+ init_waitqueue_head(&chan->hp_wait);
if (config->alloc == RING_BUFFER_ALLOC_PER_CPU) {
#if defined(CONFIG_NO_HZ) && defined(CONFIG_LIB_RING_BUFFER)
* @chan: channel to destroy
*
* Holds cpu hotplug.
- * Call "destroy" callback, finalize channels, wait for readers to release their
- * reference, then destroy ring buffer data. Note that when readers have
- * completed data consumption of finalized channels, get_subbuf() will return
- * -ENODATA. They should release their handle at that point.
- * Returns the private data pointer.
+ * Call "destroy" callback, finalize channels, and then decrement the
+ * channel reference count. Note that when readers have completed data
+ * consumption of finalized channels, get_subbuf() will return -ENODATA.
+ * They should release their handle at that point. Returns the private
+ * data pointer.
*/
void *channel_destroy(struct channel *chan)
{
int cpu;
- const struct lib_ring_buffer_config *config = chan->backend.config;
+ const struct lib_ring_buffer_config *config = &chan->backend.config;
void *priv;
channel_unregister_notifiers(chan);
ACCESS_ONCE(buf->finalized) = 1;
wake_up_interruptible(&buf->read_wait);
}
+ ACCESS_ONCE(chan->finalized) = 1;
+ wake_up_interruptible(&chan->hp_wait);
wake_up_interruptible(&chan->read_wait);
- kref_put(&chan->ref, channel_release);
priv = chan->backend.priv;
+ kref_put(&chan->ref, channel_release);
return priv;
}
EXPORT_SYMBOL_GPL(channel_destroy);
unsigned long *consumed, unsigned long *produced)
{
struct channel *chan = 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;
/**
* lib_ring_buffer_put_snapshot - move consumed counter forward
+ *
+ * Should only be called from consumer context.
* @buf: ring buffer
* @consumed_new: new consumed count value
*/
while ((long) consumed - (long) consumed_new < 0)
consumed = atomic_long_cmpxchg(&buf->consumed, consumed,
consumed_new);
+ /* Wake-up the metadata producer */
+ wake_up_interruptible(&buf->write_wait);
}
EXPORT_SYMBOL_GPL(lib_ring_buffer_move_consumer);
unsigned long consumed)
{
struct channel *chan = 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;
+ if (buf->get_subbuf) {
+ /*
+ * Reader is trying to get a subbuffer twice.
+ */
+ CHAN_WARN_ON(chan, 1);
+ return -EBUSY;
+ }
retry:
finalized = ACCESS_ONCE(buf->finalized);
/*
{
struct lib_ring_buffer_backend *bufb = &buf->backend;
struct channel *chan = 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, atomic_long_read(&buf->active_readers) != 1);
unsigned long cons_offset,
int cpu)
{
- 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);
struct channel *chan,
void *priv, int cpu)
{
- 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;
- /*
- * Can be called in the error path of allocation when
- * trans_channel_data is not yet set.
- */
- if (!chan)
- return;
/*
* No need to order commit_count, write_offset and cons_offset reads
* because we execute at teardown when no more writer nor reader
write_offset = v_read(config, &buf->offset);
cons_offset = atomic_long_read(&buf->consumed);
if (write_offset != cons_offset)
- printk(KERN_WARNING
+ printk(KERN_DEBUG
"ring buffer %s, cpu %d: "
"non-consumed data\n"
" [ %lu bytes written, %lu bytes read ]\n",
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;
+ const struct lib_ring_buffer_config *config = &chan->backend.config;
void *priv = chan->backend.priv;
- printk(KERN_DEBUG "ring buffer %s, cpu %d: %lu records written, "
- "%lu records overrun\n",
- chan->backend.name, cpu,
- v_read(config, &buf->records_count),
- v_read(config, &buf->records_overrun));
-
- if (v_read(config, &buf->records_lost_full)
- || v_read(config, &buf->records_lost_wrap)
- || v_read(config, &buf->records_lost_big))
- printk(KERN_WARNING
- "ring buffer %s, cpu %d: records were lost. Caused by:\n"
- " [ %lu buffer full, %lu nest buffer wrap-around, "
- "%lu event too big ]\n",
- chan->backend.name, cpu,
- v_read(config, &buf->records_lost_full),
- v_read(config, &buf->records_lost_wrap),
- v_read(config, &buf->records_lost_big));
-
+ if (!strcmp(chan->backend.name, "relay-metadata")) {
+ printk(KERN_DEBUG "ring buffer %s: %lu records written, "
+ "%lu records overrun\n",
+ chan->backend.name,
+ v_read(config, &buf->records_count),
+ v_read(config, &buf->records_overrun));
+ } else {
+ printk(KERN_DEBUG "ring buffer %s, cpu %d: %lu records written, "
+ "%lu records overrun\n",
+ chan->backend.name, cpu,
+ v_read(config, &buf->records_count),
+ v_read(config, &buf->records_overrun));
+
+ if (v_read(config, &buf->records_lost_full)
+ || v_read(config, &buf->records_lost_wrap)
+ || v_read(config, &buf->records_lost_big))
+ printk(KERN_WARNING
+ "ring buffer %s, cpu %d: records were lost. Caused by:\n"
+ " [ %lu buffer full, %lu nest buffer wrap-around, "
+ "%lu event too big ]\n",
+ chan->backend.name, cpu,
+ v_read(config, &buf->records_lost_full),
+ v_read(config, &buf->records_lost_wrap),
+ v_read(config, &buf->records_lost_big));
+ }
lib_ring_buffer_print_buffer_errors(buf, chan, priv, cpu);
}
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 oldidx = subbuf_index(offsets->old, chan);
unsigned long commit_count;
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 oldidx = subbuf_index(offsets->old - 1, chan);
unsigned long commit_count, padding_size, data_size;
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 beginidx = subbuf_index(offsets->begin, chan);
unsigned long commit_count;
config->cb.subbuffer_header_size());
}
-/*
- * lib_ring_buffer_switch_new_end: finish switching current subbuffer
- *
- * The only remaining threads could be the ones with pending commits. They will
- * have to do the deliver themselves.
- */
-static
-void lib_ring_buffer_switch_new_end(struct lib_ring_buffer *buf,
- struct channel *chan,
- struct switch_offsets *offsets,
- u64 tsc)
-{
- 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;
-
- data_size = subbuf_offset(offsets->end - 1, chan) + 1;
- padding_size = chan->backend.subbuf_size - data_size;
- subbuffer_set_data_size(config, &buf->backend, endidx, data_size);
-
- /*
- * Order all writes to buffer before the commit count update that will
- * determine that the subbuffer is full.
- */
- if (config->ipi == RING_BUFFER_IPI_BARRIER) {
- /*
- * Must write slot data before incrementing commit count. This
- * compiler barrier is upgraded into a smp_mb() by the IPI sent
- * by get_subbuf().
- */
- barrier();
- } else
- smp_wmb();
- v_add(config, padding_size, &buf->commit_hot[endidx].cc);
- commit_count = v_read(config, &buf->commit_hot[endidx].cc);
- lib_ring_buffer_check_deliver(config, buf, chan, offsets->end - 1,
- commit_count, endidx);
- lib_ring_buffer_write_commit_counter(config, buf, chan, endidx,
- offsets->end, commit_count,
- padding_size);
-}
-
/*
* Returns :
* 0 if ok
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);
*/
if (mode == SWITCH_FLUSH || off > 0) {
if (unlikely(off == 0)) {
+ /*
+ * A final flush that encounters an empty
+ * sub-buffer cannot switch buffer if a
+ * reader is located within this sub-buffer.
+ * Anyway, the purpose of final flushing of a
+ * sub-buffer at offset 0 is to handle the case
+ * of entirely empty stream.
+ */
+ if (unlikely(subbuf_trunc(offsets->begin, chan)
+ - subbuf_trunc((unsigned long)
+ atomic_long_read(&buf->consumed), chan)
+ >= chan->backend.buf_size))
+ return -1;
/*
* The client does not save any header information.
* Don't switch empty subbuffer on finalize, because it
void lib_ring_buffer_switch_slow(struct lib_ring_buffer *buf, enum switch_mode mode)
{
struct channel *chan = 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;
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;
unsigned long reserve_commit_diff;
offsets->begin = v_read(config, &buf->offset);
offsets->old = offsets->begin;
offsets->switch_new_start = 0;
- offsets->switch_new_end = 0;
offsets->switch_old_end = 0;
offsets->pre_header_padding = 0;
return -EIO;
if (last_tsc_overflow(config, buf, ctx->tsc))
- ctx->rflags = RING_BUFFER_RFLAG_FULL_TSC;
+ ctx->rflags |= RING_BUFFER_RFLAG_FULL_TSC;
if (unlikely(subbuf_offset(offsets->begin, ctx->chan) == 0)) {
offsets->switch_new_start = 1; /* For offsets->begin */
} else {
offsets->size = config->cb.record_header_size(config, chan,
offsets->begin,
- ctx->data_size,
&offsets->pre_header_padding,
- ctx->rflags, ctx);
+ ctx);
offsets->size +=
lib_ring_buffer_align(offsets->begin + offsets->size,
ctx->largest_align)
offsets->size =
config->cb.record_header_size(config, chan,
offsets->begin,
- ctx->data_size,
&offsets->pre_header_padding,
- ctx->rflags, ctx);
+ ctx);
offsets->size +=
lib_ring_buffer_align(offsets->begin + offsets->size,
ctx->largest_align)
*/
}
offsets->end = offsets->begin + offsets->size;
-
- if (unlikely(subbuf_offset(offsets->end, chan) == 0)) {
- /*
- * The offset_end will fall at the very beginning of the next
- * subbuffer.
- */
- offsets->switch_new_end = 1; /* For offsets->begin */
- }
return 0;
}
int lib_ring_buffer_reserve_slow(struct lib_ring_buffer_ctx *ctx)
{
struct channel *chan = ctx->chan;
- 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;
if (unlikely(offsets.switch_new_start))
lib_ring_buffer_switch_new_start(buf, chan, &offsets, ctx->tsc);
- if (unlikely(offsets.switch_new_end))
- lib_ring_buffer_switch_new_end(buf, chan, &offsets, ctx->tsc);
-
ctx->slot_size = offsets.size;
ctx->pre_offset = offsets.begin;
ctx->buf_offset = offsets.begin + offsets.pre_header_padding;
return 0;
}
EXPORT_SYMBOL_GPL(lib_ring_buffer_reserve_slow);
+
+int __init init_lib_ring_buffer_frontend(void)
+{
+ int cpu;
+
+ for_each_possible_cpu(cpu)
+ spin_lock_init(&per_cpu(ring_buffer_nohz_lock, cpu));
+ return 0;
+}
+
+module_init(init_lib_ring_buffer_frontend);
+
+void __exit exit_lib_ring_buffer_frontend(void)
+{
+}
+
+module_exit(exit_lib_ring_buffer_frontend);