X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=6cf85d96a75b0afa69994886d9a94d91a7faaa29;hp=bcede1ef5d8f3857515baeaf709ad9b78294a684;hb=5e41ebe1bcc918cb5ff5d97489ccfbae0b7086a3;hpb=30319bcbeabf573068172289850aea0bcfe23abc diff --git a/src/common/consumer.c b/src/common/consumer.c index bcede1ef5..6cf85d96a 100644 --- a/src/common/consumer.c +++ b/src/common/consumer.c @@ -291,6 +291,7 @@ void consumer_del_channel(struct lttng_consumer_channel *channel) DBG("Consumer delete channel key %" PRIu64, channel->key); pthread_mutex_lock(&consumer_data.lock); + pthread_mutex_lock(&channel->lock); /* Delete streams that might have been left in the stream list. */ cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head, @@ -324,6 +325,7 @@ void consumer_del_channel(struct lttng_consumer_channel *channel) call_rcu(&channel->node.head, free_channel_rcu); end: + pthread_mutex_unlock(&channel->lock); pthread_mutex_unlock(&consumer_data.lock); } @@ -489,6 +491,7 @@ struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key, stream->net_seq_idx = relayd_id; stream->session_id = session_id; stream->monitor = monitor; + stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE; pthread_mutex_init(&stream->lock, NULL); /* If channel is the metadata, flag this stream as metadata. */ @@ -547,6 +550,8 @@ static int add_stream(struct lttng_consumer_stream *stream, DBG3("Adding consumer stream %" PRIu64, stream->key); pthread_mutex_lock(&consumer_data.lock); + pthread_mutex_lock(&stream->chan->lock); + pthread_mutex_lock(&stream->chan->timer_lock); pthread_mutex_lock(&stream->lock); rcu_read_lock(); @@ -584,6 +589,8 @@ static int add_stream(struct lttng_consumer_stream *stream, rcu_read_unlock(); pthread_mutex_unlock(&stream->lock); + pthread_mutex_unlock(&stream->chan->timer_lock); + pthread_mutex_unlock(&stream->chan->lock); pthread_mutex_unlock(&consumer_data.lock); return ret; @@ -832,6 +839,8 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key, channel->tracefile_size = tracefile_size; channel->tracefile_count = tracefile_count; channel->monitor = monitor; + pthread_mutex_init(&channel->lock, NULL); + pthread_mutex_init(&channel->timer_lock, NULL); /* * In monitor mode, the streams associated with the channel will be put in @@ -877,6 +886,8 @@ int consumer_add_channel(struct lttng_consumer_channel *channel, struct lttng_ht_iter iter; pthread_mutex_lock(&consumer_data.lock); + pthread_mutex_lock(&channel->lock); + pthread_mutex_lock(&channel->timer_lock); rcu_read_lock(); lttng_ht_lookup(consumer_data.channel_ht, &channel->key, &iter); @@ -893,6 +904,8 @@ int consumer_add_channel(struct lttng_consumer_channel *channel, end: rcu_read_unlock(); + pthread_mutex_unlock(&channel->timer_lock); + pthread_mutex_unlock(&channel->lock); pthread_mutex_unlock(&consumer_data.lock); if (!ret && channel->wait_fd != -1 && @@ -1145,6 +1158,7 @@ struct lttng_consumer_local_data *lttng_consumer_create( ctx->consumer_error_socket = -1; ctx->consumer_metadata_socket = -1; + pthread_mutex_init(&ctx->metadata_socket_lock, NULL); /* assign the callbacks */ ctx->on_buffer_ready = buffer_ready; ctx->on_recv_channel = recv_channel; @@ -1852,6 +1866,7 @@ 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); switch (consumer_data.type) { @@ -1871,6 +1886,13 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream, break; case LTTNG_CONSUMER32_UST: case LTTNG_CONSUMER64_UST: + if (stream->monitor) { + /* close the write-side in close_metadata */ + ret = close(stream->ust_metadata_poll_pipe[0]); + if (ret < 0) { + PERROR("Close UST metadata read-side poll pipe"); + } + } lttng_ustconsumer_del_stream(stream); break; default: @@ -1939,12 +1961,13 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream, end: /* * Nullify the stream reference so it is not used after deletion. The - * consumer data lock MUST be acquired before being able to check for a - * NULL pointer value. + * channel lock MUST be acquired before being able to check for + * a NULL pointer value. */ stream->chan->metadata_stream = NULL; pthread_mutex_unlock(&stream->lock); + pthread_mutex_unlock(&stream->chan->lock); pthread_mutex_unlock(&consumer_data.lock); if (free_chan) { @@ -1972,6 +1995,8 @@ static int add_metadata_stream(struct lttng_consumer_stream *stream, DBG3("Adding metadata stream %" PRIu64 " to hash table", stream->key); pthread_mutex_lock(&consumer_data.lock); + pthread_mutex_lock(&stream->chan->lock); + pthread_mutex_lock(&stream->chan->timer_lock); pthread_mutex_lock(&stream->lock); /* @@ -2017,6 +2042,8 @@ static int add_metadata_stream(struct lttng_consumer_stream *stream, rcu_read_unlock(); pthread_mutex_unlock(&stream->lock); + pthread_mutex_unlock(&stream->chan->lock); + pthread_mutex_unlock(&stream->chan->timer_lock); pthread_mutex_unlock(&consumer_data.lock); return ret; } @@ -2240,14 +2267,21 @@ restart: DBG("Metadata available on fd %d", pollfd); assert(stream->wait_fd == pollfd); - len = ctx->on_buffer_ready(stream, ctx); + do { + len = ctx->on_buffer_ready(stream, ctx); + /* + * We don't check the return value here since if we get + * a negative len, it means an error occured thus we + * simply remove it from the poll set and free the + * stream. + */ + } while (len > 0); + /* It's ok to have an unavailable sub-buffer */ if (len < 0 && len != -EAGAIN && len != -ENODATA) { /* Clean up stream from consumer and free it. */ lttng_poll_del(&events, stream->wait_fd); consumer_del_metadata_stream(stream, metadata_ht); - } else if (len > 0) { - stream->data_read = 1; } } @@ -3457,3 +3491,23 @@ int consumer_send_status_channel(int sock, return lttcomm_send_unix_sock(sock, &msg, sizeof(msg)); } + +/* + * Using a maximum stream size with the produced and consumed position of a + * stream, computes the new consumed position to be as close as possible to the + * maximum possible stream size. + * + * If maximum stream size is lower than the possible buffer size (produced - + * consumed), the consumed_pos given is returned untouched else the new value + * is returned. + */ +unsigned long consumer_get_consumed_maxsize(unsigned long consumed_pos, + unsigned long produced_pos, uint64_t max_stream_size) +{ + if (max_stream_size && max_stream_size < (produced_pos - consumed_pos)) { + /* Offset from the produced position to get the latest buffers. */ + return produced_pos - max_stream_size; + } + + return consumed_pos; +}