X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.c;h=481bbeba04b2f91eef880335cbd3ac8ef5cb9302;hp=df12f6b8746de71a4a1ef952359abe2158268733;hb=4586339794f1139978f2b982d8124e3f5c6bb28e;hpb=8de4f9412d9681725c48f984c9e160773fdcf8c9 diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c index df12f6b87..481bbeba0 100644 --- a/src/common/ust-consumer/ust-consumer.c +++ b/src/common/ust-consumer/ust-consumer.c @@ -47,7 +47,7 @@ #include "ust-consumer.h" -#define UINT_MAX_STR_LEN 11 /* includes \0 */ +#define INT_MAX_STR_LEN 12 /* includes \0 */ extern struct lttng_consumer_global_data consumer_data; extern int consumer_poll_timeout; @@ -82,6 +82,7 @@ static void destroy_channel(struct lttng_consumer_channel *channel) */ if (channel->uchan) { lttng_ustconsumer_del_channel(channel); + lttng_ustconsumer_free_channel(channel); } free(channel); } @@ -247,14 +248,14 @@ error: static int get_stream_shm_path(char *stream_shm_path, const char *shm_path, int cpu) { - char cpu_nr[UINT_MAX_STR_LEN]; /* unsigned int max len */ + char cpu_nr[INT_MAX_STR_LEN]; /* int max len */ int ret; strncpy(stream_shm_path, shm_path, PATH_MAX); stream_shm_path[PATH_MAX - 1] = '\0'; - ret = snprintf(cpu_nr, UINT_MAX_STR_LEN, "%u", cpu); - if (ret != 1) { - ret = -1; + ret = snprintf(cpu_nr, INT_MAX_STR_LEN, "%i", cpu); + if (ret < 0) { + PERROR("snprintf"); goto end; } strncat(stream_shm_path, cpu_nr, @@ -504,7 +505,6 @@ error_open: closeret = run_as_unlink(shm_path, channel->uid, channel->gid); if (closeret) { - errno = -closeret; PERROR("unlink %s", shm_path); } } @@ -1168,7 +1168,12 @@ error: } /* - * Receive the metadata updates from the sessiond. + * Receive the metadata updates from the sessiond. Supports receiving + * overlapping metadata, but is needs to always belong to a contiguous + * range starting from 0. + * Be careful about the locks held when calling this function: it needs + * the metadata cache flush to concurrently progress in order to + * complete. */ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset, uint64_t len, struct lttng_consumer_channel *channel, @@ -1582,6 +1587,15 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, health_code_update(); + if (!len) { + /* + * There is nothing to receive. We have simply + * checked whether the channel can be found. + */ + ret_code = LTTCOMM_CONSUMERD_SUCCESS; + goto end_msg_sessiond; + } + /* Tell session daemon we are ready to receive the metadata. */ ret = consumer_send_status_msg(sock, LTTCOMM_CONSUMERD_SUCCESS); if (ret < 0) { @@ -1817,8 +1831,6 @@ void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan) if (chan->switch_timer_enabled == 1) { consumer_timer_switch_stop(chan); } - consumer_metadata_cache_destroy(chan); - ustctl_destroy_channel(chan->uchan); for (i = 0; i < chan->nr_stream_fds; i++) { int ret; @@ -1835,12 +1847,10 @@ void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan) } ret = run_as_unlink(shm_path, chan->uid, chan->gid); if (ret) { - errno = -ret; PERROR("unlink %s", shm_path); } } } - free(chan->stream_fds); /* Try to rmdir all directories under shm_path root. */ if (chan->root_shm_path[0]) { (void) run_as_recursive_rmdir(chan->root_shm_path, @@ -1848,6 +1858,16 @@ void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan) } } +void lttng_ustconsumer_free_channel(struct lttng_consumer_channel *chan) +{ + assert(chan); + assert(chan->uchan); + + consumer_metadata_cache_destroy(chan); + ustctl_destroy_channel(chan->uchan); + free(chan->stream_fds); +} + void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream) { assert(stream); @@ -1944,7 +1964,7 @@ int commit_one_metadata_packet(struct lttng_consumer_stream *stream) int ret; pthread_mutex_lock(&stream->chan->metadata_cache->lock); - if (stream->chan->metadata_cache->contiguous + if (stream->chan->metadata_cache->max_offset == stream->ust_metadata_pushed) { ret = 0; goto end; @@ -1952,7 +1972,7 @@ int commit_one_metadata_packet(struct lttng_consumer_stream *stream) write_len = ustctl_write_one_packet_to_channel(stream->chan->uchan, &stream->chan->metadata_cache->data[stream->ust_metadata_pushed], - stream->chan->metadata_cache->contiguous + stream->chan->metadata_cache->max_offset - stream->ust_metadata_pushed); assert(write_len != 0); if (write_len < 0) { @@ -1962,7 +1982,7 @@ int commit_one_metadata_packet(struct lttng_consumer_stream *stream) } stream->ust_metadata_pushed += write_len; - assert(stream->chan->metadata_cache->contiguous >= + assert(stream->chan->metadata_cache->max_offset >= stream->ust_metadata_pushed); ret = write_len; @@ -1976,7 +1996,9 @@ end: * Sync metadata meaning request them to the session daemon and snapshot to the * metadata thread can consumer them. * - * Metadata stream lock MUST be acquired. + * Metadata stream lock is held here, but we need to release it when + * interacting with sessiond, else we cause a deadlock with live + * awaiting on metadata to be pushed out. * * Return 0 if new metadatda is available, EAGAIN if the metadata stream * is empty or a negative value on error. @@ -1990,6 +2012,7 @@ int lttng_ustconsumer_sync_metadata(struct lttng_consumer_local_data *ctx, assert(ctx); assert(metadata); + pthread_mutex_unlock(&metadata->lock); /* * Request metadata from the sessiond, but don't wait for the flush * because we locked the metadata thread. @@ -1998,6 +2021,7 @@ int lttng_ustconsumer_sync_metadata(struct lttng_consumer_local_data *ctx, if (ret < 0) { goto end; } + pthread_mutex_lock(&metadata->lock); ret = commit_one_metadata_packet(metadata); if (ret <= 0) { @@ -2224,7 +2248,23 @@ retry: /* * In live, block until all the metadata is sent. */ + pthread_mutex_lock(&stream->metadata_timer_lock); + assert(!stream->missed_metadata_flush); + stream->waiting_on_metadata = true; + pthread_mutex_unlock(&stream->metadata_timer_lock); + err = consumer_stream_sync_metadata(ctx, stream->session_id); + + pthread_mutex_lock(&stream->metadata_timer_lock); + stream->waiting_on_metadata = false; + if (stream->missed_metadata_flush) { + stream->missed_metadata_flush = false; + pthread_mutex_unlock(&stream->metadata_timer_lock); + (void) consumer_flush_ust_index(stream); + } else { + pthread_mutex_unlock(&stream->metadata_timer_lock); + } + if (err < 0) { goto end; } @@ -2305,7 +2345,7 @@ int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream) uint64_t contiguous, pushed; /* Ease our life a bit. */ - contiguous = stream->chan->metadata_cache->contiguous; + contiguous = stream->chan->metadata_cache->max_offset; pushed = stream->ust_metadata_pushed; /* @@ -2434,6 +2474,10 @@ void lttng_ustconsumer_close_stream_wakeup(struct lttng_consumer_stream *stream) * function or any of its callees. Timers have a very strict locking * semantic with respect to teardown. Failure to respect this semantic * introduces deadlocks. + * + * DON'T hold the metadata lock when calling this function, else this + * can cause deadlock involving consumer awaiting for metadata to be + * pushed out due to concurrent interaction with the session daemon. */ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx, struct lttng_consumer_channel *channel, int timer, int wait)