X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=94a0cc3efb1d557da732ed9a201848b0ee4dc3e4;hp=0005b1039cb27d6ced5dfac3537e72bbf737fce0;hb=774d490c42598710944f54a53d81d1cab2a1dfaa;hpb=10a5031171c7bca5b4498c871b119e5a88b6a3fb diff --git a/src/common/consumer.c b/src/common/consumer.c index 0005b1039..94a0cc3ef 100644 --- a/src/common/consumer.c +++ b/src/common/consumer.c @@ -291,22 +291,24 @@ 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, + send_node) { + cds_list_del(&stream->send_node); + /* + * Once a stream is added to this list, the buffers were created so + * we have a guarantee that this call will succeed. + */ + consumer_stream_destroy(stream, NULL); + } switch (consumer_data.type) { case LTTNG_CONSUMER_KERNEL: break; case LTTNG_CONSUMER32_UST: case LTTNG_CONSUMER64_UST: - /* Delete streams that might have been left in the stream list. */ - cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head, - send_node) { - cds_list_del(&stream->send_node); - /* - * Once a stream is added to this list, the buffers were created so - * we have a guarantee that this call will succeed. - */ - consumer_stream_destroy(stream, NULL); - } lttng_ustconsumer_del_channel(channel); break; default: @@ -323,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); } @@ -464,7 +467,8 @@ struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key, uint64_t session_id, int cpu, int *alloc_ret, - enum consumer_channel_type type) + enum consumer_channel_type type, + unsigned int monitor) { int ret; struct lttng_consumer_stream *stream; @@ -486,6 +490,8 @@ struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key, stream->gid = gid; 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. */ @@ -537,7 +543,6 @@ static int add_stream(struct lttng_consumer_stream *stream, struct lttng_ht *ht) { int ret = 0; - struct consumer_relayd_sock_pair *relayd; assert(stream); assert(ht); @@ -545,6 +550,7 @@ 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->lock); rcu_read_lock(); @@ -563,12 +569,6 @@ static int add_stream(struct lttng_consumer_stream *stream, */ lttng_ht_add_u64(consumer_data.stream_list_ht, &stream->node_session_id); - /* Check and cleanup relayd */ - relayd = consumer_find_relayd(stream->net_seq_idx); - if (relayd != NULL) { - uatomic_inc(&relayd->refcount); - } - /* * When nb_init_stream_left reaches 0, we don't need to trigger any action * in terms of destroying the associated channel, because the action that @@ -588,6 +588,7 @@ static int add_stream(struct lttng_consumer_stream *stream, rcu_read_unlock(); pthread_mutex_unlock(&stream->lock); + pthread_mutex_unlock(&stream->chan->lock); pthread_mutex_unlock(&consumer_data.lock); return ret; @@ -706,6 +707,7 @@ int consumer_send_relayd_stream(struct lttng_consumer_stream *stream, goto end; } uatomic_inc(&relayd->refcount); + stream->sent_to_relayd = 1; } else { ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't send it.", stream->key, stream->net_seq_idx); @@ -835,6 +837,7 @@ 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); /* * In monitor mode, the streams associated with the channel will be put in @@ -880,6 +883,7 @@ 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); rcu_read_lock(); lttng_ht_lookup(consumer_data.channel_ht, &channel->key, &iter); @@ -896,6 +900,7 @@ int consumer_add_channel(struct lttng_consumer_channel *channel, end: rcu_read_unlock(); + pthread_mutex_unlock(&channel->lock); pthread_mutex_unlock(&consumer_data.lock); if (!ret && channel->wait_fd != -1 && @@ -1131,7 +1136,7 @@ struct lttng_consumer_local_data *lttng_consumer_create( struct lttng_consumer_local_data *ctx), int (*recv_channel)(struct lttng_consumer_channel *channel), int (*recv_stream)(struct lttng_consumer_stream *stream), - int (*update_stream)(int stream_key, uint32_t state)) + int (*update_stream)(uint64_t stream_key, uint32_t state)) { int ret; struct lttng_consumer_local_data *ctx; @@ -1855,6 +1860,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) { @@ -1948,6 +1954,7 @@ end: 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) { @@ -1966,7 +1973,6 @@ static int add_metadata_stream(struct lttng_consumer_stream *stream, struct lttng_ht *ht) { int ret = 0; - struct consumer_relayd_sock_pair *relayd; struct lttng_ht_iter iter; struct lttng_ht_node_u64 *node; @@ -1976,6 +1982,7 @@ 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->lock); /* @@ -1993,12 +2000,6 @@ static int add_metadata_stream(struct lttng_consumer_stream *stream, node = lttng_ht_iter_get_node_u64(&iter); assert(!node); - /* Find relayd and, if one is found, increment refcount. */ - relayd = consumer_find_relayd(stream->net_seq_idx); - if (relayd != NULL) { - uatomic_inc(&relayd->refcount); - } - /* * When nb_init_stream_left reaches 0, we don't need to trigger any action * in terms of destroying the associated channel, because the action that @@ -2027,6 +2028,7 @@ 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(&consumer_data.lock); return ret; } @@ -2301,7 +2303,11 @@ void *consumer_thread_data_poll(void *data) goto end; } - local_stream = zmalloc(sizeof(struct lttng_consumer_stream)); + local_stream = zmalloc(sizeof(struct lttng_consumer_stream *)); + if (local_stream == NULL) { + PERROR("local_stream malloc"); + goto end; + } while (1) { high_prio = 0; @@ -3041,7 +3047,7 @@ void lttng_consumer_init(void) int consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type, struct lttng_consumer_local_data *ctx, int sock, struct pollfd *consumer_sockpoll, - struct lttcomm_relayd_sock *relayd_sock, unsigned int sessiond_id) + struct lttcomm_relayd_sock *relayd_sock, uint64_t sessiond_id) { int fd = -1, ret = -1, relayd_created = 0; enum lttng_error_code ret_code = LTTNG_OK; @@ -3063,7 +3069,7 @@ int consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type, ret_code = LTTCOMM_CONSUMERD_ENOMEM; goto error; } else { - relayd->sessiond_session_id = (uint64_t) sessiond_id; + relayd->sessiond_session_id = sessiond_id; relayd_created = 1; } @@ -3463,3 +3469,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; +}