From: Mathieu Desnoyers Date: Mon, 16 Sep 2013 22:54:42 +0000 (-0500) Subject: consumerd: add health instrumentation into threads X-Git-Tag: v2.4.0-rc1~115 X-Git-Url: https://git.lttng.org./?a=commitdiff_plain;h=9ce5646a7ef9b8d7936c46649a21ee546fadd538;p=lttng-tools.git consumerd: add health instrumentation into threads Signed-off-by: Mathieu Desnoyers --- diff --git a/src/common/consumer-timer.c b/src/common/consumer-timer.c index b02ccbb11..7ece0f6a5 100644 --- a/src/common/consumer-timer.c +++ b/src/common/consumer-timer.c @@ -472,12 +472,18 @@ void *consumer_timer_thread(void *data) health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_METADATA_TIMER); + health_code_update(); + /* Only self thread will receive signal mask. */ setmask(&mask); CMM_STORE_SHARED(timer_signal.tid, pthread_self()); while (1) { + health_code_update(); + + health_poll_entry(); signr = sigwaitinfo(&mask, &info); + health_poll_exit(); if (signr == -1) { if (errno != EINTR) { PERROR("sigwaitinfo"); diff --git a/src/common/consumer.c b/src/common/consumer.c index 6abd8b1e8..2f20ffb5d 100644 --- a/src/common/consumer.c +++ b/src/common/consumer.c @@ -2196,6 +2196,8 @@ void *consumer_thread_metadata_poll(void *data) health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_METADATA); + health_code_update(); + metadata_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64); if (!metadata_ht) { /* ENOMEM at this point. Better to bail out. */ @@ -2221,6 +2223,8 @@ void *consumer_thread_metadata_poll(void *data) DBG("Metadata main loop started"); while (1) { + health_code_update(); + /* Only the metadata pipe is set */ if (LTTNG_POLL_GETNB(&events) == 0 && consumer_quit == 1) { err = 0; /* All is OK */ @@ -2229,7 +2233,9 @@ void *consumer_thread_metadata_poll(void *data) restart: DBG("Metadata poll wait with %d fd(s)", LTTNG_POLL_GETNB(&events)); + health_poll_entry(); ret = lttng_poll_wait(&events, -1); + health_poll_exit(); DBG("Metadata event catched in thread"); if (ret < 0) { if (errno == EINTR) { @@ -2243,6 +2249,8 @@ restart: /* From here, the event is a metadata wait fd */ for (i = 0; i < nb_fd; i++) { + health_code_update(); + revents = LTTNG_POLL_GETEV(&events, i); pollfd = LTTNG_POLL_GETFD(&events, i); @@ -2312,6 +2320,8 @@ restart: /* We just flushed the stream now read it. */ do { + health_code_update(); + len = ctx->on_buffer_ready(stream, ctx); /* * We don't check the return value here since if we get @@ -2334,6 +2344,8 @@ restart: assert(stream->wait_fd == pollfd); do { + health_code_update(); + len = ctx->on_buffer_ready(stream, ctx); /* * We don't check the return value here since if we get @@ -2394,6 +2406,8 @@ void *consumer_thread_data_poll(void *data) health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_DATA); + health_code_update(); + data_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64); if (data_ht == NULL) { /* ENOMEM at this point. Better to bail out. */ @@ -2407,6 +2421,8 @@ void *consumer_thread_data_poll(void *data) } while (1) { + health_code_update(); + high_prio = 0; num_hup = 0; @@ -2459,7 +2475,9 @@ void *consumer_thread_data_poll(void *data) /* poll on the array of fds */ restart: DBG("polling on %d fd", nb_fd + 1); + health_poll_entry(); num_rdy = poll(pollfd, nb_fd + 1, -1); + health_poll_exit(); DBG("poll num_rdy : %d", num_rdy); if (num_rdy == -1) { /* @@ -2509,6 +2527,8 @@ void *consumer_thread_data_poll(void *data) /* Take care of high priority channels first. */ for (i = 0; i < nb_fd; i++) { + health_code_update(); + if (local_stream[i] == NULL) { continue; } @@ -2537,6 +2557,8 @@ void *consumer_thread_data_poll(void *data) /* Take care of low priority channels. */ for (i = 0; i < nb_fd; i++) { + health_code_update(); + if (local_stream[i] == NULL) { continue; } @@ -2557,6 +2579,8 @@ void *consumer_thread_data_poll(void *data) /* Handle hangup and errors */ for (i = 0; i < nb_fd; i++) { + health_code_update(); + if (local_stream[i] == NULL) { continue; } @@ -2721,6 +2745,8 @@ void *consumer_thread_channel_poll(void *data) health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_CHANNEL); + health_code_update(); + channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64); if (!channel_ht) { /* ENOMEM at this point. Better to bail out. */ @@ -2745,6 +2771,8 @@ void *consumer_thread_channel_poll(void *data) DBG("Channel main loop started"); while (1) { + health_code_update(); + /* Only the channel pipe is set */ if (LTTNG_POLL_GETNB(&events) == 0 && consumer_quit == 1) { err = 0; /* All is OK */ @@ -2753,7 +2781,9 @@ void *consumer_thread_channel_poll(void *data) restart: DBG("Channel poll wait with %d fd(s)", LTTNG_POLL_GETNB(&events)); + health_poll_entry(); ret = lttng_poll_wait(&events, -1); + health_poll_exit(); DBG("Channel event catched in thread"); if (ret < 0) { if (errno == EINTR) { @@ -2767,6 +2797,8 @@ restart: /* From here, the event is a channel wait fd */ for (i = 0; i < nb_fd; i++) { + health_code_update(); + revents = LTTNG_POLL_GETEV(&events, i); pollfd = LTTNG_POLL_GETFD(&events, i); @@ -2833,6 +2865,8 @@ restart: /* Delete streams that might have been left in the stream list. */ cds_list_for_each_entry_safe(stream, stmp, &chan->streams.head, send_node) { + health_code_update(); + cds_list_del(&stream->send_node); lttng_ustconsumer_del_stream(stream); uatomic_sub(&stream->chan->refcount, 1); @@ -2967,6 +3001,8 @@ void *consumer_thread_sessiond_poll(void *data) health_register(health_consumerd, HEALTH_CONSUMERD_TYPE_SESSIOND); + health_code_update(); + DBG("Creating command socket %s", ctx->consumer_command_sock_path); unlink(ctx->consumer_command_sock_path); client_socket = lttcomm_create_unix_sock(ctx->consumer_command_sock_path); @@ -3027,7 +3063,12 @@ void *consumer_thread_sessiond_poll(void *data) consumer_sockpoll[1].events = POLLIN | POLLPRI; while (1) { - if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) { + health_code_update(); + + health_poll_entry(); + ret = lttng_consumer_poll_socket(consumer_sockpoll); + health_poll_exit(); + if (ret < 0) { goto end; } DBG("Incoming command on sock"); diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c index d02e8502d..a506737ab 100644 --- a/src/common/kernel-consumer/kernel-consumer.c +++ b/src/common/kernel-consumer/kernel-consumer.c @@ -42,6 +42,7 @@ #include #include "kernel-consumer.h" +#include "../../bin/lttng-consumerd/health-consumerd.h" extern struct lttng_consumer_global_data consumer_data; extern int consumer_poll_timeout; @@ -139,6 +140,9 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, } cds_list_for_each_entry(stream, &channel->streams.head, send_node) { + + health_code_update(); + /* * Lock stream because we are about to change its state. */ @@ -221,6 +225,8 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, ssize_t read_len; unsigned long len, padded_len; + health_code_update(); + DBG("Kernel consumer taking snapshot at pos %lu", consumed_pos); ret = kernctl_get_subbuf(stream->wait_fd, &consumed_pos); @@ -362,6 +368,8 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, } do { + health_code_update(); + ret_read = lttng_kconsumer_read_subbuffer(metadata_stream, ctx); if (ret_read < 0) { if (ret_read != -EAGAIN) { @@ -413,6 +421,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, enum lttng_error_code ret_code = LTTNG_OK; struct lttcomm_consumer_msg msg; + health_code_update(); + ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg)); if (ret != sizeof(msg)) { if (ret > 0) { @@ -421,6 +431,9 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } return ret; } + + health_code_update(); + if (msg.cmd_type == LTTNG_CONSUMER_STOP) { /* * Notify the session daemon that the command is completed. @@ -433,6 +446,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, return -ENOENT; } + health_code_update(); + /* relayd needs RCU read-side protection */ rcu_read_lock(); @@ -451,12 +466,17 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, struct lttng_consumer_channel *new_channel; int ret_recv; + health_code_update(); + /* First send a status message before receiving the fds. */ ret = consumer_send_status_msg(sock, ret_code); if (ret < 0) { /* Somehow, the session daemon is not responding anymore. */ goto error_fatal; } + + health_code_update(); + DBG("consumer_add_channel %" PRIu64, msg.u.channel.channel_key); new_channel = consumer_allocate_channel(msg.u.channel.channel_key, msg.u.channel.session_id, msg.u.channel.pathname, @@ -494,6 +514,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, goto end_nosignal; }; + health_code_update(); + if (ctx->on_recv_channel != NULL) { ret_recv = ctx->on_recv_channel(new_channel); if (ret_recv == 0) { @@ -509,6 +531,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, msg.u.channel.live_timer_interval); } + health_code_update(); + /* If we received an error in add_channel, we need to report it. */ if (ret < 0) { ret = consumer_send_status_msg(sock, ret); @@ -542,23 +566,33 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, ret_code = LTTNG_ERR_KERN_CHAN_NOT_FOUND; } + health_code_update(); + /* First send a status message before receiving the fds. */ ret = consumer_send_status_msg(sock, ret_code); if (ret < 0) { /* Somehow, the session daemon is not responding anymore. */ goto error_fatal; } + + health_code_update(); + if (ret_code != LTTNG_OK) { /* Channel was not found. */ goto end_nosignal; } /* Blocking call */ - if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) { + health_poll_entry(); + ret = lttng_consumer_poll_socket(consumer_sockpoll); + health_poll_exit(); + if (ret < 0) { rcu_read_unlock(); return -EINTR; } + health_code_update(); + /* Get stream file descriptor from socket */ ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1); if (ret != sizeof(fd)) { @@ -567,6 +601,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, return ret; } + health_code_update(); + /* * Send status code to session daemon only if the recv works. If the * above recv() failed, the session daemon is notified through the @@ -578,6 +614,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, goto end_nosignal; } + health_code_update(); + new_stream = consumer_allocate_stream(channel->key, fd, LTTNG_CONSUMER_ACTIVE_STREAM, @@ -635,6 +673,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, */ new_stream->hangup_flush_done = 0; + health_code_update(); + if (ctx->on_recv_stream) { ret = ctx->on_recv_stream(new_stream); if (ret < 0) { @@ -643,6 +683,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } } + health_code_update(); + if (new_stream->metadata_flag) { channel->metadata_stream = new_stream; } @@ -690,6 +732,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, /* Vitible to other threads */ new_stream->globally_visible = 1; + health_code_update(); + ret = lttng_pipe_write(stream_pipe, &new_stream, sizeof(new_stream)); if (ret < 0) { ERR("Consumer write %s stream to pipe %d", @@ -740,6 +784,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, consumer_flag_relayd_for_destroy(relayd); } + health_code_update(); + ret = consumer_send_status_msg(sock, ret_code); if (ret < 0) { /* Somehow, the session daemon is not responding anymore. */ @@ -757,6 +803,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, ret = consumer_data_pending(id); + health_code_update(); + /* Send back returned value to session daemon */ ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret)); if (ret < 0) { @@ -792,6 +840,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } } + health_code_update(); + ret = consumer_send_status_msg(sock, ret_code); if (ret < 0) { /* Somehow, the session daemon is not responding anymore. */ @@ -810,12 +860,16 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, ret_code = LTTNG_ERR_KERN_CHAN_NOT_FOUND; } + 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(); + /* * This command should ONLY be issued for channel with streams set in * no monitor mode. @@ -843,6 +897,7 @@ end_nosignal: * Return 1 to indicate success since the 0 value can be a socket * shutdown during the recv() or send() call. */ + health_code_update(); return 1; error_fatal: diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c index 6a508143e..27b40b84a 100644 --- a/src/common/ust-consumer/ust-consumer.c +++ b/src/common/ust-consumer/ust-consumer.c @@ -43,6 +43,7 @@ #include #include "ust-consumer.h" +#include "../../bin/lttng-consumerd/health-consumerd.h" extern struct lttng_consumer_global_data consumer_data; extern int consumer_poll_timeout; @@ -63,6 +64,9 @@ static void destroy_channel(struct lttng_consumer_channel *channel) 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); @@ -257,6 +261,8 @@ static int create_ust_streams(struct lttng_consumer_channel *channel, 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) { @@ -412,6 +418,9 @@ static int send_sessiond_channel(int sock, 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) { @@ -450,6 +459,9 @@ static int send_sessiond_channel(int sock, /* 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) { @@ -551,6 +563,9 @@ static int send_streams_to_thread(struct lttng_consumer_channel *channel, /* 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) { @@ -601,6 +616,9 @@ static int flush_channel(uint64_t chan_key) 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: @@ -796,6 +814,8 @@ static int snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id, } assert(!metadata_channel->monitor); + health_code_update(); + /* * Ask the sessiond if we have new metadata waiting and update the * consumer metadata cache. @@ -805,6 +825,8 @@ static int snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id, 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. @@ -836,6 +858,8 @@ static int snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id, } do { + health_code_update(); + ret = lttng_consumer_read_subbuffer(metadata_stream, ctx); if (ret < 0) { goto error_stream; @@ -889,6 +913,9 @@ static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id, 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; @@ -946,6 +973,8 @@ static int snapshot_channel(uint64_t key, char *path, uint64_t 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); @@ -1033,6 +1062,8 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset, goto end; } + health_code_update(); + /* Receive metadata string. */ ret = lttcomm_recv_unix_sock(sock, metadata_str, len); if (ret < 0) { @@ -1041,6 +1072,8 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset, 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) { @@ -1061,6 +1094,9 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset, } 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); } @@ -1083,6 +1119,8 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, 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)", @@ -1097,6 +1135,9 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } return ret; } + + health_code_update(); + if (msg.cmd_type == LTTNG_CONSUMER_STOP) { /* * Notify the session daemon that the command is completed. @@ -1109,6 +1150,8 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, return -ENOENT; } + health_code_update(); + /* relayd needs RCU read-side lock */ rcu_read_lock(); @@ -1238,6 +1281,8 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, goto error_fatal; }; + health_code_update(); + ret = ask_channel(ctx, sock, channel, &attr); if (ret < 0) { goto end_channel_error; @@ -1256,6 +1301,8 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, 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 @@ -1277,6 +1324,8 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, 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 @@ -1305,6 +1354,8 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, goto end_msg_sessiond; } + health_code_update(); + /* Send everything to sessiond. */ ret = send_sessiond_channel(sock, channel, ctx, &relayd_err); if (ret < 0) { @@ -1324,6 +1375,8 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, 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. @@ -1396,6 +1449,8 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, 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) { @@ -1403,11 +1458,18 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, 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) { @@ -1451,11 +1513,13 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } } + 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: @@ -1465,6 +1529,8 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, 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. @@ -1482,6 +1548,9 @@ end_msg_sessiond: goto error_fatal; } rcu_read_unlock(); + + health_code_update(); + return 1; end_channel_error: if (channel) { @@ -1498,6 +1567,9 @@ end_channel_error: goto error_fatal; } rcu_read_unlock(); + + health_code_update(); + return 1; error_fatal: rcu_read_unlock(); @@ -2052,6 +2124,9 @@ void lttng_ustconsumer_close_metadata(struct lttng_ht *metadata_ht) 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 @@ -2123,6 +2198,9 @@ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx, 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) { @@ -2130,6 +2208,8 @@ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx, goto end; } + health_code_update(); + /* Receive the metadata from sessiond */ ret = lttcomm_recv_unix_sock(ctx->consumer_metadata_socket, &msg, sizeof(msg)); @@ -2144,6 +2224,8 @@ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx, goto end; } + health_code_update(); + if (msg.cmd_type == LTTNG_ERR_UND) { /* No registry found */ (void) consumer_send_status_msg(ctx->consumer_metadata_socket, @@ -2165,6 +2247,8 @@ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx, 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); @@ -2176,6 +2260,8 @@ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx, 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) { @@ -2188,6 +2274,8 @@ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx, ret = 0; end: + health_code_update(); + pthread_mutex_unlock(&ctx->metadata_socket_lock); return ret; }