fix sampling positions after the put_next + fix data_rotate_pipe handling

[lttng-tools.git] / src / common / kernel-consumer / kernel-consumer.c

diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c
index 2f15f6a25067493be6b0e9ec22ef73d33debcf61..f3d074c7c606748f133719a5163d341ebcc04a30 100644 (file)
--- a/src/common/kernel-consumer/kernel-consumer.c
+++ b/src/common/kernel-consumer/kernel-consumer.c
@@ -170,7 +170,8 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path,
                stream->net_seq_idx = relayd_id;
                channel->relayd_id = relayd_id;
                if (relayd_id != (uint64_t) -1ULL) {
-                       ret = consumer_send_relayd_stream(stream, path);
+                       ret = consumer_send_relayd_stream(stream, path,
+                                       LTTNG_DOMAIN_KERNEL);
                        if (ret < 0) {
                                ERR("sending stream to relayd");
                                goto end_unlock;
@@ -367,7 +368,8 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path,
        }
 
        if (use_relayd) {
-               ret = consumer_send_relayd_stream(metadata_stream, path);
+               ret = consumer_send_relayd_stream(metadata_stream, path,
+                               LTTNG_DOMAIN_KERNEL);
                if (ret < 0) {
                        goto error;
                }
@@ -723,7 +725,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                /* Send stream to relayd if the stream has an ID. */
                if (new_stream->net_seq_idx != (uint64_t) -1ULL) {
                        ret = consumer_send_relayd_stream(new_stream,
-                                       new_stream->chan->pathname);
+                                       new_stream->chan->pathname,
+                                       LTTNG_DOMAIN_KERNEL);
                        if (ret < 0) {
                                consumer_stream_free(new_stream);
                                goto end_nosignal;
@@ -1138,6 +1141,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                                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");
@@ -1152,19 +1156,14 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                        goto end_nosignal;
                }
 
-               /*
-                * Rotate the streams that are ready right now.
-                * FIXME: this is a second consecutive iteration over the
-                * streams in a channel, there is probably a better way to
-                * handle this, but it needs to be after the
-                * consumer_send_status_msg() call.
-                */
+               /* 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 channel failed");
+                       ERR("Rotate ready streams failed");
                        ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
                }
+
                break;
        }
        case LTTNG_CONSUMER_ROTATE_RENAME:
@@ -1173,9 +1172,9 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                                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.create,
                                msg.u.rotate_rename.uid,
-                               msg.u.rotate_rename.gid);
+                               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;
@@ -1188,7 +1187,38 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                        /* Somehow, the session daemon is not responding anymore. */
                        goto end_nosignal;
                }
+               break;
+       }
+       case LTTNG_CONSUMER_ROTATE_PENDING_RELAY:
+       {
+               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);
+               if (ret < 0) {
+                       ERR("Rotate pending relay 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;
+               }
+
+               /* Send back returned value to session daemon */
+               ret = lttcomm_send_unix_sock(sock, &pending, sizeof(pending));
+               if (ret < 0) {
+                       PERROR("send data pending ret code");
+                       goto error_fatal;
+               }
+               break;
        }
        default:
                goto end_nosignal;
@@ -1432,7 +1462,7 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
                struct lttng_consumer_local_data *ctx)
 {
        unsigned long len, subbuf_size, padding;
-       int err, write_index = 1, rotation_ret, rotate_ready;
+       int err, write_index = 1, rotation_ret;
        ssize_t ret = 0;
        int infd = stream->wait_fd;
        struct ctf_packet_index index;
@@ -1473,24 +1503,6 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
                goto error;
        }
 
-       rotate_ready = lttng_consumer_stream_is_rotate_ready(stream, len);
-       if (rotate_ready < 0) {
-               ERR("Failed to check if stream is ready for rotation");
-               err = kernctl_put_subbuf(infd);
-               if (err != 0) {
-                       if (err == -EFAULT) {
-                               PERROR("Error in unreserving sub buffer\n");
-                       } else if (err == -EIO) {
-                               /* Should never happen with newer LTTng versions */
-                               PERROR("Reader has been pushed by the writer, last sub-buffer corrupted.");
-                       }
-                       ret = err;
-                       goto error;
-               }
-               ret = -1;
-               goto error;
-       }
-
        if (!stream->metadata_flag) {
                ret = get_index_values(&index, infd);
                if (ret < 0) {
@@ -1630,6 +1642,7 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
                ret = err;
                goto error;
        }
+       fprintf(stderr, "consumer read stream %lu done\n", stream->key);
 
        /* Write index if needed. */
        if (!write_index) {
@@ -1667,13 +1680,18 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
        }
 
 rotate:
-       if (rotate_ready) {
+       rotation_ret = lttng_consumer_stream_is_rotate_ready(stream);
+       if (rotation_ret == 1) {
                rotation_ret = lttng_consumer_rotate_stream(ctx, stream);
                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: