*/
static struct lttng_consumer_stream *allocate_stream(int cpu, int key,
struct lttng_consumer_channel *channel,
- struct lttng_consumer_local_data *ctx, int *_alloc_ret)
+ struct lttng_consumer_local_data *ctx, int *_alloc_ret,
+ uint64_t trace_archive_id)
{
int alloc_ret;
struct lttng_consumer_stream *stream = NULL;
cpu,
&alloc_ret,
channel->type,
- channel->monitor);
+ channel->monitor,
+ trace_archive_id);
if (stream == NULL) {
switch (alloc_ret) {
case -ENOENT:
* Return 0 on success else a negative value.
*/
static int create_ust_streams(struct lttng_consumer_channel *channel,
- struct lttng_consumer_local_data *ctx)
+ struct lttng_consumer_local_data *ctx,
+ uint64_t trace_archive_id)
{
int ret, cpu = 0;
struct ustctl_consumer_stream *ustream;
}
/* Allocate consumer stream object. */
- stream = allocate_stream(cpu, wait_fd, channel, ctx, &ret);
+ stream = allocate_stream(cpu, wait_fd, channel, ctx, &ret,
+ trace_archive_id);
if (!stream) {
goto error_alloc;
}
*/
static int ask_channel(struct lttng_consumer_local_data *ctx,
struct lttng_consumer_channel *channel,
- struct ustctl_consumer_channel_attr *attr)
+ struct ustctl_consumer_channel_attr *attr,
+ uint64_t trace_archive_id)
{
int ret;
}
/* Open all streams for this channel. */
- ret = create_ust_streams(channel, ctx);
+ ret = create_ust_streams(channel, ctx, trace_archive_id);
if (ret < 0) {
goto end;
}
/*
* Snapshot the whole metadata.
+ * RCU read-side lock must be held across this function to ensure existence of
+ * metadata_channel.
*
* Returns 0 on success, < 0 on error
*/
-static int snapshot_metadata(uint64_t key, char *path, uint64_t relayd_id,
- struct lttng_consumer_local_data *ctx)
+static int snapshot_metadata(struct lttng_consumer_channel *metadata_channel,
+ uint64_t key, char *path, uint64_t relayd_id,
+ struct lttng_consumer_local_data *ctx,
+ uint64_t trace_archive_id)
{
int ret = 0;
- struct lttng_consumer_channel *metadata_channel;
struct lttng_consumer_stream *metadata_stream;
assert(path);
rcu_read_lock();
- metadata_channel = consumer_find_channel(key);
- if (!metadata_channel) {
- ERR("UST snapshot metadata channel not found for key %" PRIu64,
- key);
- ret = -1;
- goto error;
- }
assert(!metadata_channel->monitor);
health_code_update();
* The metadata stream is NOT created in no monitor mode when the channel
* is created on a sessiond ask channel command.
*/
- ret = create_ust_streams(metadata_channel, ctx);
+ ret = create_ust_streams(metadata_channel, ctx, trace_archive_id);
if (ret < 0) {
goto error;
}
/*
* Take a snapshot of all the stream of a channel.
+ * RCU read-side lock must be held across this function to ensure existence of
+ * channel.
*
* Returns 0 on success, < 0 on error
*/
-static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id,
- uint64_t nb_packets_per_stream, struct lttng_consumer_local_data *ctx)
+static int snapshot_channel(struct lttng_consumer_channel *channel,
+ uint64_t key, char *path, uint64_t relayd_id,
+ uint64_t nb_packets_per_stream,
+ struct lttng_consumer_local_data *ctx)
{
int ret;
unsigned use_relayd = 0;
unsigned long consumed_pos, produced_pos;
- struct lttng_consumer_channel *channel;
struct lttng_consumer_stream *stream;
assert(path);
use_relayd = 1;
}
- channel = consumer_find_channel(key);
- if (!channel) {
- ERR("UST snapshot channel not found for key %" PRIu64, key);
- ret = -1;
- goto error;
- }
assert(!channel->monitor);
DBG("UST consumer snapshot channel %" PRIu64, key);
consumer_stream_close(stream);
error_unlock:
pthread_mutex_unlock(&stream->lock);
-error:
rcu_read_unlock();
return ret;
}
health_code_update();
- ret = ask_channel(ctx, channel, &attr);
+ ret = ask_channel(ctx, channel, &attr,
+ msg.u.ask_channel.trace_archive_id);
if (ret < 0) {
goto end_channel_error;
}
}
case LTTNG_CONSUMER_SNAPSHOT_CHANNEL:
{
- if (msg.u.snapshot_channel.metadata) {
- ret = snapshot_metadata(msg.u.snapshot_channel.key,
- msg.u.snapshot_channel.pathname,
- msg.u.snapshot_channel.relayd_id,
- ctx);
- if (ret < 0) {
- ERR("Snapshot metadata failed");
- ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA;
- }
+ struct lttng_consumer_channel *channel;
+ uint64_t key = msg.u.snapshot_channel.key;
+
+ channel = consumer_find_channel(key);
+ if (!channel) {
+ DBG("UST snapshot channel not found for key %" PRIu64, key);
+ ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
} else {
- ret = snapshot_channel(msg.u.snapshot_channel.key,
- msg.u.snapshot_channel.pathname,
- msg.u.snapshot_channel.relayd_id,
- msg.u.snapshot_channel.nb_packets_per_stream,
- ctx);
- if (ret < 0) {
- ERR("Snapshot channel failed");
- ret_code = LTTCOMM_CONSUMERD_CHANNEL_FAIL;
+ if (msg.u.snapshot_channel.metadata) {
+ ret = snapshot_metadata(channel, key,
+ msg.u.snapshot_channel.pathname,
+ msg.u.snapshot_channel.relayd_id,
+ ctx,
+ msg.u.snapshot_channel.trace_archive_id);
+ if (ret < 0) {
+ ERR("Snapshot metadata failed");
+ ret_code = LTTCOMM_CONSUMERD_SNAPSHOT_FAILED;
+ }
+ } else {
+ ret = snapshot_channel(channel, key,
+ msg.u.snapshot_channel.pathname,
+ msg.u.snapshot_channel.relayd_id,
+ msg.u.snapshot_channel.nb_packets_per_stream,
+ ctx);
+ if (ret < 0) {
+ ERR("Snapshot channel failed");
+ ret_code = LTTCOMM_CONSUMERD_SNAPSHOT_FAILED;
+ }
}
}
-
health_code_update();
ret = consumer_send_status_msg(sock, ret_code);
if (ret < 0) {
}
goto end_msg_sessiond;
}
- case LTTNG_CONSUMER_SET_CHANNEL_ROTATE_PIPE:
- {
- int channel_rotate_pipe;
- int flags;
-
- ret_code = LTTCOMM_CONSUMERD_SUCCESS;
- /* Successfully received the command's type. */
- ret = consumer_send_status_msg(sock, ret_code);
- if (ret < 0) {
- goto error_fatal;
- }
-
- ret = lttcomm_recv_fds_unix_sock(sock, &channel_rotate_pipe, 1);
- if (ret != sizeof(channel_rotate_pipe)) {
- ERR("Failed to receive channel rotate pipe");
- goto error_fatal;
- }
-
- DBG("Received channel rotate pipe (%d)", channel_rotate_pipe);
- ctx->channel_rotate_pipe = channel_rotate_pipe;
- /* Set the pipe as non-blocking. */
- ret = fcntl(channel_rotate_pipe, F_GETFL, 0);
- if (ret == -1) {
- PERROR("fcntl get flags of the channel rotate pipe");
- goto error_fatal;
- }
- flags = ret;
-
- ret = fcntl(channel_rotate_pipe, F_SETFL, flags | O_NONBLOCK);
- if (ret == -1) {
- PERROR("fcntl set O_NONBLOCK flag of the channel rotate pipe");
- goto error_fatal;
- }
- DBG("Channel rotate pipe set as non-blocking");
- ret_code = LTTCOMM_CONSUMERD_SUCCESS;
- ret = consumer_send_status_msg(sock, ret_code);
- if (ret < 0) {
- goto error_fatal;
- }
- break;
- }
case LTTNG_CONSUMER_ROTATE_CHANNEL:
{
- /*
- * Sample the rotate position of all the streams in this channel.
- */
- 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");
- ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
- }
+ struct lttng_consumer_channel *channel;
+ uint64_t key = msg.u.rotate_channel.key;
- health_code_update();
+ channel = consumer_find_channel(key);
+ if (!channel) {
+ DBG("Channel %" PRIu64 " not found", key);
+ ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
+ } else {
+ /*
+ * Sample the rotate position of all the streams in
+ * this channel.
+ */
+ ret = lttng_consumer_rotate_channel(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");
+ ret_code = LTTCOMM_CONSUMERD_ROTATION_FAIL;
+ }
+ health_code_update();
+ }
ret = consumer_send_status_msg(sock, ret_code);
if (ret < 0) {
/* Somehow, the session daemon is not responding anymore. */
* handle this, but it needs to be after the
* consumer_send_status_msg() call.
*/
- ret = lttng_consumer_rotate_ready_streams(
- msg.u.rotate_channel.key, ctx);
- if (ret < 0) {
- ERR("Rotate channel failed");
- ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
+ if (channel) {
+ ret = lttng_consumer_rotate_ready_streams(
+ channel, key, ctx);
+ if (ret < 0) {
+ ERR("Rotate channel failed");
+ }
}
break;
}
msg.u.rotate_rename.relayd_id);
if (ret < 0) {
ERR("Rotate rename failed");
- ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL;
+ ret_code = LTTCOMM_CONSUMERD_ROTATE_RENAME_FAILED;
}
health_code_update();
}
break;
}
- case LTTNG_CONSUMER_ROTATE_PENDING_RELAY:
+ case LTTNG_CONSUMER_CHECK_ROTATION_PENDING_LOCAL:
{
- uint32_t pending;
-
- DBG("Consumer rotate pending on relay for session %" PRIu64,
- msg.u.rotate_pending_relay.session_id);
- pending = 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);
+ int pending;
+ uint32_t pending_reply;
+
+ DBG("Perform local check of pending rotation for session id %" PRIu64,
+ msg.u.check_rotation_pending_local.session_id);
+ pending = lttng_consumer_check_rotation_pending_local(
+ msg.u.check_rotation_pending_local.session_id,
+ msg.u.check_rotation_pending_local.chunk_id);
if (pending < 0) {
- ERR("Rotate pending relay failed");
- ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL;
+ ERR("Local rotation pending check failed with code %i", pending);
+ ret_code = LTTCOMM_CONSUMERD_ROTATION_PENDING_LOCAL_FAILED;
+ } else {
+ pending_reply = !!pending;
+ }
+
+ 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;
+ }
+
+ if (pending < 0) {
+ /*
+ * An error occured while running the command;
+ * don't send the 'pending' flag as the sessiond
+ * will not read it.
+ */
+ break;
+ }
+
+ /* Send back returned value to session daemon */
+ ret = lttcomm_send_unix_sock(sock, &pending_reply,
+ sizeof(pending_reply));
+ if (ret < 0) {
+ PERROR("Failed to send rotation pending return code");
+ goto error_fatal;
+ }
+ break;
+ }
+ case LTTNG_CONSUMER_CHECK_ROTATION_PENDING_RELAY:
+ {
+ int pending;
+ uint32_t pending_reply;
+
+ DBG("Perform relayd check of pending rotation for session id %" PRIu64,
+ msg.u.check_rotation_pending_relay.session_id);
+ pending = lttng_consumer_check_rotation_pending_relay(
+ msg.u.check_rotation_pending_relay.session_id,
+ msg.u.check_rotation_pending_relay.relayd_id,
+ msg.u.check_rotation_pending_relay.chunk_id);
+ if (pending < 0) {
+ ERR("Relayd rotation pending check failed with code %i", pending);
+ ret_code = LTTCOMM_CONSUMERD_ROTATION_PENDING_RELAY_FAILED;
+ } else {
+ pending_reply = !!pending;
}
health_code_update();
goto end_nosignal;
}
+ if (pending < 0) {
+ /*
+ * An error occured while running the command;
+ * don't send the 'pending' flag as the sessiond
+ * will not read it.
+ */
+ break;
+ }
+
/* Send back returned value to session daemon */
- ret = lttcomm_send_unix_sock(sock, &pending, sizeof(pending));
+ ret = lttcomm_send_unix_sock(sock, &pending_reply,
+ sizeof(pending_reply));
if (ret < 0) {
- PERROR("send data pending ret code");
+ PERROR("Failed to send rotation pending return code");
goto error_fatal;
}
break;
msg.u.mkdir.relayd_id);
if (ret < 0) {
ERR("consumer mkdir failed");
- ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL;
+ ret_code = LTTCOMM_CONSUMERD_MKDIR_FAILED;
}
health_code_update();
request.key = channel->key;
DBG("Sending metadata request to sessiond, session id %" PRIu64
- ", per-pid %" PRIu64 ", app UID %u and channek key %" PRIu64,
+ ", per-pid %" PRIu64 ", app UID %u and channel key %" PRIu64,
request.session_id, request.session_id_per_pid, request.uid,
request.key);
projects / lttng-tools.git / blobdiff