X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.c;h=c794b93f8e9e9574036bd4cef79468be9a853f1a;hp=e933f9227c9bd48fc0ce46562cd2994565684157;hb=309167d2a6f59d0c8cbf64eb23ba912cdea76a34;hpb=843c71c5b9f7f6b04c5a1751a4679310dd07033b diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c index e933f9227..c794b93f8 100644 --- a/src/common/ust-consumer/ust-consumer.c +++ b/src/common/ust-consumer/ust-consumer.c @@ -40,6 +40,7 @@ #include #include #include +#include #include "ust-consumer.h" @@ -194,18 +195,41 @@ 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)); + if (stream->metadata_flag) { + consumer_del_stream_for_metadata(stream); + } else { + consumer_del_stream_for_data(stream); + } } - +error: return ret; } @@ -533,17 +557,14 @@ 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: @@ -579,12 +600,51 @@ 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(); return ret; } +/* + * Close metadata stream wakeup_fd using the given key to retrieve the channel. + * RCU read side lock MUST be acquired before calling this function. + * + * NOTE: This function does NOT take any channel nor stream lock. + * + * Return 0 on success else LTTng error code. + */ +static int _close_metadata(struct lttng_consumer_channel *channel) +{ + int ret = LTTNG_OK; + + assert(channel); + assert(channel->type == CONSUMER_CHANNEL_TYPE_METADATA); + + if (channel->switch_timer_enabled == 1) { + DBG("Deleting timer on metadata channel"); + consumer_timer_switch_stop(channel); + } + + if (channel->metadata_stream) { + ret = ustctl_stream_close_wakeup_fd(channel->metadata_stream->ustream); + if (ret < 0) { + ERR("UST consumer unable to close fd of metadata (ret: %d)", ret); + ret = LTTCOMM_CONSUMERD_ERROR_METADATA; + } + + 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"); + ret = LTTCOMM_CONSUMERD_ERROR_METADATA; + } + } + } + + return ret; +} /* * Close metadata stream wakeup_fd using the given key to retrieve the channel. @@ -619,26 +679,7 @@ static int close_metadata(uint64_t chan_key) goto error_unlock; } - if (channel->switch_timer_enabled == 1) { - DBG("Deleting timer on metadata channel"); - consumer_timer_switch_stop(channel); - } - - if (channel->metadata_stream) { - ret = ustctl_stream_close_wakeup_fd(channel->metadata_stream->ustream); - if (ret < 0) { - ERR("UST consumer unable to close fd of metadata (ret: %d)", ret); - 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"); - } - } - } + ret = _close_metadata(channel); error_unlock: pthread_mutex_unlock(&channel->lock); @@ -747,7 +788,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; } @@ -757,7 +799,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; } @@ -784,7 +826,7 @@ static int snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id, ret = utils_create_stream_file(path, metadata_stream->name, metadata_stream->chan->tracefile_size, metadata_stream->tracefile_count_current, - metadata_stream->uid, metadata_stream->gid); + metadata_stream->uid, metadata_stream->gid, NULL); if (ret < 0) { goto error_stream; } @@ -843,12 +885,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. */ @@ -864,7 +906,7 @@ static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id, ret = utils_create_stream_file(path, stream->name, stream->chan->tracefile_size, stream->tracefile_count_current, - stream->uid, stream->gid); + stream->uid, stream->gid, NULL); if (ret < 0) { goto error_unlock; } @@ -934,15 +976,15 @@ static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id, } read_len = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len, - padded_len - len); + padded_len - len, NULL); if (use_relayd) { if (read_len != len) { - ret = -1; + ret = -EPERM; goto error_put_subbuf; } } else { if (read_len != padded_len) { - ret = -1; + ret = -EPERM; goto error_put_subbuf; } } @@ -980,7 +1022,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; @@ -1017,7 +1060,7 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset, } pthread_mutex_unlock(&channel->metadata_cache->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); } @@ -1156,6 +1199,13 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, goto end_channel_error; } + /* + * Assign UST application UID to the channel. This value is ignored for + * per PID buffers. This is specific to UST thus setting this after the + * allocation. + */ + channel->ust_app_uid = msg.u.ask_channel.ust_app_uid; + /* Build channel attributes from received message. */ attr.subbuf_size = msg.u.ask_channel.subbuf_size; attr.num_subbuf = msg.u.ask_channel.num_subbuf; @@ -1352,7 +1402,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; @@ -1551,14 +1601,72 @@ void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream) ustctl_destroy_stream(stream->ustream); } +/* + * Populate index values of a UST stream. Values are set in big endian order. + * + * Return 0 on success or else a negative value. + */ +static int get_index_values(struct lttng_packet_index *index, + struct ustctl_consumer_stream *ustream) +{ + int ret; + + ret = ustctl_get_timestamp_begin(ustream, &index->timestamp_begin); + if (ret < 0) { + PERROR("ustctl_get_timestamp_begin"); + goto error; + } + index->timestamp_begin = htobe64(index->timestamp_begin); + + ret = ustctl_get_timestamp_end(ustream, &index->timestamp_end); + if (ret < 0) { + PERROR("ustctl_get_timestamp_end"); + goto error; + } + index->timestamp_end = htobe64(index->timestamp_end); + + ret = ustctl_get_events_discarded(ustream, &index->events_discarded); + if (ret < 0) { + PERROR("ustctl_get_events_discarded"); + goto error; + } + index->events_discarded = htobe64(index->events_discarded); + + ret = ustctl_get_content_size(ustream, &index->content_size); + if (ret < 0) { + PERROR("ustctl_get_content_size"); + goto error; + } + index->content_size = htobe64(index->content_size); + + ret = ustctl_get_packet_size(ustream, &index->packet_size); + if (ret < 0) { + PERROR("ustctl_get_packet_size"); + goto error; + } + index->packet_size = htobe64(index->packet_size); + + ret = ustctl_get_stream_id(ustream, &index->stream_id); + if (ret < 0) { + PERROR("ustctl_get_stream_id"); + goto error; + } + index->stream_id = htobe64(index->stream_id); + +error: + return ret; +} + + int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream, struct lttng_consumer_local_data *ctx) { unsigned long len, subbuf_size, padding; - int err; + int err, write_index = 0; long ret = 0; char dummy; struct ustctl_consumer_stream *ustream; + struct lttng_packet_index index; assert(stream); assert(stream->ustream); @@ -1570,6 +1678,11 @@ int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream, /* Ease our life for what's next. */ ustream = stream->ustream; + /* Indicate that for this stream we have to write the index. */ + if (stream->index_fd >= 0) { + write_index = 1; + } + /* We can consume the 1 byte written into the wait_fd by UST */ if (stream->monitor && !stream->hangup_flush_done) { ssize_t readlen; @@ -1627,6 +1740,15 @@ retry: goto end; } assert(stream->chan->output == CONSUMER_CHANNEL_MMAP); + + if (!stream->metadata_flag && write_index) { + index.offset = htobe64(stream->out_fd_offset); + ret = get_index_values(&index, ustream); + if (ret < 0) { + goto end; + } + } + /* Get the full padded subbuffer size */ err = ustctl_get_padded_subbuf_size(ustream, &len); assert(err == 0); @@ -1640,7 +1762,7 @@ retry: padding = len - subbuf_size; /* write the subbuffer to the tracefile */ - ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size, padding); + ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size, padding, &index); /* * The mmap operation should write subbuf_size amount of data when network * streaming or the full padding (len) size when we are _not_ streaming. @@ -1658,10 +1780,20 @@ retry: DBG("Error writing to tracefile " "(ret: %ld != len: %lu != subbuf_size: %lu)", ret, len, subbuf_size); + write_index = 0; } err = ustctl_put_next_subbuf(ustream); assert(err == 0); + /* Write index if needed. */ + if (write_index) { + err = index_write(stream->index_fd, &index, sizeof(index)); + if (err < 0) { + ret = -1; + goto end; + } + } + end: return ret; } @@ -1681,12 +1813,23 @@ int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream) if (stream->net_seq_idx == (uint64_t) -1ULL && stream->chan->monitor) { ret = utils_create_stream_file(stream->chan->pathname, stream->name, stream->chan->tracefile_size, stream->tracefile_count_current, - stream->uid, stream->gid); + stream->uid, stream->gid, NULL); if (ret < 0) { goto error; } stream->out_fd = ret; stream->tracefile_size_current = 0; + + if (!stream->metadata_flag) { + ret = index_create_file(stream->chan->pathname, + stream->name, stream->uid, stream->gid, + stream->chan->tracefile_size, + stream->tracefile_count_current); + if (ret < 0) { + goto error; + } + stream->index_fd = ret; + } } ret = 0; @@ -1717,6 +1860,12 @@ int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream) } if (stream->chan->type == CONSUMER_CHANNEL_TYPE_METADATA) { + uint64_t contiguous, pushed; + + /* Ease our life a bit. */ + contiguous = stream->chan->metadata_cache->contiguous; + pushed = stream->ust_metadata_pushed; + /* * We can simply check whether all contiguously available data * has been pushed to the ring buffer, since the push operation @@ -1728,10 +1877,10 @@ int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream) * 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) { + contiguous, pushed); + assert(((int64_t) contiguous - pushed) >= 0); + if ((contiguous != pushed) || + (((int64_t) contiguous - pushed) > 0 || contiguous == 0)) { ret = 1; /* Data is pending */ goto end; } @@ -1767,7 +1916,6 @@ end: */ void lttng_ustconsumer_close_metadata(struct lttng_ht *metadata_ht) { - int ret; struct lttng_ht_iter iter; struct lttng_consumer_stream *stream; @@ -1779,17 +1927,16 @@ void lttng_ustconsumer_close_metadata(struct lttng_ht *metadata_ht) rcu_read_lock(); cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream, node.node) { - int fd = stream->wait_fd; - + pthread_mutex_lock(&stream->chan->lock); /* - * Whatever happens here we have to continue to try to close every - * streams. Let's report at least the error on failure. + * Whatever returned value, we must continue to try to close everything + * so ignore it. */ - ret = ustctl_stream_close_wakeup_fd(stream->ustream); - if (ret) { - ERR("Unable to close metadata stream fd %d ret %d", fd, ret); - } - DBG("Metadata wait fd %d closed", fd); + (void) _close_metadata(stream->chan); + DBG("Metadata wait fd %d and poll pipe fd %d closed", stream->wait_fd, + stream->ust_metadata_poll_pipe[1]); + pthread_mutex_unlock(&stream->chan->lock); + } rcu_read_unlock(); } @@ -1804,8 +1951,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; @@ -1831,12 +1984,18 @@ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx, request.session_id = channel->session_id; request.session_id_per_pid = channel->session_id_per_pid; - request.uid = channel->uid; + /* + * Request the application UID here so the metadata of that application can + * be sent back. The channel UID corresponds to the user UID of the session + * used for the rights on the stream file(s). + */ + request.uid = channel->ust_app_uid; request.key = channel->key; + DBG("Sending metadata request to sessiond, session id %" PRIu64 - ", per-pid %" PRIu64, - channel->session_id, - channel->session_id_per_pid); + ", per-pid %" PRIu64 ", app UID %u and channek key %" PRIu64, + request.session_id, request.session_id_per_pid, request.uid, + request.key); pthread_mutex_lock(&ctx->metadata_socket_lock); ret = lttcomm_send_unix_sock(ctx->consumer_metadata_socket, &request, @@ -1893,7 +2052,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