/*
- * Copyright (C) 2011 Julien Desfossez <julien.desfossez@polymtl.ca>
+ * Copyright (C) 2011 EfficiOS Inc.
* Copyright (C) 2011 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
* Copyright (C) 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
*
#include <stdbool.h>
#include <stdint.h>
-#include <bin/lttng-consumerd/health-consumerd.h>
-#include <common/common.h>
-#include <common/sessiond-comm/sessiond-comm.h>
-#include <common/relayd/relayd.h>
-#include <common/compat/fcntl.h>
-#include <common/compat/endian.h>
-#include <common/consumer/consumer-metadata-cache.h>
-#include <common/consumer/consumer-stream.h>
-#include <common/consumer/consumer-timer.h>
-#include <common/utils.h>
-#include <common/index/index.h>
-#include <common/consumer/consumer.h>
-#include <common/shm.h>
-#include <common/optional.h>
-
-#include "ust-consumer.h"
+#include <bin/lttng-consumerd/health-consumerd.hpp>
+#include <common/common.hpp>
+#include <common/sessiond-comm/sessiond-comm.hpp>
+#include <common/relayd/relayd.hpp>
+#include <common/compat/fcntl.hpp>
+#include <common/compat/endian.hpp>
+#include <common/consumer/consumer-metadata-cache.hpp>
+#include <common/consumer/consumer-stream.hpp>
+#include <common/consumer/consumer-timer.hpp>
+#include <common/utils.hpp>
+#include <common/index/index.hpp>
+#include <common/consumer/consumer.hpp>
+#include <common/shm.hpp>
+#include <common/optional.hpp>
+
+#include "ust-consumer.hpp"
#define INT_MAX_STR_LEN 12 /* includes \0 */
LTTNG_EXPORT DEFINE_LTTNG_UST_SIGBUS_STATE();
-/*
- * Free channel object and all streams associated with it. This MUST be used
- * only and only if the channel has _NEVER_ been added to the global channel
- * hash table.
- */
-static void destroy_channel(struct lttng_consumer_channel *channel)
-{
- struct lttng_consumer_stream *stream, *stmp;
-
- LTTNG_ASSERT(channel);
-
- DBG("UST consumer cleaning stream list");
-
- cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
- send_node) {
-
- health_code_update();
-
- cds_list_del(&stream->send_node);
- lttng_ust_ctl_destroy_stream(stream->ustream);
- lttng_trace_chunk_put(stream->trace_chunk);
- free(stream);
- }
-
- /*
- * If a channel is available meaning that was created before the streams
- * were, delete it.
- */
- if (channel->uchan) {
- lttng_ustconsumer_del_channel(channel);
- lttng_ustconsumer_free_channel(channel);
- }
-
- if (channel->trace_chunk) {
- lttng_trace_chunk_put(channel->trace_chunk);
- }
-
- free(channel);
-}
-
/*
* Add channel to internal consumer state.
*
* global.
*/
stream->globally_visible = 1;
- cds_list_del(&stream->send_node);
+ cds_list_del_init(&stream->send_node);
ret = lttng_pipe_write(stream_pipe, &stream, sizeof(stream));
if (ret < 0) {
nr_stream_fds = 1;
else
nr_stream_fds = lttng_ust_ctl_get_nr_stream_per_channel();
- stream_fds = (int *) zmalloc(nr_stream_fds * sizeof(*stream_fds));
+ stream_fds = calloc<int>(nr_stream_fds);
if (!stream_fds) {
ret = -1;
goto error_alloc;
next:
pthread_mutex_unlock(&stream->lock);
}
+
+ /*
+ * Send one last buffer statistics update to the session daemon. This
+ * ensures that the session daemon gets at least one statistics update
+ * per channel even in the case of short-lived channels, such as when a
+ * short-lived app is traced in per-pid mode.
+ */
+ sample_and_send_channel_buffer_stats(channel);
error:
rcu_read_unlock();
return ret;
int ret;
struct lttng_consumer_channel *metadata;
+ ASSERT_RCU_READ_LOCKED();
+
DBG("UST consumer setup metadata key %" PRIu64, key);
metadata = consumer_find_channel(key);
* will make sure to clean that list.
*/
consumer_stream_destroy(metadata->metadata_stream, NULL);
- cds_list_del(&metadata->metadata_stream->send_node);
metadata->metadata_stream = NULL;
send_streams_error:
error_no_stream:
LTTNG_ASSERT(path);
LTTNG_ASSERT(ctx);
+ ASSERT_RCU_READ_LOCKED();
DBG("UST consumer snapshot metadata with key %" PRIu64 " at path %s",
key, path);
* new metadata stream.
*/
consumer_stream_destroy(metadata_stream, NULL);
- cds_list_del(&metadata_stream->send_node);
metadata_channel->metadata_stream = NULL;
error:
LTTNG_ASSERT(path);
LTTNG_ASSERT(ctx);
+ ASSERT_RCU_READ_LOCKED();
rcu_read_lock();
if (use_relayd) {
ret = consumer_send_relayd_stream(stream, path);
if (ret < 0) {
- goto error_unlock;
+ goto error_close_stream;
}
} else {
ret = consumer_stream_create_output_files(stream,
false);
if (ret < 0) {
- goto error_unlock;
+ goto error_close_stream;
}
DBG("UST consumer snapshot stream (%" PRIu64 ")",
stream->key);
ret = lttng_ustconsumer_take_snapshot(stream);
if (ret < 0) {
ERR("Taking UST snapshot");
- goto error_unlock;
+ goto error_close_stream;
}
ret = lttng_ustconsumer_get_produced_snapshot(stream, &produced_pos);
if (ret < 0) {
ERR("Produced UST snapshot position");
- goto error_unlock;
+ goto error_close_stream;
}
ret = lttng_ustconsumer_get_consumed_snapshot(stream, &consumed_pos);
if (ret < 0) {
ERR("Consumerd UST snapshot position");
- goto error_unlock;
+ goto error_close_stream;
}
/*
DBG("UST consumer push metadata key %" PRIu64 " of len %" PRIu64, key, len);
- metadata_str = (char *) zmalloc(len * sizeof(char));
+ metadata_str = calloc<char>(len);
if (!metadata_str) {
PERROR("zmalloc metadata string");
ret_code = LTTCOMM_CONSUMERD_ENOMEM;
switch (msg.cmd_type) {
case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
{
+ uint32_t major = msg.u.relayd_sock.major;
+ uint32_t minor = msg.u.relayd_sock.minor;
+ enum lttcomm_sock_proto protocol =
+ (enum lttcomm_sock_proto) msg.u.relayd_sock
+ .relayd_socket_protocol;
+
/* Session daemon status message are handled in the following call. */
consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
- msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
- &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id,
- msg.u.relayd_sock.relayd_session_id);
+ msg.u.relayd_sock.type, ctx, sock,
+ consumer_sockpoll, msg.u.relayd_sock.session_id,
+ msg.u.relayd_sock.relayd_session_id, major,
+ minor, protocol);
goto end_nosignal;
}
case LTTNG_CONSUMER_DESTROY_RELAYD:
*/
rotate_channel = lttng_consumer_rotate_channel(
found_channel, key,
- msg.u.rotate_channel.relayd_id,
- msg.u.rotate_channel.metadata, ctx);
+ msg.u.rotate_channel.relayd_id);
if (rotate_channel < 0) {
ERR("Rotate channel failed");
ret_code = LTTCOMM_CONSUMERD_ROTATION_FAIL;
ret_rotate_read_streams =
lttng_consumer_rotate_ready_streams(
- found_channel, key,
- ctx);
+ found_channel, key);
if (ret_rotate_read_streams < 0) {
ERR("Rotate channel failed");
}
case LTTNG_CONSUMER_INIT:
{
int ret_send_status;
+ lttng_uuid sessiond_uuid;
- ret_code = lttng_consumer_init_command(ctx,
- msg.u.init.sessiond_uuid);
+ std::copy(std::begin(msg.u.init.sessiond_uuid), std::end(msg.u.init.sessiond_uuid),
+ sessiond_uuid.begin());
+ ret_code = lttng_consumer_init_command(ctx, sessiond_uuid);
health_code_update();
ret_send_status = consumer_send_status_msg(sock, ret_code);
if (ret_send_status < 0) {
end_channel_error:
if (channel) {
- /*
- * Free channel here since no one has a reference to it. We don't
- * free after that because a stream can store this pointer.
- */
- destroy_channel(channel);
+ consumer_del_channel(channel);
}
/* We have to send a status channel message indicating an error. */
{
stream->quiescent = true;
}
}
- pthread_mutex_unlock(&stream->lock);
+
stream->hangup_flush_done = 1;
+ pthread_mutex_unlock(&stream->lock);
}
void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan)
LTTNG_ASSERT(ctx);
LTTNG_ASSERT(metadata_stream);
+ ASSERT_RCU_READ_LOCKED();
metadata_channel = metadata_stream->chan;
pthread_mutex_unlock(&metadata_stream->lock);
}
static int put_next_subbuffer(struct lttng_consumer_stream *stream,
- struct stream_subbuffer *subbuffer)
+ struct stream_subbuffer *subbuffer __attribute__((unused)))
{
const int ret = lttng_ust_ctl_put_next_subbuf(stream->ustream);
}
static int signal_metadata(struct lttng_consumer_stream *stream,
- struct lttng_consumer_local_data *ctx)
+ struct lttng_consumer_local_data *ctx __attribute__((unused)))
{
ASSERT_LOCKED(stream->metadata_rdv_lock);
return pthread_cond_broadcast(&stream->metadata_rdv) ? -errno : 0;
projects / lttng-tools.git / blobdiff