X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;ds=sidebyside;f=src%2Fcommon%2Fconsumer%2Fconsumer.c;h=ccf9b40386ab306b8130ea073b62d8a07bd343e7;hb=fee30cf85f3b58b329d669db781cd7aac22f7785;hp=5c14dd1d80be07eeda87fae696cbde7a450de8ec;hpb=83f4233dde73280d45fcfb799add96cce32098ce;p=lttng-tools.git diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c index 5c14dd1d8..ccf9b4038 100644 --- a/src/common/consumer/consumer.c +++ b/src/common/consumer/consumer.c @@ -47,6 +47,7 @@ #include #include #include +#include struct lttng_consumer_global_data consumer_data = { .stream_count = 0, @@ -66,13 +67,16 @@ struct consumer_channel_msg { uint64_t key; /* del */ }; +/* Flag used to temporarily pause data consumption from testpoints. */ +int data_consumption_paused; + /* * Flag to inform the polling thread to quit when all fd hung up. Updated by * the consumer_thread_receive_fds when it notices that all fds has hung up. * Also updated by the signal handler (consumer_should_exit()). Read by the * polling threads. */ -volatile int consumer_quit; +int consumer_quit; /* * Global hash table containing respectively metadata and data streams. The @@ -367,6 +371,9 @@ void consumer_del_channel(struct lttng_consumer_channel *channel) if (channel->live_timer_enabled == 1) { consumer_timer_live_stop(channel); } + if (channel->monitor_timer_enabled == 1) { + consumer_timer_monitor_stop(channel); + } switch (consumer_data.type) { case LTTNG_CONSUMER_KERNEL: @@ -570,7 +577,7 @@ struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key, stream->session_id = session_id; stream->monitor = monitor; stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE; - stream->index_fd = -1; + stream->index_file = NULL; stream->last_sequence_number = -1ULL; pthread_mutex_init(&stream->lock, NULL); pthread_mutex_init(&stream->metadata_timer_lock, NULL); @@ -1220,7 +1227,7 @@ void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx) { ssize_t ret; - consumer_quit = 1; + CMM_STORE_SHARED(consumer_quit, 1); ret = lttng_write(ctx->consumer_should_quit[1], "4", 1); if (ret < 1) { PERROR("write consumer quit"); @@ -1347,6 +1354,8 @@ struct lttng_consumer_local_data *lttng_consumer_create( goto error_metadata_pipe; } + ctx->channel_monitor_pipe = -1; + return ctx; error_metadata_pipe: @@ -1530,7 +1539,6 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap( mmap_base = stream->mmap_base; ret = kernctl_get_mmap_read_offset(stream->wait_fd, &mmap_offset); if (ret < 0) { - ret = -errno; PERROR("tracer ctl get_mmap_read_offset"); goto end; } @@ -1625,21 +1633,16 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap( } outfd = stream->out_fd; - if (stream->index_fd >= 0) { - ret = close(stream->index_fd); - if (ret < 0) { - PERROR("Closing index"); - goto end; - } - stream->index_fd = -1; - ret = index_create_file(stream->chan->pathname, + 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); - if (ret < 0) { + stream->tracefile_count_current, + CTF_INDEX_MAJOR, CTF_INDEX_MINOR); + if (!stream->index_file) { goto end; } - stream->index_fd = ret; } /* Reset current size because we just perform a rotation. */ @@ -1832,22 +1835,16 @@ ssize_t lttng_consumer_on_read_subbuffer_splice( } outfd = stream->out_fd; - if (stream->index_fd >= 0) { - ret = close(stream->index_fd); - if (ret < 0) { - PERROR("Closing index"); - goto end; - } - stream->index_fd = -1; - ret = index_create_file(stream->chan->pathname, + 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); - if (ret < 0) { - written = ret; + stream->tracefile_count_current, + CTF_INDEX_MAJOR, CTF_INDEX_MINOR); + if (!stream->index_file) { goto end; } - stream->index_fd = ret; } /* Reset current size because we just perform a rotation. */ @@ -2064,6 +2061,10 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream, pthread_mutex_lock(&consumer_data.lock); pthread_mutex_lock(&stream->chan->lock); pthread_mutex_lock(&stream->lock); + if (stream->chan->metadata_cache) { + /* Only applicable to userspace consumers. */ + pthread_mutex_lock(&stream->chan->metadata_cache->lock); + } /* Remove any reference to that stream. */ consumer_stream_delete(stream, ht); @@ -2087,6 +2088,9 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream, */ stream->chan->metadata_stream = NULL; + if (stream->chan->metadata_cache) { + pthread_mutex_unlock(&stream->chan->metadata_cache->lock); + } pthread_mutex_unlock(&stream->lock); pthread_mutex_unlock(&stream->chan->lock); pthread_mutex_unlock(&consumer_data.lock); @@ -2522,13 +2526,16 @@ void *consumer_thread_data_poll(void *data) pthread_mutex_unlock(&consumer_data.lock); /* No FDs and consumer_quit, consumer_cleanup the thread */ - if (nb_fd == 0 && consumer_quit == 1) { + if (nb_fd == 0 && CMM_LOAD_SHARED(consumer_quit) == 1) { err = 0; /* All is OK */ goto end; } /* poll on the array of fds */ restart: DBG("polling on %d fd", nb_fd + 2); + if (testpoint(consumerd_thread_data_poll)) { + goto end; + } health_poll_entry(); num_rdy = poll(pollfd, nb_fd + 2, -1); health_poll_exit(); @@ -2548,6 +2555,12 @@ void *consumer_thread_data_poll(void *data) goto end; } + if (caa_unlikely(data_consumption_paused)) { + DBG("Data consumption paused, sleeping..."); + sleep(1); + goto restart; + } + /* * If the consumer_data_pipe triggered poll go directly to the * beginning of the loop to update the array. We want to prioritize @@ -3190,7 +3203,7 @@ void *consumer_thread_sessiond_poll(void *data) err = 0; goto end; } - if (consumer_quit) { + if (CMM_LOAD_SHARED(consumer_quit)) { DBG("consumer_thread_receive_fds received quit from signal"); err = 0; /* All is OK */ goto end; @@ -3215,7 +3228,7 @@ end: * when all fds have hung up, the polling thread * can exit cleanly */ - consumer_quit = 1; + CMM_STORE_SHARED(consumer_quit, 1); /* * Notify the data poll thread to poll back again and test the @@ -3369,7 +3382,6 @@ int consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type, /* Not found. Allocate one. */ relayd = consumer_allocate_relayd_sock_pair(net_seq_idx); if (relayd == NULL) { - ret = -ENOMEM; ret_code = LTTCOMM_CONSUMERD_ENOMEM; goto error; } else { @@ -3402,14 +3414,12 @@ int consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type, if (ret) { /* Needing to exit in the middle of a command: error. */ lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR); - ret = -EINTR; goto error_nosignal; } /* Get relayd socket from session daemon */ ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1); if (ret != sizeof(fd)) { - ret = -1; fd = -1; /* Just in case it gets set with an invalid value. */ /* @@ -3483,7 +3493,6 @@ int consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type, break; default: ERR("Unknown relayd socket type (%d)", sock_type); - ret = -1; ret_code = LTTCOMM_CONSUMERD_FATAL; goto error; }