X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fkernel-consumer%2Fkernel-consumer.c;h=b890416cac1d96db4fe481c27f9f395e4a14707d;hp=7c01bc772a008aa38a05a399c7acadf4849fd0a2;hb=3eb928aad0c1f249830d16d6744a5a23b145a26c;hpb=02d02e31d47c091a38154c9c188c08387902d97b diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c index 7c01bc772..b890416ca 100644 --- a/src/common/kernel-consumer/kernel-consumer.c +++ b/src/common/kernel-consumer/kernel-consumer.c @@ -124,28 +124,22 @@ int lttng_kconsumer_get_consumed_snapshot(struct lttng_consumer_stream *stream, /* * 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 */ -int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, - uint64_t relayd_id, uint64_t nb_packets_per_stream, +int lttng_kconsumer_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; - struct lttng_consumer_channel *channel; struct lttng_consumer_stream *stream; DBG("Kernel consumer snapshot 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; - } - /* Splice is not supported yet for channel snapshot. */ if (channel->output != CONSUMER_CHANNEL_MMAP) { ERR("Unsupported output %d", channel->output); @@ -333,15 +327,17 @@ end: /* * Read the whole metadata available for a snapshot. + * RCU read-side lock must be held across this function to ensure existence of + * metadata_channel. * * Returns 0 on success, < 0 on error */ -int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, - uint64_t relayd_id, struct lttng_consumer_local_data *ctx) +static int lttng_kconsumer_snapshot_metadata(struct lttng_consumer_channel *metadata_channel, + uint64_t key, char *path, uint64_t relayd_id, + struct lttng_consumer_local_data *ctx) { int ret, use_relayd = 0; ssize_t ret_read; - struct lttng_consumer_channel *metadata_channel; struct lttng_consumer_stream *metadata_stream; assert(ctx); @@ -351,15 +347,9 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, rcu_read_lock(); - metadata_channel = consumer_find_channel(key); - if (!metadata_channel) { - ERR("Kernel snapshot metadata not found for key %" PRIu64, key); - ret = -1; - goto error; - } - metadata_stream = metadata_channel->metadata_stream; assert(metadata_stream); + pthread_mutex_lock(&metadata_stream->lock); /* Flag once that we have a valid relayd for the stream. */ if (relayd_id != (uint64_t) -1ULL) { @@ -369,7 +359,7 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, if (use_relayd) { ret = consumer_send_relayd_stream(metadata_stream, path); if (ret < 0) { - goto error; + goto error_snapshot; } } else { ret = utils_create_stream_file(path, metadata_stream->name, @@ -377,7 +367,7 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, metadata_stream->tracefile_count_current, metadata_stream->uid, metadata_stream->gid, NULL); if (ret < 0) { - goto error; + goto error_snapshot; } metadata_stream->out_fd = ret; } @@ -390,7 +380,8 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, if (ret_read != -EAGAIN) { ERR("Kernel snapshot reading metadata subbuffer (ret: %zd)", ret_read); - goto error; + ret = ret_read; + goto error_snapshot; } /* ret_read is negative at this point so we will exit the loop. */ continue; @@ -415,11 +406,11 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, } ret = 0; - +error_snapshot: + pthread_mutex_unlock(&metadata_stream->lock); cds_list_del(&metadata_stream->send_node); consumer_stream_destroy(metadata_stream, NULL); metadata_channel->metadata_stream = NULL; -error: rcu_read_unlock(); return ret; } @@ -644,7 +635,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, msg.u.stream.cpu, &alloc_ret, channel->type, - channel->monitor); + channel->monitor, + msg.u.stream.trace_archive_id); if (new_stream == NULL) { switch (alloc_ret) { case -ENOMEM: @@ -896,26 +888,34 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } case LTTNG_CONSUMER_SNAPSHOT_CHANNEL: { - if (msg.u.snapshot_channel.metadata == 1) { - ret = lttng_kconsumer_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) { + ERR("Channel %" PRIu64 " not found", key); + ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; } else { - ret = lttng_kconsumer_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_CHAN_NOT_FOUND; + if (msg.u.snapshot_channel.metadata == 1) { + ret = lttng_kconsumer_snapshot_metadata(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_SNAPSHOT_FAILED; + } + } else { + ret = lttng_kconsumer_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); @@ -1078,45 +1078,48 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } break; } - case LTTNG_CONSUMER_SET_CHANNEL_ROTATE_PIPE: + case LTTNG_CONSUMER_ROTATE_CHANNEL: { - 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; - } + struct lttng_consumer_channel *channel; + uint64_t key = msg.u.rotate_channel.key; - ret = lttcomm_recv_fds_unix_sock(sock, &channel_rotate_pipe, 1); - if (ret != (ssize_t) sizeof(channel_rotate_pipe)) { - ERR("Failed to receive channel rotate pipe"); - goto error_fatal; - } + DBG("Consumer rotate channel %" PRIu64, key); - 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; + channel = consumer_find_channel(key); + if (!channel) { + ERR("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; + } - 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; + health_code_update(); } - 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; + /* Somehow, the session daemon is not responding anymore. */ + goto end_nosignal; } + if (channel) { + /* Rotate the streams that are ready right now. */ + ret = lttng_consumer_rotate_ready_streams( + channel, key, ctx); + if (ret < 0) { + ERR("Rotate ready streams failed"); + } + } + break; } case LTTNG_CONSUMER_ROTATE_RENAME: @@ -1132,7 +1135,77 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, msg.u.rotate_rename.relayd_id); if (ret < 0) { ERR("Rotate rename failed"); - ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; + ret_code = LTTCOMM_CONSUMERD_ROTATE_RENAME_FAILED; + } + + 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; + } + break; + } + case LTTNG_CONSUMER_CHECK_ROTATION_PENDING_LOCAL: + { + 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("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(); @@ -1142,6 +1215,23 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, /* 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_MKDIR: @@ -1155,7 +1245,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, msg.u.mkdir.relayd_id); if (ret < 0) { ERR("consumer mkdir failed"); - ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; + ret_code = LTTCOMM_CONSUMERD_MKDIR_FAILED; } health_code_update();