X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.c;h=676aa0d9e2f066bec6b439f61e151579f4758bb5;hp=3330ff522cdeea40fd47b1cc8dd60f72aedf7f37;hb=6a00837f8cb0431a2ad90974d67fae138ea97dd5;hpb=4891ece8d4eeb2645efdf1467680b037c63ab425 diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c index 3330ff522..676aa0d9e 100644 --- a/src/common/ust-consumer/ust-consumer.c +++ b/src/common/ust-consumer/ust-consumer.c @@ -194,53 +194,40 @@ static int send_stream_to_thread(struct lttng_consumer_stream *stream, /* Get the right pipe where the stream will be sent. */ if (stream->metadata_flag) { + ret = consumer_add_metadata_stream(stream); + if (ret) { + ERR("Consumer add metadata stream %" PRIu64 " failed.", + stream->key); + goto error; + } stream_pipe = ctx->consumer_metadata_pipe; } else { + ret = consumer_add_data_stream(stream); + if (ret) { + ERR("Consumer add stream %" PRIu64 " failed.", + stream->key); + goto error; + } stream_pipe = ctx->consumer_data_pipe; } + /* + * From this point on, the stream's ownership has been moved away from + * the channel and becomes globally visible. + */ + stream->globally_visible = 1; + ret = lttng_pipe_write(stream_pipe, &stream, sizeof(stream)); if (ret < 0) { ERR("Consumer write %s stream to pipe %d", stream->metadata_flag ? "metadata" : "data", lttng_pipe_get_writefd(stream_pipe)); - } - - 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; + if (stream->metadata_flag) { + consumer_del_stream_for_metadata(stream); + } else { + consumer_del_stream_for_data(stream); } - } 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; } @@ -266,8 +253,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); @@ -321,6 +318,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]; } } @@ -379,7 +378,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); @@ -409,18 +408,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; } } @@ -555,60 +556,20 @@ static int send_streams_to_thread(struct lttng_consumer_channel *channel, * If we are unable to send the stream to the thread, there is * a big problem so just stop everything. */ + /* Remove node from the channel stream list. */ + cds_list_del(&stream->send_node); goto error; } /* Remove node from the channel stream list. */ cds_list_del(&stream->send_node); - /* - * From this point on, the stream's ownership has been moved away from - * the channel and becomes globally visible. - */ - stream->globally_visible = 1; } 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; - } - - ustctl_flush_buffer(metadata->metadata_stream->ustream, 1); - -error: - return ret; -} - /* * Flush channel's streams using the given key to retrieve the channel. * @@ -638,7 +599,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(); @@ -672,6 +633,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; @@ -689,9 +651,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; @@ -737,10 +707,13 @@ static int setup_metadata(struct lttng_consumer_local_data *ctx, uint64_t key) } /* 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); @@ -781,8 +754,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; @@ -796,7 +767,8 @@ static int snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id, metadata_channel = consumer_find_channel(key); if (!metadata_channel) { - ERR("UST snapshot metadata channel not found for key %lu", key); + ERR("UST snapshot metadata channel not found for key %" PRIu64, + key); ret = -1; goto error; } @@ -806,7 +778,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; } @@ -842,30 +814,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); @@ -890,7 +845,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; @@ -909,12 +864,12 @@ static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id, channel = consumer_find_channel(key); if (!channel) { - ERR("UST snapshot channel not found for key %lu", key); + ERR("UST snapshot channel not found for key %" PRIu64, key); ret = -1; goto error; } assert(!channel->monitor); - DBG("UST consumer snapshot channel %lu", key); + DBG("UST consumer snapshot channel %" PRIu64, key); cds_list_for_each_entry(stream, &channel->streams.head, send_node) { /* Lock stream because we are about to change its state. */ @@ -961,6 +916,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; @@ -1037,7 +1001,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; @@ -1059,17 +1024,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) { @@ -1081,13 +1035,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); } @@ -1115,12 +1067,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; @@ -1422,7 +1374,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; @@ -1456,6 +1408,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"); @@ -1633,28 +1586,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, @@ -1751,15 +1733,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; @@ -1815,8 +1826,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; @@ -1849,6 +1866,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) { @@ -1903,7 +1921,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 @@ -1914,5 +1932,6 @@ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx, ret = 0; end: + pthread_mutex_unlock(&ctx->metadata_socket_lock); return ret; }