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 76c0310635a402e78dd37277126d46ca2433cbce..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,68 @@ 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:
+       {
+               DBG("Consumer rename session %" PRIu64 " after rotation",
+                               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.uid,
+                               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;
+               }
+
+               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_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:
@@ -1589,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) {
@@ -1626,9 +1680,17 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
        }
 
 rotate:
-       rotation_ret = lttng_consumer_rotate_stream(ctx, stream);
-       if (rotation_ret < 0) {
-               ERR("Stream rotation error");
+       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;
        }