*/
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);
/* 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];
}
}
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);
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 = 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;
+ 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;
}
}
}
pthread_mutex_lock(&consumer_data.lock);
+ pthread_mutex_lock(&channel->lock);
if (cds_lfht_is_node_deleted(&channel->node.node)) {
goto error_unlock;
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;
}
/* Send metadata stream to relayd if needed. */
- ret = consumer_send_relayd_stream(metadata->metadata_stream,
- metadata->pathname);
- 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);
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;
}
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);
* 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;
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;
* and ultimately try to get rid of this global consumer data lock.
*/
pthread_mutex_lock(&consumer_data.lock);
-
+ pthread_mutex_lock(&channel->lock);
pthread_mutex_lock(&channel->metadata_cache->lock);
ret = consumer_metadata_cache_write(channel, offset, len, metadata_str);
if (ret < 0) {
* waiting for the metadata cache to be flushed.
*/
pthread_mutex_unlock(&channel->metadata_cache->lock);
+ pthread_mutex_unlock(&channel->lock);
pthread_mutex_unlock(&consumer_data.lock);
goto end_free;
}
pthread_mutex_unlock(&channel->metadata_cache->lock);
+ pthread_mutex_unlock(&channel->lock);
pthread_mutex_unlock(&consumer_data.lock);
while (consumer_metadata_cache_flushed(channel, offset + len)) {
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");
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,
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;
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) {
ret = 0;
end:
+ pthread_mutex_unlock(&ctx->metadata_socket_lock);
return ret;
}
projects / lttng-tools.git / blobdiff