#include "common/index/ctf-index.h"
#define _LGPL_SOURCE
-#include <assert.h>
#include <poll.h>
#include <pthread.h>
#include <stdlib.h>
#include <common/string-utils/format.h>
#include <common/dynamic-array.h>
-struct lttng_consumer_global_data consumer_data = {
- .stream_count = 0,
- .need_update = 1,
- .type = LTTNG_CONSUMER_UNKNOWN,
+struct lttng_consumer_global_data the_consumer_data = {
+ .stream_count = 0,
+ .need_update = 1,
+ .type = LTTNG_CONSUMER_UNKNOWN,
};
enum consumer_channel_action {
static const char *get_consumer_domain(void)
{
- switch (consumer_data.type) {
+ switch (the_consumer_data.type) {
case LTTNG_CONSUMER_KERNEL:
return DEFAULT_KERNEL_TRACE_DIR;
case LTTNG_CONSUMER64_UST:
{
struct lttng_consumer_stream *null_stream = NULL;
- assert(pipe);
+ LTTNG_ASSERT(pipe);
(void) lttng_pipe_write(pipe, &null_stream, sizeof(null_stream));
}
{
struct lttng_consumer_stream *stream, *stmp;
- assert(channel);
+ LTTNG_ASSERT(channel);
/* Delete streams that might have been left in the stream list. */
cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
struct lttng_ht_node_u64 *node;
struct lttng_consumer_stream *stream = NULL;
- assert(ht);
+ LTTNG_ASSERT(ht);
/* -1ULL keys are lookup failures */
if (key == (uint64_t) -1ULL) {
return NULL;
}
- lttng_ht_lookup(consumer_data.channel_ht, &key, &iter);
+ lttng_ht_lookup(the_consumer_data.channel_ht, &key, &iter);
node = lttng_ht_iter_get_node_u64(&iter);
if (node != NULL) {
channel = caa_container_of(node, struct lttng_consumer_channel, node);
struct lttng_consumer_channel *channel =
caa_container_of(node, struct lttng_consumer_channel, node);
- switch (consumer_data.type) {
+ switch (the_consumer_data.type) {
case LTTNG_CONSUMER_KERNEL:
break;
case LTTNG_CONSUMER32_UST:
DBG("Consumer destroy and close relayd socket pair");
iter.iter.node = &relayd->node.node;
- ret = lttng_ht_del(consumer_data.relayd_ht, &iter);
+ ret = lttng_ht_del(the_consumer_data.relayd_ht, &iter);
if (ret != 0) {
/* We assume the relayd is being or is destroyed */
return;
DBG("Consumer delete channel key %" PRIu64, channel->key);
- pthread_mutex_lock(&consumer_data.lock);
+ pthread_mutex_lock(&the_consumer_data.lock);
pthread_mutex_lock(&channel->lock);
/* Destroy streams that might have been left in the stream list. */
consumer_timer_monitor_stop(channel);
}
- switch (consumer_data.type) {
+ switch (the_consumer_data.type) {
case LTTNG_CONSUMER_KERNEL:
break;
case LTTNG_CONSUMER32_UST:
break;
default:
ERR("Unknown consumer_data type");
- assert(0);
+ abort();
goto end;
}
rcu_read_lock();
iter.iter.node = &channel->node.node;
- ret = lttng_ht_del(consumer_data.channel_ht, &iter);
- assert(!ret);
+ ret = lttng_ht_del(the_consumer_data.channel_ht, &iter);
+ LTTNG_ASSERT(!ret);
iter.iter.node = &channel->channels_by_session_id_ht_node.node;
- ret = lttng_ht_del(consumer_data.channels_by_session_id_ht,
+ ret = lttng_ht_del(the_consumer_data.channels_by_session_id_ht,
&iter);
- assert(!ret);
+ LTTNG_ASSERT(!ret);
rcu_read_unlock();
}
call_rcu(&channel->node.head, free_channel_rcu);
end:
pthread_mutex_unlock(&channel->lock);
- pthread_mutex_unlock(&consumer_data.lock);
+ pthread_mutex_unlock(&the_consumer_data.lock);
}
/*
rcu_read_lock();
- cds_lfht_for_each_entry(consumer_data.relayd_ht->ht, &iter.iter, relayd,
- node.node) {
+ cds_lfht_for_each_entry(the_consumer_data.relayd_ht->ht, &iter.iter,
+ relayd, node.node) {
consumer_destroy_relayd(relayd);
}
rcu_read_unlock();
- lttng_ht_destroy(consumer_data.relayd_ht);
+ lttng_ht_destroy(the_consumer_data.relayd_ht);
}
/*
{
uint64_t netidx;
- assert(relayd);
+ LTTNG_ASSERT(relayd);
DBG("Cleaning up relayd object ID %"PRIu64, relayd->net_seq_idx);
*/
void consumer_flag_relayd_for_destroy(struct consumer_relayd_sock_pair *relayd)
{
- assert(relayd);
+ LTTNG_ASSERT(relayd);
/* Set destroy flag for this object */
uatomic_set(&relayd->destroy_flag, 1);
{
struct lttng_ht *ht = data_ht;
- assert(stream);
- assert(ht);
+ LTTNG_ASSERT(stream);
+ LTTNG_ASSERT(ht);
DBG3("Adding consumer stream %" PRIu64, stream->key);
- pthread_mutex_lock(&consumer_data.lock);
+ pthread_mutex_lock(&the_consumer_data.lock);
pthread_mutex_lock(&stream->chan->lock);
pthread_mutex_lock(&stream->chan->timer_lock);
pthread_mutex_lock(&stream->lock);
lttng_ht_add_unique_u64(ht, &stream->node);
- lttng_ht_add_u64(consumer_data.stream_per_chan_id_ht,
+ lttng_ht_add_u64(the_consumer_data.stream_per_chan_id_ht,
&stream->node_channel_id);
/*
* the key since the HT does not use it and we allow to add redundant keys
* into this table.
*/
- lttng_ht_add_u64(consumer_data.stream_list_ht, &stream->node_session_id);
+ lttng_ht_add_u64(the_consumer_data.stream_list_ht,
+ &stream->node_session_id);
/*
* When nb_init_stream_left reaches 0, we don't need to trigger any action
}
/* Update consumer data once the node is inserted. */
- consumer_data.stream_count++;
- consumer_data.need_update = 1;
+ the_consumer_data.stream_count++;
+ the_consumer_data.need_update = 1;
rcu_read_unlock();
pthread_mutex_unlock(&stream->lock);
pthread_mutex_unlock(&stream->chan->timer_lock);
pthread_mutex_unlock(&stream->chan->lock);
- pthread_mutex_unlock(&consumer_data.lock);
+ pthread_mutex_unlock(&the_consumer_data.lock);
}
/*
struct lttng_ht_node_u64 *node;
struct lttng_ht_iter iter;
- assert(relayd);
+ LTTNG_ASSERT(relayd);
- lttng_ht_lookup(consumer_data.relayd_ht,
- &relayd->net_seq_idx, &iter);
+ lttng_ht_lookup(the_consumer_data.relayd_ht, &relayd->net_seq_idx,
+ &iter);
node = lttng_ht_iter_get_node_u64(&iter);
if (node != NULL) {
goto end;
}
- lttng_ht_add_unique_u64(consumer_data.relayd_ht, &relayd->node);
+ lttng_ht_add_unique_u64(the_consumer_data.relayd_ht, &relayd->node);
end:
return ret;
goto error;
}
- lttng_ht_lookup(consumer_data.relayd_ht, &key,
- &iter);
+ lttng_ht_lookup(the_consumer_data.relayd_ht, &key, &iter);
node = lttng_ht_iter_get_node_u64(&iter);
if (node != NULL) {
relayd = caa_container_of(node, struct consumer_relayd_sock_pair, node);
int ret = 0;
struct consumer_relayd_sock_pair *relayd;
- assert(stream);
- assert(stream->net_seq_idx != -1ULL);
- assert(path);
+ LTTNG_ASSERT(stream);
+ LTTNG_ASSERT(stream->net_seq_idx != -1ULL);
+ LTTNG_ASSERT(path);
/* The stream is not metadata. Get relayd reference if exists. */
rcu_read_lock();
int ret = 0;
struct consumer_relayd_sock_pair *relayd;
- assert(net_seq_idx != -1ULL);
+ LTTNG_ASSERT(net_seq_idx != -1ULL);
/* The stream is not metadata. Get relayd reference if exists. */
rcu_read_lock();
struct lttcomm_relayd_data_hdr data_hdr;
/* Safety net */
- assert(stream);
- assert(relayd);
+ LTTNG_ASSERT(stream);
+ LTTNG_ASSERT(relayd);
/* Reset data header */
memset(&data_hdr, 0, sizeof(data_hdr));
return outfd;
}
+/*
+ * Write a character on the metadata poll pipe to wake the metadata thread.
+ * Returns 0 on success, -1 on error.
+ */
+int consumer_metadata_wakeup_pipe(const struct lttng_consumer_channel *channel)
+{
+ int ret = 0;
+
+ DBG("Waking up metadata poll thread (writing to pipe): channel name = '%s'",
+ channel->name);
+ if (channel->monitor && channel->metadata_stream) {
+ const char dummy = 'c';
+ const ssize_t write_ret = lttng_write(
+ channel->metadata_stream->ust_metadata_poll_pipe[1],
+ &dummy, 1);
+
+ if (write_ret < 1) {
+ if (errno == EWOULDBLOCK) {
+ /*
+ * This is fine, the metadata poll thread
+ * is having a hard time keeping-up, but
+ * it will eventually wake-up and consume
+ * the available data.
+ */
+ ret = 0;
+ } else {
+ PERROR("Failed to write to UST metadata pipe while attempting to wake-up the metadata poll thread");
+ ret = -1;
+ goto end;
+ }
+ }
+ }
+
+end:
+ return ret;
+}
+
/*
* Trigger a dump of the metadata content. Following/during the succesful
* completion of this call, the metadata poll thread will start receiving
ASSERT_LOCKED(stream->chan->lock);
ASSERT_LOCKED(stream->lock);
- assert(stream->metadata_flag);
- assert(stream->chan->trace_chunk);
+ LTTNG_ASSERT(stream->metadata_flag);
+ LTTNG_ASSERT(stream->chan->trace_chunk);
- switch (consumer_data.type) {
+ switch (the_consumer_data.type) {
case LTTNG_CONSUMER_KERNEL:
/*
* Reset the position of what has been read from the
const bool acquired_reference = lttng_trace_chunk_get(
new_trace_chunk);
- assert(acquired_reference);
+ LTTNG_ASSERT(acquired_reference);
}
lttng_trace_chunk_put(channel->trace_chunk);
if (chunk_id) {
trace_chunk = lttng_trace_chunk_registry_find_chunk(
- consumer_data.chunk_registry, session_id,
+ the_consumer_data.chunk_registry, session_id,
*chunk_id);
if (!trace_chunk) {
ERR("Failed to find trace chunk reference during creation of channel");
channel->output = CONSUMER_CHANNEL_MMAP;
break;
default:
- assert(0);
+ abort();
free(channel);
channel = NULL;
goto end;
int consumer_add_channel(struct lttng_consumer_channel *channel,
struct lttng_consumer_local_data *ctx)
{
- pthread_mutex_lock(&consumer_data.lock);
+ pthread_mutex_lock(&the_consumer_data.lock);
pthread_mutex_lock(&channel->lock);
pthread_mutex_lock(&channel->timer_lock);
steal_channel_key(channel->key);
rcu_read_lock();
- lttng_ht_add_unique_u64(consumer_data.channel_ht, &channel->node);
- lttng_ht_add_u64(consumer_data.channels_by_session_id_ht,
+ lttng_ht_add_unique_u64(the_consumer_data.channel_ht, &channel->node);
+ lttng_ht_add_u64(the_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);
- pthread_mutex_unlock(&consumer_data.lock);
+ pthread_mutex_unlock(&the_consumer_data.lock);
if (channel->wait_fd != -1 && channel->type == CONSUMER_CHANNEL_TYPE_DATA) {
notify_channel_pipe(ctx, channel, -1, CONSUMER_CHANNEL_ADD);
struct lttng_ht_iter iter;
struct lttng_consumer_stream *stream;
- assert(ctx);
- assert(ht);
- assert(pollfd);
- assert(local_stream);
+ LTTNG_ASSERT(ctx);
+ LTTNG_ASSERT(ht);
+ LTTNG_ASSERT(pollfd);
+ LTTNG_ASSERT(local_stream);
DBG("Updating poll fd array");
*nb_inactive_fd = 0;
rcu_read_lock();
- cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter, channel,
- node.node) {
+ cds_lfht_for_each_entry(the_consumer_data.channel_ht->ht, &iter.iter,
+ channel, node.node) {
consumer_del_channel(channel);
}
rcu_read_unlock();
- lttng_ht_destroy(consumer_data.channel_ht);
- lttng_ht_destroy(consumer_data.channels_by_session_id_ht);
+ lttng_ht_destroy(the_consumer_data.channel_ht);
+ lttng_ht_destroy(the_consumer_data.channels_by_session_id_ht);
cleanup_relayd_ht();
- lttng_ht_destroy(consumer_data.stream_per_chan_id_ht);
+ lttng_ht_destroy(the_consumer_data.stream_per_chan_id_ht);
/*
* This HT contains streams that are freed by either the metadata thread or
* the data thread so we do *nothing* on the hash table and simply destroy
* it.
*/
- lttng_ht_destroy(consumer_data.stream_list_ht);
+ lttng_ht_destroy(the_consumer_data.stream_list_ht);
/*
* Trace chunks in the registry may still exist if the session
* to hit.
*/
trace_chunks_left = lttng_trace_chunk_registry_put_each_chunk(
- consumer_data.chunk_registry);
+ the_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.",
}
/* Run all callbacks freeing each chunk. */
rcu_barrier();
- lttng_trace_chunk_registry_destroy(consumer_data.chunk_registry);
+ lttng_trace_chunk_registry_destroy(the_consumer_data.chunk_registry);
}
/*
int ret;
struct lttng_consumer_local_data *ctx;
- assert(consumer_data.type == LTTNG_CONSUMER_UNKNOWN ||
- consumer_data.type == type);
- consumer_data.type = type;
+ LTTNG_ASSERT(the_consumer_data.type == LTTNG_CONSUMER_UNKNOWN ||
+ the_consumer_data.type == type);
+ the_consumer_data.type = type;
ctx = zmalloc(sizeof(struct lttng_consumer_local_data));
if (ctx == NULL) {
/* RCU lock for the relayd pointer */
rcu_read_lock();
- assert(stream->net_seq_idx != (uint64_t) -1ULL ||
+ LTTNG_ASSERT(stream->net_seq_idx != (uint64_t) -1ULL ||
stream->trace_chunk);
/* Flag that the current stream if set for network streaming. */
int *splice_pipe;
unsigned int relayd_hang_up = 0;
- switch (consumer_data.type) {
+ switch (the_consumer_data.type) {
case LTTNG_CONSUMER_KERNEL:
break;
case LTTNG_CONSUMER32_UST:
return -ENOSYS;
default:
ERR("Unknown consumer_data type");
- assert(0);
+ abort();
}
/* RCU lock for the relayd pointer */
*/
int lttng_consumer_sample_snapshot_positions(struct lttng_consumer_stream *stream)
{
- switch (consumer_data.type) {
+ switch (the_consumer_data.type) {
case LTTNG_CONSUMER_KERNEL:
return lttng_kconsumer_sample_snapshot_positions(stream);
case LTTNG_CONSUMER32_UST:
return lttng_ustconsumer_sample_snapshot_positions(stream);
default:
ERR("Unknown consumer_data type");
- assert(0);
+ abort();
return -ENOSYS;
}
}
*/
int lttng_consumer_take_snapshot(struct lttng_consumer_stream *stream)
{
- switch (consumer_data.type) {
+ switch (the_consumer_data.type) {
case LTTNG_CONSUMER_KERNEL:
return lttng_kconsumer_take_snapshot(stream);
case LTTNG_CONSUMER32_UST:
return lttng_ustconsumer_take_snapshot(stream);
default:
ERR("Unknown consumer_data type");
- assert(0);
+ abort();
return -ENOSYS;
}
}
int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
unsigned long *pos)
{
- switch (consumer_data.type) {
+ switch (the_consumer_data.type) {
case LTTNG_CONSUMER_KERNEL:
return lttng_kconsumer_get_produced_snapshot(stream, pos);
case LTTNG_CONSUMER32_UST:
return lttng_ustconsumer_get_produced_snapshot(stream, pos);
default:
ERR("Unknown consumer_data type");
- assert(0);
+ abort();
return -ENOSYS;
}
}
int lttng_consumer_get_consumed_snapshot(struct lttng_consumer_stream *stream,
unsigned long *pos)
{
- switch (consumer_data.type) {
+ switch (the_consumer_data.type) {
case LTTNG_CONSUMER_KERNEL:
return lttng_kconsumer_get_consumed_snapshot(stream, pos);
case LTTNG_CONSUMER32_UST:
return lttng_ustconsumer_get_consumed_snapshot(stream, pos);
default:
ERR("Unknown consumer_data type");
- assert(0);
+ abort();
return -ENOSYS;
}
}
int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
int sock, struct pollfd *consumer_sockpoll)
{
- switch (consumer_data.type) {
+ switch (the_consumer_data.type) {
case LTTNG_CONSUMER_KERNEL:
return lttng_kconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
case LTTNG_CONSUMER32_UST:
return lttng_ustconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
default:
ERR("Unknown consumer_data type");
- assert(0);
+ abort();
return -ENOSYS;
}
}
static
void lttng_consumer_close_all_metadata(void)
{
- switch (consumer_data.type) {
+ switch (the_consumer_data.type) {
case LTTNG_CONSUMER_KERNEL:
/*
* The Kernel consumer has a different metadata scheme so we don't
break;
default:
ERR("Unknown consumer_data type");
- assert(0);
+ abort();
}
}
struct lttng_consumer_channel *channel = NULL;
bool free_channel = false;
- assert(stream);
+ LTTNG_ASSERT(stream);
/*
* This call should NEVER receive regular stream. It must always be
* metadata stream and this is crucial for data structure synchronization.
*/
- assert(stream->metadata_flag);
+ LTTNG_ASSERT(stream->metadata_flag);
DBG3("Consumer delete metadata stream %d", stream->wait_fd);
- pthread_mutex_lock(&consumer_data.lock);
+ pthread_mutex_lock(&the_consumer_data.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
}
pthread_mutex_unlock(&stream->lock);
pthread_mutex_unlock(&channel->lock);
- pthread_mutex_unlock(&consumer_data.lock);
+ pthread_mutex_unlock(&the_consumer_data.lock);
if (free_channel) {
consumer_del_channel(channel);
struct lttng_ht_iter iter;
struct lttng_ht_node_u64 *node;
- assert(stream);
- assert(ht);
+ LTTNG_ASSERT(stream);
+ LTTNG_ASSERT(ht);
DBG3("Adding metadata stream %" PRIu64 " to hash table", stream->key);
- pthread_mutex_lock(&consumer_data.lock);
+ pthread_mutex_lock(&the_consumer_data.lock);
pthread_mutex_lock(&stream->chan->lock);
pthread_mutex_lock(&stream->chan->timer_lock);
pthread_mutex_lock(&stream->lock);
*/
lttng_ht_lookup(ht, &stream->key, &iter);
node = lttng_ht_iter_get_node_u64(&iter);
- assert(!node);
+ LTTNG_ASSERT(!node);
/*
* When nb_init_stream_left reaches 0, we don't need to trigger any action
lttng_ht_add_unique_u64(ht, &stream->node);
- lttng_ht_add_u64(consumer_data.stream_per_chan_id_ht,
- &stream->node_channel_id);
+ lttng_ht_add_u64(the_consumer_data.stream_per_chan_id_ht,
+ &stream->node_channel_id);
/*
* Add stream to the stream_list_ht of the consumer data. No need to steal
* the key since the HT does not use it and we allow to add redundant keys
* into this table.
*/
- lttng_ht_add_u64(consumer_data.stream_list_ht, &stream->node_session_id);
+ lttng_ht_add_u64(the_consumer_data.stream_list_ht,
+ &stream->node_session_id);
rcu_read_unlock();
pthread_mutex_unlock(&stream->lock);
pthread_mutex_unlock(&stream->chan->lock);
pthread_mutex_unlock(&stream->chan->timer_lock);
- pthread_mutex_unlock(&consumer_data.lock);
+ pthread_mutex_unlock(&the_consumer_data.lock);
}
/*
DBG("Consumer delete flagged metadata stream");
- assert(pollset);
+ LTTNG_ASSERT(pollset);
rcu_read_lock();
cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream, node.node) {
lttng_ht_lookup(metadata_ht, &tmp_id, &iter);
}
node = lttng_ht_iter_get_node_u64(&iter);
- assert(node);
+ LTTNG_ASSERT(node);
stream = caa_container_of(node, struct lttng_consumer_stream,
node);
if (revents & (LPOLLIN | LPOLLPRI)) {
/* Get the data out of the metadata file descriptor */
DBG("Metadata available on fd %d", pollfd);
- assert(stream->wait_fd == pollfd);
+ LTTNG_ASSERT(stream->wait_fd == pollfd);
do {
health_code_update();
}
} else if (revents & (LPOLLERR | LPOLLHUP)) {
DBG("Metadata fd %d is hup|err.", pollfd);
- if (!stream->hangup_flush_done
- && (consumer_data.type == LTTNG_CONSUMER32_UST
- || consumer_data.type == LTTNG_CONSUMER64_UST)) {
+ if (!stream->hangup_flush_done &&
+ (the_consumer_data.type == LTTNG_CONSUMER32_UST ||
+ the_consumer_data.type ==
+ LTTNG_CONSUMER64_UST)) {
DBG("Attempting to flush and consume the UST buffers");
lttng_ustconsumer_on_stream_hangup(stream);
* the fds set has been updated, we need to update our
* local array as well
*/
- pthread_mutex_lock(&consumer_data.lock);
- if (consumer_data.need_update) {
+ pthread_mutex_lock(&the_consumer_data.lock);
+ if (the_consumer_data.need_update) {
free(pollfd);
pollfd = NULL;
local_stream = NULL;
/* Allocate for all fds */
- pollfd = zmalloc((consumer_data.stream_count + nb_pipes_fd) * sizeof(struct pollfd));
+ pollfd = zmalloc((the_consumer_data.stream_count +
+ nb_pipes_fd) *
+ sizeof(struct pollfd));
if (pollfd == NULL) {
PERROR("pollfd malloc");
- pthread_mutex_unlock(&consumer_data.lock);
+ pthread_mutex_unlock(&the_consumer_data.lock);
goto end;
}
- local_stream = zmalloc((consumer_data.stream_count + nb_pipes_fd) *
+ local_stream = zmalloc((the_consumer_data.stream_count +
+ nb_pipes_fd) *
sizeof(struct lttng_consumer_stream *));
if (local_stream == NULL) {
PERROR("local_stream malloc");
- pthread_mutex_unlock(&consumer_data.lock);
+ pthread_mutex_unlock(&the_consumer_data.lock);
goto end;
}
ret = update_poll_array(ctx, &pollfd, local_stream,
if (ret < 0) {
ERR("Error in allocating pollfd or local_outfds");
lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
- pthread_mutex_unlock(&consumer_data.lock);
+ pthread_mutex_unlock(&the_consumer_data.lock);
goto end;
}
nb_fd = ret;
- consumer_data.need_update = 0;
+ the_consumer_data.need_update = 0;
}
- pthread_mutex_unlock(&consumer_data.lock);
+ pthread_mutex_unlock(&the_consumer_data.lock);
/* No FDs and consumer_quit, consumer_cleanup the thread */
if (nb_fd == 0 && nb_inactive_fd == 0 &&
}
if (!local_stream[i]->hangup_flush_done
&& (pollfd[i].revents & (POLLHUP | POLLERR | POLLNVAL))
- && (consumer_data.type == LTTNG_CONSUMER32_UST
- || consumer_data.type == LTTNG_CONSUMER64_UST)) {
+ && (the_consumer_data.type == LTTNG_CONSUMER32_UST
+ || the_consumer_data.type == LTTNG_CONSUMER64_UST)) {
DBG("fd %d is hup|err|nval. Attempting flush and read.",
pollfd[i].fd);
lttng_ustconsumer_on_stream_hangup(local_stream[i]);
struct lttng_consumer_stream *stream;
struct lttng_ht_iter iter;
- ht = consumer_data.stream_per_chan_id_ht;
+ ht = the_consumer_data.stream_per_chan_id_ht;
rcu_read_lock();
cds_lfht_for_each_entry_duplicate(ht->ht,
if (cds_lfht_is_node_deleted(&stream->node.node)) {
goto next;
}
- switch (consumer_data.type) {
+ switch (the_consumer_data.type) {
case LTTNG_CONSUMER_KERNEL:
break;
case LTTNG_CONSUMER32_UST:
break;
default:
ERR("Unknown consumer_data type");
- assert(0);
+ abort();
}
next:
pthread_mutex_unlock(&stream->lock);
rcu_read_lock();
cds_lfht_for_each_entry(ht->ht, &iter.iter, channel, wait_fd_node.node) {
ret = lttng_ht_del(ht, &iter);
- assert(ret != 0);
+ LTTNG_ASSERT(ret != 0);
}
rcu_read_unlock();
lttng_poll_del(&events, chan->wait_fd);
iter.iter.node = &chan->wait_fd_node.node;
ret = lttng_ht_del(channel_ht, &iter);
- assert(ret == 0);
+ LTTNG_ASSERT(ret == 0);
- switch (consumer_data.type) {
+ switch (the_consumer_data.type) {
case LTTNG_CONSUMER_KERNEL:
break;
case LTTNG_CONSUMER32_UST:
break;
default:
ERR("Unknown consumer_data type");
- assert(0);
+ abort();
}
/*
lttng_ht_lookup(channel_ht, &tmp_id, &iter);
}
node = lttng_ht_iter_get_node_u64(&iter);
- assert(node);
+ LTTNG_ASSERT(node);
chan = caa_container_of(node, struct lttng_consumer_channel,
wait_fd_node);
lttng_poll_del(&events, chan->wait_fd);
ret = lttng_ht_del(channel_ht, &iter);
- assert(ret == 0);
+ LTTNG_ASSERT(ret == 0);
/*
* This will close the wait fd for each stream associated to
{
int ret;
- assert(ctx);
- assert(sockpoll);
+ LTTNG_ASSERT(ctx);
+ LTTNG_ASSERT(sockpoll);
ret = lttng_consumer_poll_socket(sockpoll);
if (ret) {
err = 0; /* All is OK */
goto end;
}
- DBG("received command on sock");
+ DBG("Received command on sock");
}
/* All is OK */
err = 0;
return NULL;
}
+static int post_consume(struct lttng_consumer_stream *stream,
+ const struct stream_subbuffer *subbuffer,
+ struct lttng_consumer_local_data *ctx)
+{
+ size_t i;
+ int ret = 0;
+ const size_t count = lttng_dynamic_array_get_count(
+ &stream->read_subbuffer_ops.post_consume_cbs);
+
+ 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);
+
+ ret = op(stream, subbuffer, ctx);
+ if (ret) {
+ goto end;
+ }
+ }
+end:
+ return ret;
+}
+
ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
struct lttng_consumer_local_data *ctx,
bool locked_by_caller)
{
- ssize_t ret, written_bytes;
+ ssize_t ret, written_bytes = 0;
int rotation_ret;
struct stream_subbuffer subbuffer = {};
+ enum get_next_subbuffer_status get_next_status;
if (!locked_by_caller) {
stream->read_subbuffer_ops.lock(stream);
}
}
- ret = stream->read_subbuffer_ops.get_next_subbuffer(stream, &subbuffer);
- if (ret) {
- if (ret == -ENODATA) {
- /* Not an error. */
- ret = 0;
- }
+ get_next_status = stream->read_subbuffer_ops.get_next_subbuffer(
+ stream, &subbuffer);
+ switch (get_next_status) {
+ case GET_NEXT_SUBBUFFER_STATUS_OK:
+ break;
+ case GET_NEXT_SUBBUFFER_STATUS_NO_DATA:
+ /* Not an error. */
+ ret = 0;
+ goto sleep_stream;
+ case GET_NEXT_SUBBUFFER_STATUS_ERROR:
+ ret = -1;
goto end;
+ default:
+ abort();
}
ret = stream->read_subbuffer_ops.pre_consume_subbuffer(
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);
+ if (written_bytes <= 0) {
+ ERR("Error consuming subbuffer: (%zd)", written_bytes);
+ ret = (int) written_bytes;
+ goto error_put_subbuf;
}
ret = stream->read_subbuffer_ops.put_next_subbuffer(stream, &subbuffer);
goto end;
}
- if (stream->read_subbuffer_ops.post_consume) {
- ret = stream->read_subbuffer_ops.post_consume(stream, &subbuffer, ctx);
- if (ret) {
- goto end;
- }
+ ret = post_consume(stream, &subbuffer, ctx);
+ if (ret) {
+ goto end;
}
/*
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);
}
int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
{
- switch (consumer_data.type) {
+ switch (the_consumer_data.type) {
case LTTNG_CONSUMER_KERNEL:
return lttng_kconsumer_on_recv_stream(stream);
case LTTNG_CONSUMER32_UST:
return lttng_ustconsumer_on_recv_stream(stream);
default:
ERR("Unknown consumer_data type");
- assert(0);
+ abort();
return -ENOSYS;
}
}
*/
int lttng_consumer_init(void)
{
- consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
- if (!consumer_data.channel_ht) {
+ the_consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+ if (!the_consumer_data.channel_ht) {
goto error;
}
- consumer_data.channels_by_session_id_ht =
+ the_consumer_data.channels_by_session_id_ht =
lttng_ht_new(0, LTTNG_HT_TYPE_U64);
- if (!consumer_data.channels_by_session_id_ht) {
+ if (!the_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) {
+ the_consumer_data.relayd_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+ if (!the_consumer_data.relayd_ht) {
goto error;
}
- consumer_data.stream_list_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
- if (!consumer_data.stream_list_ht) {
+ the_consumer_data.stream_list_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+ if (!the_consumer_data.stream_list_ht) {
goto error;
}
- consumer_data.stream_per_chan_id_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
- if (!consumer_data.stream_per_chan_id_ht) {
+ the_consumer_data.stream_per_chan_id_ht =
+ lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+ if (!the_consumer_data.stream_per_chan_id_ht) {
goto error;
}
goto error;
}
- consumer_data.chunk_registry = lttng_trace_chunk_registry_create();
- if (!consumer_data.chunk_registry) {
+ the_consumer_data.chunk_registry = lttng_trace_chunk_registry_create();
+ if (!the_consumer_data.chunk_registry) {
goto error;
}
enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
struct consumer_relayd_sock_pair *relayd = NULL;
- assert(ctx);
- assert(relayd_sock);
+ LTTNG_ASSERT(ctx);
+ LTTNG_ASSERT(relayd_sock);
DBG("Consumer adding relayd socket (idx: %" PRIu64 ")", net_seq_idx);
/* Get relayd reference if exists. */
relayd = consumer_find_relayd(net_seq_idx);
if (relayd == NULL) {
- assert(sock_type == LTTNG_STREAM_CONTROL);
+ LTTNG_ASSERT(sock_type == LTTNG_STREAM_CONTROL);
/* Not found. Allocate one. */
relayd = consumer_allocate_relayd_sock_pair(net_seq_idx);
if (relayd == NULL) {
/*
* relayd key should never be found for control socket.
*/
- assert(sock_type != LTTNG_STREAM_CONTROL);
+ LTTNG_ASSERT(sock_type != LTTNG_STREAM_CONTROL);
}
/* First send a status message before receiving the fds. */
struct consumer_relayd_sock_pair *relayd = NULL;
/* Iterate over all relayd since they are indexed by net_seq_idx. */
- cds_lfht_for_each_entry(consumer_data.relayd_ht->ht, &iter.iter, relayd,
- node.node) {
+ cds_lfht_for_each_entry(the_consumer_data.relayd_ht->ht, &iter.iter,
+ relayd, node.node) {
/*
* Check by sessiond id which is unique here where the relayd session
* id might not be when having multiple relayd.
DBG("Consumer data pending command on session id %" PRIu64, id);
rcu_read_lock();
- pthread_mutex_lock(&consumer_data.lock);
+ pthread_mutex_lock(&the_consumer_data.lock);
- switch (consumer_data.type) {
+ switch (the_consumer_data.type) {
case LTTNG_CONSUMER_KERNEL:
data_pending = lttng_kconsumer_data_pending;
break;
break;
default:
ERR("Unknown consumer data type");
- assert(0);
+ abort();
}
/* Ease our life a bit */
- ht = consumer_data.stream_list_ht;
+ ht = the_consumer_data.stream_list_ht;
cds_lfht_for_each_entry_duplicate(ht->ht,
ht->hash_fct(&id, lttng_ht_seed),
data_not_pending:
/* Data is available to be read by a viewer. */
- pthread_mutex_unlock(&consumer_data.lock);
+ pthread_mutex_unlock(&the_consumer_data.lock);
rcu_read_unlock();
return 0;
data_pending:
/* Data is still being extracted from buffers. */
- pthread_mutex_unlock(&consumer_data.lock);
+ pthread_mutex_unlock(&the_consumer_data.lock);
rcu_read_unlock();
return 1;
}
{
struct lttcomm_consumer_status_channel msg;
- assert(sock >= 0);
+ LTTNG_ASSERT(sock >= 0);
memset(&msg, 0, sizeof(msg));
if (!channel) {
return start_pos;
}
-static
-int consumer_flush_buffer(struct lttng_consumer_stream *stream, int producer_active)
+/* Stream lock must be held by the caller. */
+static int sample_stream_positions(struct lttng_consumer_stream *stream,
+ unsigned long *produced, unsigned long *consumed)
{
- int ret = 0;
+ int ret;
- switch (consumer_data.type) {
- case LTTNG_CONSUMER_KERNEL:
- 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_ustconsumer_flush_buffer(stream, producer_active);
- break;
- default:
- ERR("Unknown consumer_data type");
- abort();
+ ASSERT_LOCKED(stream->lock);
+
+ ret = lttng_consumer_sample_snapshot_positions(stream);
+ if (ret < 0) {
+ ERR("Failed to sample snapshot positions");
+ goto end;
+ }
+
+ ret = lttng_consumer_get_produced_snapshot(stream, produced);
+ if (ret < 0) {
+ ERR("Failed to sample produced position");
+ goto end;
+ }
+
+ ret = lttng_consumer_get_consumed_snapshot(stream, consumed);
+ if (ret < 0) {
+ ERR("Failed to sample consumed position");
+ goto end;
}
end:
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_ht *ht = the_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);
- assert(channel->trace_chunk);
+ LTTNG_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) {
}
/*
- * Do not flush an empty packet when rotating from a NULL trace
+ * Do not flush a 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.
+ * rotation which rotated to NULL performed that side-effect
+ * already. No new data can be produced when a stream has no
+ * associated trace chunk (e.g. a stop followed by a rotate).
*/
if (stream->trace_chunk) {
+ bool flush_active;
+
+ if (stream->metadata_flag) {
+ /*
+ * Don't produce an empty metadata packet,
+ * simply close the current one.
+ *
+ * Metadata is regenerated on every trace chunk
+ * switch; there is no concern that no data was
+ * produced.
+ */
+ flush_active = true;
+ } else {
+ /*
+ * Only flush an empty packet if the "packet
+ * open" could not be performed on transition
+ * to a new trace chunk and no packets were
+ * consumed within the chunk's lifetime.
+ */
+ if (stream->opened_packet_in_current_trace_chunk) {
+ flush_active = true;
+ } else {
+ /*
+ * Stream could have been full at the
+ * time of rotation, but then have had
+ * no activity at all.
+ *
+ * It is important to flush a packet
+ * to prevent 0-length files from being
+ * produced as most viewers choke on
+ * them.
+ *
+ * Unfortunately viewers will not be
+ * able to know that tracing was active
+ * for this stream during this trace
+ * chunk's lifetime.
+ */
+ ret = sample_stream_positions(stream, &produced_pos, &consumed_pos);
+ if (ret) {
+ goto end_unlock_stream;
+ }
+
+ /*
+ * Don't flush an empty packet if data
+ * was produced; it will be consumed
+ * before the rotation completes.
+ */
+ flush_active = produced_pos != consumed_pos;
+ if (!flush_active) {
+ const char *trace_chunk_name;
+ uint64_t trace_chunk_id;
+
+ chunk_status = lttng_trace_chunk_get_name(
+ stream->trace_chunk,
+ &trace_chunk_name,
+ NULL);
+ if (chunk_status == LTTNG_TRACE_CHUNK_STATUS_NONE) {
+ trace_chunk_name = "none";
+ }
+
+ /*
+ * Consumer trace chunks are
+ * never anonymous.
+ */
+ chunk_status = lttng_trace_chunk_get_id(
+ stream->trace_chunk,
+ &trace_chunk_id);
+ LTTNG_ASSERT(chunk_status ==
+ LTTNG_TRACE_CHUNK_STATUS_OK);
+
+ DBG("Unable to open packet for stream during trace chunk's lifetime. "
+ "Flushing an empty packet to prevent an empty file from being created: "
+ "stream id = %" PRIu64 ", trace chunk name = `%s`, trace chunk id = %" PRIu64,
+ stream->key, trace_chunk_name, trace_chunk_id);
+ }
+ }
+ }
+
/*
- * 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.
+ * Close the current packet before sampling the
+ * ring buffer positions.
*/
- ret = consumer_flush_buffer(stream, stream->metadata_flag ? 1 : 0);
+ ret = consumer_stream_flush_buffer(stream, flush_active);
if (ret < 0) {
ERR("Failed to flush stream %" PRIu64 " during channel rotation",
stream->key);
}
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;
- pthread_mutex_unlock(&channel->lock);
- if (is_local_trace) {
- ret = 0;
- goto end;
- }
+ 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;
+ }
- relayd = consumer_find_relayd(relayd_id);
- if (!relayd) {
- ERR("Failed to find relayd %" PRIu64, relayd_id);
- ret = -1;
- goto end;
+ 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("Relayd rotate stream failed. Cleaning up relayd %" PRIu64,
+ relayd->net_seq_idx);
+ lttng_consumer_cleanup_relayd(relayd);
+ goto end_unlock_channel;
+ }
}
- 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("Relayd rotate stream failed. Cleaning up relayd %" PRIu64,
- relayd->net_seq_idx);
- lttng_consumer_cleanup_relayd(relayd);
- goto end;
+ 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);
ret = 0;
goto end;
end:
rcu_read_unlock();
lttng_dynamic_array_reset(&stream_rotation_positions);
+ lttng_dynamic_pointer_array_reset(&streams_packet_to_open);
return ret;
}
goto end;
}
- switch (consumer_data.type) {
+ switch (the_consumer_data.type) {
case LTTNG_CONSUMER_KERNEL:
ret = kernctl_buffer_clear(stream->wait_fd);
if (ret < 0) {
break;
case LTTNG_CONSUMER32_UST:
case LTTNG_CONSUMER64_UST:
- lttng_ustconsumer_clear_buffer(stream);
+ ret = lttng_ustconsumer_clear_buffer(stream);
+ if (ret < 0) {
+ ERR("Failed to clear ust stream (ret = %d)", ret);
+ goto end;
+ }
break;
default:
ERR("Unknown consumer_data type");
{
int ret;
- ret = consumer_flush_buffer(stream, 1);
+ ret = consumer_stream_flush_buffer(stream, 1);
if (ret < 0) {
ERR("Failed to flush stream %" PRIu64 " during channel clear",
stream->key);
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_ht *ht = the_consumer_data.stream_per_chan_id_ht;
rcu_read_lock();
} else {
relayd_id_str = "(formatting error)";
}
- }
+ }
- /* Local protocol error. */
- assert(chunk_creation_timestamp);
+ /* Local protocol error. */
+ LTTNG_ASSERT(chunk_creation_timestamp);
ret = time_to_iso8601_str(chunk_creation_timestamp,
creation_timestamp_buffer,
sizeof(creation_timestamp_buffer));
}
published_chunk = lttng_trace_chunk_registry_publish_chunk(
- consumer_data.chunk_registry, session_id,
+ the_consumer_data.chunk_registry, session_id,
created_chunk);
lttng_trace_chunk_put(created_chunk);
created_chunk = NULL;
}
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(
+ cds_lfht_for_each_entry_duplicate(
+ the_consumer_data.channels_by_session_id_ht->ht,
+ the_consumer_data.channels_by_session_id_ht->hash_fct(
&session_id, lttng_ht_seed),
- consumer_data.channels_by_session_id_ht->match_fct,
+ the_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,
close_command_name);
chunk = lttng_trace_chunk_registry_find_chunk(
- consumer_data.chunk_registry, session_id, chunk_id);
+ the_consumer_data.chunk_registry, session_id, chunk_id);
if (!chunk) {
ERR("Failed to find chunk: session_id = %" PRIu64
", chunk_id = %" PRIu64,
* current chunk found in the session's channels.
*/
rcu_read_lock();
- cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter,
+ cds_lfht_for_each_entry(the_consumer_data.channel_ht->ht, &iter.iter,
channel, node.node) {
int ret;
bool chunk_exists_local, chunk_exists_remote;
if (relayd_id) {
- int ret;
-
/* Only used for logging purposes. */
ret = snprintf(relayd_id_buffer, sizeof(relayd_id_buffer),
"%" PRIu64, *relayd_id);
} 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);
+ the_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");
struct lttng_ht_iter iter;
int ret;
- ht = consumer_data.stream_per_chan_id_ht;
+ ht = the_consumer_data.stream_per_chan_id_ht;
rcu_read_lock();
cds_lfht_for_each_entry_duplicate(ht->ht,
end:
return ret;
}
+
+enum lttcomm_return_code lttng_consumer_open_channel_packets(
+ struct lttng_consumer_channel *channel)
+{
+ 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;
+}
+
+void lttng_consumer_sigbus_handle(void *addr)
+{
+ lttng_ustconsumer_sigbus_handle(addr);
+}