X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fcommon%2Fkernel-consumer%2Fkernel-consumer.c;h=e5595a1c3c106973bbbd285997612b98d422bca0;hb=8265f19e4002c6826f9b96e55075d3382602e0c3;hp=4618ccedcd750d18d651c4ace97477dd620f9843;hpb=d3e2ba59faddb31870e2ce29b6a881f7ad5ad883;p=lttng-tools.git diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c index 4618ccedc..e5595a1c3 100644 --- a/src/common/kernel-consumer/kernel-consumer.c +++ b/src/common/kernel-consumer/kernel-consumer.c @@ -29,11 +29,13 @@ #include #include +#include #include #include #include #include #include +#include #include #include #include @@ -139,6 +141,9 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, } cds_list_for_each_entry(stream, &channel->streams.head, send_node) { + + health_code_update(); + /* * Lock stream because we are about to change its state. */ @@ -172,6 +177,13 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, DBG("Kernel consumer snapshot stream %s/%s (%" PRIu64 ")", path, stream->name, stream->key); } + if (relayd_id != -1ULL) { + ret = consumer_send_relayd_streams_sent(relayd_id); + if (ret < 0) { + ERR("sending streams sent to relayd"); + goto end_unlock; + } + } ret = kernctl_buffer_flush(stream->wait_fd); if (ret < 0) { @@ -221,6 +233,8 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path, ssize_t read_len; unsigned long len, padded_len; + health_code_update(); + DBG("Kernel consumer taking snapshot at pos %lu", consumed_pos); ret = kernctl_get_subbuf(stream->wait_fd, &consumed_pos); @@ -362,6 +376,8 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, } do { + health_code_update(); + ret_read = lttng_kconsumer_read_subbuffer(metadata_stream, ctx); if (ret_read < 0) { if (ret_read != -EAGAIN) { @@ -410,9 +426,11 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, int sock, struct pollfd *consumer_sockpoll) { ssize_t ret; - enum lttng_error_code ret_code = LTTNG_OK; + enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS; struct lttcomm_consumer_msg msg; + health_code_update(); + ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg)); if (ret != sizeof(msg)) { if (ret > 0) { @@ -421,17 +439,13 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } return ret; } - if (msg.cmd_type == LTTNG_CONSUMER_STOP) { - /* - * Notify the session daemon that the command is completed. - * - * On transport layer error, the function call will print an error - * message so handling the returned code is a bit useless since we - * return an error code anyway. - */ - (void) consumer_send_status_msg(sock, ret_code); - return -ENOENT; - } + + health_code_update(); + + /* Deprecated command */ + assert(msg.cmd_type != LTTNG_CONSUMER_STOP); + + health_code_update(); /* relayd needs RCU read-side protection */ rcu_read_lock(); @@ -451,12 +465,17 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, struct lttng_consumer_channel *new_channel; int ret_recv; + health_code_update(); + /* First send a status message before receiving the fds. */ ret = consumer_send_status_msg(sock, ret_code); if (ret < 0) { /* Somehow, the session daemon is not responding anymore. */ goto error_fatal; } + + health_code_update(); + DBG("consumer_add_channel %" PRIu64, msg.u.channel.channel_key); new_channel = consumer_allocate_channel(msg.u.channel.channel_key, msg.u.channel.session_id, msg.u.channel.pathname, @@ -494,6 +513,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, goto end_nosignal; }; + health_code_update(); + if (ctx->on_recv_channel != NULL) { ret_recv = ctx->on_recv_channel(new_channel); if (ret_recv == 0) { @@ -504,7 +525,12 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } else { ret = consumer_add_channel(new_channel, ctx); } - consumer_timer_live_start(new_channel, msg.u.channel.live_timer_interval); + if (CONSUMER_CHANNEL_TYPE_DATA) { + consumer_timer_live_start(new_channel, + msg.u.channel.live_timer_interval); + } + + health_code_update(); /* If we received an error in add_channel, we need to report it. */ if (ret < 0) { @@ -536,26 +562,35 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, * happens while tearing down. */ ERR("Unable to find channel key %" PRIu64, msg.u.stream.channel_key); - ret_code = LTTNG_ERR_KERN_CHAN_NOT_FOUND; + ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; } + health_code_update(); + /* First send a status message before receiving the fds. */ ret = consumer_send_status_msg(sock, ret_code); if (ret < 0) { /* Somehow, the session daemon is not responding anymore. */ goto error_fatal; } - if (ret_code != LTTNG_OK) { + + health_code_update(); + + if (ret_code != LTTCOMM_CONSUMERD_SUCCESS) { /* Channel was not found. */ goto end_nosignal; } /* Blocking call */ - if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) { - rcu_read_unlock(); - return -EINTR; + health_poll_entry(); + ret = lttng_consumer_poll_socket(consumer_sockpoll); + health_poll_exit(); + if (ret) { + goto error_fatal; } + health_code_update(); + /* Get stream file descriptor from socket */ ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1); if (ret != sizeof(fd)) { @@ -564,6 +599,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, return ret; } + health_code_update(); + /* * Send status code to session daemon only if the recv works. If the * above recv() failed, the session daemon is notified through the @@ -575,6 +612,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, goto end_nosignal; } + health_code_update(); + new_stream = consumer_allocate_stream(channel->key, fd, LTTNG_CONSUMER_ACTIVE_STREAM, @@ -632,6 +671,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, */ new_stream->hangup_flush_done = 0; + health_code_update(); + if (ctx->on_recv_stream) { ret = ctx->on_recv_stream(new_stream); if (ret < 0) { @@ -640,6 +681,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } } + health_code_update(); + if (new_stream->metadata_flag) { channel->metadata_stream = new_stream; } @@ -687,6 +730,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, /* Vitible to other threads */ new_stream->globally_visible = 1; + health_code_update(); + ret = lttng_pipe_write(stream_pipe, &new_stream, sizeof(new_stream)); if (ret < 0) { ERR("Consumer write %s stream to pipe %d", @@ -704,6 +749,57 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, new_stream->name, fd, new_stream->relayd_stream_id); break; } + case LTTNG_CONSUMER_STREAMS_SENT: + { + struct lttng_consumer_channel *channel; + + /* + * Get stream's channel reference. Needed when adding the stream to the + * global hash table. + */ + channel = consumer_find_channel(msg.u.sent_streams.channel_key); + if (!channel) { + /* + * We could not find the channel. Can happen if cpu hotplug + * happens while tearing down. + */ + ERR("Unable to find channel key %" PRIu64, + msg.u.sent_streams.channel_key); + ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; + } + + health_code_update(); + + /* + * Send status code to session daemon. + */ + ret = consumer_send_status_msg(sock, ret_code); + if (ret < 0) { + /* Somehow, the session daemon is not responding anymore. */ + goto end_nosignal; + } + + health_code_update(); + + /* + * We should not send this message if we don't monitor the + * streams in this channel. + */ + if (!channel->monitor) { + break; + } + + health_code_update(); + /* Send stream to relayd if the stream has an ID. */ + if (msg.u.sent_streams.net_seq_idx != (uint64_t) -1ULL) { + ret = consumer_send_relayd_streams_sent( + msg.u.sent_streams.net_seq_idx); + if (ret < 0) { + goto end_nosignal; + } + } + break; + } case LTTNG_CONSUMER_UPDATE_STREAM: { rcu_read_unlock(); @@ -720,7 +816,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, relayd = consumer_find_relayd(index); if (relayd == NULL) { DBG("Unable to find relayd %" PRIu64, index); - ret_code = LTTNG_ERR_NO_CONSUMER; + ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL; } /* @@ -737,6 +833,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, consumer_flag_relayd_for_destroy(relayd); } + health_code_update(); + ret = consumer_send_status_msg(sock, ret_code); if (ret < 0) { /* Somehow, the session daemon is not responding anymore. */ @@ -754,6 +852,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, ret = consumer_data_pending(id); + health_code_update(); + /* Send back returned value to session daemon */ ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret)); if (ret < 0) { @@ -775,7 +875,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, msg.u.snapshot_channel.relayd_id, ctx); if (ret < 0) { ERR("Snapshot metadata failed"); - ret_code = LTTNG_ERR_KERN_META_FAIL; + ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA; } } else { ret = lttng_kconsumer_snapshot_channel(msg.u.snapshot_channel.key, @@ -785,10 +885,12 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, ctx); if (ret < 0) { ERR("Snapshot channel failed"); - ret_code = LTTNG_ERR_KERN_CHAN_FAIL; + 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. */ @@ -804,15 +906,24 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, channel = consumer_find_channel(key); if (!channel) { ERR("Kernel consumer destroy channel %" PRIu64 " not found", key); - ret_code = LTTNG_ERR_KERN_CHAN_NOT_FOUND; + 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; } + health_code_update(); + + /* Stop right now if no channel was found. */ + if (!channel) { + goto end_nosignal; + } + /* * This command should ONLY be issued for channel with streams set in * no monitor mode. @@ -840,6 +951,7 @@ end_nosignal: * Return 1 to indicate success since the 0 value can be a socket * shutdown during the recv() or send() call. */ + health_code_update(); return 1; error_fatal: @@ -853,7 +965,7 @@ error_fatal: * * Return 0 on success or else a negative value. */ -static int get_index_values(struct lttng_packet_index *index, int infd) +static int get_index_values(struct ctf_packet_index *index, int infd) { int ret; @@ -902,6 +1014,42 @@ static int get_index_values(struct lttng_packet_index *index, int infd) error: return ret; } +/* + * Sync metadata meaning request them to the session daemon and snapshot to the + * metadata thread can consumer them. + * + * Metadata stream lock MUST be acquired. + * + * Return 0 if new metadatda is available, EAGAIN if the metadata stream + * is empty or a negative value on error. + */ +int lttng_kconsumer_sync_metadata(struct lttng_consumer_stream *metadata) +{ + int ret; + + assert(metadata); + + ret = kernctl_buffer_flush(metadata->wait_fd); + if (ret < 0) { + ERR("Failed to flush kernel stream"); + goto end; + } + + ret = kernctl_snapshot(metadata->wait_fd); + if (ret < 0) { + if (errno != EAGAIN) { + ERR("Sync metadata, taking kernel snapshot failed."); + goto end; + } + DBG("Sync metadata, no new kernel metadata"); + /* No new metadata, exit. */ + ret = ENODATA; + goto end; + } + +end: + return ret; +} /* * Consume data on a file descriptor and write it on a trace file. @@ -913,7 +1061,7 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, int err, write_index = 1; ssize_t ret = 0; int infd = stream->wait_fd; - struct lttng_packet_index index; + struct ctf_packet_index index; DBG("In read_subbuffer (infd : %d)", infd); @@ -936,6 +1084,17 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, err = kernctl_get_padded_subbuf_size(infd, &len); if (err != 0) { perror("Getting sub-buffer len failed."); + err = kernctl_put_subbuf(infd); + if (err != 0) { + if (errno == EFAULT) { + perror("Error in unreserving sub buffer\n"); + } else if (errno == EIO) { + /* Should never happen with newer LTTng versions */ + perror("Reader has been pushed by the writer, last sub-buffer corrupted."); + } + ret = -errno; + goto end; + } ret = -errno; goto end; } @@ -943,6 +1102,17 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, if (!stream->metadata_flag) { ret = get_index_values(&index, infd); if (ret < 0) { + err = kernctl_put_subbuf(infd); + if (err != 0) { + if (errno == EFAULT) { + perror("Error in unreserving sub buffer\n"); + } else if (errno == EIO) { + /* Should never happen with newer LTTng versions */ + perror("Reader has been pushed by the writer, last sub-buffer corrupted."); + } + ret = -errno; + goto end; + } goto end; } } else { @@ -981,6 +1151,17 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, err = kernctl_get_subbuf_size(infd, &subbuf_size); if (err != 0) { perror("Getting sub-buffer len failed."); + err = kernctl_put_subbuf(infd); + if (err != 0) { + if (errno == EFAULT) { + perror("Error in unreserving sub buffer\n"); + } else if (errno == EIO) { + /* Should never happen with newer LTTng versions */ + perror("Reader has been pushed by the writer, last sub-buffer corrupted."); + } + ret = -errno; + goto end; + } ret = -errno; goto end; } @@ -1002,9 +1183,10 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, (ret != len && stream->net_seq_idx == (uint64_t) -1ULL)) { /* * Display the error but continue processing to try to release the - * subbuffer + * subbuffer. This is a DBG statement since this is possible to + * happen without being a critical error. */ - ERR("Error writing to tracefile " + DBG("Error writing to tracefile " "(ret: %zd != len: %lu != subbuf_size: %lu)", ret, len, subbuf_size); write_index = 0; @@ -1032,6 +1214,16 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, goto end; } + if (stream->chan->live_timer_interval && !stream->metadata_flag) { + /* + * In live, block until all the metadata is sent. + */ + err = consumer_stream_sync_metadata(ctx, stream->session_id); + if (err < 0) { + goto end; + } + } + err = consumer_stream_write_index(stream, &index); if (err < 0) { goto end;