Fix: kernel consumer: data_pending check if endpoint active

[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 5c0dce45d27e87337c3c3190551522f122c4b49f..4f3e2871daeaa1c66faadaf64a754dccf0711b94 100644 (file)
--- a/src/common/kernel-consumer/kernel-consumer.c
+++ b/src/common/kernel-consumer/kernel-consumer.c
@@ -110,7 +110,8 @@ 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;
@@ -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;
@@ -656,6 +666,8 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                        consumer_stream_free(new_stream);
                        goto end_nosignal;
                }
+               /* Successfully sent to the right thread. */
+               new_stream->globally_visible = 1;
 
                DBG("Kernel consumer ADD_STREAM %s (fd: %d) with relayd id %" PRIu64,
                                new_stream->name, fd, new_stream->relayd_stream_id);
@@ -737,7 +749,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;
@@ -993,6 +1007,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. */