Fix: format string type mismatch

[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 bdf0eff7c4a2b56a72bdda1f5bd06ea35e213a98..8416bc5a3716fb5400cbeb49ae373858a2137144 100644 (file)
--- a/src/common/kernel-consumer/kernel-consumer.c
+++ b/src/common/kernel-consumer/kernel-consumer.c
@@ -110,20 +110,21 @@ int lttng_kconsumer_get_consumed_snapshot(struct lttng_consumer_stream *stream,
  * Returns 0 on success, < 0 on error
  */
 int lttng_kconsumer_snapshot_channel(uint64_t key, char *path,
-               uint64_t relayd_id, struct lttng_consumer_local_data *ctx)
+               uint64_t relayd_id, uint64_t max_stream_size,
+               struct lttng_consumer_local_data *ctx)
 {
        int ret;
        unsigned long consumed_pos, produced_pos;
        struct lttng_consumer_channel *channel;
        struct lttng_consumer_stream *stream;
 
-       DBG("Kernel consumer snapshot channel %lu", key);
+       DBG("Kernel consumer snapshot channel %" PRIu64, key);
 
        rcu_read_lock();
 
        channel = consumer_find_channel(key);
        if (!channel) {
-               ERR("No channel found for key %lu", key);
+               ERR("No channel found for key %" PRIu64, key);
                ret = -1;
                goto end;
        }
@@ -203,6 +204,15 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path,
                        }
                }
 
+               /*
+                * The original value is sent back if max stream size is larger than
+                * the possible size of the snapshot. Also, we asume that the session
+                * daemon should never send a maximum stream size that is lower than
+                * subbuffer size.
+                */
+               consumed_pos = consumer_get_consumed_maxsize(consumed_pos,
+                               produced_pos, max_stream_size);
+
                while (consumed_pos < produced_pos) {
                        ssize_t read_len;
                        unsigned long len, padded_len;
@@ -260,12 +270,14 @@ int lttng_kconsumer_snapshot_channel(uint64_t key, char *path,
                }
 
                if (relayd_id == (uint64_t) -1ULL) {
-                       ret = close(stream->out_fd);
-                       if (ret < 0) {
-                               PERROR("Kernel consumer snapshot close out_fd");
-                               goto end_unlock;
+                       if (stream->out_fd >= 0) {
+                               ret = close(stream->out_fd);
+                               if (ret < 0) {
+                                       PERROR("Kernel consumer snapshot close out_fd");
+                                       goto end_unlock;
+                               }
+                               stream->out_fd = -1;
                        }
-                       stream->out_fd = -1;
                } else {
                        close_relayd_stream(stream);
                        stream->net_seq_idx = (uint64_t) -1ULL;
@@ -344,7 +356,7 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path,
                ret_read = lttng_kconsumer_read_subbuffer(metadata_stream, ctx);
                if (ret_read < 0) {
                        if (ret_read != -EPERM) {
-                               ERR("Kernel snapshot reading metadata subbuffer (ret: %ld)",
+                               ERR("Kernel snapshot reading metadata subbuffer (ret: %zd)",
                                                ret_read);
                                goto error;
                        }
@@ -357,15 +369,17 @@ int lttng_kconsumer_snapshot_metadata(uint64_t key, char *path,
                close_relayd_stream(metadata_stream);
                metadata_stream->net_seq_idx = (uint64_t) -1ULL;
        } else {
-               ret = close(metadata_stream->out_fd);
-               if (ret < 0) {
-                       PERROR("Kernel consumer snapshot metadata close out_fd");
-                       /*
-                        * Don't go on error here since the snapshot was successful at this
-                        * point but somehow the close failed.
-                        */
+               if (metadata_stream->out_fd >= 0) {
+                       ret = close(metadata_stream->out_fd);
+                       if (ret < 0) {
+                               PERROR("Kernel consumer snapshot metadata close out_fd");
+                               /*
+                                * Don't go on error here since the snapshot was successful at this
+                                * point but somehow the close failed.
+                                */
+                       }
+                       metadata_stream->out_fd = -1;
                }
-               metadata_stream->out_fd = -1;
        }
 
        ret = 0;
@@ -392,8 +406,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 
        ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
        if (ret != sizeof(msg)) {
-               lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
                if (ret > 0) {
+                       lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
                        ret = -1;
                }
                return ret;
@@ -629,7 +643,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, NULL);
+                       ret = consumer_send_relayd_stream(new_stream,
+                                       new_stream->chan->pathname);
                        if (ret < 0) {
                                consumer_stream_free(new_stream);
                                goto end_nosignal;
@@ -638,17 +653,38 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 
                /* Get the right pipe where the stream will be sent. */
                if (new_stream->metadata_flag) {
+                       ret = consumer_add_metadata_stream(new_stream);
+                       if (ret) {
+                               ERR("Consumer add metadata stream %" PRIu64 " failed. Continuing",
+                                               new_stream->key);
+                               consumer_stream_free(new_stream);
+                               goto end_nosignal;
+                       }
                        stream_pipe = ctx->consumer_metadata_pipe;
                } else {
+                       ret = consumer_add_data_stream(new_stream);
+                       if (ret) {
+                               ERR("Consumer add stream %" PRIu64 " failed. Continuing",
+                                               new_stream->key);
+                               consumer_stream_free(new_stream);
+                               goto end_nosignal;
+                       }
                        stream_pipe = ctx->consumer_data_pipe;
                }
 
+               /* Vitible to other threads */
+               new_stream->globally_visible = 1;
+
                ret = lttng_pipe_write(stream_pipe, &new_stream, sizeof(new_stream));
                if (ret < 0) {
                        ERR("Consumer write %s stream to pipe %d",
                                        new_stream->metadata_flag ? "metadata" : "data",
                                        lttng_pipe_get_writefd(stream_pipe));
-                       consumer_stream_free(new_stream);
+                       if (new_stream->metadata_flag) {
+                               consumer_del_stream_for_metadata(new_stream);
+                       } else {
+                               consumer_del_stream_for_data(new_stream);
+                       }
                        goto end_nosignal;
                }
 
@@ -732,7 +768,9 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                } else {
                        ret = lttng_kconsumer_snapshot_channel(msg.u.snapshot_channel.key,
                                        msg.u.snapshot_channel.pathname,
-                                       msg.u.snapshot_channel.relayd_id, ctx);
+                                       msg.u.snapshot_channel.relayd_id,
+                                       msg.u.snapshot_channel.max_stream_size,
+                                       ctx);
                        if (ret < 0) {
                                ERR("Snapshot channel failed");
                                ret_code = LTTNG_ERR_KERN_CHAN_FAIL;
@@ -988,6 +1026,11 @@ int lttng_kconsumer_data_pending(struct lttng_consumer_stream *stream)
 
        assert(stream);
 
+       if (stream->endpoint_status != CONSUMER_ENDPOINT_ACTIVE) {
+               ret = 0;
+               goto end;
+       }
+
        ret = kernctl_get_next_subbuf(stream->wait_fd);
        if (ret == 0) {
                /* There is still data so let's put back this subbuffer. */