/*
- * 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>
#include <bin/lttng-consumerd/health-consumerd.h>
#include <common/common.h>
#include <common/utils.h>
+#include <common/time.h>
#include <common/compat/poll.h>
#include <common/compat/endian.h>
#include <common/index/index.h>
#include <common/consumer/consumer-testpoint.h>
#include <common/align.h>
#include <common/consumer/consumer-metadata-cache.h>
+#include <common/trace-chunk.h>
+#include <common/trace-chunk-registry.h>
+#include <common/string-utils/format.h>
+#include <common/dynamic-array.h>
struct lttng_consumer_global_data consumer_data = {
.stream_count = 0,
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
*/
void consumer_del_channel(struct lttng_consumer_channel *channel)
{
- int ret;
struct lttng_ht_iter iter;
DBG("Consumer delete channel key %" PRIu64, channel->key);
goto end;
}
- rcu_read_lock();
- iter.iter.node = &channel->node.node;
- ret = lttng_ht_del(consumer_data.channel_ht, &iter);
- assert(!ret);
+ lttng_trace_chunk_put(channel->trace_chunk);
+ channel->trace_chunk = NULL;
- iter.iter.node = &channel->channels_by_session_id_ht_node.node;
- ret = lttng_ht_del(consumer_data.channels_by_session_id_ht, &iter);
- assert(!ret);
- rcu_read_unlock();
+ if (channel->is_published) {
+ int ret;
+
+ rcu_read_lock();
+ iter.iter.node = &channel->node.node;
+ ret = lttng_ht_del(consumer_data.channel_ht, &iter);
+ assert(!ret);
+
+ iter.iter.node = &channel->channels_by_session_id_ht_node.node;
+ ret = lttng_ht_del(consumer_data.channels_by_session_id_ht,
+ &iter);
+ assert(!ret);
+ rcu_read_unlock();
+ }
+ channel->is_deleted = true;
call_rcu(&channel->node.head, free_channel_rcu);
end:
pthread_mutex_unlock(&channel->lock);
{
stream->channel_read_only_attributes.tracefile_size =
channel->tracefile_size;
- memcpy(stream->channel_read_only_attributes.path, channel->pathname,
- sizeof(stream->channel_read_only_attributes.path));
-}
-
-struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
- uint64_t stream_key,
- enum lttng_consumer_stream_state state,
- const char *channel_name,
- uid_t uid,
- gid_t gid,
- uint64_t relayd_id,
- uint64_t session_id,
- int cpu,
- int *alloc_ret,
- enum consumer_channel_type type,
- unsigned int monitor,
- uint64_t trace_archive_id)
-{
- int ret;
- struct lttng_consumer_stream *stream;
-
- stream = zmalloc(sizeof(*stream));
- if (stream == NULL) {
- PERROR("malloc struct lttng_consumer_stream");
- ret = -ENOMEM;
- goto end;
- }
-
- rcu_read_lock();
-
- stream->key = stream_key;
- stream->out_fd = -1;
- stream->out_fd_offset = 0;
- stream->output_written = 0;
- stream->state = state;
- stream->uid = uid;
- stream->gid = gid;
- 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;
- stream->trace_archive_id = trace_archive_id;
- 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();
- free(stream);
-end:
- if (alloc_ret) {
- *alloc_ret = ret;
- }
- return NULL;
}
/*
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,
- stream->chan->tracefile_size, stream->chan->tracefile_count,
- stream->trace_archive_id);
+ get_consumer_domain(), path, &stream->relayd_stream_id,
+ stream->chan->tracefile_size,
+ stream->chan->tracefile_count,
+ stream->trace_chunk);
pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret < 0) {
ERR("Relayd add stream failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
data_hdr.stream_id = htobe64(stream->relayd_stream_id);
data_hdr.data_size = htobe32(data_size);
data_hdr.padding_size = htobe32(padding);
+
/*
* Note that net_seq_num below is assigned with the *current* value of
* next_net_seq_num and only after that the next_net_seq_num will be
return outfd;
}
+/*
+ * Trigger a dump of the metadata content. Following/during the succesful
+ * completion of this call, the metadata poll thread will start receiving
+ * metadata packets to consume.
+ *
+ * The caller must hold the channel and stream locks.
+ */
+static
+int consumer_metadata_stream_dump(struct lttng_consumer_stream *stream)
+{
+ int ret;
+
+ ASSERT_LOCKED(stream->chan->lock);
+ ASSERT_LOCKED(stream->lock);
+ assert(stream->metadata_flag);
+ assert(stream->chan->trace_chunk);
+
+ switch (consumer_data.type) {
+ case LTTNG_CONSUMER_KERNEL:
+ /*
+ * Reset the position of what has been read from the
+ * metadata cache to 0 so we can dump it again.
+ */
+ ret = kernctl_metadata_cache_dump(stream->wait_fd);
+ break;
+ case LTTNG_CONSUMER32_UST:
+ case LTTNG_CONSUMER64_UST:
+ /*
+ * Reset the position pushed from the metadata cache so it
+ * will write from the beginning on the next push.
+ */
+ stream->ust_metadata_pushed = 0;
+ ret = consumer_metadata_wakeup_pipe(stream->chan);
+ break;
+ default:
+ ERR("Unknown consumer_data type");
+ abort();
+ }
+ if (ret < 0) {
+ ERR("Failed to dump the metadata cache");
+ }
+ return ret;
+}
+
+static
+int lttng_consumer_channel_set_trace_chunk(
+ struct lttng_consumer_channel *channel,
+ struct lttng_trace_chunk *new_trace_chunk)
+{
+ pthread_mutex_lock(&channel->lock);
+ if (channel->is_deleted) {
+ /*
+ * The channel has been logically deleted and should no longer
+ * be used. It has released its reference to its current trace
+ * chunk and should not acquire a new one.
+ *
+ * Return success as there is nothing for the caller to do.
+ */
+ goto end;
+ }
+
+ /*
+ * The acquisition of the reference cannot fail (barring
+ * a severe internal error) since a reference to the published
+ * chunk is already held by the caller.
+ */
+ if (new_trace_chunk) {
+ const bool acquired_reference = lttng_trace_chunk_get(
+ new_trace_chunk);
+
+ assert(acquired_reference);
+ }
+
+ lttng_trace_chunk_put(channel->trace_chunk);
+ channel->trace_chunk = new_trace_chunk;
+end:
+ pthread_mutex_unlock(&channel->lock);
+ return 0;
+}
+
/*
* Allocate and return a new lttng_consumer_channel object using the given key
* to initialize the hash table node.
*/
struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
uint64_t session_id,
+ const uint64_t *chunk_id,
const char *pathname,
const char *name,
- uid_t uid,
- gid_t gid,
uint64_t relayd_id,
enum lttng_event_output output,
uint64_t tracefile_size,
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)
{
- struct lttng_consumer_channel *channel;
+ struct lttng_consumer_channel *channel = NULL;
+ struct lttng_trace_chunk *trace_chunk = NULL;
+
+ if (chunk_id) {
+ trace_chunk = lttng_trace_chunk_registry_find_chunk(
+ consumer_data.chunk_registry, session_id,
+ *chunk_id);
+ if (!trace_chunk) {
+ ERR("Failed to find trace chunk reference during creation of channel");
+ goto end;
+ }
+ }
channel = zmalloc(sizeof(*channel));
if (channel == NULL) {
channel->refcount = 0;
channel->session_id = session_id;
channel->session_id_per_pid = session_id_per_pid;
- channel->uid = uid;
- channel->gid = gid;
channel->relayd_id = relayd_id;
channel->tracefile_size = tracefile_size;
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);
channel->session_id);
channel->wait_fd = -1;
-
CDS_INIT_LIST_HEAD(&channel->streams.head);
+ if (trace_chunk) {
+ int ret = lttng_consumer_channel_set_trace_chunk(channel,
+ trace_chunk);
+ if (ret) {
+ goto error;
+ }
+ }
+
DBG("Allocated channel (key %" PRIu64 ")", channel->key);
end:
+ lttng_trace_chunk_put(trace_chunk);
return channel;
+error:
+ consumer_del_channel(channel);
+ channel = NULL;
+ goto end;
}
/*
lttng_ht_add_u64(consumer_data.channels_by_session_id_ht,
&channel->channels_by_session_id_ht_node);
rcu_read_unlock();
+ channel->is_published = true;
pthread_mutex_unlock(&channel->timer_lock);
pthread_mutex_unlock(&channel->lock);
* closed by the polling thread after a wakeup on the data_pipe or
* metadata_pipe.
*/
- if (stream->state != LTTNG_CONSUMER_ACTIVE_STREAM ||
- stream->endpoint_status == CONSUMER_ENDPOINT_INACTIVE) {
+ if (stream->endpoint_status == CONSUMER_ENDPOINT_INACTIVE) {
(*nb_inactive_fd)++;
continue;
}
/*
* This clobbers way too much the debug output. Uncomment that if you
* need it for debugging purposes.
- *
- * DBG("Active FD %d", stream->wait_fd);
*/
(*pollfd)[i].fd = stream->wait_fd;
(*pollfd)[i].events = POLLIN | POLLPRI;
{
struct lttng_ht_iter iter;
struct lttng_consumer_channel *channel;
+ unsigned int trace_chunks_left;
rcu_read_lock();
* it.
*/
lttng_ht_destroy(consumer_data.stream_list_ht);
+
+ /*
+ * Trace chunks in the registry may still exist if the session
+ * daemon has encountered an internal error and could not
+ * tear down its sessions and/or trace chunks properly.
+ *
+ * Release the session daemon's implicit reference to any remaining
+ * trace chunk and print an error if any trace chunk was found. Note
+ * that there are _no_ legitimate cases for trace chunks to be left,
+ * it is a leak. However, it can happen following a crash of the
+ * session daemon and not emptying the registry would cause an assertion
+ * to hit.
+ */
+ trace_chunks_left = lttng_trace_chunk_registry_put_each_chunk(
+ consumer_data.chunk_registry);
+ if (trace_chunks_left) {
+ ERR("%u trace chunks are leaked by lttng-consumerd. "
+ "This can be caused by an internal error of the session daemon.",
+ trace_chunks_left);
+ }
+ /* Run all callbacks freeing each chunk. */
+ rcu_barrier();
+ lttng_trace_chunk_registry_destroy(consumer_data.chunk_registry);
}
/*
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))
* core function for writing trace buffers to either the local filesystem or
* the network.
*
- * It must be called with the stream lock held.
+ * It must be called with the stream and the channel lock held.
*
* Careful review MUST be put if any changes occur!
*
* 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();
+ assert(stream->net_seq_idx != (uint64_t) -1ULL ||
+ stream->trace_chunk);
/* Flag that the current stream if set for network streaming. */
if (stream->net_seq_idx != (uint64_t) -1ULL) {
}
}
- /* 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 = utils_rotate_stream_file(stream->chan->pathname,
- stream->name, stream->chan->tracefile_size,
- stream->chan->tracefile_count, stream->uid, stream->gid,
- stream->out_fd, &(stream->tracefile_count_current),
- &stream->out_fd);
- if (ret < 0) {
- ERR("Rotating output file");
+ ret = consumer_stream_rotate_output_files(stream);
+ if (ret) {
goto end;
}
outfd = stream->out_fd;
-
- if (stream->index_file) {
- lttng_index_file_put(stream->index_file);
- stream->index_file = lttng_index_file_create(stream->chan->pathname,
- stream->name, stream->uid, stream->gid,
- stream->chan->tracefile_size,
- stream->tracefile_count_current,
- CTF_INDEX_MAJOR, CTF_INDEX_MINOR);
- if (!stream->index_file) {
- goto end;
- }
- }
-
- /* Reset current size because we just perform a rotation. */
- stream->tracefile_size_current = 0;
- stream->out_fd_offset = 0;
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.
relayd_hang_up = 1;
/* Socket operation failed. We consider the relayd dead */
- if (errno == EPIPE || errno == EINVAL || errno == EBADF) {
+ if (errno == EPIPE) {
/*
* This is possible if the fd is closed on the other side
* (outfd) or any write problem. It can be verbose a bit for a
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;
if (stream->chan->tracefile_size > 0 &&
(stream->tracefile_size_current + len) >
stream->chan->tracefile_size) {
- ret = utils_rotate_stream_file(stream->chan->pathname,
- stream->name, stream->chan->tracefile_size,
- stream->chan->tracefile_count, stream->uid, stream->gid,
- stream->out_fd, &(stream->tracefile_count_current),
- &stream->out_fd);
+ ret = consumer_stream_rotate_output_files(stream);
if (ret < 0) {
written = ret;
- ERR("Rotating output file");
goto end;
}
outfd = stream->out_fd;
-
- if (stream->index_file) {
- lttng_index_file_put(stream->index_file);
- stream->index_file = lttng_index_file_create(stream->chan->pathname,
- stream->name, stream->uid, stream->gid,
- stream->chan->tracefile_size,
- stream->tracefile_count_current,
- CTF_INDEX_MAJOR, CTF_INDEX_MINOR);
- if (!stream->index_file) {
- goto end;
- }
- }
-
- /* Reset current size because we just perform a rotation. */
- stream->tracefile_size_current = 0;
- stream->out_fd_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) {
void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
struct lttng_ht *ht)
{
- struct lttng_consumer_channel *free_chan = NULL;
+ struct lttng_consumer_channel *channel = NULL;
+ bool free_channel = false;
assert(stream);
/*
DBG3("Consumer delete metadata stream %d", stream->wait_fd);
pthread_mutex_lock(&consumer_data.lock);
- pthread_mutex_lock(&stream->chan->lock);
+ /*
+ * Note that this assumes that a stream's channel is never changed and
+ * that the stream's lock doesn't need to be taken to sample its
+ * channel.
+ */
+ channel = stream->chan;
+ pthread_mutex_lock(&channel->lock);
pthread_mutex_lock(&stream->lock);
- if (stream->chan->metadata_cache) {
+ if (channel->metadata_cache) {
/* Only applicable to userspace consumers. */
- pthread_mutex_lock(&stream->chan->metadata_cache->lock);
+ pthread_mutex_lock(&channel->metadata_cache->lock);
}
/* Remove any reference to that stream. */
consumer_stream_destroy_buffers(stream);
/* Atomically decrement channel refcount since other threads can use it. */
- if (!uatomic_sub_return(&stream->chan->refcount, 1)
- && !uatomic_read(&stream->chan->nb_init_stream_left)) {
+ if (!uatomic_sub_return(&channel->refcount, 1)
+ && !uatomic_read(&channel->nb_init_stream_left)) {
/* Go for channel deletion! */
- free_chan = stream->chan;
+ free_channel = true;
}
+ stream->chan = NULL;
/*
* Nullify the stream reference so it is not used after deletion. The
* channel lock MUST be acquired before being able to check for a NULL
* pointer value.
*/
- stream->chan->metadata_stream = NULL;
+ channel->metadata_stream = NULL;
- if (stream->chan->metadata_cache) {
- pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
+ if (channel->metadata_cache) {
+ pthread_mutex_unlock(&channel->metadata_cache->lock);
}
pthread_mutex_unlock(&stream->lock);
- pthread_mutex_unlock(&stream->chan->lock);
+ pthread_mutex_unlock(&channel->lock);
pthread_mutex_unlock(&consumer_data.lock);
- if (free_chan) {
- consumer_del_channel(free_chan);
+ if (free_channel) {
+ consumer_del_channel(channel);
}
+ lttng_trace_chunk_put(stream->trace_chunk);
+ stream->trace_chunk = NULL;
consumer_stream_free(stream);
}
rcu_read_unlock();
}
-/*
- * Perform operations that need to be done after a stream has
- * rotated and released the stream lock.
- *
- * Multiple rotations cannot occur simultaneously, so we know the state of the
- * "rotated" stream flag cannot change.
- *
- * This MUST be called WITHOUT the stream lock held.
- */
-static
-int consumer_post_rotation(struct lttng_consumer_stream *stream,
- struct lttng_consumer_local_data *ctx)
-{
- int ret = 0;
-
- pthread_mutex_lock(&stream->chan->lock);
-
- switch (consumer_data.type) {
- case LTTNG_CONSUMER_KERNEL:
- break;
- case LTTNG_CONSUMER32_UST:
- case LTTNG_CONSUMER64_UST:
- /*
- * The ust_metadata_pushed counter has been reset to 0, so now
- * we can wakeup the metadata thread so it dumps the metadata
- * cache to the new file.
- */
- if (stream->metadata_flag) {
- consumer_metadata_wakeup_pipe(stream->chan);
- }
- break;
- default:
- ERR("Unknown consumer_data type");
- abort();
- }
-
- pthread_mutex_unlock(&stream->chan->lock);
- return ret;
-}
-
/*
* Thread polls on metadata file descriptor and write them on disk or on the
* network.
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. */
return NULL;
}
-ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
+static int post_consume(struct lttng_consumer_stream *stream,
+ const struct stream_subbuffer *subbuffer,
struct lttng_consumer_local_data *ctx)
{
- ssize_t ret;
- int rotate_ret;
- bool rotated = false;
-
- pthread_mutex_lock(&stream->lock);
- if (stream->metadata_flag) {
- pthread_mutex_lock(&stream->metadata_rdv_lock);
- }
+ size_t i;
+ int ret = 0;
+ const size_t count = lttng_dynamic_array_get_count(
+ &stream->read_subbuffer_ops.post_consume_cbs);
- switch (consumer_data.type) {
- case LTTNG_CONSUMER_KERNEL:
- ret = lttng_kconsumer_read_subbuffer(stream, ctx, &rotated);
- break;
- case LTTNG_CONSUMER32_UST:
- case LTTNG_CONSUMER64_UST:
- ret = lttng_ustconsumer_read_subbuffer(stream, ctx, &rotated);
- break;
- default:
- ERR("Unknown consumer_data type");
- assert(0);
- ret = -ENOSYS;
- break;
- }
+ for (i = 0; i < count; i++) {
+ const post_consume_cb op = *(post_consume_cb *) lttng_dynamic_array_get_element(
+ &stream->read_subbuffer_ops.post_consume_cbs,
+ i);
- if (stream->metadata_flag) {
- pthread_cond_broadcast(&stream->metadata_rdv);
- pthread_mutex_unlock(&stream->metadata_rdv_lock);
- }
- pthread_mutex_unlock(&stream->lock);
- if (rotated) {
- rotate_ret = consumer_post_rotation(stream, ctx);
- if (rotate_ret < 0) {
- ERR("Failed after a rotation");
- ret = -1;
+ ret = op(stream, subbuffer, ctx);
+ if (ret) {
+ goto end;
}
}
-
+end:
return ret;
}
-int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
+ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
+ struct lttng_consumer_local_data *ctx,
+ bool locked_by_caller)
{
- switch (consumer_data.type) {
- case LTTNG_CONSUMER_KERNEL:
- return lttng_kconsumer_on_recv_stream(stream);
- case LTTNG_CONSUMER32_UST:
- case LTTNG_CONSUMER64_UST:
- return lttng_ustconsumer_on_recv_stream(stream);
- default:
- ERR("Unknown consumer_data type");
- assert(0);
- return -ENOSYS;
- }
-}
+ ssize_t ret, written_bytes = 0;
+ int rotation_ret;
+ struct stream_subbuffer subbuffer = {};
-/*
- * Allocate and set consumer data hash tables.
- */
-int lttng_consumer_init(void)
-{
- consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
- if (!consumer_data.channel_ht) {
- goto error;
+ if (!locked_by_caller) {
+ stream->read_subbuffer_ops.lock(stream);
}
- consumer_data.channels_by_session_id_ht =
- lttng_ht_new(0, LTTNG_HT_TYPE_U64);
- if (!consumer_data.channels_by_session_id_ht) {
- goto error;
+ if (stream->read_subbuffer_ops.on_wake_up) {
+ ret = stream->read_subbuffer_ops.on_wake_up(stream);
+ if (ret) {
+ goto end;
+ }
}
- consumer_data.relayd_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+ /*
+ * 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.padded_subbuf_size,
+ subbuffer.info.data.subbuf_size);
+ }
+
+ ret = stream->read_subbuffer_ops.put_next_subbuffer(stream, &subbuffer);
+ if (ret) {
+ goto end;
+ }
+
+ ret = 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;
+ }
+
+sleep_stream:
+ 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);
+ }
+
+ 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:
+ return lttng_kconsumer_on_recv_stream(stream);
+ case LTTNG_CONSUMER32_UST:
+ case LTTNG_CONSUMER64_UST:
+ return lttng_ustconsumer_on_recv_stream(stream);
+ default:
+ ERR("Unknown consumer_data type");
+ assert(0);
+ return -ENOSYS;
+ }
+}
+
+/*
+ * Allocate and set consumer data hash tables.
+ */
+int lttng_consumer_init(void)
+{
+ consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+ if (!consumer_data.channel_ht) {
+ goto error;
+ }
+
+ consumer_data.channels_by_session_id_ht =
+ lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+ if (!consumer_data.channels_by_session_id_ht) {
+ goto error;
+ }
+
+ consumer_data.relayd_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
if (!consumer_data.relayd_ht) {
goto error;
}
goto error;
}
+ consumer_data.chunk_registry = lttng_trace_chunk_registry_create();
+ if (!consumer_data.chunk_registry) {
+ goto error;
+ }
+
return 0;
error:
return start_pos;
}
-static
-int consumer_flush_buffer(struct lttng_consumer_stream *stream, int producer_active)
-{
- int ret = 0;
-
- 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;
- }
- break;
- case LTTNG_CONSUMER32_UST:
- case LTTNG_CONSUMER64_UST:
- lttng_ustctl_flush_buffer(stream, producer_active);
- break;
- default:
- ERR("Unknown consumer_data type");
- abort();
- }
-
-end:
- return ret;
-}
-
/*
* Sample the rotate position for all the streams of a channel. If a stream
* is already at the rotate position (produced == consumed), we flag it as
* Returns 0 on success, < 0 on error
*/
int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel,
- uint64_t key, const char *path, uint64_t relayd_id,
- uint32_t metadata, uint64_t new_chunk_id,
+ uint64_t key, uint64_t relayd_id, uint32_t metadata,
struct lttng_consumer_local_data *ctx)
{
int ret;
struct lttng_consumer_stream *stream;
struct lttng_ht_iter iter;
struct lttng_ht *ht = consumer_data.stream_per_chan_id_ht;
+ struct lttng_dynamic_array stream_rotation_positions;
+ uint64_t next_chunk_id, stream_count = 0;
+ enum lttng_trace_chunk_status chunk_status;
+ const bool is_local_trace = relayd_id == -1ULL;
+ struct consumer_relayd_sock_pair *relayd = NULL;
+ bool rotating_to_new_chunk = true;
+ /* Array of `struct lttng_consumer_stream *` */
+ struct lttng_dynamic_pointer_array streams_packet_to_open;
+ size_t stream_idx;
DBG("Consumer sample rotate position for channel %" PRIu64, key);
+ lttng_dynamic_array_init(&stream_rotation_positions,
+ sizeof(struct relayd_stream_rotation_position), NULL);
+ lttng_dynamic_pointer_array_init(&streams_packet_to_open, NULL);
+
rcu_read_lock();
pthread_mutex_lock(&channel->lock);
- channel->current_chunk_id = new_chunk_id;
-
- ret = lttng_strncpy(channel->pathname, path, sizeof(channel->pathname));
- if (ret) {
- ERR("Failed to copy new path to channel during channel rotation");
+ assert(channel->trace_chunk);
+ chunk_status = lttng_trace_chunk_get_id(channel->trace_chunk,
+ &next_chunk_id);
+ if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
ret = -1;
goto end_unlock_channel;
}
- if (relayd_id == -1ULL) {
- /*
- * The domain path (/ust or /kernel) has been created before, we
- * now need to create the last part of the path: the application/user
- * specific section (uid/1000/64-bit).
- */
- ret = utils_mkdir_recursive(channel->pathname, S_IRWXU | S_IRWXG,
- channel->uid, channel->gid);
- if (ret < 0) {
- ERR("Failed to create trace directory at %s during rotation",
- channel->pathname);
- ret = -1;
- goto end_unlock_channel;
- }
- }
-
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) {
- unsigned long consumed_pos;
+ unsigned long produced_pos = 0, consumed_pos = 0;
health_code_update();
*/
pthread_mutex_lock(&stream->lock);
- ret = lttng_strncpy(stream->channel_read_only_attributes.path,
- channel->pathname,
- sizeof(stream->channel_read_only_attributes.path));
- if (ret) {
- ERR("Failed to sample channel path name during channel rotation");
- goto end_unlock_stream;
+ if (stream->trace_chunk == stream->chan->trace_chunk) {
+ rotating_to_new_chunk = false;
}
- ret = lttng_consumer_sample_snapshot_positions(stream);
- if (ret < 0) {
+
+ /*
+ * 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_stream_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_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;
+ }
- 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_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);
+ }
+ /*
+ * 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.
+ */
+ 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);
- lttng_consumer_get_consumed_snapshot(stream,
- &consumed_pos);
- if (consumed_pos == stream->rotate_position) {
- stream->rotate_ready = true;
+ 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".
+ */
+ const struct relayd_stream_rotation_position position = {
+ .stream_id = stream->relayd_stream_id,
+ .rotate_at_seq_num = stream->rotate_position,
+ };
+
+ ret = lttng_dynamic_array_add_element(
+ &stream_rotation_positions,
+ &position);
+ if (ret) {
+ ERR("Failed to allocate stream rotation position");
+ goto end_unlock_stream;
+ }
+ stream_count++;
+ }
+
+ stream->opened_packet_in_current_trace_chunk = false;
+
+ if (rotating_to_new_chunk && !stream->metadata_flag) {
+ /*
+ * Attempt to flush an empty packet as close to the
+ * rotation point as possible. In the event where a
+ * stream remains inactive after the rotation point,
+ * this ensures that the new trace chunk has a
+ * beginning timestamp set at the begining of the
+ * trace chunk instead of only creating an empty
+ * packet when the trace chunk is stopped.
+ *
+ * This indicates to the viewers that the stream
+ * was being recorded, but more importantly it
+ * allows viewers to determine a useable trace
+ * intersection.
+ *
+ * This presents a problem in the case where the
+ * ring-buffer is completely full.
+ *
+ * Consider the following scenario:
+ * - The consumption of data is slow (slow network,
+ * for instance),
+ * - The ring buffer is full,
+ * - A rotation is initiated,
+ * - The flush below does nothing (no space left to
+ * open a new packet),
+ * - The other streams rotate very soon, and new
+ * data is produced in the new chunk,
+ * - This stream completes its rotation long after the
+ * rotation was initiated
+ * - The session is stopped before any event can be
+ * produced in this stream's buffers.
+ *
+ * The resulting trace chunk will have a single packet
+ * temporaly at the end of the trace chunk for this
+ * stream making the stream intersection more narrow
+ * than it should be.
+ *
+ * To work-around this, an empty flush is performed
+ * after the first consumption of a packet during a
+ * rotation if open_packet fails. The idea is that
+ * consuming a packet frees enough space to switch
+ * packets in this scenario and allows the tracer to
+ * "stamp" the beginning of the new trace chunk at the
+ * earliest possible point.
+ *
+ * The packet open is performed after the channel
+ * rotation to ensure that no attempt to open a packet
+ * is performed in a stream that has no active trace
+ * chunk.
+ */
+ ret = lttng_dynamic_pointer_array_add_pointer(
+ &streams_packet_to_open, stream);
+ if (ret) {
+ PERROR("Failed to add a stream pointer to array of streams in which to open a packet");
+ ret = -1;
+ goto end_unlock_stream;
+ }
+ }
+
+ pthread_mutex_unlock(&stream->lock);
+ }
+ stream = NULL;
+
+ if (!is_local_trace) {
+ relayd = consumer_find_relayd(relayd_id);
+ if (!relayd) {
+ ERR("Failed to find relayd %" PRIu64, relayd_id);
+ ret = -1;
+ goto end_unlock_channel;
}
- ret = consumer_flush_buffer(stream, 1);
+ pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+ ret = relayd_rotate_streams(&relayd->control_sock, stream_count,
+ rotating_to_new_chunk ? &next_chunk_id : NULL,
+ (const struct relayd_stream_rotation_position *)
+ stream_rotation_positions.buffer
+ .data);
+ pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret < 0) {
- ERR("Failed to flush stream %" PRIu64 " during channel rotation",
- stream->key);
- goto end_unlock_stream;
+ ERR("Relayd rotate stream failed. Cleaning up relayd %" PRIu64,
+ relayd->net_seq_idx);
+ lttng_consumer_cleanup_relayd(relayd);
+ goto end_unlock_channel;
}
+ }
+ for (stream_idx = 0;
+ stream_idx < lttng_dynamic_pointer_array_get_count(
+ &streams_packet_to_open);
+ stream_idx++) {
+ enum consumer_stream_open_packet_status status;
+
+ stream = lttng_dynamic_pointer_array_get_pointer(
+ &streams_packet_to_open, stream_idx);
+
+ pthread_mutex_lock(&stream->lock);
+ status = consumer_stream_open_packet(stream);
pthread_mutex_unlock(&stream->lock);
+ switch (status) {
+ case CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED:
+ DBG("Opened a packet after a rotation: stream id = %" PRIu64
+ ", channel name = %s, session id = %" PRIu64,
+ stream->key, stream->chan->name,
+ stream->chan->session_id);
+ break;
+ case CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE:
+ /*
+ * Can't open a packet as there is no space left
+ * in the buffer. A new packet will be opened
+ * once one has been consumed.
+ */
+ DBG("No space left to open a packet after a rotation: stream id = %" PRIu64
+ ", channel name = %s, session id = %" PRIu64,
+ stream->key, stream->chan->name,
+ stream->chan->session_id);
+ break;
+ case CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR:
+ /* Logged by callee. */
+ ret = -1;
+ goto end_unlock_channel;
+ default:
+ abort();
+ }
}
- pthread_mutex_unlock(&channel->lock);
+ pthread_mutex_unlock(&channel->lock);
ret = 0;
goto end;
pthread_mutex_unlock(&channel->lock);
end:
rcu_read_unlock();
+ lttng_dynamic_array_reset(&stream_rotation_positions);
+ lttng_dynamic_pointer_array_reset(&streams_packet_to_open);
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;
}
-
- /* Rotate position not reached yet (with check for overflow). */
- if ((long) (consumed_pos - stream->rotate_position) < 0) {
- ret = 0;
- goto end;
- }
- ret = 1;
-
+ DBG("clear: before: %lu after: %lu", consumed_pos_before, consumed_pos_after);
end:
return ret;
}
-/*
- * 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;
- stream->rotate_ready = false;
-}
-
-/*
- * Perform the rotation a local stream file.
- */
-int rotate_local_stream(struct lttng_consumer_local_data *ctx,
- struct lttng_consumer_stream *stream)
+static
+int consumer_clear_stream(struct lttng_consumer_stream *stream)
{
int ret;
- DBG("Rotate local stream: stream key %" PRIu64 ", channel key %" PRIu64 " at path %s",
- stream->key,
- stream->chan->key,
- stream->channel_read_only_attributes.path);
-
- ret = close(stream->out_fd);
+ ret = consumer_stream_flush_buffer(stream, 1);
if (ret < 0) {
- PERROR("Closing trace file (fd %d), stream %" PRIu64,
- stream->out_fd, stream->key);
- assert(0);
+ ERR("Failed to flush stream %" PRIu64 " during channel clear",
+ stream->key);
+ ret = LTTCOMM_CONSUMERD_FATAL;
goto error;
}
- ret = utils_create_stream_file(
- stream->channel_read_only_attributes.path,
- stream->name,
- stream->channel_read_only_attributes.tracefile_size,
- stream->tracefile_count_current,
- stream->uid, stream->gid, NULL);
+ ret = consumer_clear_buffer(stream);
if (ret < 0) {
- ERR("Rotate create stream file");
+ ERR("Failed to clear stream %" PRIu64 " during channel clear",
+ stream->key);
+ ret = LTTCOMM_CONSUMERD_FATAL;
goto error;
}
- stream->out_fd = ret;
- stream->tracefile_size_current = 0;
-
- if (!stream->metadata_flag) {
- struct lttng_index_file *index_file;
-
- lttng_index_file_put(stream->index_file);
-
- index_file = lttng_index_file_create(
- stream->channel_read_only_attributes.path,
- stream->name, stream->uid, stream->gid,
- stream->channel_read_only_attributes.tracefile_size,
- stream->tracefile_count_current,
- CTF_INDEX_MAJOR, CTF_INDEX_MINOR);
- if (!index_file) {
- ERR("Create index file during rotation");
- goto error;
- }
- stream->index_file = index_file;
- stream->out_fd_offset = 0;
- }
-
- ret = 0;
- goto end;
+ ret = LTTCOMM_CONSUMERD_SUCCESS;
error:
- ret = -1;
-end:
return ret;
-
}
-/*
- * Perform the rotation a stream file on the relay.
- */
-int rotate_relay_stream(struct lttng_consumer_local_data *ctx,
- struct lttng_consumer_stream *stream)
+static
+int consumer_clear_unmonitored_channel(struct lttng_consumer_channel *channel)
{
int ret;
- struct consumer_relayd_sock_pair *relayd;
+ struct lttng_consumer_stream *stream;
- DBG("Rotate relay stream");
- relayd = consumer_find_relayd(stream->net_seq_idx);
- if (!relayd) {
- ERR("Failed to find relayd");
- ret = -1;
- goto end;
+ 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;
- pthread_mutex_lock(&relayd->ctrl_sock_mutex);
- ret = relayd_rotate_stream(&relayd->control_sock,
- stream->relayd_stream_id,
- stream->channel_read_only_attributes.path,
- stream->chan->current_chunk_id,
- stream->last_sequence_number);
- pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
- if (ret < 0) {
- ERR("Relayd rotate stream failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
- lttng_consumer_cleanup_relayd(relayd);
+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 (ret) {
- ERR("Rotate relay stream");
+
+ /*
+ * 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)
+{
+ DBG("lttng_consumer_reset_stream_rotate_state for stream %" PRIu64,
+ stream->key);
+ stream->rotate_position = -1ULL;
+ stream->rotate_ready = false;
+}
+
+/*
+ * Perform the rotation a local stream file.
+ */
+static
+int rotate_local_stream(struct lttng_consumer_local_data *ctx,
+ struct lttng_consumer_stream *stream)
+{
+ int ret = 0;
+
+ DBG("Rotate local stream: stream key %" PRIu64 ", channel key %" PRIu64,
+ stream->key,
+ stream->chan->key);
+ stream->tracefile_size_current = 0;
+ stream->tracefile_count_current = 0;
+
+ if (stream->out_fd >= 0) {
+ ret = close(stream->out_fd);
+ if (ret) {
+ PERROR("Failed to close stream out_fd of channel \"%s\"",
+ stream->chan->name);
+ }
+ stream->out_fd = -1;
+ }
+
+ if (stream->index_file) {
+ lttng_index_file_put(stream->index_file);
+ stream->index_file = NULL;
+ }
+
+ if (!stream->trace_chunk) {
+ goto end;
+ }
+
+ ret = consumer_stream_create_output_files(stream, true);
end:
return ret;
}
/*
* Performs the stream rotation for the rotate session feature if needed.
- * It must be called with the stream lock held.
+ * It must be called with the channel and stream locks held.
*
* Return 0 on success, a negative number of error.
*/
int lttng_consumer_rotate_stream(struct lttng_consumer_local_data *ctx,
- struct lttng_consumer_stream *stream, bool *rotated)
+ struct lttng_consumer_stream *stream)
{
int ret;
DBG("Consumer rotate stream %" PRIu64, stream->key);
- if (stream->net_seq_idx != (uint64_t) -1ULL) {
- ret = rotate_relay_stream(ctx, stream);
- } else {
- ret = rotate_local_stream(ctx, stream);
- }
- stream->trace_archive_id++;
- if (ret < 0) {
- ERR("Failed to rotate stream, ret = %i", ret);
+ /*
+ * Update the stream's 'current' chunk to the session's (channel)
+ * now-current chunk.
+ */
+ lttng_trace_chunk_put(stream->trace_chunk);
+ if (stream->chan->trace_chunk == stream->trace_chunk) {
+ /*
+ * A channel can be rotated and not have a "next" chunk
+ * to transition to. In that case, the channel's "current chunk"
+ * has not been closed yet, but it has not been updated to
+ * a "next" trace chunk either. Hence, the stream, like its
+ * parent channel, becomes part of no chunk and can't output
+ * anything until a new trace chunk is created.
+ */
+ stream->trace_chunk = NULL;
+ } else if (stream->chan->trace_chunk &&
+ !lttng_trace_chunk_get(stream->chan->trace_chunk)) {
+ ERR("Failed to acquire a reference to channel's trace chunk during stream rotation");
+ ret = -1;
goto error;
+ } else {
+ /*
+ * Update the stream's trace chunk to its parent channel's
+ * current trace chunk.
+ */
+ stream->trace_chunk = stream->chan->trace_chunk;
}
- if (stream->metadata_flag) {
- switch (consumer_data.type) {
- case LTTNG_CONSUMER_KERNEL:
- /*
- * Reset the position of what has been read from the metadata
- * cache to 0 so we can dump it again.
- */
- ret = kernctl_metadata_cache_dump(stream->wait_fd);
- if (ret < 0) {
- ERR("Failed to dump the kernel metadata cache after rotation");
- goto error;
- }
- break;
- case LTTNG_CONSUMER32_UST:
- case LTTNG_CONSUMER64_UST:
- /*
- * Reset the position pushed from the metadata cache so it
- * will write from the beginning on the next push.
- */
- stream->ust_metadata_pushed = 0;
- break;
- default:
- ERR("Unknown consumer_data type");
- abort();
+ if (stream->net_seq_idx == (uint64_t) -1ULL) {
+ ret = rotate_local_stream(ctx, stream);
+ if (ret < 0) {
+ ERR("Failed to rotate stream, ret = %i", ret);
+ goto error;
}
}
- lttng_consumer_reset_stream_rotate_state(stream);
- if (rotated) {
- *rotated = true;
+ if (stream->metadata_flag && stream->trace_chunk) {
+ /*
+ * If the stream has transitioned to a new trace
+ * chunk, the metadata should be re-dumped to the
+ * newest chunk.
+ *
+ * However, it is possible for a stream to transition to
+ * a "no-chunk" state. This can happen if a rotation
+ * occurs on an inactive session. In such cases, the metadata
+ * regeneration will happen when the next trace chunk is
+ * created.
+ */
+ ret = consumer_metadata_stream_dump(stream);
+ if (ret) {
+ goto error;
+ }
}
+ lttng_consumer_reset_stream_rotate_state(stream);
ret = 0;
stream, node_channel_id.node) {
health_code_update();
+ pthread_mutex_lock(&stream->chan->lock);
pthread_mutex_lock(&stream->lock);
if (!stream->rotate_ready) {
pthread_mutex_unlock(&stream->lock);
+ pthread_mutex_unlock(&stream->chan->lock);
continue;
}
DBG("Consumer rotate ready stream %" PRIu64, stream->key);
- ret = lttng_consumer_rotate_stream(ctx, stream, NULL);
+ ret = lttng_consumer_rotate_stream(ctx, stream);
pthread_mutex_unlock(&stream->lock);
- if (ret) {
- goto end;
- }
-
- ret = consumer_post_rotation(stream, ctx);
+ pthread_mutex_unlock(&stream->chan->lock);
if (ret) {
goto end;
}
return ret;
}
-static
-int rotate_rename_local(const char *old_path, const char *new_path,
- uid_t uid, gid_t gid)
+enum lttcomm_return_code lttng_consumer_init_command(
+ struct lttng_consumer_local_data *ctx,
+ const lttng_uuid sessiond_uuid)
{
- int ret;
+ enum lttcomm_return_code ret;
+ char uuid_str[LTTNG_UUID_STR_LEN];
- assert(old_path);
- assert(new_path);
-
- ret = utils_mkdir_recursive(new_path, S_IRWXU | S_IRWXG, uid, gid);
- if (ret < 0) {
- ERR("Create directory on rotate");
+ if (ctx->sessiond_uuid.is_set) {
+ ret = LTTCOMM_CONSUMERD_ALREADY_SET;
goto end;
}
- ret = rename(old_path, new_path);
- if (ret < 0 && errno != ENOENT) {
- PERROR("Rename completed rotation chunk");
- goto end;
- }
-
- ret = 0;
+ ctx->sessiond_uuid.is_set = true;
+ memcpy(ctx->sessiond_uuid.value, sessiond_uuid, sizeof(lttng_uuid));
+ ret = LTTCOMM_CONSUMERD_SUCCESS;
+ lttng_uuid_to_str(sessiond_uuid, uuid_str);
+ DBG("Received session daemon UUID: %s", uuid_str);
end:
return ret;
}
-static
-int rotate_rename_relay(const char *old_path, const char *new_path,
- uint64_t relayd_id)
+enum lttcomm_return_code lttng_consumer_create_trace_chunk(
+ const uint64_t *relayd_id, uint64_t session_id,
+ uint64_t chunk_id,
+ time_t chunk_creation_timestamp,
+ const char *chunk_override_name,
+ const struct lttng_credentials *credentials,
+ struct lttng_directory_handle *chunk_directory_handle)
{
int ret;
- struct consumer_relayd_sock_pair *relayd;
+ enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
+ struct lttng_trace_chunk *created_chunk = NULL, *published_chunk = NULL;
+ enum lttng_trace_chunk_status chunk_status;
+ char relayd_id_buffer[MAX_INT_DEC_LEN(*relayd_id)];
+ char creation_timestamp_buffer[ISO8601_STR_LEN];
+ const char *relayd_id_str = "(none)";
+ const char *creation_timestamp_str;
+ struct lttng_ht_iter iter;
+ struct lttng_consumer_channel *channel;
- relayd = consumer_find_relayd(relayd_id);
- if (!relayd) {
- ERR("Failed to find relayd while running rotate_rename_relay command");
- ret = -1;
- goto end;
+ if (relayd_id) {
+ /* Only used for logging purposes. */
+ ret = snprintf(relayd_id_buffer, sizeof(relayd_id_buffer),
+ "%" PRIu64, *relayd_id);
+ if (ret > 0 && ret < sizeof(relayd_id_buffer)) {
+ relayd_id_str = relayd_id_buffer;
+ } else {
+ relayd_id_str = "(formatting error)";
+ }
+ }
+
+ /* Local protocol error. */
+ assert(chunk_creation_timestamp);
+ ret = time_to_iso8601_str(chunk_creation_timestamp,
+ creation_timestamp_buffer,
+ sizeof(creation_timestamp_buffer));
+ creation_timestamp_str = !ret ? creation_timestamp_buffer :
+ "(formatting error)";
+
+ DBG("Consumer create trace chunk command: relay_id = %s"
+ ", session_id = %" PRIu64 ", chunk_id = %" PRIu64
+ ", chunk_override_name = %s"
+ ", chunk_creation_timestamp = %s",
+ relayd_id_str, session_id, chunk_id,
+ chunk_override_name ? : "(none)",
+ creation_timestamp_str);
+
+ /*
+ * The trace chunk registry, as used by the consumer daemon, implicitly
+ * owns the trace chunks. This is only needed in the consumer since
+ * the consumer has no notion of a session beyond session IDs being
+ * used to identify other objects.
+ *
+ * The lttng_trace_chunk_registry_publish() call below provides a
+ * reference which is not released; it implicitly becomes the session
+ * daemon's reference to the chunk in the consumer daemon.
+ *
+ * The lifetime of trace chunks in the consumer daemon is managed by
+ * the session daemon through the LTTNG_CONSUMER_CREATE_TRACE_CHUNK
+ * and LTTNG_CONSUMER_DESTROY_TRACE_CHUNK commands.
+ */
+ created_chunk = lttng_trace_chunk_create(chunk_id,
+ chunk_creation_timestamp, NULL);
+ if (!created_chunk) {
+ ERR("Failed to create trace chunk");
+ ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
+ goto error;
}
- pthread_mutex_lock(&relayd->ctrl_sock_mutex);
- ret = relayd_rotate_rename(&relayd->control_sock, old_path, new_path);
- if (ret < 0) {
- ERR("Relayd rotate rename failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
- lttng_consumer_cleanup_relayd(relayd);
+ if (chunk_override_name) {
+ chunk_status = lttng_trace_chunk_override_name(created_chunk,
+ chunk_override_name);
+ if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
+ ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
+ goto error;
+ }
}
- pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
-end:
- return ret;
-}
-int lttng_consumer_rotate_rename(const char *old_path, const char *new_path,
- uid_t uid, gid_t gid, uint64_t relayd_id)
-{
- if (relayd_id != -1ULL) {
- return rotate_rename_relay(old_path, new_path, relayd_id);
- } else {
- return rotate_rename_local(old_path, new_path, uid, gid);
+ if (chunk_directory_handle) {
+ chunk_status = lttng_trace_chunk_set_credentials(created_chunk,
+ credentials);
+ if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
+ ERR("Failed to set trace chunk credentials");
+ ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
+ goto error;
+ }
+ /*
+ * The consumer daemon has no ownership of the chunk output
+ * directory.
+ */
+ 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;
+ goto error;
+ }
}
-}
-/* Stream lock must be acquired by the caller. */
-static
-bool check_stream_rotation_pending(const struct lttng_consumer_stream *stream,
- uint64_t session_id, uint64_t chunk_id)
-{
- bool pending = false;
+ published_chunk = lttng_trace_chunk_registry_publish_chunk(
+ consumer_data.chunk_registry, session_id,
+ created_chunk);
+ lttng_trace_chunk_put(created_chunk);
+ created_chunk = NULL;
+ if (!published_chunk) {
+ ERR("Failed to publish trace chunk");
+ ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
+ goto error;
+ }
- if (stream->session_id != session_id) {
- /* Skip. */
- goto end;
+ rcu_read_lock();
+ cds_lfht_for_each_entry_duplicate(consumer_data.channels_by_session_id_ht->ht,
+ consumer_data.channels_by_session_id_ht->hash_fct(
+ &session_id, lttng_ht_seed),
+ consumer_data.channels_by_session_id_ht->match_fct,
+ &session_id, &iter.iter, channel,
+ channels_by_session_id_ht_node.node) {
+ ret = lttng_consumer_channel_set_trace_chunk(channel,
+ published_chunk);
+ if (ret) {
+ /*
+ * Roll-back the creation of this chunk.
+ *
+ * This is important since the session daemon will
+ * assume that the creation of this chunk failed and
+ * will never ask for it to be closed, resulting
+ * in a leak and an inconsistent state for some
+ * channels.
+ */
+ enum lttcomm_return_code close_ret;
+ char path[LTTNG_PATH_MAX];
+
+ DBG("Failed to set new trace chunk on existing channels, rolling back");
+ close_ret = lttng_consumer_close_trace_chunk(relayd_id,
+ session_id, chunk_id,
+ chunk_creation_timestamp, NULL,
+ path);
+ if (close_ret != LTTCOMM_CONSUMERD_SUCCESS) {
+ ERR("Failed to roll-back the creation of new chunk: session_id = %" PRIu64 ", chunk_id = %" PRIu64,
+ session_id, chunk_id);
+ }
+
+ ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
+ break;
+ }
}
- /*
- * If the stream's archive_id belongs to the chunk being rotated (or an
- * even older one), it means that the consumer has not consumed all the
- * buffers that belong to the chunk being rotated. Therefore, the
- * rotation is considered as ongoing/pending.
- */
- pending = stream->trace_archive_id <= chunk_id;
-end:
- return pending;
+ if (relayd_id) {
+ struct consumer_relayd_sock_pair *relayd;
+
+ relayd = consumer_find_relayd(*relayd_id);
+ if (relayd) {
+ pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+ ret = relayd_create_trace_chunk(
+ &relayd->control_sock, published_chunk);
+ pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
+ } else {
+ ERR("Failed to find relay daemon socket: relayd_id = %" PRIu64, *relayd_id);
+ }
+
+ if (!relayd || ret) {
+ enum lttcomm_return_code close_ret;
+ char path[LTTNG_PATH_MAX];
+
+ close_ret = lttng_consumer_close_trace_chunk(relayd_id,
+ session_id,
+ chunk_id,
+ chunk_creation_timestamp,
+ NULL, path);
+ if (close_ret != LTTCOMM_CONSUMERD_SUCCESS) {
+ ERR("Failed to roll-back the creation of new chunk: session_id = %" PRIu64 ", chunk_id = %" PRIu64,
+ session_id,
+ chunk_id);
+ }
+
+ ret_code = LTTCOMM_CONSUMERD_CREATE_TRACE_CHUNK_FAILED;
+ goto error_unlock;
+ }
+ }
+error_unlock:
+ rcu_read_unlock();
+error:
+ /* Release the reference returned by the "publish" operation. */
+ lttng_trace_chunk_put(published_chunk);
+ lttng_trace_chunk_put(created_chunk);
+ return ret_code;
}
-/* RCU read lock must be acquired by the caller. */
-int lttng_consumer_check_rotation_pending_local(uint64_t session_id,
- uint64_t chunk_id)
+enum lttcomm_return_code lttng_consumer_close_trace_chunk(
+ const uint64_t *relayd_id, uint64_t session_id,
+ uint64_t chunk_id, time_t chunk_close_timestamp,
+ const enum lttng_trace_chunk_command_type *close_command,
+ char *path)
{
+ enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
+ struct lttng_trace_chunk *chunk;
+ char relayd_id_buffer[MAX_INT_DEC_LEN(*relayd_id)];
+ const char *relayd_id_str = "(none)";
+ const char *close_command_name = "none";
struct lttng_ht_iter iter;
- struct lttng_consumer_stream *stream;
- bool rotation_pending = false;
+ struct lttng_consumer_channel *channel;
+ enum lttng_trace_chunk_status chunk_status;
- /* Start with the metadata streams... */
- cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream, node.node) {
- pthread_mutex_lock(&stream->lock);
- rotation_pending = check_stream_rotation_pending(stream,
- session_id, chunk_id);
- pthread_mutex_unlock(&stream->lock);
- if (rotation_pending) {
- goto end;
+ if (relayd_id) {
+ int ret;
+
+ /* Only used for logging purposes. */
+ ret = snprintf(relayd_id_buffer, sizeof(relayd_id_buffer),
+ "%" PRIu64, *relayd_id);
+ if (ret > 0 && ret < sizeof(relayd_id_buffer)) {
+ relayd_id_str = relayd_id_buffer;
+ } else {
+ relayd_id_str = "(formatting error)";
}
}
+ if (close_command) {
+ close_command_name = lttng_trace_chunk_command_type_get_name(
+ *close_command);
+ }
- /* ... followed by the data streams. */
- cds_lfht_for_each_entry(data_ht->ht, &iter.iter, stream, node.node) {
- pthread_mutex_lock(&stream->lock);
- rotation_pending = check_stream_rotation_pending(stream,
+ DBG("Consumer close trace chunk command: relayd_id = %s"
+ ", session_id = %" PRIu64 ", chunk_id = %" PRIu64
+ ", close command = %s",
+ relayd_id_str, session_id, chunk_id,
+ close_command_name);
+
+ chunk = lttng_trace_chunk_registry_find_chunk(
+ consumer_data.chunk_registry, session_id, chunk_id);
+ if (!chunk) {
+ ERR("Failed to find chunk: session_id = %" PRIu64
+ ", chunk_id = %" PRIu64,
session_id, chunk_id);
- pthread_mutex_unlock(&stream->lock);
- if (rotation_pending) {
+ ret_code = LTTCOMM_CONSUMERD_UNKNOWN_TRACE_CHUNK;
+ goto end;
+ }
+
+ chunk_status = lttng_trace_chunk_set_close_timestamp(chunk,
+ chunk_close_timestamp);
+ if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
+ ret_code = LTTCOMM_CONSUMERD_CLOSE_TRACE_CHUNK_FAILED;
+ goto end;
+ }
+
+ if (close_command) {
+ chunk_status = lttng_trace_chunk_set_close_command(
+ chunk, *close_command);
+ if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
+ ret_code = LTTCOMM_CONSUMERD_CLOSE_TRACE_CHUNK_FAILED;
goto end;
}
}
+ /*
+ * chunk is now invalid to access as we no longer hold a reference to
+ * it; it is only kept around to compare it (by address) to the
+ * current chunk found in the session's channels.
+ */
+ rcu_read_lock();
+ cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter,
+ channel, node.node) {
+ int ret;
+
+ /*
+ * Only change the channel's chunk to NULL if it still
+ * references the chunk being closed. The channel may
+ * reference a newer channel in the case of a session
+ * rotation. When a session rotation occurs, the "next"
+ * chunk is created before the "current" chunk is closed.
+ */
+ if (channel->trace_chunk != chunk) {
+ continue;
+ }
+ ret = lttng_consumer_channel_set_trace_chunk(channel, NULL);
+ if (ret) {
+ /*
+ * Attempt to close the chunk on as many channels as
+ * possible.
+ */
+ ret_code = LTTCOMM_CONSUMERD_CLOSE_TRACE_CHUNK_FAILED;
+ }
+ }
+
+ if (relayd_id) {
+ int ret;
+ struct consumer_relayd_sock_pair *relayd;
+
+ relayd = consumer_find_relayd(*relayd_id);
+ if (relayd) {
+ pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+ ret = relayd_close_trace_chunk(
+ &relayd->control_sock, chunk,
+ path);
+ pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
+ } else {
+ ERR("Failed to find relay daemon socket: relayd_id = %" PRIu64,
+ *relayd_id);
+ }
+
+ if (!relayd || ret) {
+ ret_code = LTTCOMM_CONSUMERD_CLOSE_TRACE_CHUNK_FAILED;
+ goto error_unlock;
+ }
+ }
+error_unlock:
+ rcu_read_unlock();
end:
- return !!rotation_pending;
+ /*
+ * Release the reference returned by the "find" operation and
+ * the session daemon's implicit reference to the chunk.
+ */
+ lttng_trace_chunk_put(chunk);
+ lttng_trace_chunk_put(chunk);
+
+ return ret_code;
}
-int lttng_consumer_check_rotation_pending_relay(uint64_t session_id,
- uint64_t relayd_id, uint64_t chunk_id)
+enum lttcomm_return_code lttng_consumer_trace_chunk_exists(
+ const uint64_t *relayd_id, uint64_t session_id,
+ uint64_t chunk_id)
{
int ret;
- struct consumer_relayd_sock_pair *relayd;
+ enum lttcomm_return_code ret_code;
+ char relayd_id_buffer[MAX_INT_DEC_LEN(*relayd_id)];
+ const char *relayd_id_str = "(none)";
+ const bool is_local_trace = !relayd_id;
+ struct consumer_relayd_sock_pair *relayd = NULL;
+ bool chunk_exists_local, chunk_exists_remote;
- relayd = consumer_find_relayd(relayd_id);
- if (!relayd) {
- ERR("Failed to find relayd id %" PRIu64, relayd_id);
- ret = -1;
+ if (relayd_id) {
+ int ret;
+
+ /* Only used for logging purposes. */
+ ret = snprintf(relayd_id_buffer, sizeof(relayd_id_buffer),
+ "%" PRIu64, *relayd_id);
+ if (ret > 0 && ret < sizeof(relayd_id_buffer)) {
+ relayd_id_str = relayd_id_buffer;
+ } else {
+ relayd_id_str = "(formatting error)";
+ }
+ }
+
+ DBG("Consumer trace chunk exists command: relayd_id = %s"
+ ", chunk_id = %" PRIu64, relayd_id_str,
+ chunk_id);
+ ret = lttng_trace_chunk_registry_chunk_exists(
+ consumer_data.chunk_registry, session_id,
+ chunk_id, &chunk_exists_local);
+ if (ret) {
+ /* Internal error. */
+ ERR("Failed to query the existence of a trace chunk");
+ ret_code = LTTCOMM_CONSUMERD_FATAL;
+ goto end;
+ }
+ DBG("Trace chunk %s locally",
+ chunk_exists_local ? "exists" : "does not exist");
+ if (chunk_exists_local) {
+ ret_code = LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_LOCAL;
+ goto end;
+ } else if (is_local_trace) {
+ ret_code = LTTCOMM_CONSUMERD_UNKNOWN_TRACE_CHUNK;
goto end;
}
+ rcu_read_lock();
+ relayd = consumer_find_relayd(*relayd_id);
+ if (!relayd) {
+ ERR("Failed to find relayd %" PRIu64, *relayd_id);
+ ret_code = LTTCOMM_CONSUMERD_INVALID_PARAMETERS;
+ goto end_rcu_unlock;
+ }
+ DBG("Looking up existence of trace chunk on relay daemon");
pthread_mutex_lock(&relayd->ctrl_sock_mutex);
- ret = relayd_rotate_pending(&relayd->control_sock, chunk_id);
+ ret = relayd_trace_chunk_exists(&relayd->control_sock, chunk_id,
+ &chunk_exists_remote);
+ pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
if (ret < 0) {
- ERR("Relayd rotate pending failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
- lttng_consumer_cleanup_relayd(relayd);
+ ERR("Failed to look-up the existence of trace chunk on relay daemon");
+ ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL;
+ goto end_rcu_unlock;
}
- pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
+ ret_code = chunk_exists_remote ?
+ LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_REMOTE :
+ LTTCOMM_CONSUMERD_UNKNOWN_TRACE_CHUNK;
+ DBG("Trace chunk %s on relay daemon",
+ chunk_exists_remote ? "exists" : "does not exist");
+
+end_rcu_unlock:
+ rcu_read_unlock();
end:
- return ret;
+ return ret_code;
}
static
-int mkdir_local(const char *path, uid_t uid, gid_t gid)
+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;
- ret = utils_mkdir_recursive(path, S_IRWXU | S_IRWXG, uid, gid);
- if (ret < 0) {
- /* utils_mkdir_recursive logs an error. */
- goto end;
+ 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;
- ret = 0;
-end:
+error_unlock:
+ pthread_mutex_unlock(&stream->lock);
+ rcu_read_unlock();
return ret;
}
-static
-int mkdir_relay(const char *path, uint64_t relayd_id)
+int lttng_consumer_clear_channel(struct lttng_consumer_channel *channel)
{
int ret;
- struct consumer_relayd_sock_pair *relayd;
- relayd = consumer_find_relayd(relayd_id);
- if (!relayd) {
- ERR("Failed to find relayd");
- ret = -1;
+ 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;
}
- pthread_mutex_lock(&relayd->ctrl_sock_mutex);
- ret = relayd_mkdir(&relayd->control_sock, path);
- if (ret < 0) {
- ERR("Relayd mkdir failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
- lttng_consumer_cleanup_relayd(relayd);
+ if (!channel->monitor) {
+ ret = consumer_clear_unmonitored_channel(channel);
+ } else {
+ ret = consumer_clear_monitored_channel(channel);
}
- pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
-
end:
return ret;
-
}
-int lttng_consumer_mkdir(const char *path, uid_t uid, gid_t gid,
- uint64_t relayd_id)
+enum lttcomm_return_code lttng_consumer_open_channel_packets(
+ struct lttng_consumer_channel *channel)
{
- if (relayd_id != -1ULL) {
- return mkdir_relay(path, relayd_id);
- } else {
- return mkdir_local(path, uid, gid);
+ struct lttng_consumer_stream *stream;
+ enum lttcomm_return_code ret = LTTCOMM_CONSUMERD_SUCCESS;
+
+ if (channel->metadata_stream) {
+ ERR("Open channel packets command attempted on a metadata channel");
+ ret = LTTCOMM_CONSUMERD_INVALID_PARAMETERS;
+ goto end;
}
+
+ rcu_read_lock();
+ cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
+ enum consumer_stream_open_packet_status status;
+
+ pthread_mutex_lock(&stream->lock);
+ if (cds_lfht_is_node_deleted(&stream->node.node)) {
+ goto next;
+ }
+
+ status = consumer_stream_open_packet(stream);
+ switch (status) {
+ case CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED:
+ DBG("Opened a packet in \"open channel packets\" command: stream id = %" PRIu64
+ ", channel name = %s, session id = %" PRIu64,
+ stream->key, stream->chan->name,
+ stream->chan->session_id);
+ stream->opened_packet_in_current_trace_chunk = true;
+ break;
+ case CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE:
+ DBG("No space left to open a packet in \"open channel packets\" command: stream id = %" PRIu64
+ ", channel name = %s, session id = %" PRIu64,
+ stream->key, stream->chan->name,
+ stream->chan->session_id);
+ break;
+ case CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR:
+ /*
+ * Only unexpected internal errors can lead to this
+ * failing. Report an unknown error.
+ */
+ ERR("Failed to flush empty buffer in \"open channel packets\" command: stream id = %" PRIu64
+ ", channel id = %" PRIu64
+ ", channel name = %s"
+ ", session id = %" PRIu64,
+ stream->key, channel->key,
+ channel->name, channel->session_id);
+ ret = LTTCOMM_CONSUMERD_UNKNOWN_ERROR;
+ goto error_unlock;
+ default:
+ abort();
+ }
+
+ next:
+ pthread_mutex_unlock(&stream->lock);
+ }
+
+end_rcu_unlock:
+ rcu_read_unlock();
+end:
+ return ret;
+
+error_unlock:
+ pthread_mutex_unlock(&stream->lock);
+ goto end_rcu_unlock;
}
projects / lttng-tools.git / blobdiff