X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.c;h=5bcca4e17a3d94622f423d8a30e410e05473d741;hp=f318af153ca3aca1dcdd1573590d7447de61490c;hb=b8086166dbf40d53a4a1e3f0cddbf291fac97bd3;hpb=10a5031171c7bca5b4498c871b119e5a88b6a3fb diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c index f318af153..5bcca4e17 100644 --- a/src/common/ust-consumer/ust-consumer.c +++ b/src/common/ust-consumer/ust-consumer.c @@ -152,7 +152,8 @@ static struct lttng_consumer_stream *allocate_stream(int cpu, int key, channel->session_id, cpu, &alloc_ret, - channel->type); + channel->type, + channel->monitor); if (stream == NULL) { switch (alloc_ret) { case -ENOENT: @@ -208,42 +209,6 @@ static int send_stream_to_thread(struct lttng_consumer_stream *stream, return ret; } -/* - * Search for a relayd object related to the stream. If found, send the stream - * to the relayd. - * - * On success, returns 0 else a negative value. - */ -static int send_stream_to_relayd(struct lttng_consumer_stream *stream) -{ - int ret = 0; - struct consumer_relayd_sock_pair *relayd; - - assert(stream); - - relayd = consumer_find_relayd(stream->net_seq_idx); - if (relayd != NULL) { - pthread_mutex_lock(&relayd->ctrl_sock_mutex); - /* Add stream on the relayd */ - ret = relayd_add_stream(&relayd->control_sock, stream->name, - stream->chan->pathname, &stream->relayd_stream_id, - stream->chan->tracefile_size, - stream->chan->tracefile_count); - pthread_mutex_unlock(&relayd->ctrl_sock_mutex); - if (ret < 0) { - goto error; - } - } else if (stream->net_seq_idx != (uint64_t) -1ULL) { - ERR("Network sequence index %" PRIu64 " unknown. Not adding stream.", - stream->net_seq_idx); - ret = -1; - goto error; - } - -error: - return ret; -} - /* * Create streams for the given channel using liblttng-ust-ctl. * @@ -265,8 +230,18 @@ static int create_ust_streams(struct lttng_consumer_channel *channel, */ while ((ustream = ustctl_create_stream(channel->uchan, cpu))) { int wait_fd; + int ust_metadata_pipe[2]; - wait_fd = ustctl_stream_get_wait_fd(ustream); + if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA && channel->monitor) { + ret = utils_create_pipe_cloexec_nonblock(ust_metadata_pipe); + if (ret < 0) { + ERR("Create ust metadata poll pipe"); + goto error; + } + wait_fd = ust_metadata_pipe[0]; + } else { + wait_fd = ustctl_stream_get_wait_fd(ustream); + } /* Allocate consumer stream object. */ stream = allocate_stream(cpu, wait_fd, channel, ctx, &ret); @@ -320,6 +295,8 @@ static int create_ust_streams(struct lttng_consumer_channel *channel, /* Keep stream reference when creating metadata. */ if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA) { channel->metadata_stream = stream; + stream->ust_metadata_poll_pipe[0] = ust_metadata_pipe[0]; + stream->ust_metadata_poll_pipe[1] = ust_metadata_pipe[1]; } } @@ -378,7 +355,7 @@ static int send_sessiond_stream(int sock, struct lttng_consumer_stream *stream) assert(stream); assert(sock >= 0); - DBG2("UST consumer sending stream %" PRIu64 " to sessiond", stream->key); + DBG("UST consumer sending stream %" PRIu64 " to sessiond", stream->key); /* Send stream to session daemon. */ ret = ustctl_send_stream_to_sessiond(sock, stream->ustream); @@ -408,18 +385,20 @@ static int send_sessiond_channel(int sock, DBG("UST consumer sending channel %s to sessiond", channel->name); - cds_list_for_each_entry(stream, &channel->streams.head, send_node) { - /* Try to send the stream to the relayd if one is available. */ - ret = send_stream_to_relayd(stream); - if (ret < 0) { - /* - * Flag that the relayd was the problem here probably due to a - * communicaton error on the socket. - */ - if (relayd_error) { - *relayd_error = 1; + if (channel->relayd_id != (uint64_t) -1ULL) { + cds_list_for_each_entry(stream, &channel->streams.head, send_node) { + /* Try to send the stream to the relayd if one is available. */ + ret = consumer_send_relayd_stream(stream, stream->chan->pathname); + if (ret < 0) { + /* + * Flag that the relayd was the problem here probably due to a + * communicaton error on the socket. + */ + if (relayd_error) { + *relayd_error = 1; + } + ret_code = LTTNG_ERR_RELAYD_CONNECT_FAIL; } - ret_code = LTTNG_ERR_RELAYD_CONNECT_FAIL; } } @@ -559,45 +538,14 @@ static int send_streams_to_thread(struct lttng_consumer_channel *channel, /* Remove node from the channel stream list. */ cds_list_del(&stream->send_node); - } - -error: - return ret; -} -/* - * Write metadata to the given channel using ustctl to convert the string to - * the ringbuffer. - * Called only from consumer_metadata_cache_write. - * The metadata cache lock MUST be acquired to write in the cache. - * - * Return 0 on success else a negative value. - */ -int lttng_ustconsumer_push_metadata(struct lttng_consumer_channel *metadata, - const char *metadata_str, uint64_t target_offset, uint64_t len) -{ - int ret; - - assert(metadata); - assert(metadata_str); - - DBG("UST consumer writing metadata to channel %s", metadata->name); - - if (!metadata->metadata_stream) { - ret = 0; - goto error; - } - - assert(target_offset <= metadata->metadata_cache->max_offset); - ret = ustctl_write_metadata_to_channel(metadata->uchan, - metadata_str + target_offset, len); - if (ret < 0) { - ERR("ustctl write metadata fail with ret %d, len %" PRIu64, ret, len); - goto error; + /* + * From this point on, the stream's ownership has been moved away from + * the channel and becomes globally visible. + */ + stream->globally_visible = 1; } - ustctl_flush_buffer(metadata->metadata_stream->ustream, 1); - error: return ret; } @@ -631,7 +579,7 @@ 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) { - ustctl_flush_buffer(stream->ustream, 1); + ustctl_flush_buffer(stream->ustream, 1); } error: rcu_read_unlock(); @@ -665,6 +613,7 @@ static int close_metadata(uint64_t chan_key) } pthread_mutex_lock(&consumer_data.lock); + pthread_mutex_lock(&channel->lock); if (cds_lfht_is_node_deleted(&channel->node.node)) { goto error_unlock; @@ -682,9 +631,17 @@ static int close_metadata(uint64_t chan_key) ret = LTTCOMM_CONSUMERD_ERROR_METADATA; goto error_unlock; } + if (channel->monitor) { + /* close the read-side in consumer_del_metadata_stream */ + ret = close(channel->metadata_stream->ust_metadata_poll_pipe[1]); + if (ret < 0) { + PERROR("Close UST metadata write-side poll pipe"); + } + } } error_unlock: + pthread_mutex_unlock(&channel->lock); pthread_mutex_unlock(&consumer_data.lock); error: return ret; @@ -726,14 +683,17 @@ static int setup_metadata(struct lttng_consumer_local_data *ctx, uint64_t key) if (cds_list_empty(&metadata->streams.head)) { ERR("Metadata channel key %" PRIu64 ", no stream available.", key); ret = LTTCOMM_CONSUMERD_ERROR_METADATA; - goto error; + goto error_no_stream; } /* Send metadata stream to relayd if needed. */ - ret = send_stream_to_relayd(metadata->metadata_stream); - if (ret < 0) { - ret = LTTCOMM_CONSUMERD_ERROR_METADATA; - goto error; + if (metadata->metadata_stream->net_seq_idx != (uint64_t) -1ULL) { + ret = consumer_send_relayd_stream(metadata->metadata_stream, + metadata->pathname); + if (ret < 0) { + ret = LTTCOMM_CONSUMERD_ERROR_METADATA; + goto error; + } } ret = send_streams_to_thread(metadata, ctx); @@ -758,7 +718,9 @@ error: * the stream is still in the local stream list of the channel. This call * will make sure to clean that list. */ - consumer_del_channel(metadata); + cds_list_del(&metadata->metadata_stream->send_node); + consumer_stream_destroy(metadata->metadata_stream, NULL); +error_no_stream: end: return ret; } @@ -772,8 +734,6 @@ static int snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id, struct lttng_consumer_local_data *ctx) { int ret = 0; - ssize_t write_len; - uint64_t total_len = 0; struct lttng_consumer_channel *metadata_channel; struct lttng_consumer_stream *metadata_stream; @@ -797,7 +757,7 @@ static int snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id, * Ask the sessiond if we have new metadata waiting and update the * consumer metadata cache. */ - ret = lttng_ustconsumer_request_metadata(ctx, metadata_channel); + ret = lttng_ustconsumer_request_metadata(ctx, metadata_channel, 0); if (ret < 0) { goto error; } @@ -833,30 +793,13 @@ static int snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id, } pthread_mutex_lock(&metadata_channel->metadata_cache->lock); - while (total_len < metadata_channel->metadata_cache->total_bytes_written) { - /* - * Write at most one packet of metadata into the channel - * to avoid blocking here. - */ - write_len = ustctl_write_one_packet_to_channel(metadata_channel->uchan, - metadata_channel->metadata_cache->data, - metadata_channel->metadata_cache->total_bytes_written); - if (write_len < 0) { - ERR("UST consumer snapshot writing metadata packet"); - ret = -1; - goto error_unlock; - } - total_len += write_len; - DBG("Written %" PRIu64 " bytes to metadata (left: %" PRIu64 ")", - write_len, - metadata_channel->metadata_cache->total_bytes_written - write_len); - ustctl_flush_buffer(metadata_stream->ustream, 1); + do { ret = lttng_consumer_read_subbuffer(metadata_stream, ctx); if (ret < 0) { goto error_unlock; } - } + } while (ret > 0); error_unlock: pthread_mutex_unlock(&metadata_channel->metadata_cache->lock); @@ -881,7 +824,7 @@ error: * Returns 0 on success, < 0 on error */ static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id, - struct lttng_consumer_local_data *ctx) + uint64_t max_stream_size, struct lttng_consumer_local_data *ctx) { int ret; unsigned use_relayd = 0; @@ -952,6 +895,15 @@ static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id, goto error_unlock; } + /* + * The original value is sent back if max stream size is larger than + * the possible size of the snapshot. Also, we asume that the session + * daemon should never send a maximum stream size that is lower than + * subbuffer size. + */ + consumed_pos = consumer_get_consumed_maxsize(consumed_pos, + produced_pos, max_stream_size); + while (consumed_pos < produced_pos) { ssize_t read_len; unsigned long len, padded_len; @@ -1028,7 +980,8 @@ error: * Receive the metadata updates from the sessiond. */ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset, - uint64_t len, struct lttng_consumer_channel *channel) + uint64_t len, struct lttng_consumer_channel *channel, + int timer) { int ret, ret_code = LTTNG_OK; char *metadata_str; @@ -1050,17 +1003,6 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset, goto end_free; } - /* - * XXX: The consumer data lock is acquired before calling metadata cache - * write which calls push metadata that MUST be protected by the consumer - * lock in order to be able to check the validity of the metadata stream of - * the channel. - * - * Note that this will be subject to change to better fine grained locking - * and ultimately try to get rid of this global consumer data lock. - */ - pthread_mutex_lock(&consumer_data.lock); - pthread_mutex_lock(&channel->metadata_cache->lock); ret = consumer_metadata_cache_write(channel, offset, len, metadata_str); if (ret < 0) { @@ -1072,13 +1014,11 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset, * waiting for the metadata cache to be flushed. */ pthread_mutex_unlock(&channel->metadata_cache->lock); - pthread_mutex_unlock(&consumer_data.lock); goto end_free; } pthread_mutex_unlock(&channel->metadata_cache->lock); - pthread_mutex_unlock(&consumer_data.lock); - while (consumer_metadata_cache_flushed(channel, offset + len)) { + while (consumer_metadata_cache_flushed(channel, offset + len, timer)) { DBG("Waiting for metadata to be flushed"); usleep(DEFAULT_METADATA_AVAILABILITY_WAIT_TIME); } @@ -1106,12 +1046,12 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, if (ret != sizeof(msg)) { DBG("Consumer received unexpected message size %zd (expects %zu)", ret, sizeof(msg)); - lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD); /* * The ret value might 0 meaning an orderly shutdown but this is ok * since the caller handles this. */ if (ret > 0) { + lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD); ret = -1; } return ret; @@ -1413,7 +1353,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } ret = lttng_ustconsumer_recv_metadata(sock, key, offset, - len, channel); + len, channel, 0); if (ret < 0) { /* error receiving from sessiond */ goto error_fatal; @@ -1447,6 +1387,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, ret = snapshot_channel(msg.u.snapshot_channel.key, msg.u.snapshot_channel.pathname, msg.u.snapshot_channel.relayd_id, + msg.u.snapshot_channel.max_stream_size, ctx); if (ret < 0) { ERR("Snapshot channel failed"); @@ -1624,28 +1565,57 @@ int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream, assert(stream->ustream); assert(ctx); - DBG2("In UST read_subbuffer (wait_fd: %d, name: %s)", stream->wait_fd, + DBG("In UST read_subbuffer (wait_fd: %d, name: %s)", stream->wait_fd, stream->name); /* Ease our life for what's next. */ ustream = stream->ustream; /* We can consume the 1 byte written into the wait_fd by UST */ - if (!stream->hangup_flush_done) { + if (stream->monitor && !stream->hangup_flush_done) { ssize_t readlen; do { readlen = read(stream->wait_fd, &dummy, 1); } while (readlen == -1 && errno == EINTR); - if (readlen == -1) { + if (readlen == -1 && errno != EAGAIN && errno != EWOULDBLOCK) { ret = readlen; goto end; } } +retry: /* Get the next subbuffer */ err = ustctl_get_next_subbuf(ustream); if (err != 0) { + /* + * Populate metadata info if the existing info has + * already been read. + */ + if (stream->metadata_flag) { + ssize_t write_len; + + if (stream->chan->metadata_cache->contiguous + == stream->ust_metadata_pushed) { + ret = 0; + goto end; + } + + 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->ust_metadata_pushed); + assert(write_len != 0); + if (write_len < 0) { + ERR("Writing one metadata packet"); + ret = -1; + goto end; + } + stream->ust_metadata_pushed += write_len; + ustctl_flush_buffer(stream->ustream, 1); + goto retry; + } + ret = err; /* ustctl_get_next_subbuf returns negative, caller expect positive. */ /* * This is a debug message even for single-threaded consumer, @@ -1742,15 +1712,44 @@ int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream) DBG("UST consumer checking data pending"); - ret = ustctl_get_next_subbuf(stream->ustream); - if (ret == 0) { - /* There is still data so let's put back this subbuffer. */ - ret = ustctl_put_subbuf(stream->ustream); - assert(ret == 0); - ret = 1; /* Data is pending */ + if (stream->endpoint_status != CONSUMER_ENDPOINT_ACTIVE) { + ret = 0; goto end; } + if (stream->chan->type == CONSUMER_CHANNEL_TYPE_METADATA) { + /* + * We can simply check whether all contiguously available data + * has been pushed to the ring buffer, since the push operation + * is performed within get_next_subbuf(), and because both + * get_next_subbuf() and put_next_subbuf() are issued atomically + * thanks to the stream lock within + * lttng_ustconsumer_read_subbuffer(). This basically means that + * whetnever ust_metadata_pushed is incremented, the associated + * metadata has been consumed from the metadata stream. + */ + DBG("UST consumer metadata pending check: contiguous %" PRIu64 " vs pushed %" PRIu64, + stream->chan->metadata_cache->contiguous, + stream->ust_metadata_pushed); + if (stream->chan->metadata_cache->contiguous + != stream->ust_metadata_pushed) { + ret = 1; /* Data is pending */ + goto end; + } + } else { + ret = ustctl_get_next_subbuf(stream->ustream); + if (ret == 0) { + /* + * There is still data so let's put back this + * subbuffer. + */ + ret = ustctl_put_subbuf(stream->ustream); + assert(ret == 0); + ret = 1; /* Data is pending */ + goto end; + } + } + /* Data is NOT pending so ready to be read. */ ret = 0; @@ -1806,8 +1805,14 @@ void lttng_ustconsumer_close_stream_wakeup(struct lttng_consumer_stream *stream) } } +/* + * Please refer to consumer-timer.c before adding any lock within this + * function or any of its callees. Timers have a very strict locking + * semantic with respect to teardown. Failure to respect this semantic + * introduces deadlocks. + */ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx, - struct lttng_consumer_channel *channel) + struct lttng_consumer_channel *channel, int timer) { struct lttcomm_metadata_request_msg request; struct lttcomm_consumer_msg msg; @@ -1840,6 +1845,7 @@ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx, channel->session_id, channel->session_id_per_pid); + pthread_mutex_lock(&ctx->metadata_socket_lock); ret = lttcomm_send_unix_sock(ctx->consumer_metadata_socket, &request, sizeof(request)); if (ret < 0) { @@ -1894,7 +1900,7 @@ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx, } ret_code = lttng_ustconsumer_recv_metadata(ctx->consumer_metadata_socket, - key, offset, len, channel); + key, offset, len, channel, timer); if (ret_code >= 0) { /* * Only send the status msg if the sessiond is alive meaning a positive @@ -1905,5 +1911,6 @@ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx, ret = 0; end: + pthread_mutex_unlock(&ctx->metadata_socket_lock); return ret; }