/*
- * Copyright (C) 2011 - Julien Desfossez <julien.desfossez@polymtl.ca>
- * Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
- * 2012 - David Goulet <dgoulet@efficios.com>
+ * Copyright (C) 2011 Julien Desfossez <julien.desfossez@polymtl.ca>
+ * Copyright (C) 2011 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
+ * Copyright (C) 2012 David Goulet <dgoulet@efficios.com>
*
- * This program is free software; you can redistribute it and/or modify
- * it under the terms of the GNU General Public License, version 2 only,
- * as published by the Free Software Foundation.
+ * SPDX-License-Identifier: GPL-2.0-only
*
- * This program 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 General Public License for
- * more details.
- *
- * You should have received a copy of the GNU General Public License along
- * with this program; if not, write to the Free Software Foundation, Inc.,
- * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*/
+#include "common/index/ctf-index.h"
#define _LGPL_SOURCE
#include <assert.h>
#include <poll.h>
static struct lttng_ht *metadata_ht;
static struct lttng_ht *data_ht;
+static const char *get_consumer_domain(void)
+{
+ switch (consumer_data.type) {
+ case LTTNG_CONSUMER_KERNEL:
+ return DEFAULT_KERNEL_TRACE_DIR;
+ case LTTNG_CONSUMER64_UST:
+ /* Fall-through. */
+ case LTTNG_CONSUMER32_UST:
+ return DEFAULT_UST_TRACE_DIR;
+ default:
+ abort();
+ }
+}
+
/*
* Notify a thread lttng pipe to poll back again. This usually means that some
* global state has changed so we just send back the thread in a poll wait
channel->tracefile_size;
}
-struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
- uint64_t stream_key,
- const char *channel_name,
- uint64_t relayd_id,
- uint64_t session_id,
- struct lttng_trace_chunk *trace_chunk,
- int cpu,
- int *alloc_ret,
- enum consumer_channel_type type,
- unsigned int monitor)
-{
- int ret;
- struct lttng_consumer_stream *stream;
-
- stream = zmalloc(sizeof(*stream));
- if (stream == NULL) {
- PERROR("malloc struct lttng_consumer_stream");
- ret = -ENOMEM;
- goto end;
- }
-
- if (trace_chunk && !lttng_trace_chunk_get(trace_chunk)) {
- ERR("Failed to acquire trace chunk reference during the creation of a stream");
- ret = -1;
- goto error;
- }
-
- rcu_read_lock();
- stream->key = stream_key;
- stream->trace_chunk = trace_chunk;
- stream->out_fd = -1;
- stream->out_fd_offset = 0;
- stream->output_written = 0;
- stream->net_seq_idx = relayd_id;
- stream->session_id = session_id;
- stream->monitor = monitor;
- stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
- stream->index_file = NULL;
- stream->last_sequence_number = -1ULL;
- pthread_mutex_init(&stream->lock, NULL);
- pthread_mutex_init(&stream->metadata_timer_lock, NULL);
-
- /* If channel is the metadata, flag this stream as metadata. */
- if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
- stream->metadata_flag = 1;
- /* Metadata is flat out. */
- strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
- /* Live rendez-vous point. */
- pthread_cond_init(&stream->metadata_rdv, NULL);
- pthread_mutex_init(&stream->metadata_rdv_lock, NULL);
- } else {
- /* Format stream name to <channel_name>_<cpu_number> */
- ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
- channel_name, cpu);
- if (ret < 0) {
- PERROR("snprintf stream name");
- goto error;
- }
- }
-
- /* Key is always the wait_fd for streams. */
- lttng_ht_node_init_u64(&stream->node, stream->key);
-
- /* Init node per channel id key */
- lttng_ht_node_init_u64(&stream->node_channel_id, channel_key);
-
- /* Init session id node with the stream session id */
- lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id);
-
- DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64
- " relayd_id %" PRIu64 ", session_id %" PRIu64,
- stream->name, stream->key, channel_key,
- stream->net_seq_idx, stream->session_id);
-
- rcu_read_unlock();
- return stream;
-
-error:
- rcu_read_unlock();
- lttng_trace_chunk_put(stream->trace_chunk);
- free(stream);
-end:
- if (alloc_ret) {
- *alloc_ret = ret;
- }
- return NULL;
-}
-
/*
* Add a stream to the global list protected by a mutex.
*/
pthread_mutex_unlock(&consumer_data.lock);
}
-void consumer_del_data_stream(struct lttng_consumer_stream *stream)
-{
- consumer_del_stream(stream, data_ht);
-}
-
/*
* Add relayd socket to global consumer data hashtable. RCU read side lock MUST
* be acquired before calling this.
/* Add stream on the relayd */
pthread_mutex_lock(&relayd->ctrl_sock_mutex);
ret = relayd_add_stream(&relayd->control_sock, stream->name,
- path, &stream->relayd_stream_id,
+ get_consumer_domain(), path, &stream->relayd_stream_id,
stream->chan->tracefile_size,
stream->chan->tracefile_count,
stream->trace_chunk);
struct lttng_consumer_channel *channel,
struct lttng_trace_chunk *new_trace_chunk)
{
- int ret = 0;
- const bool is_local_trace = channel->relayd_id == -1ULL;
- bool update_stream_trace_chunk;
- struct cds_lfht_iter iter;
- struct lttng_consumer_stream *stream;
- unsigned long channel_hash;
-
pthread_mutex_lock(&channel->lock);
if (channel->is_deleted) {
/*
*/
goto end;
}
- /*
- * A stream can transition to a state where it and its channel
- * no longer belong to a trace chunk. For instance, this happens when
- * a session is rotated while it is inactive. After the rotation
- * of an inactive session completes, the channel and its streams no
- * longer belong to a trace chunk.
- *
- * However, if a session is stopped, rotated, and started again,
- * the session daemon will create a new chunk and send it to its peers.
- * In that case, the streams' transition to a new chunk can be performed
- * immediately.
- *
- * This trace chunk transition could also be performed lazily when
- * a buffer is consumed. However, creating the files here allows the
- * consumer daemon to report any creation error to the session daemon
- * and cause the start of the tracing session to fail.
- */
- update_stream_trace_chunk = !channel->trace_chunk && new_trace_chunk;
/*
* The acquisition of the reference cannot fail (barring
lttng_trace_chunk_put(channel->trace_chunk);
channel->trace_chunk = new_trace_chunk;
- if (!is_local_trace || !new_trace_chunk) {
- /* Not an error. */
- goto end;
- }
-
- if (!update_stream_trace_chunk) {
- goto end;
- }
-
- channel_hash = consumer_data.stream_per_chan_id_ht->hash_fct(
- &channel->key, lttng_ht_seed);
- rcu_read_lock();
- cds_lfht_for_each_entry_duplicate(consumer_data.stream_per_chan_id_ht->ht,
- channel_hash,
- consumer_data.stream_per_chan_id_ht->match_fct,
- &channel->key, &iter, stream, node_channel_id.node) {
- bool acquired_reference, should_regenerate_metadata = false;
-
- acquired_reference = lttng_trace_chunk_get(channel->trace_chunk);
- assert(acquired_reference);
-
- pthread_mutex_lock(&stream->lock);
-
- /*
- * On a transition from "no-chunk" to a new chunk, a metadata
- * stream's content must be entirely dumped. This must occcur
- * _after_ the creation of the metadata stream's output files
- * as the consumption thread (not necessarily the one executing
- * this) may start to consume during the call to
- * consumer_metadata_stream_dump().
- */
- should_regenerate_metadata =
- stream->metadata_flag &&
- !stream->trace_chunk && channel->trace_chunk;
- stream->trace_chunk = channel->trace_chunk;
- ret = consumer_stream_create_output_files(stream, true);
- if (ret) {
- pthread_mutex_unlock(&stream->lock);
- goto end_rcu_unlock;
- }
- if (should_regenerate_metadata) {
- ret = consumer_metadata_stream_dump(stream);
- }
- pthread_mutex_unlock(&stream->lock);
- if (ret) {
- goto end_rcu_unlock;
- }
- }
-end_rcu_unlock:
- rcu_read_unlock();
end:
pthread_mutex_unlock(&channel->lock);
- return ret;
+ return 0;
}
/*
uint64_t session_id_per_pid,
unsigned int monitor,
unsigned int live_timer_interval,
+ bool is_in_live_session,
const char *root_shm_path,
const char *shm_path)
{
channel->tracefile_count = tracefile_count;
channel->monitor = monitor;
channel->live_timer_interval = live_timer_interval;
+ channel->is_live = is_in_live_session;
pthread_mutex_init(&channel->lock, NULL);
pthread_mutex_init(&channel->timer_lock, NULL);
struct lttng_consumer_local_data *lttng_consumer_create(
enum lttng_consumer_type type,
ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream,
- struct lttng_consumer_local_data *ctx),
+ struct lttng_consumer_local_data *ctx, bool locked_by_caller),
int (*recv_channel)(struct lttng_consumer_channel *channel),
int (*recv_stream)(struct lttng_consumer_stream *stream),
int (*update_stream)(uint64_t stream_key, uint32_t state))
* Returns the number of bytes written
*/
ssize_t lttng_consumer_on_read_subbuffer_mmap(
- struct lttng_consumer_local_data *ctx,
- struct lttng_consumer_stream *stream, unsigned long len,
- unsigned long padding,
- struct ctf_packet_index *index)
+ struct lttng_consumer_stream *stream,
+ const struct lttng_buffer_view *buffer,
+ unsigned long padding)
{
- unsigned long mmap_offset;
- void *mmap_base;
ssize_t ret = 0;
off_t orig_offset = stream->out_fd_offset;
/* Default is on the disk */
int outfd = stream->out_fd;
struct consumer_relayd_sock_pair *relayd = NULL;
unsigned int relayd_hang_up = 0;
+ const size_t subbuf_content_size = buffer->size - padding;
+ size_t write_len;
/* RCU lock for the relayd pointer */
rcu_read_lock();
}
}
- /* get the offset inside the fd to mmap */
- switch (consumer_data.type) {
- case LTTNG_CONSUMER_KERNEL:
- mmap_base = stream->mmap_base;
- ret = kernctl_get_mmap_read_offset(stream->wait_fd, &mmap_offset);
- if (ret < 0) {
- PERROR("tracer ctl get_mmap_read_offset");
- goto end;
- }
- break;
- case LTTNG_CONSUMER32_UST:
- case LTTNG_CONSUMER64_UST:
- mmap_base = lttng_ustctl_get_mmap_base(stream);
- if (!mmap_base) {
- ERR("read mmap get mmap base for stream %s", stream->name);
- ret = -EPERM;
- goto end;
- }
- ret = lttng_ustctl_get_mmap_read_offset(stream, &mmap_offset);
- if (ret != 0) {
- PERROR("tracer ctl get_mmap_read_offset");
- ret = -EINVAL;
- goto end;
- }
- break;
- default:
- ERR("Unknown consumer_data type");
- assert(0);
- }
-
/* Handle stream on the relayd if the output is on the network */
if (relayd) {
- unsigned long netlen = len;
+ unsigned long netlen = subbuf_content_size;
/*
* Lock the control socket for the complete duration of the function
goto write_error;
}
}
- } else {
- /* No streaming, we have to set the len with the full padding */
- len += padding;
+ write_len = subbuf_content_size;
+ } else {
+ /* No streaming; we have to write the full padding. */
if (stream->metadata_flag && stream->reset_metadata_flag) {
ret = utils_truncate_stream_file(stream->out_fd, 0);
if (ret < 0) {
* Check if we need to change the tracefile before writing the packet.
*/
if (stream->chan->tracefile_size > 0 &&
- (stream->tracefile_size_current + len) >
+ (stream->tracefile_size_current + buffer->size) >
stream->chan->tracefile_size) {
ret = consumer_stream_rotate_output_files(stream);
if (ret) {
outfd = stream->out_fd;
orig_offset = 0;
}
- stream->tracefile_size_current += len;
- if (index) {
- index->offset = htobe64(stream->out_fd_offset);
- }
+ stream->tracefile_size_current += buffer->size;
+ write_len = buffer->size;
}
/*
* This call guarantee that len or less is returned. It's impossible to
* receive a ret value that is bigger than len.
*/
- ret = lttng_write(outfd, mmap_base + mmap_offset, len);
- DBG("Consumer mmap write() ret %zd (len %lu)", ret, len);
- if (ret < 0 || ((size_t) ret != len)) {
+ ret = lttng_write(outfd, buffer->data, write_len);
+ DBG("Consumer mmap write() ret %zd (len %zu)", ret, write_len);
+ if (ret < 0 || ((size_t) ret != write_len)) {
/*
* Report error to caller if nothing was written else at least send the
* amount written.
DBG("Consumer mmap write detected relayd hang up");
} else {
/* Unhandled error, print it and stop function right now. */
- PERROR("Error in write mmap (ret %zd != len %lu)", ret, len);
+ PERROR("Error in write mmap (ret %zd != write_len %zu)", ret,
+ write_len);
}
goto write_error;
}
/* This call is useless on a socket so better save a syscall. */
if (!relayd) {
/* This won't block, but will start writeout asynchronously */
- lttng_sync_file_range(outfd, stream->out_fd_offset, len,
+ lttng_sync_file_range(outfd, stream->out_fd_offset, write_len,
SYNC_FILE_RANGE_WRITE);
- stream->out_fd_offset += len;
+ stream->out_fd_offset += write_len;
lttng_consumer_sync_trace_file(stream, orig_offset);
}
ssize_t lttng_consumer_on_read_subbuffer_splice(
struct lttng_consumer_local_data *ctx,
struct lttng_consumer_stream *stream, unsigned long len,
- unsigned long padding,
- struct ctf_packet_index *index)
+ unsigned long padding)
{
ssize_t ret = 0, written = 0, ret_splice = 0;
loff_t offset = 0;
orig_offset = 0;
}
stream->tracefile_size_current += len;
- index->offset = htobe64(stream->out_fd_offset);
}
while (len > 0) {
}
}
+static
void lttng_consumer_close_all_metadata(void)
{
switch (consumer_data.type) {
do {
health_code_update();
- len = ctx->on_buffer_ready(stream, ctx);
+ len = ctx->on_buffer_ready(stream, ctx, false);
/*
* We don't check the return value here since if we get
* a negative len, it means an error occurred thus we
do {
health_code_update();
- len = ctx->on_buffer_ready(stream, ctx);
+ len = ctx->on_buffer_ready(stream, ctx, false);
/*
* We don't check the return value here since if we get
* a negative len, it means an error occurred thus we
if (pollfd[i].revents & POLLPRI) {
DBG("Urgent read on fd %d", pollfd[i].fd);
high_prio = 1;
- len = ctx->on_buffer_ready(local_stream[i], ctx);
+ len = ctx->on_buffer_ready(local_stream[i], ctx, false);
/* it's ok to have an unavailable sub-buffer */
if (len < 0 && len != -EAGAIN && len != -ENODATA) {
/* Clean the stream and free it. */
local_stream[i]->hangup_flush_done ||
local_stream[i]->has_data) {
DBG("Normal read on fd %d", pollfd[i].fd);
- len = ctx->on_buffer_ready(local_stream[i], ctx);
+ len = ctx->on_buffer_ready(local_stream[i], ctx, false);
/* it's ok to have an unavailable sub-buffer */
if (len < 0 && len != -EAGAIN && len != -ENODATA) {
/* Clean the stream and free it. */
}
ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
- struct lttng_consumer_local_data *ctx)
+ struct lttng_consumer_local_data *ctx,
+ bool locked_by_caller)
{
- ssize_t ret;
+ ssize_t ret, written_bytes;
+ int rotation_ret;
+ struct stream_subbuffer subbuffer = {};
- pthread_mutex_lock(&stream->chan->lock);
- pthread_mutex_lock(&stream->lock);
- if (stream->metadata_flag) {
- pthread_mutex_lock(&stream->metadata_rdv_lock);
+ if (!locked_by_caller) {
+ stream->read_subbuffer_ops.lock(stream);
}
- switch (consumer_data.type) {
- case LTTNG_CONSUMER_KERNEL:
- ret = lttng_kconsumer_read_subbuffer(stream, ctx);
- break;
- case LTTNG_CONSUMER32_UST:
- case LTTNG_CONSUMER64_UST:
- ret = lttng_ustconsumer_read_subbuffer(stream, ctx);
- break;
- default:
- ERR("Unknown consumer_data type");
- assert(0);
- ret = -ENOSYS;
- break;
+ if (stream->read_subbuffer_ops.on_wake_up) {
+ ret = stream->read_subbuffer_ops.on_wake_up(stream);
+ if (ret) {
+ goto end;
+ }
}
- if (stream->metadata_flag) {
- pthread_cond_broadcast(&stream->metadata_rdv);
- pthread_mutex_unlock(&stream->metadata_rdv_lock);
+ /*
+ * If the stream was flagged to be ready for rotation before we extract
+ * the next packet, rotate it now.
+ */
+ if (stream->rotate_ready) {
+ DBG("Rotate stream before consuming data");
+ ret = lttng_consumer_rotate_stream(ctx, stream);
+ if (ret < 0) {
+ ERR("Stream rotation error before consuming data");
+ goto end;
+ }
+ }
+
+ ret = stream->read_subbuffer_ops.get_next_subbuffer(stream, &subbuffer);
+ if (ret) {
+ if (ret == -ENODATA) {
+ /* Not an error. */
+ ret = 0;
+ goto sleep_stream;
+ }
+ goto end;
+ }
+
+ ret = stream->read_subbuffer_ops.pre_consume_subbuffer(
+ stream, &subbuffer);
+ if (ret) {
+ goto error_put_subbuf;
+ }
+
+ written_bytes = stream->read_subbuffer_ops.consume_subbuffer(
+ ctx, stream, &subbuffer);
+ /*
+ * Should write subbuf_size amount of data when network streaming or
+ * the full padded size when we are not streaming.
+ */
+ if ((written_bytes != subbuffer.info.data.subbuf_size &&
+ stream->net_seq_idx != (uint64_t) -1ULL) ||
+ (written_bytes != subbuffer.info.data.padded_subbuf_size &&
+ stream->net_seq_idx ==
+ (uint64_t) -1ULL)) {
+ /*
+ * Display the error but continue processing to try to
+ * release the subbuffer. This is a DBG statement
+ * since this can happen without being a critical
+ * error.
+ */
+ DBG("Failed to write to tracefile (written_bytes: %zd != padded subbuffer size: %lu, subbuffer size: %lu)",
+ written_bytes, subbuffer.info.data.subbuf_size,
+ subbuffer.info.data.padded_subbuf_size);
+ }
+
+ ret = stream->read_subbuffer_ops.put_next_subbuffer(stream, &subbuffer);
+ if (ret) {
+ goto end;
+ }
+
+ if (stream->read_subbuffer_ops.post_consume) {
+ ret = stream->read_subbuffer_ops.post_consume(stream, &subbuffer, ctx);
+ if (ret) {
+ goto end;
+ }
+ }
+
+ /*
+ * After extracting the packet, we check if the stream is now ready to
+ * be rotated and perform the action immediately.
+ *
+ * Don't overwrite `ret` as callers expect the number of bytes
+ * consumed to be returned on success.
+ */
+ rotation_ret = lttng_consumer_stream_is_rotate_ready(stream);
+ if (rotation_ret == 1) {
+ rotation_ret = lttng_consumer_rotate_stream(ctx, stream);
+ if (rotation_ret < 0) {
+ ret = rotation_ret;
+ ERR("Stream rotation error after consuming data");
+ goto end;
+ }
+ } else if (rotation_ret < 0) {
+ ret = rotation_ret;
+ ERR("Failed to check if stream was ready to rotate after consuming data");
+ goto end;
+ }
+
+stream_sleep:
+ if (stream->read_subbuffer_ops.on_sleep) {
+ stream->read_subbuffer_ops.on_sleep(stream, ctx);
+ }
+
+ ret = written_bytes;
+end:
+ if (!locked_by_caller) {
+ stream->read_subbuffer_ops.unlock(stream);
}
- pthread_mutex_unlock(&stream->lock);
- pthread_mutex_unlock(&stream->chan->lock);
return ret;
+error_put_subbuf:
+ (void) stream->read_subbuffer_ops.put_next_subbuffer(stream, &subbuffer);
+ goto end;
}
int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
switch (consumer_data.type) {
case LTTNG_CONSUMER_KERNEL:
- ret = kernctl_buffer_flush(stream->wait_fd);
- if (ret < 0) {
- ERR("Failed to flush kernel stream");
- goto end;
+ if (producer_active) {
+ ret = kernctl_buffer_flush(stream->wait_fd);
+ if (ret < 0) {
+ ERR("Failed to flush kernel stream");
+ goto end;
+ }
+ } else {
+ ret = kernctl_buffer_flush_empty(stream->wait_fd);
+ if (ret < 0) {
+ /*
+ * Doing a buffer flush which does not take into
+ * account empty packets. This is not perfect,
+ * but required as a fall-back when
+ * "flush_empty" is not implemented by
+ * lttng-modules.
+ */
+ ret = kernctl_buffer_flush(stream->wait_fd);
+ if (ret < 0) {
+ ERR("Failed to flush kernel stream");
+ goto end;
+ }
+ }
}
break;
case LTTNG_CONSUMER32_UST:
case LTTNG_CONSUMER64_UST:
- lttng_ustctl_flush_buffer(stream, producer_active);
+ lttng_ustconsumer_flush_buffer(stream, producer_active);
break;
default:
ERR("Unknown consumer_data type");
ht->hash_fct(&channel->key, lttng_ht_seed),
ht->match_fct, &channel->key, &iter.iter,
stream, node_channel_id.node) {
- unsigned long consumed_pos;
+ unsigned long produced_pos = 0, consumed_pos = 0;
health_code_update();
rotating_to_new_chunk = false;
}
- ret = lttng_consumer_sample_snapshot_positions(stream);
- if (ret < 0) {
- ERR("Failed to sample snapshot position during channel rotation");
- goto end_unlock_stream;
+ /*
+ * Do not flush an empty packet when rotating from a NULL trace
+ * chunk. The stream has no means to output data, and the prior
+ * rotation which rotated to NULL performed that side-effect already.
+ */
+ if (stream->trace_chunk) {
+ /*
+ * For metadata stream, do an active flush, which does not
+ * produce empty packets. For data streams, empty-flush;
+ * ensures we have at least one packet in each stream per trace
+ * chunk, even if no data was produced.
+ */
+ ret = consumer_flush_buffer(stream, stream->metadata_flag ? 1 : 0);
+ if (ret < 0) {
+ ERR("Failed to flush stream %" PRIu64 " during channel rotation",
+ stream->key);
+ goto end_unlock_stream;
+ }
}
- ret = lttng_consumer_get_produced_snapshot(stream,
- &stream->rotate_position);
- if (ret < 0) {
- ERR("Failed to sample produced position during channel rotation");
+ ret = lttng_consumer_take_snapshot(stream);
+ if (ret < 0 && ret != -ENODATA && ret != -EAGAIN) {
+ ERR("Failed to sample snapshot position during channel rotation");
goto end_unlock_stream;
}
+ if (!ret) {
+ ret = lttng_consumer_get_produced_snapshot(stream,
+ &produced_pos);
+ if (ret < 0) {
+ ERR("Failed to sample produced position during channel rotation");
+ goto end_unlock_stream;
+ }
- lttng_consumer_get_consumed_snapshot(stream,
- &consumed_pos);
- if (consumed_pos == stream->rotate_position) {
+ ret = lttng_consumer_get_consumed_snapshot(stream,
+ &consumed_pos);
+ if (ret < 0) {
+ ERR("Failed to sample consumed position during channel rotation");
+ goto end_unlock_stream;
+ }
+ }
+ /*
+ * Align produced position on the start-of-packet boundary of the first
+ * packet going into the next trace chunk.
+ */
+ produced_pos = ALIGN_FLOOR(produced_pos, stream->max_sb_size);
+ if (consumed_pos == produced_pos) {
+ DBG("Set rotate ready for stream %" PRIu64 " produced = %lu consumed = %lu",
+ stream->key, produced_pos, consumed_pos);
stream->rotate_ready = true;
+ } else {
+ DBG("Different consumed and produced positions "
+ "for stream %" PRIu64 " produced = %lu consumed = %lu",
+ stream->key, produced_pos, consumed_pos);
}
-
/*
- * Active flush; has no effect if the production position
- * is at a packet boundary.
+ * The rotation position is based on the packet_seq_num of the
+ * packet following the last packet that was consumed for this
+ * stream, incremented by the offset between produced and
+ * consumed positions. This rotation position is a lower bound
+ * (inclusive) at which the next trace chunk starts. Since it
+ * is a lower bound, it is OK if the packet_seq_num does not
+ * correspond exactly to the same packet identified by the
+ * consumed_pos, which can happen in overwrite mode.
*/
- ret = consumer_flush_buffer(stream, 1);
- if (ret < 0) {
- ERR("Failed to flush stream %" PRIu64 " during channel rotation",
+ if (stream->sequence_number_unavailable) {
+ /*
+ * Rotation should never be performed on a session which
+ * interacts with a pre-2.8 lttng-modules, which does
+ * not implement packet sequence number.
+ */
+ ERR("Failure to rotate stream %" PRIu64 ": sequence number unavailable",
stream->key);
+ ret = -1;
goto end_unlock_stream;
}
+ stream->rotate_position = stream->last_sequence_number + 1 +
+ ((produced_pos - consumed_pos) / stream->max_sb_size);
+ DBG("Set rotation position for stream %" PRIu64 " at position %" PRIu64,
+ stream->key, stream->rotate_position);
if (!is_local_trace) {
/*
* The relay daemon control protocol expects a rotation
* position as "the sequence number of the first packet
- * _after_ the current trace chunk.
- *
- * At the moment when the positions of the buffers are
- * sampled, the production position does not necessarily
- * sit at a packet boundary. The 'active' flush
- * operation above will push the production position to
- * the next packet boundary _if_ it is not already
- * sitting at such a boundary.
- *
- * Assuming a current production position that is not
- * on the bound of a packet, the 'target' sequence
- * number is
- * (consumed_pos / subbuffer_size) + 1
- * Note the '+ 1' to ensure the current packet is
- * part of the current trace chunk.
- *
- * However, if the production position is already at
- * a packet boundary, the '+ 1' is not necessary as the
- * last packet of the current chunk is already
- * 'complete'.
+ * _after_ the current trace chunk".
*/
const struct relayd_stream_rotation_position position = {
.stream_id = stream->relayd_stream_id,
- .rotate_at_seq_num = (stream->rotate_position / stream->max_sb_size) +
- !!(stream->rotate_position % stream->max_sb_size),
+ .rotate_at_seq_num = stream->rotate_position,
};
ret = lttng_dynamic_array_add_element(
return ret;
}
-/*
- * Check if a stream is ready to be rotated after extracting it.
- *
- * Return 1 if it is ready for rotation, 0 if it is not, a negative value on
- * error. Stream lock must be held.
- */
-int lttng_consumer_stream_is_rotate_ready(struct lttng_consumer_stream *stream)
+static
+int consumer_clear_buffer(struct lttng_consumer_stream *stream)
{
- int ret;
- unsigned long consumed_pos;
+ int ret = 0;
+ unsigned long consumed_pos_before, consumed_pos_after;
- if (!stream->rotate_position && !stream->rotate_ready) {
- ret = 0;
+ ret = lttng_consumer_sample_snapshot_positions(stream);
+ if (ret < 0) {
+ ERR("Taking snapshot positions");
goto end;
}
- if (stream->rotate_ready) {
- ret = 1;
+ ret = lttng_consumer_get_consumed_snapshot(stream, &consumed_pos_before);
+ if (ret < 0) {
+ ERR("Consumed snapshot position");
goto end;
}
- /*
- * If we don't have the rotate_ready flag, check the consumed position
- * to determine if we need to rotate.
- */
+ switch (consumer_data.type) {
+ case LTTNG_CONSUMER_KERNEL:
+ ret = kernctl_buffer_clear(stream->wait_fd);
+ if (ret < 0) {
+ ERR("Failed to clear kernel stream (ret = %d)", ret);
+ goto end;
+ }
+ break;
+ case LTTNG_CONSUMER32_UST:
+ case LTTNG_CONSUMER64_UST:
+ lttng_ustconsumer_clear_buffer(stream);
+ break;
+ default:
+ ERR("Unknown consumer_data type");
+ abort();
+ }
+
ret = lttng_consumer_sample_snapshot_positions(stream);
if (ret < 0) {
ERR("Taking snapshot positions");
goto end;
}
-
- ret = lttng_consumer_get_consumed_snapshot(stream, &consumed_pos);
+ ret = lttng_consumer_get_consumed_snapshot(stream, &consumed_pos_after);
if (ret < 0) {
ERR("Consumed snapshot position");
goto end;
}
+ DBG("clear: before: %lu after: %lu", consumed_pos_before, consumed_pos_after);
+end:
+ return ret;
+}
- /* Rotate position not reached yet (with check for overflow). */
- if ((long) (consumed_pos - stream->rotate_position) < 0) {
- ret = 0;
- goto end;
+static
+int consumer_clear_stream(struct lttng_consumer_stream *stream)
+{
+ int ret;
+
+ ret = consumer_flush_buffer(stream, 1);
+ if (ret < 0) {
+ ERR("Failed to flush stream %" PRIu64 " during channel clear",
+ stream->key);
+ ret = LTTCOMM_CONSUMERD_FATAL;
+ goto error;
}
- ret = 1;
-end:
+ ret = consumer_clear_buffer(stream);
+ if (ret < 0) {
+ ERR("Failed to clear stream %" PRIu64 " during channel clear",
+ stream->key);
+ ret = LTTCOMM_CONSUMERD_FATAL;
+ goto error;
+ }
+
+ ret = LTTCOMM_CONSUMERD_SUCCESS;
+error:
+ return ret;
+}
+
+static
+int consumer_clear_unmonitored_channel(struct lttng_consumer_channel *channel)
+{
+ int ret;
+ struct lttng_consumer_stream *stream;
+
+ rcu_read_lock();
+ pthread_mutex_lock(&channel->lock);
+ cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
+ health_code_update();
+ pthread_mutex_lock(&stream->lock);
+ ret = consumer_clear_stream(stream);
+ if (ret) {
+ goto error_unlock;
+ }
+ pthread_mutex_unlock(&stream->lock);
+ }
+ pthread_mutex_unlock(&channel->lock);
+ rcu_read_unlock();
+ return 0;
+
+error_unlock:
+ pthread_mutex_unlock(&stream->lock);
+ pthread_mutex_unlock(&channel->lock);
+ rcu_read_unlock();
return ret;
}
+/*
+ * Check if a stream is ready to be rotated after extracting it.
+ *
+ * Return 1 if it is ready for rotation, 0 if it is not, a negative value on
+ * error. Stream lock must be held.
+ */
+int lttng_consumer_stream_is_rotate_ready(struct lttng_consumer_stream *stream)
+{
+ DBG("Check is rotate ready for stream %" PRIu64
+ " ready %u rotate_position %" PRIu64
+ " last_sequence_number %" PRIu64,
+ stream->key, stream->rotate_ready,
+ stream->rotate_position, stream->last_sequence_number);
+ if (stream->rotate_ready) {
+ return 1;
+ }
+
+ /*
+ * If packet seq num is unavailable, it means we are interacting
+ * with a pre-2.8 lttng-modules which does not implement the
+ * sequence number. Rotation should never be used by sessiond in this
+ * scenario.
+ */
+ if (stream->sequence_number_unavailable) {
+ ERR("Internal error: rotation used on stream %" PRIu64
+ " with unavailable sequence number",
+ stream->key);
+ return -1;
+ }
+
+ if (stream->rotate_position == -1ULL ||
+ stream->last_sequence_number == -1ULL) {
+ return 0;
+ }
+
+ /*
+ * Rotate position not reached yet. The stream rotate position is
+ * the position of the next packet belonging to the next trace chunk,
+ * but consumerd considers rotation ready when reaching the last
+ * packet of the current chunk, hence the "rotate_position - 1".
+ */
+
+ DBG("Check is rotate ready for stream %" PRIu64
+ " last_sequence_number %" PRIu64
+ " rotate_position %" PRIu64,
+ stream->key, stream->last_sequence_number,
+ stream->rotate_position);
+ if (stream->last_sequence_number >= stream->rotate_position - 1) {
+ return 1;
+ }
+
+ return 0;
+}
+
/*
* Reset the state for a stream after a rotation occurred.
*/
void lttng_consumer_reset_stream_rotate_state(struct lttng_consumer_stream *stream)
{
- stream->rotate_position = 0;
+ DBG("lttng_consumer_reset_stream_rotate_state for stream %" PRIu64,
+ stream->key);
+ stream->rotate_position = -1ULL;
stream->rotate_ready = false;
}
const lttng_uuid sessiond_uuid)
{
enum lttcomm_return_code ret;
- char uuid_str[UUID_STR_LEN];
+ char uuid_str[LTTNG_UUID_STR_LEN];
if (ctx->sessiond_uuid.is_set) {
ret = LTTCOMM_CONSUMERD_ALREADY_SET;
* and LTTNG_CONSUMER_DESTROY_TRACE_CHUNK commands.
*/
created_chunk = lttng_trace_chunk_create(chunk_id,
- chunk_creation_timestamp);
+ chunk_creation_timestamp, NULL);
if (!created_chunk) {
ERR("Failed to create trace chunk");
ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
*/
chunk_status = lttng_trace_chunk_set_as_user(created_chunk,
chunk_directory_handle);
+ chunk_directory_handle = NULL;
if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
ERR("Failed to set trace chunk's directory handle");
ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
end:
return ret_code;
}
+
+static
+int consumer_clear_monitored_channel(struct lttng_consumer_channel *channel)
+{
+ struct lttng_ht *ht;
+ struct lttng_consumer_stream *stream;
+ struct lttng_ht_iter iter;
+ int ret;
+
+ ht = consumer_data.stream_per_chan_id_ht;
+
+ rcu_read_lock();
+ cds_lfht_for_each_entry_duplicate(ht->ht,
+ ht->hash_fct(&channel->key, lttng_ht_seed),
+ ht->match_fct, &channel->key,
+ &iter.iter, stream, node_channel_id.node) {
+ /*
+ * Protect against teardown with mutex.
+ */
+ pthread_mutex_lock(&stream->lock);
+ if (cds_lfht_is_node_deleted(&stream->node.node)) {
+ goto next;
+ }
+ ret = consumer_clear_stream(stream);
+ if (ret) {
+ goto error_unlock;
+ }
+ next:
+ pthread_mutex_unlock(&stream->lock);
+ }
+ rcu_read_unlock();
+ return LTTCOMM_CONSUMERD_SUCCESS;
+
+error_unlock:
+ pthread_mutex_unlock(&stream->lock);
+ rcu_read_unlock();
+ return ret;
+}
+
+int lttng_consumer_clear_channel(struct lttng_consumer_channel *channel)
+{
+ int ret;
+
+ DBG("Consumer clear channel %" PRIu64, channel->key);
+
+ if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA) {
+ /*
+ * Nothing to do for the metadata channel/stream.
+ * Snapshot mechanism already take care of the metadata
+ * handling/generation, and monitored channels only need to
+ * have their data stream cleared..
+ */
+ ret = LTTCOMM_CONSUMERD_SUCCESS;
+ goto end;
+ }
+
+ if (!channel->monitor) {
+ ret = consumer_clear_unmonitored_channel(channel);
+ } else {
+ ret = consumer_clear_monitored_channel(channel);
+ }
+end:
+ return ret;
+}
projects / lttng-tools.git / blobdiff