#include <common/index/index.h>
#include "ust-consumer.h"
+#include "../../bin/lttng-consumerd/health-consumerd.h"
extern struct lttng_consumer_global_data consumer_data;
extern int consumer_poll_timeout;
cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
send_node) {
+
+ health_code_update();
+
cds_list_del(&stream->send_node);
ustctl_destroy_stream(stream->ustream);
free(stream);
int wait_fd;
int ust_metadata_pipe[2];
+ health_code_update();
+
if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA && channel->monitor) {
ret = utils_create_pipe_cloexec_nonblock(ust_metadata_pipe);
if (ret < 0) {
if (channel->relayd_id != (uint64_t) -1ULL) {
cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
+
+ health_code_update();
+
/* Try to send the stream to the relayd if one is available. */
ret = consumer_send_relayd_stream(stream, stream->chan->pathname);
if (ret < 0) {
/* The channel was sent successfully to the sessiond at this point. */
cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
+
+ health_code_update();
+
/* Send stream to session daemon. */
ret = send_sessiond_stream(sock, stream);
if (ret < 0) {
/* Send streams to the corresponding thread. */
cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
send_node) {
+
+ health_code_update();
+
/* Sending the stream to the thread. */
ret = send_stream_to_thread(stream, ctx);
if (ret < 0) {
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) {
+
+ health_code_update();
+
ustctl_flush_buffer(stream->ustream, 1);
}
error:
}
assert(!metadata_channel->monitor);
+ health_code_update();
+
/*
* Ask the sessiond if we have new metadata waiting and update the
* consumer metadata cache.
goto error;
}
+ health_code_update();
+
/*
* The metadata stream is NOT created in no monitor mode when the channel
* is created on a sessiond ask channel command.
}
do {
+ health_code_update();
+
ret = lttng_consumer_read_subbuffer(metadata_stream, ctx);
if (ret < 0) {
goto error_stream;
DBG("UST consumer snapshot channel %" PRIu64, key);
cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
+
+ health_code_update();
+
/* Lock stream because we are about to change its state. */
pthread_mutex_lock(&stream->lock);
stream->net_seq_idx = relayd_id;
ssize_t read_len;
unsigned long len, padded_len;
+ health_code_update();
+
DBG("UST consumer taking snapshot at pos %lu", consumed_pos);
ret = ustctl_get_subbuf(stream->ustream, &consumed_pos);
goto end;
}
+ health_code_update();
+
/* Receive metadata string. */
ret = lttcomm_recv_unix_sock(sock, metadata_str, len);
if (ret < 0) {
goto end_free;
}
+ health_code_update();
+
pthread_mutex_lock(&channel->metadata_cache->lock);
ret = consumer_metadata_cache_write(channel, offset, len, metadata_str);
if (ret < 0) {
}
while (consumer_metadata_cache_flushed(channel, offset + len, timer)) {
DBG("Waiting for metadata to be flushed");
+
+ health_code_update();
+
usleep(DEFAULT_METADATA_AVAILABILITY_WAIT_TIME);
}
struct lttcomm_consumer_msg msg;
struct lttng_consumer_channel *channel = NULL;
+ health_code_update();
+
ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
if (ret != sizeof(msg)) {
DBG("Consumer received unexpected message size %zd (expects %zu)",
}
return ret;
}
+
+ health_code_update();
+
if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
/*
* Notify the session daemon that the command is completed.
return -ENOENT;
}
+ health_code_update();
+
/* relayd needs RCU read-side lock */
rcu_read_lock();
goto error_fatal;
};
+ health_code_update();
+
ret = ask_channel(ctx, sock, channel, &attr);
if (ret < 0) {
goto end_channel_error;
msg.u.ask_channel.live_timer_interval);
}
+ health_code_update();
+
/*
* Add the channel to the internal state AFTER all streams were created
* and successfully sent to session daemon. This way, all streams must
goto end_channel_error;
}
+ health_code_update();
+
/*
* Channel and streams are now created. Inform the session daemon that
* everything went well and should wait to receive the channel and
goto end_msg_sessiond;
}
+ health_code_update();
+
/* Send everything to sessiond. */
ret = send_sessiond_channel(sock, channel, ctx, &relayd_err);
if (ret < 0) {
goto error_fatal;
}
+ health_code_update();
+
/*
* In no monitor mode, the streams ownership is kept inside the channel
* so don't send them to the data thread.
goto end_msg_sessiond;
}
+ health_code_update();
+
/* Tell session daemon we are ready to receive the metadata. */
ret = consumer_send_status_msg(sock, LTTNG_OK);
if (ret < 0) {
goto error_fatal;
}
+ health_code_update();
+
/* Wait for more data. */
- if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
+ health_poll_entry();
+ ret = lttng_consumer_poll_socket(consumer_sockpoll);
+ health_poll_exit();
+ if (ret < 0) {
goto error_fatal;
}
+ health_code_update();
+
ret = lttng_ustconsumer_recv_metadata(sock, key, offset,
len, channel, 0, 1);
if (ret < 0) {
}
}
+ health_code_update();
ret = consumer_send_status_msg(sock, ret_code);
if (ret < 0) {
/* Somehow, the session daemon is not responding anymore. */
goto end_nosignal;
}
+ health_code_update();
break;
}
default:
end_nosignal:
rcu_read_unlock();
+ health_code_update();
+
/*
* Return 1 to indicate success since the 0 value can be a socket
* shutdown during the recv() or send() call.
goto error_fatal;
}
rcu_read_unlock();
+
+ health_code_update();
+
return 1;
end_channel_error:
if (channel) {
goto error_fatal;
}
rcu_read_unlock();
+
+ health_code_update();
+
return 1;
error_fatal:
rcu_read_unlock();
rcu_read_lock();
cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream,
node.node) {
+
+ health_code_update();
+
pthread_mutex_lock(&stream->chan->lock);
/*
* Whatever returned value, we must continue to try to close everything
request.key);
pthread_mutex_lock(&ctx->metadata_socket_lock);
+
+ health_code_update();
+
ret = lttcomm_send_unix_sock(ctx->consumer_metadata_socket, &request,
sizeof(request));
if (ret < 0) {
goto end;
}
+ health_code_update();
+
/* Receive the metadata from sessiond */
ret = lttcomm_recv_unix_sock(ctx->consumer_metadata_socket, &msg,
sizeof(msg));
goto end;
}
+ health_code_update();
+
if (msg.cmd_type == LTTNG_ERR_UND) {
/* No registry found */
(void) consumer_send_status_msg(ctx->consumer_metadata_socket,
DBG("No new metadata to receive for key %" PRIu64, key);
}
+ health_code_update();
+
/* Tell session daemon we are ready to receive the metadata. */
ret = consumer_send_status_msg(ctx->consumer_metadata_socket,
LTTNG_OK);
goto end;
}
+ health_code_update();
+
ret_code = lttng_ustconsumer_recv_metadata(ctx->consumer_metadata_socket,
key, offset, len, channel, timer, wait);
if (ret_code >= 0) {
ret = 0;
end:
+ health_code_update();
+
pthread_mutex_unlock(&ctx->metadata_socket_lock);
return ret;
}
projects / lttng-tools.git / blobdiff