}
goto error;
}
-
+ consumer_stream_copy_ro_channel_values(stream, channel);
stream->chan = channel;
error:
health_code_update();
/* Try to send the stream to the relayd if one is available. */
- ret = consumer_send_relayd_stream(stream, stream->chan->pathname);
+ ret = consumer_send_relayd_stream(stream, stream->chan->pathname,
+ LTTNG_DOMAIN_UST);
if (ret < 0) {
/*
* Flag that the relayd was the problem here probably due to a
/* Send metadata stream to relayd if needed. */
if (metadata->metadata_stream->net_seq_idx != (uint64_t) -1ULL) {
ret = consumer_send_relayd_stream(metadata->metadata_stream,
- metadata->pathname);
+ metadata->pathname, LTTNG_DOMAIN_UST);
if (ret < 0) {
ret = LTTCOMM_CONSUMERD_ERROR_METADATA;
goto error;
if (relayd_id != (uint64_t) -1ULL) {
metadata_stream->net_seq_idx = relayd_id;
- ret = consumer_send_relayd_stream(metadata_stream, path);
+ ret = consumer_send_relayd_stream(metadata_stream, path,
+ LTTNG_DOMAIN_UST);
if (ret < 0) {
goto error_stream;
}
stream->net_seq_idx = relayd_id;
if (use_relayd) {
- ret = consumer_send_relayd_stream(stream, path);
+ ret = consumer_send_relayd_stream(stream, path,
+ LTTNG_DOMAIN_UST);
if (ret < 0) {
goto error_unlock;
}
return ret;
}
-/*
- * When a channel has finished the rotation of all its streams, inform the
- * session daemon.
- */
-static
-int rotate_notify_sessiond(struct lttng_consumer_local_data *ctx,
- uint64_t key)
-{
- int ret;
-
- do {
- ret = write(ctx->channel_rotate_pipe, &key, sizeof(key));
- } while (ret == -1 && errno == EINTR);
- if (ret == -1) {
- PERROR("write to the channel rotate pipe");
- } else {
- DBG("Sent channel rotation notification for channel key %"
- PRIu64, key);
- }
-
- return ret;
-}
-
-/*
- * Performs the stream rotation for the rotate session feature if needed.
- * It must be called with the stream and channel locks held.
- *
- * FIXME: find a way to lock the chan without deadlock. same for kernel.
- *
- * Return 0 on success, a negative number of error.
- */
-static
-int stream_rotation(struct lttng_consumer_local_data *ctx,
- struct lttng_consumer_stream *stream)
-{
- int ret;
- unsigned long consumed_pos;
-
- if (!stream->rotate_position && !stream->rotate_ready) {
- ret = 0;
- goto end;
- }
-
- /*
- * If we don't have the rotate_ready flag, check the consumed position
- * to determine if we need to rotate.
- */
- if (!stream->rotate_ready) {
- ret = lttng_ustconsumer_sample_snapshot_positions(stream);
- if (ret < 0) {
- ERR("Taking UST snapshot positions");
- goto error;
- }
-
- ret = lttng_ustconsumer_get_consumed_snapshot(stream, &consumed_pos);
- if (ret < 0) {
- ERR("Produced UST snapshot position");
- goto error;
- }
-
- fprintf(stderr, "packet %lu, pos %lu\n", stream->key, consumed_pos);
- /* Rotate position not reached yet. */
- if (consumed_pos < stream->rotate_position) {
- ret = 0;
- goto end;
- }
- fprintf(stderr, "Rotate position %lu (expected %lu) reached for stream %lu\n",
- consumed_pos, stream->rotate_position, stream->key);
- } else {
- fprintf(stderr, "Rotate position reached for stream %lu\n",
- stream->key);
- }
-
- ret = close(stream->out_fd);
- if (ret < 0) {
- PERROR("Closing tracefile");
- goto error;
- }
-
- fprintf(stderr, "Rotating stream %lu to %s/%s\n", stream->key,
- stream->chan->pathname, stream->name);
- ret = utils_create_stream_file(stream->chan->pathname, stream->name,
- stream->chan->tracefile_size, stream->tracefile_count_current,
- stream->uid, stream->gid, NULL);
- if (ret < 0) {
- goto error;
- }
- stream->out_fd = ret;
- stream->tracefile_size_current = 0;
-
- if (!stream->metadata_flag) {
- struct lttng_index_file *index_file;
-
- lttng_index_file_put(stream->index_file);
-
- index_file = lttng_index_file_create(stream->chan->pathname,
- stream->name, stream->uid, stream->gid,
- stream->chan->tracefile_size,
- stream->tracefile_count_current,
- CTF_INDEX_MAJOR, CTF_INDEX_MINOR);
- if (!index_file) {
- goto error;
- }
- stream->index_file = index_file;
- stream->out_fd_offset = 0;
- } else {
- /*
- * Reset the position pushed from the metadata cache so it
- * will write from the beginning on the next push.
- */
- stream->ust_metadata_pushed = 0;
- /*
- * Wakeup the metadata thread so it dumps the metadata cache
- * to file again.
- */
- consumer_metadata_wakeup_pipe(stream->chan);
- }
-
- stream->rotate_position = 0;
- stream->rotate_ready = 0;
-
- if (--stream->chan->nr_stream_rotate_pending == 0) {
- rotate_notify_sessiond(ctx, stream->chan->key);
- fprintf(stderr, "SENT %lu\n", stream->chan->key);
- }
-
- ret = 0;
- goto end;
-
-error:
- ret = -1;
-end:
- return ret;
-}
-
-/*
- * Sample the rotate position for all the streams of a channel.
- *
- * Returns 0 on success, < 0 on error
- */
-static
-int lttng_ustconsumer_rotate_channel(uint64_t key, char *path,
- uint64_t relayd_id, uint32_t metadata,
- struct lttng_consumer_local_data *ctx)
-{
- int ret;
- struct lttng_consumer_channel *channel;
- struct lttng_consumer_stream *stream;
- struct lttng_ht_iter iter;
- struct lttng_ht *ht = consumer_data.stream_per_chan_id_ht;
-
- /* FIXME: metadata param is useless */
-
- DBG("UST consumer sample rotate position for channel %" PRIu64, key);
-
- rcu_read_lock();
-
- channel = consumer_find_channel(key);
- if (!channel) {
- ERR("No channel found for key %" PRIu64, key);
- ret = -1;
- goto end;
- }
- pthread_mutex_lock(&channel->lock);
- snprintf(channel->pathname, PATH_MAX, "%s", path);
-
- 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) {
- uint64_t consumed_pos;
- health_code_update();
-
- /*
- * Lock stream because we are about to change its state.
- */
- pthread_mutex_lock(&stream->lock);
- ret = lttng_ustconsumer_sample_snapshot_positions(stream);
- if (ret < 0) {
- ERR("Taking UST snapshot positions");
- goto end_unlock;
- }
-
- ret = lttng_ustconsumer_get_produced_snapshot(stream,
- &stream->rotate_position);
- if (ret < 0) {
- ERR("Produced UST snapshot position");
- goto end_unlock;
- }
- fprintf(stderr, "Stream %lu should rotate after %lu to %s\n",
- stream->key, stream->rotate_position,
- channel->pathname);
- lttng_ustconsumer_get_consumed_snapshot(stream,
- &consumed_pos);
- fprintf(stderr, "consumed %lu\n", consumed_pos);
- if (consumed_pos == stream->rotate_position) {
- stream->rotate_ready = 1;
- fprintf(stderr, "Stream %lu ready to rotate to %s\n",
- stream->key, channel->pathname);
- }
- channel->nr_stream_rotate_pending++;
-
- ustctl_flush_buffer(stream->ustream, 1);
-
- pthread_mutex_unlock(&stream->lock);
- }
-
- ret = 0;
- goto end_unlock_channel;
-
-end_unlock:
- pthread_mutex_unlock(&stream->lock);
-end_unlock_channel:
- pthread_mutex_unlock(&channel->lock);
-end:
- rcu_read_unlock();
- return ret;
-}
-
-/*
- * Rotate all the ready streams.
- *
- * This is especially important for low throughput streams that have already
- * been consumed, we cannot wait for their next packet to perform the
- * rotation.
- *
- * Returns 0 on success, < 0 on error
- */
-static
-int lttng_ustconsumer_rotate_ready_streams(uint64_t key,
- struct lttng_consumer_local_data *ctx)
-{
- int ret;
- struct lttng_consumer_channel *channel;
- struct lttng_consumer_stream *stream;
- struct lttng_ht_iter iter;
- struct lttng_ht *ht = consumer_data.stream_per_chan_id_ht;
-
- rcu_read_lock();
-
- channel = consumer_find_channel(key);
- if (!channel) {
- ERR("No channel found for key %" PRIu64, key);
- ret = -1;
- goto end;
- }
- 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();
-
- /*
- * Lock stream because we are about to change its state.
- */
- pthread_mutex_lock(&stream->lock);
- if (stream->rotate_ready == 0) {
- pthread_mutex_unlock(&stream->lock);
- continue;
- }
- ret = stream_rotation(ctx, stream);
- if (ret < 0) {
- pthread_mutex_unlock(&stream->lock);
- ERR("Stream rotation error");
- goto end;
- }
-
- pthread_mutex_unlock(&stream->lock);
- }
-
- ret = 0;
-
-end:
- rcu_read_unlock();
- return ret;
-}
-
/*
* Receive the metadata updates from the sessiond. Supports receiving
* overlapping metadata, but is needs to always belong to a contiguous
}
case LTTNG_CONSUMER_ROTATE_CHANNEL:
{
- ret = lttng_ustconsumer_rotate_channel(msg.u.rotate_channel.key,
+ ret = lttng_consumer_rotate_channel(msg.u.rotate_channel.key,
msg.u.rotate_channel.pathname,
msg.u.rotate_channel.relayd_id,
msg.u.rotate_channel.metadata,
+ msg.u.rotate_channel.new_chunk_id,
ctx);
if (ret < 0) {
ERR("Rotate channel failed");
* handle this, but it needs to be after the
* consumer_send_status_msg() call.
*/
- ret = lttng_ustconsumer_rotate_ready_streams(
+ ret = lttng_consumer_rotate_ready_streams(
msg.u.rotate_channel.key, ctx);
if (ret < 0) {
ERR("Rotate channel failed");
}
break;
}
+ case LTTNG_CONSUMER_ROTATE_RENAME:
+ {
+ DBG("Consumer rename session %" PRIu64 " after rotation",
+ msg.u.rotate_rename.session_id);
+ ret = lttng_consumer_rotate_rename(msg.u.rotate_rename.current_path,
+ msg.u.rotate_rename.new_path,
+ msg.u.rotate_rename.uid,
+ msg.u.rotate_rename.gid,
+ msg.u.rotate_rename.relayd_id);
+ if (ret < 0) {
+ ERR("Rotate rename failed");
+ ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
+ }
+
+ health_code_update();
+
+ ret = consumer_send_status_msg(sock, ret_code);
+ if (ret < 0) {
+ /* Somehow, the session daemon is not responding anymore. */
+ goto end_nosignal;
+ }
+
+ }
+ case LTTNG_CONSUMER_ROTATE_PENDING_RELAY:
+ {
+ DBG("Consumer rotate pending on relay for session %" PRIu64,
+ msg.u.rotate_pending_relay.session_id);
+ ret = lttng_consumer_rotate_pending_relay(
+ msg.u.rotate_pending_relay.session_id,
+ msg.u.rotate_pending_relay.relayd_id,
+ msg.u.rotate_pending_relay.chunk_id);
+ if (ret < 0) {
+ ERR("Rotate pending relay failed");
+ ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
+ }
+
+ health_code_update();
+ ret_code = ret;
+
+ ret = consumer_send_status_msg(sock, ret_code);
+ if (ret < 0) {
+ /* Somehow, the session daemon is not responding anymore. */
+ goto end_nosignal;
+ }
+
+ }
default:
break;
}
return ustctl_get_mmap_base(stream->ustream);
}
+void lttng_ustctl_flush_buffer(struct lttng_consumer_stream *stream,
+ int producer_active)
+{
+ assert(stream);
+ assert(stream->ustream);
+
+ ustctl_flush_buffer(stream->ustream, producer_active);
+}
+
/*
* Take a snapshot for a specific stream.
*
struct lttng_consumer_local_data *ctx)
{
unsigned long len, subbuf_size, padding;
- int err, write_index = 1, rotation_ret;
+ int err, write_index = 1, rotation_ret, rotate_ready;
long ret = 0;
struct ustctl_consumer_stream *ustream;
struct ctf_packet_index index;
readlen = lttng_read(stream->wait_fd, &dummy, 1);
if (readlen < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {
ret = readlen;
- goto end;
+ goto error;
}
}
if (stream->metadata_flag) {
ret = commit_one_metadata_packet(stream);
if (ret <= 0) {
- goto end;
+ goto error;
}
ustctl_flush_buffer(stream->ustream, 1);
goto retry;
*/
DBG("Reserving sub buffer failed (everything is normal, "
"it is due to concurrency) [ret: %d]", err);
- goto end;
+ goto error;
}
assert(stream->chan->output == CONSUMER_CHANNEL_MMAP);
if (ret < 0) {
err = ustctl_put_subbuf(ustream);
assert(err == 0);
- goto end;
+ goto error;
}
/* Update the stream's sequence and discarded events count. */
PERROR("kernctl_get_events_discarded");
err = ustctl_put_subbuf(ustream);
assert(err == 0);
- goto end;
+ goto error;
}
} else {
write_index = 0;
assert(len >= subbuf_size);
padding = len - subbuf_size;
+
+ rotate_ready = lttng_consumer_stream_is_rotate_ready(stream, len);
+ if (rotate_ready < 0) {
+ ERR("Failed to check if stream is ready for rotation");
+ ret = -1;
+ err = ustctl_put_subbuf(ustream);
+ assert(err == 0);
+ goto error;
+ }
+ stream->rotate_ready = rotate_ready;
+
/* write the subbuffer to the tracefile */
ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size, padding, &index);
/*
if (!stream->metadata_flag) {
ret = notify_if_more_data(stream, ctx);
if (ret < 0) {
- goto end;
+ goto error;
}
}
/* Write index if needed. */
if (!write_index) {
- goto end;
+ goto rotate;
}
if (stream->chan->live_timer_interval && !stream->metadata_flag) {
}
if (err < 0) {
- goto end;
+ goto error;
}
}
assert(!stream->metadata_flag);
err = consumer_stream_write_index(stream, &index);
if (err < 0) {
- goto end;
+ goto error;
}
-end:
- /* FIXME: do we need this lock, it causes deadlocks when called
- * at the same time with lttng_ustconsumer_rotate_channel ? */
-// pthread_mutex_lock(&stream->chan->lock);
- rotation_ret = stream_rotation(ctx, stream);
- if (rotation_ret < 0) {
-// pthread_mutex_unlock(&stream->chan->lock);
- ret = -1;
- ERR("Stream rotation error");
- goto end;
+rotate:
+ if (stream->rotate_ready) {
+ rotation_ret = lttng_consumer_rotate_stream(ctx, stream);
+ if (rotation_ret < 0) {
+ ret = -1;
+ ERR("Stream rotation error");
+ goto error;
+ }
}
-// pthread_mutex_unlock(&stream->chan->lock);
+error:
return ret;
}
projects / deliverable / lttng-tools.git / blobdiff