X-Git-Url: https://git.liburcu.org/?a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.c;h=2d3d89c7c66e4349aa9e5141a5229ade6ba32451;hb=94d4914075c61cd1ee2ec00d8b61eacff105fc47;hp=c794b93f8e9e9574036bd4cef79468be9a853f1a;hpb=309167d2a6f59d0c8cbf64eb23ba912cdea76a34;p=lttng-tools.git diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c index c794b93f8..2d3d89c7c 100644 --- a/src/common/ust-consumer/ust-consumer.c +++ b/src/common/ust-consumer/ust-consumer.c @@ -117,14 +117,15 @@ static struct lttng_consumer_channel *allocate_channel(uint64_t session_id, const char *pathname, const char *name, uid_t uid, gid_t gid, uint64_t relayd_id, uint64_t key, enum lttng_event_output output, uint64_t tracefile_size, uint64_t tracefile_count, - uint64_t session_id_per_pid, unsigned int monitor) + uint64_t session_id_per_pid, unsigned int monitor, + unsigned int live_timer_interval) { assert(pathname); assert(name); return consumer_allocate_channel(key, session_id, pathname, name, uid, gid, relayd_id, output, tracefile_size, - tracefile_count, session_id_per_pid, monitor); + tracefile_count, session_id_per_pid, monitor, live_timer_interval); } /* @@ -799,7 +800,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, 0); + ret = lttng_ustconsumer_request_metadata(ctx, metadata_channel, 0, 1); if (ret < 0) { goto error; } @@ -834,18 +835,13 @@ static int snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id, metadata_stream->tracefile_size_current = 0; } - pthread_mutex_lock(&metadata_channel->metadata_cache->lock); - do { ret = lttng_consumer_read_subbuffer(metadata_stream, ctx); if (ret < 0) { - goto error_unlock; + goto error_stream; } } while (ret > 0); -error_unlock: - pthread_mutex_unlock(&metadata_channel->metadata_cache->lock); - error_stream: /* * Clean up the stream completly because the next snapshot will use a new @@ -1023,7 +1019,7 @@ error: */ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset, uint64_t len, struct lttng_consumer_channel *channel, - int timer) + int timer, int wait) { int ret, ret_code = LTTNG_OK; char *metadata_str; @@ -1060,6 +1056,9 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset, } pthread_mutex_unlock(&channel->metadata_cache->lock); + if (!wait) { + goto end_free; + } while (consumer_metadata_cache_flushed(channel, offset + len, timer)) { DBG("Waiting for metadata to be flushed"); usleep(DEFAULT_METADATA_AVAILABILITY_WAIT_TIME); @@ -1119,7 +1118,8 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, /* Session daemon status message are handled in the following call. */ ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index, msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll, - &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id); + &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id, + msg.u.relayd_sock.relayd_session_id); goto end_nosignal; } case LTTNG_CONSUMER_DESTROY_RELAYD: @@ -1194,7 +1194,8 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, msg.u.ask_channel.tracefile_size, msg.u.ask_channel.tracefile_count, msg.u.ask_channel.session_id_per_pid, - msg.u.ask_channel.monitor); + msg.u.ask_channel.monitor, + msg.u.ask_channel.live_timer_interval); if (!channel) { goto end_channel_error; } @@ -1250,6 +1251,9 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } consumer_timer_switch_start(channel, attr.switch_timer_interval); attr.switch_timer_interval = 0; + } else { + consumer_timer_live_start(channel, + msg.u.ask_channel.live_timer_interval); } /* @@ -1267,6 +1271,9 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } consumer_metadata_cache_destroy(channel); } + if (channel->live_timer_enabled == 1) { + consumer_timer_live_stop(channel); + } goto end_channel_error; } @@ -1402,7 +1409,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } ret = lttng_ustconsumer_recv_metadata(sock, key, offset, - len, channel, 0); + len, channel, 0, 1); if (ret < 0) { /* error receiving from sessiond */ goto error_fatal; @@ -1657,12 +1664,117 @@ error: return ret; } +/* + * Write up to one packet from the metadata cache to the channel. + * + * Returns the number of bytes pushed in the cache, or a negative value + * on error. + */ +static +int commit_one_metadata_packet(struct lttng_consumer_stream *stream) +{ + ssize_t write_len; + int ret; + + pthread_mutex_lock(&stream->chan->metadata_cache->lock); + 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; + + assert(stream->chan->metadata_cache->contiguous >= + stream->ust_metadata_pushed); + ret = write_len; + +end: + pthread_mutex_unlock(&stream->chan->metadata_cache->lock); + return ret; +} + + +/* + * Sync metadata meaning request them to the session daemon and snapshot to the + * metadata thread can consumer them. + * + * Metadata stream lock MUST be acquired. + * + * Return 0 if new metadatda is available, EAGAIN if the metadata stream + * is empty or a negative value on error. + */ +int lttng_ustconsumer_sync_metadata(struct lttng_consumer_local_data *ctx, + struct lttng_consumer_stream *metadata) +{ + int ret; + int retry = 0; + + assert(ctx); + assert(metadata); + + /* + * Request metadata from the sessiond, but don't wait for the flush + * because we locked the metadata thread. + */ + ret = lttng_ustconsumer_request_metadata(ctx, metadata->chan, 0, 0); + if (ret < 0) { + goto end; + } + + ret = commit_one_metadata_packet(metadata); + if (ret <= 0) { + goto end; + } else if (ret > 0) { + retry = 1; + } + + ustctl_flush_buffer(metadata->ustream, 1); + ret = ustctl_snapshot(metadata->ustream); + if (ret < 0) { + if (errno != EAGAIN) { + ERR("Sync metadata, taking UST snapshot"); + goto end; + } + DBG("No new metadata when syncing them."); + /* No new metadata, exit. */ + ret = ENODATA; + goto end; + } + /* + * After this flush, we still need to extract metadata. + */ + if (retry) { + ret = EAGAIN; + } + +end: + return ret; +} + +/* + * Read subbuffer from the given stream. + * + * Stream lock MUST be acquired. + * + * Return 0 on success else a negative value. + */ int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream, struct lttng_consumer_local_data *ctx) { unsigned long len, subbuf_size, padding; - int err, write_index = 0; + int err, write_index = 1; long ret = 0; char dummy; struct ustctl_consumer_stream *ustream; @@ -1678,11 +1790,6 @@ 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; @@ -1705,25 +1812,10 @@ retry: * already been read. */ if (stream->metadata_flag) { - ssize_t write_len; - - if (stream->chan->metadata_cache->contiguous - == stream->ust_metadata_pushed) { - ret = 0; + ret = commit_one_metadata_packet(stream); + if (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; } @@ -1741,12 +1833,14 @@ retry: } assert(stream->chan->output == CONSUMER_CHANNEL_MMAP); - if (!stream->metadata_flag && write_index) { + if (!stream->metadata_flag) { index.offset = htobe64(stream->out_fd_offset); ret = get_index_values(&index, ustream); if (ret < 0) { goto end; } + } else { + write_index = 0; } /* Get the full padded subbuffer size */ @@ -1786,14 +1880,26 @@ retry: assert(err == 0); /* Write index if needed. */ - if (write_index) { - err = index_write(stream->index_fd, &index, sizeof(index)); + if (!write_index) { + goto end; + } + + if (stream->chan->live_timer_interval && !stream->metadata_flag) { + /* + * In live, block until all the metadata is sent. + */ + err = consumer_stream_sync_metadata(ctx, stream->session_id); if (err < 0) { - ret = -1; goto end; } } + assert(!stream->metadata_flag); + err = consumer_stream_write_index(stream, &index); + if (err < 0) { + goto end; + } + end: return ret; } @@ -1958,7 +2064,7 @@ void lttng_ustconsumer_close_stream_wakeup(struct lttng_consumer_stream *stream) * introduces deadlocks. */ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx, - struct lttng_consumer_channel *channel, int timer) + struct lttng_consumer_channel *channel, int timer, int wait) { struct lttcomm_metadata_request_msg request; struct lttcomm_consumer_msg msg; @@ -2052,7 +2158,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, timer); + key, offset, len, channel, timer, wait); if (ret_code >= 0) { /* * Only send the status msg if the sessiond is alive meaning a positive