X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fkernel-consumer%2Fkernel-consumer.c;h=66ae16bc8f09c224ffd3e1130bf9c1fa0ca0fa13;hb=e098433c90550d74288498f8c4474ef4c2daea68;hp=f7704b46564a63273dabe03b9e181b4846ed4292;hpb=62c43103c60bd704cd8ed7acaaa22465802f5673;p=lttng-tools.git diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c index f7704b465..66ae16bc8 100644 --- a/src/common/kernel-consumer/kernel-consumer.c +++ b/src/common/kernel-consumer/kernel-consumer.c @@ -385,7 +385,7 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path, do { health_code_update(); - ret_read = lttng_kconsumer_read_subbuffer(metadata_stream, ctx); + ret_read = lttng_kconsumer_read_subbuffer(metadata_stream, ctx, NULL); if (ret_read < 0) { if (ret_read != -EAGAIN) { ERR("Kernel snapshot reading metadata subbuffer (ret: %zd)", @@ -644,7 +644,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: @@ -771,8 +772,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, goto end_nosignal; } - DBG("Kernel consumer ADD_STREAM %s (fd: %d) with relayd id %" PRIu64, - new_stream->name, fd, new_stream->relayd_stream_id); + DBG("Kernel consumer ADD_STREAM %s (fd: %d) %s with relayd id %" PRIu64, + new_stream->name, fd, new_stream->chan->pathname, new_stream->relayd_stream_id); break; } case LTTNG_CONSUMER_STREAMS_SENT: @@ -1119,6 +1120,42 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } break; } + case LTTNG_CONSUMER_ROTATE_CHANNEL: + { + DBG("Consumer rotate channel %" PRIu64, msg.u.rotate_channel.key); + + /* + * 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; + } + + 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; + } + + /* Rotate the streams that are ready right now. */ + ret = lttng_consumer_rotate_ready_streams( + msg.u.rotate_channel.key, ctx); + if (ret < 0) { + ERR("Rotate ready streams failed"); + ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; + } + + break; + } case LTTNG_CONSUMER_ROTATE_RENAME: { DBG("Consumer rename session %" PRIu64 " after rotation, old path = \"%s\", new path = \"%s\"", @@ -1144,6 +1181,40 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } break; } + case LTTNG_CONSUMER_ROTATE_PENDING_RELAY: + { + uint32_t pending_reply; + + 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; + } else { + pending_reply = !!ret; + } + + 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; + } + + /* Send back returned value to session daemon */ + ret = lttcomm_send_unix_sock(sock, &pending_reply, + sizeof(pending_reply)); + if (ret < 0) { + PERROR("send data pending ret code"); + goto error_fatal; + } + break; + } case LTTNG_CONSUMER_MKDIR: { DBG("Consumer mkdir %s in session %" PRIu64, @@ -1406,16 +1477,30 @@ end: * Consume data on a file descriptor and write it on a trace file. */ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, - struct lttng_consumer_local_data *ctx) + struct lttng_consumer_local_data *ctx, bool *rotated) { unsigned long len, subbuf_size, padding; - int err, write_index = 1; + int err, write_index = 1, rotation_ret; ssize_t ret = 0; int infd = stream->wait_fd; struct ctf_packet_index index; DBG("In read_subbuffer (infd : %d)", infd); + /* + * If the stream was flagged to be ready for rotation before we extract the + * next packet, rotate it now. + */ + if (stream->rotate_ready) { + DBG("Rotate stream before extracting data"); + rotation_ret = lttng_consumer_rotate_stream(ctx, stream, rotated); + if (rotation_ret < 0) { + ERR("Stream rotation error"); + ret = -1; + goto error; + } + } + /* Get the next subbuffer */ err = kernctl_get_next_subbuf(infd); if (err != 0) { @@ -1428,7 +1513,7 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, DBG("Reserving sub buffer failed (everything is normal, " "it is due to concurrency)"); ret = err; - goto end; + goto error; } /* Get the full subbuffer size including padding */ @@ -1444,10 +1529,10 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); } ret = err; - goto end; + goto error; } ret = err; - goto end; + goto error; } if (!stream->metadata_flag) { @@ -1462,9 +1547,9 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); } ret = err; - goto end; + goto error; } - goto end; + goto error; } ret = update_stream_stats(stream); if (ret < 0) { @@ -1477,9 +1562,9 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); } ret = err; - goto end; + goto error; } - goto end; + goto error; } } else { write_index = 0; @@ -1494,9 +1579,9 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); } ret = err; - goto end; + goto error; } - goto end; + goto error; } } @@ -1541,10 +1626,10 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); } ret = err; - goto end; + goto error; } ret = err; - goto end; + goto error; } /* Make sure the tracer is not gone mad on us! */ @@ -1587,12 +1672,12 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, PERROR("Reader has been pushed by the writer, last sub-buffer corrupted."); } ret = err; - goto end; + goto error; } /* Write index if needed. */ if (!write_index) { - goto end; + goto rotate; } if (stream->chan->live_timer_interval && !stream->metadata_flag) { @@ -1616,16 +1701,35 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream, pthread_mutex_unlock(&stream->metadata_timer_lock); } if (err < 0) { - goto end; + goto error; } } err = consumer_stream_write_index(stream, &index); if (err < 0) { - goto end; + goto error; } -end: +rotate: + /* + * After extracting the packet, we check if the stream is now ready to be + * rotated and perform the action immediately. + */ + rotation_ret = lttng_consumer_stream_is_rotate_ready(stream); + if (rotation_ret == 1) { + rotation_ret = lttng_consumer_rotate_stream(ctx, stream, rotated); + if (rotation_ret < 0) { + ERR("Stream rotation error"); + ret = -1; + goto error; + } + } else if (rotation_ret < 0) { + ERR("Checking if stream is ready to rotate"); + ret = -1; + goto error; + } + +error: return ret; }