*/
if (channel->uchan) {
lttng_ustconsumer_del_channel(channel);
+ lttng_ustconsumer_free_channel(channel);
}
free(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];
+ if (channel->monitor) {
+ /* Set metadata poll pipe if we created one */
+ memcpy(stream->ust_metadata_poll_pipe,
+ ust_metadata_pipe,
+ sizeof(ust_metadata_pipe));
+ }
}
}
health_code_update();
- ustctl_flush_buffer(stream->ustream, 1);
+ pthread_mutex_lock(&stream->lock);
+ if (!stream->quiescent) {
+ ustctl_flush_buffer(stream->ustream, 0);
+ stream->quiescent = true;
+ }
+ pthread_mutex_unlock(&stream->lock);
+ }
+error:
+ rcu_read_unlock();
+ return ret;
+}
+
+/*
+ * Clear quiescent state from channel's streams using the given key to
+ * retrieve the channel.
+ *
+ * Return 0 on success else an LTTng error code.
+ */
+static int clear_quiescent_channel(uint64_t chan_key)
+{
+ int ret = 0;
+ struct lttng_consumer_channel *channel;
+ struct lttng_consumer_stream *stream;
+ struct lttng_ht *ht;
+ struct lttng_ht_iter iter;
+
+ DBG("UST consumer clear quiescent channel key %" PRIu64, chan_key);
+
+ rcu_read_lock();
+ channel = consumer_find_channel(chan_key);
+ if (!channel) {
+ ERR("UST consumer clear quiescent channel %" PRIu64 " not found", chan_key);
+ ret = LTTNG_ERR_UST_CHAN_NOT_FOUND;
+ goto error;
+ }
+
+ ht = consumer_data.stream_per_chan_id_ht;
+
+ /* For each stream of the channel id, clear quiescent state. */
+ 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) {
+
+ health_code_update();
+
+ pthread_mutex_lock(&stream->lock);
+ stream->quiescent = false;
+ pthread_mutex_unlock(&stream->lock);
}
error:
rcu_read_unlock();
* Returns 0 on success, < 0 on error
*/
static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id,
- uint64_t max_stream_size, struct lttng_consumer_local_data *ctx)
+ uint64_t nb_packets_per_stream, struct lttng_consumer_local_data *ctx)
{
int ret;
unsigned use_relayd = 0;
}
}
- ustctl_flush_buffer(stream->ustream, 1);
+ /*
+ * If tracing is active, we want to perform a "full" buffer flush.
+ * Else, if quiescent, it has already been done by the prior stop.
+ */
+ if (!stream->quiescent) {
+ ustctl_flush_buffer(stream->ustream, 0);
+ }
ret = lttng_ustconsumer_take_snapshot(stream);
if (ret < 0) {
/*
* 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
+ * the possible size of the snapshot. Also, we assume 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);
+ consumed_pos = consumer_get_consume_start_pos(consumed_pos,
+ produced_pos, nb_packets_per_stream,
+ stream->max_sb_size);
while (consumed_pos < produced_pos) {
ssize_t read_len;
}
/*
- * Receive the metadata updates from the sessiond.
+ * Receive the metadata updates from the sessiond. Supports receiving
+ * overlapping metadata, but is needs to always belong to a contiguous
+ * range starting from 0.
+ * Be careful about the locks held when calling this function: it needs
+ * the metadata cache flush to concurrently progress in order to
+ * complete.
*/
int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset,
uint64_t len, struct lttng_consumer_channel *channel,
goto end_msg_sessiond;
}
+ case LTTNG_CONSUMER_CLEAR_QUIESCENT_CHANNEL:
+ {
+ int ret;
+
+ ret = clear_quiescent_channel(
+ msg.u.clear_quiescent_channel.key);
+ if (ret != 0) {
+ ret_code = ret;
+ }
+
+ goto end_msg_sessiond;
+ }
case LTTNG_CONSUMER_PUSH_METADATA:
{
int ret;
health_code_update();
+ if (!len) {
+ /*
+ * There is nothing to receive. We have simply
+ * checked whether the channel can be found.
+ */
+ ret_code = LTTCOMM_CONSUMERD_SUCCESS;
+ goto end_msg_sessiond;
+ }
+
/* Tell session daemon we are ready to receive the metadata. */
ret = consumer_send_status_msg(sock, LTTCOMM_CONSUMERD_SUCCESS);
if (ret < 0) {
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,
+ msg.u.snapshot_channel.nb_packets_per_stream,
ctx);
if (ret < 0) {
ERR("Snapshot channel failed");
}
/*
- * Called when the stream signal the consumer that it has hang up.
+ * Called when the stream signals the consumer that it has hung up.
*/
void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream)
{
assert(stream);
assert(stream->ustream);
- ustctl_flush_buffer(stream->ustream, 0);
+ pthread_mutex_lock(&stream->lock);
+ if (!stream->quiescent) {
+ ustctl_flush_buffer(stream->ustream, 0);
+ stream->quiescent = true;
+ }
+ pthread_mutex_unlock(&stream->lock);
stream->hangup_flush_done = 1;
}
if (chan->switch_timer_enabled == 1) {
consumer_timer_switch_stop(chan);
}
+}
+
+void lttng_ustconsumer_free_channel(struct lttng_consumer_channel *chan)
+{
+ assert(chan);
+ assert(chan->uchan);
+
consumer_metadata_cache_destroy(chan);
ustctl_destroy_channel(chan->uchan);
}
int ret;
pthread_mutex_lock(&stream->chan->metadata_cache->lock);
- if (stream->chan->metadata_cache->contiguous
+ if (stream->chan->metadata_cache->max_offset
== 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->chan->metadata_cache->max_offset
- stream->ust_metadata_pushed);
assert(write_len != 0);
if (write_len < 0) {
}
stream->ust_metadata_pushed += write_len;
- assert(stream->chan->metadata_cache->contiguous >=
+ assert(stream->chan->metadata_cache->max_offset >=
stream->ust_metadata_pushed);
ret = write_len;
* Sync metadata meaning request them to the session daemon and snapshot to the
* metadata thread can consumer them.
*
- * Metadata stream lock MUST be acquired.
+ * Metadata stream lock is held here, but we need to release it when
+ * interacting with sessiond, else we cause a deadlock with live
+ * awaiting on metadata to be pushed out.
*
* Return 0 if new metadatda is available, EAGAIN if the metadata stream
* is empty or a negative value on error.
assert(ctx);
assert(metadata);
+ pthread_mutex_unlock(&metadata->lock);
/*
* Request metadata from the sessiond, but don't wait for the flush
* because we locked the metadata thread.
if (ret < 0) {
goto end;
}
+ pthread_mutex_lock(&metadata->lock);
ret = commit_one_metadata_packet(metadata);
if (ret <= 0) {
goto end;
}
- ret = ustctl_put_next_subbuf(ustream);
+ ret = ustctl_put_subbuf(ustream);
assert(!ret);
/* This stream still has data. Flag it and wake up the data thread. */
/*
* In live, block until all the metadata is sent.
*/
+ pthread_mutex_lock(&stream->metadata_timer_lock);
+ assert(!stream->missed_metadata_flush);
+ stream->waiting_on_metadata = true;
+ pthread_mutex_unlock(&stream->metadata_timer_lock);
+
err = consumer_stream_sync_metadata(ctx, stream->session_id);
+
+ pthread_mutex_lock(&stream->metadata_timer_lock);
+ stream->waiting_on_metadata = false;
+ if (stream->missed_metadata_flush) {
+ stream->missed_metadata_flush = false;
+ pthread_mutex_unlock(&stream->metadata_timer_lock);
+ (void) consumer_flush_ust_index(stream);
+ } else {
+ pthread_mutex_unlock(&stream->metadata_timer_lock);
+ }
+
if (err < 0) {
goto end;
}
uint64_t contiguous, pushed;
/* Ease our life a bit. */
- contiguous = stream->chan->metadata_cache->contiguous;
+ contiguous = stream->chan->metadata_cache->max_offset;
pushed = stream->ust_metadata_pushed;
/*
* function or any of its callees. Timers have a very strict locking
* semantic with respect to teardown. Failure to respect this semantic
* introduces deadlocks.
+ *
+ * DON'T hold the metadata lock when calling this function, else this
+ * can cause deadlock involving consumer awaiting for metadata to be
+ * pushed out due to concurrent interaction with the session daemon.
*/
int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx,
struct lttng_consumer_channel *channel, int timer, int wait)
projects / lttng-tools.git / blobdiff