/*
- * ring_buffer_frontend.c
+ * SPDX-License-Identifier: LGPL-2.1-only
*
* 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:
*
* - put_subbuf
*/
-#define _GNU_SOURCE
#define _LGPL_SOURCE
#include <sys/types.h>
#include <sys/mman.h>
#include <fcntl.h>
#include <signal.h>
#include <time.h>
+#include <stdbool.h>
+#include <stdint.h>
#include <urcu/compiler.h>
#include <urcu/ref.h>
#include <urcu/tls-compat.h>
#include "backend.h"
#include "frontend.h"
#include "shm.h"
-#include "tlsfixup.h"
+#include "rb-init.h"
#include "../liblttng-ust/compat.h" /* For ENODATA */
/* Print DBG() messages about events lost only every 1048576 hits */
#define CLOCKID CLOCK_MONOTONIC
#define LTTNG_UST_RING_BUFFER_GET_RETRY 10
#define LTTNG_UST_RING_BUFFER_RETRY_DELAY_MS 10
+#define RETRY_DELAY_MS 100 /* 100 ms. */
/*
* Non-static to ensure the compiler does not optimize away the xor.
.lock = PTHREAD_MUTEX_INITIALIZER,
};
+static bool lttng_ust_allow_blocking;
+
+void lttng_ust_ringbuffer_set_allow_blocking(void)
+{
+ lttng_ust_allow_blocking = true;
+}
+
+/* Get blocking timeout, in ms */
+static int lttng_ust_ringbuffer_get_timeout(struct channel *chan)
+{
+ if (!lttng_ust_allow_blocking)
+ return 0;
+ return chan->u.s.blocking_timeout_ms;
+}
+
/**
* lib_ring_buffer_reset - Reset ring buffer to initial values.
* @buf: Ring buffer.
for (i = 0; i < chan->backend.num_subbuf; i++) {
struct commit_counters_hot *cc_hot;
struct commit_counters_cold *cc_cold;
+ uint64_t *ts_end;
cc_hot = shmp_index(handle, buf->commit_hot, i);
if (!cc_hot)
cc_cold = shmp_index(handle, buf->commit_cold, i);
if (!cc_cold)
return;
+ ts_end = shmp_index(handle, buf->ts_end, i);
+ if (!ts_end)
+ return;
v_set(config, &cc_hot->cc, 0);
v_set(config, &cc_hot->seq, 0);
v_set(config, &cc_cold->cc_sb, 0);
+ *ts_end = 0;
}
uatomic_set(&buf->consumed, 0);
uatomic_set(&buf->record_disabled, 0);
goto free_commit;
}
+ align_shm(shmobj, __alignof__(uint64_t));
+ set_shmp(buf->ts_end,
+ zalloc_shm(shmobj,
+ sizeof(uint64_t) * chan->backend.num_subbuf));
+ if (!shmp(handle, buf->ts_end)) {
+ ret = -ENOMEM;
+ goto free_commit_cold;
+ }
+
+
ret = lib_ring_buffer_backend_create(&buf->backend, &chan->backend,
cpu, handle, shmobj);
if (ret) {
/* Error handling */
free_init:
+ /* ts_end will be freed by shm teardown */
+free_commit_cold:
/* commit_cold will be freed by shm teardown */
free_commit:
/* commit_hot will be freed by shm teardown */
lib_ring_buffer_setup_timer_thread();
+ memset(&sev, 0, sizeof(sev));
sev.sigev_notify = SIGEV_SIGNAL;
sev.sigev_signo = LTTNG_UST_RB_SIG_FLUSH;
sev.sigev_value.sival_ptr = chan;
}
static void channel_free(struct channel *chan,
- struct lttng_ust_shm_handle *handle)
+ struct lttng_ust_shm_handle *handle,
+ int consumer)
{
channel_backend_free(&chan->backend, handle);
/* chan is freed by shm teardown */
- shm_object_table_destroy(handle->table);
+ shm_object_table_destroy(handle->table, consumer);
free(handle);
}
void *buf_addr, size_t subbuf_size,
size_t num_subbuf, unsigned int switch_timer_interval,
unsigned int read_timer_interval,
- const int *stream_fds, int nr_stream_fds)
+ const int *stream_fds, int nr_stream_fds,
+ int64_t blocking_timeout)
{
int ret;
size_t shmsize, chansize;
struct lttng_ust_shm_handle *handle;
struct shm_object *shmobj;
unsigned int nr_streams;
+ int64_t blocking_timeout_ms;
if (config->alloc == RING_BUFFER_ALLOC_PER_CPU)
nr_streams = num_possible_cpus();
if (nr_stream_fds != nr_streams)
return NULL;
+ if (blocking_timeout < -1) {
+ return NULL;
+ }
+ /* usec to msec */
+ if (blocking_timeout == -1) {
+ blocking_timeout_ms = -1;
+ } else {
+ blocking_timeout_ms = blocking_timeout / 1000;
+ if (blocking_timeout_ms != (int32_t) blocking_timeout_ms) {
+ return NULL;
+ }
+ }
+
if (lib_ring_buffer_check_config(config, switch_timer_interval,
read_timer_interval))
return NULL;
/* Calculate the shm allocation layout */
shmsize = sizeof(struct channel);
- shmsize += offset_align(shmsize, __alignof__(struct lttng_ust_lib_ring_buffer_shmp));
+ shmsize += lttng_ust_offset_align(shmsize, __alignof__(struct lttng_ust_lib_ring_buffer_shmp));
shmsize += sizeof(struct lttng_ust_lib_ring_buffer_shmp) * nr_streams;
chansize = shmsize;
if (priv_data_align)
- shmsize += offset_align(shmsize, priv_data_align);
+ shmsize += lttng_ust_offset_align(shmsize, priv_data_align);
shmsize += priv_data_size;
/* Allocate normal memory for channel (not shared) */
shmobj = shm_object_table_alloc(handle->table, shmsize, SHM_OBJECT_MEM,
- -1);
+ -1, -1);
if (!shmobj)
goto error_append;
/* struct channel is at object 0, offset 0 (hardcoded) */
*priv_data = NULL;
}
+ chan->u.s.blocking_timeout_ms = (int32_t) blocking_timeout_ms;
+
ret = channel_backend_init(&chan->backend, name, config,
subbuf_size, num_subbuf, handle,
stream_fds);
error_backend_init:
error_append:
- shm_object_table_destroy(handle->table);
+ shm_object_table_destroy(handle->table, 1);
error_table_alloc:
free(handle);
return NULL;
return handle;
error_table_object:
- shm_object_table_destroy(handle->table);
+ shm_object_table_destroy(handle->table, 0);
error_table_alloc:
free(handle);
return NULL;
}
static
-void channel_release(struct channel *chan, struct lttng_ust_shm_handle *handle)
+void channel_release(struct channel *chan, struct lttng_ust_shm_handle *handle,
+ int consumer)
{
- channel_free(chan, handle);
+ channel_free(chan, handle, consumer);
}
/**
* Call "destroy" callback, finalize channels, 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.
+ * They should release their handle at that point.
*/
void channel_destroy(struct channel *chan, struct lttng_ust_shm_handle *handle,
int consumer)
* sessiond/consumer are keeping a reference on the shm file
* descriptor directly. No need to refcount.
*/
- channel_release(chan, handle);
+ channel_release(chan, handle, consumer);
return;
}
return -EAGAIN;
}
+/**
+ * Performs the same function as lib_ring_buffer_snapshot(), but the positions
+ * are saved regardless of whether the consumed and produced positions are
+ * in the same subbuffer.
+ * @buf: ring buffer
+ * @consumed: consumed byte count indicating the last position read
+ * @produced: produced byte count indicating the last position written
+ *
+ * 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 lttng_ust_lib_ring_buffer *buf,
+ unsigned long *consumed, unsigned long *produced,
+ struct lttng_ust_shm_handle *handle)
+{
+ struct channel *chan;
+ const struct lttng_ust_lib_ring_buffer_config *config;
+
+ chan = shmp(handle, buf->backend.chan);
+ if (!chan)
+ return -EPERM;
+ config = &chan->backend.config;
+ cmm_smp_rmb();
+ *consumed = uatomic_read(&buf->consumed);
+ /*
+ * No need to issue a memory barrier between consumed count read and
+ * write offset read, because consumed count can only change
+ * concurrently in overwrite mode, and we keep a sequence counter
+ * identifier derived from the write offset to check we are getting
+ * the same sub-buffer we are expecting (the sub-buffers are atomically
+ * "tagged" upon writes, tags are checked upon read).
+ */
+ *produced = v_read(config, &buf->offset);
+ return 0;
+}
+
/**
* lib_ring_buffer_move_consumer - move consumed counter forward
* @buf: ring buffer
unsigned long oldidx = subbuf_index(offsets->old - 1, chan);
unsigned long commit_count, padding_size, data_size;
struct commit_counters_hot *cc_hot;
+ uint64_t *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,
handle);
+ ts_end = shmp_index(handle, buf->ts_end, oldidx);
+ if (!ts_end)
+ return;
/*
- * 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.
*/
cmm_smp_wmb();
cc_hot = shmp_index(handle, buf->commit_hot, oldidx);
{
const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
unsigned long endidx, data_size;
+ uint64_t *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,
handle);
+ ts_end = shmp_index(handle, buf->ts_end, endidx);
+ if (!ts_end)
+ return;
+ /*
+ * 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;
}
/*
* Force a sub-buffer switch. This operation is completely reentrant : can be
* called while tracing is active with absolutely no lock held.
*
- * Note, however, that as a v_cmpxchg is used for some atomic
- * operations, this function must be called from the CPU which owns the buffer
- * for a ACTIVE flush.
+ * For RING_BUFFER_SYNC_PER_CPU ring buffers, as a v_cmpxchg is used for
+ * some atomic operations, this function must be called from the CPU
+ * which owns the buffer for a ACTIVE flush. However, for
+ * RING_BUFFER_SYNC_GLOBAL ring buffers, this function can be called
+ * from any CPU.
*/
void lib_ring_buffer_switch_slow(struct lttng_ust_lib_ring_buffer *buf, enum switch_mode mode,
struct lttng_ust_shm_handle *handle)
lib_ring_buffer_switch_old_end(buf, chan, &offsets, tsc, handle);
}
+static
+bool handle_blocking_retry(int *timeout_left_ms)
+{
+ int timeout = *timeout_left_ms, delay;
+
+ if (caa_likely(!timeout))
+ return false; /* Do not retry, discard event. */
+ if (timeout < 0) /* Wait forever. */
+ delay = RETRY_DELAY_MS;
+ else
+ delay = min_t(int, timeout, RETRY_DELAY_MS);
+ (void) poll(NULL, 0, delay);
+ if (timeout > 0)
+ *timeout_left_ms -= delay;
+ return true; /* Retry. */
+}
+
/*
* Returns :
* 0 if ok
int lib_ring_buffer_try_reserve_slow(struct lttng_ust_lib_ring_buffer *buf,
struct channel *chan,
struct switch_offsets *offsets,
- struct lttng_ust_lib_ring_buffer_ctx *ctx)
+ struct lttng_ust_lib_ring_buffer_ctx *ctx,
+ void *client_ctx)
{
const struct lttng_ust_lib_ring_buffer_config *config = &chan->backend.config;
struct lttng_ust_shm_handle *handle = ctx->handle;
unsigned long reserve_commit_diff, offset_cmp;
+ int timeout_left_ms = lttng_ust_ringbuffer_get_timeout(chan);
retry:
offsets->begin = offset_cmp = v_read(config, &buf->offset);
offsets->size = config->cb.record_header_size(config, chan,
offsets->begin,
&offsets->pre_header_padding,
- ctx);
+ ctx, client_ctx);
offsets->size +=
lib_ring_buffer_align(offsets->begin + offsets->size,
ctx->largest_align)
>= chan->backend.buf_size)) {
unsigned long nr_lost;
+ if (handle_blocking_retry(&timeout_left_ms))
+ goto retry;
+
/*
* We do not overwrite non consumed buffers
* and we are full : record is lost.
config->cb.record_header_size(config, chan,
offsets->begin,
&offsets->pre_header_padding,
- ctx);
+ ctx, client_ctx);
offsets->size +=
lib_ring_buffer_align(offsets->begin + offsets->size,
ctx->largest_align)
* -EIO for other errors, else returns 0.
* It will take care of sub-buffer switching.
*/
-int lib_ring_buffer_reserve_slow(struct lttng_ust_lib_ring_buffer_ctx *ctx)
+int lib_ring_buffer_reserve_slow(struct lttng_ust_lib_ring_buffer_ctx *ctx,
+ void *client_ctx)
{
struct channel *chan = ctx->chan;
struct lttng_ust_shm_handle *handle = ctx->handle;
do {
ret = lib_ring_buffer_try_reserve_slow(buf, chan, &offsets,
- ctx);
+ ctx, client_ctx);
if (caa_unlikely(ret))
return ret;
} while (caa_unlikely(v_cmpxchg(config, &buf->offset, offsets.old,
if (caa_likely(v_cmpxchg(config, &cc_cold->cc_sb,
old_commit_count, old_commit_count + 1)
== old_commit_count)) {
+ uint64_t *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.
*/
+ cmm_smp_mb();
+ ts_end = shmp_index(handle, buf->ts_end, idx);
+ if (!ts_end)
+ return;
deliver_count_events(config, buf, idx, handle);
- config->cb.buffer_end(buf, tsc, idx,
+ config->cb.buffer_end(buf, *ts_end, idx,
lib_ring_buffer_get_data_size(config,
buf,
idx,
projects / lttng-ust.git / blobdiff