Fix: relayd metadata size

[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 8ed279f44e16db7e73a14d8947c9e87b143f6946..22bf1002097a1e33a846f2a5c04d6305f6308d69 100644 (file)
--- a/src/common/kernel-consumer/kernel-consumer.c
+++ b/src/common/kernel-consumer/kernel-consumer.c
@@ -59,6 +59,9 @@ ssize_t lttng_kconsumer_on_read_subbuffer_mmap(
        uint64_t metadata_id;
        struct consumer_relayd_sock_pair *relayd = NULL;
 
+       /* RCU lock for the relayd pointer */
+       rcu_read_lock();
+
        /* Flag that the current stream if set for network streaming. */
        if (stream->net_seq_idx != -1) {
                relayd = consumer_find_relayd(stream->net_seq_idx);
@@ -76,11 +79,10 @@ ssize_t lttng_kconsumer_on_read_subbuffer_mmap(
                goto end;
        }
 
-       /* RCU lock for the relayd pointer */
-       rcu_read_lock();
-
        /* Handle stream on the relayd if the output is on the network */
        if (relayd) {
+               unsigned long netlen = len;
+
                /*
                 * Lock the control socket for the complete duration of the function
                 * since from this point on we will use the socket.
@@ -88,9 +90,10 @@ ssize_t lttng_kconsumer_on_read_subbuffer_mmap(
                if (stream->metadata_flag) {
                        /* Metadata requires the control socket. */
                        pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+                       netlen += sizeof(stream->relayd_stream_id);
                }
 
-               ret = consumer_handle_stream_before_relayd(stream, len);
+               ret = consumer_handle_stream_before_relayd(stream, netlen);
                if (ret >= 0) {
                        /* Use the returned socket. */
                        outfd = ret;
@@ -101,12 +104,12 @@ ssize_t lttng_kconsumer_on_read_subbuffer_mmap(
                                do {
                                        ret = write(outfd, (void *) &metadata_id,
                                                        sizeof(stream->relayd_stream_id));
-                                       if (ret < 0) {
-                                               PERROR("write metadata stream id");
-                                               written = ret;
-                                               goto end;
-                                       }
-                               } while (errno == EINTR);
+                               } while (ret < 0 && errno == EINTR);
+                               if (ret < 0) {
+                                       PERROR("write metadata stream id");
+                                       written = ret;
+                                       goto end;
+                               }
                                DBG("Metadata stream id %zu written before data",
                                                stream->relayd_stream_id);
                                /*
@@ -120,18 +123,15 @@ ssize_t lttng_kconsumer_on_read_subbuffer_mmap(
        }
 
        while (len > 0) {
-               ret = write(outfd, stream->mmap_base + mmap_offset, len);
+               do {
+                       ret = write(outfd, stream->mmap_base + mmap_offset, len);
+               } while (ret < 0 && errno == EINTR);
                if (ret < 0) {
-                       if (errno == EINTR) {
-                               /* restart the interrupted system call */
-                               continue;
-                       } else {
-                               perror("Error in file write");
-                               if (written == 0) {
-                                       written = ret;
-                               }
-                               goto end;
+                       perror("Error in file write");
+                       if (written == 0) {
+                               written = ret;
                        }
+                       goto end;
                } else if (ret > len) {
                        perror("Error in file write");
                        written += ret;
@@ -181,6 +181,9 @@ ssize_t lttng_kconsumer_on_read_subbuffer_splice(
        uint64_t metadata_id;
        struct consumer_relayd_sock_pair *relayd = NULL;
 
+       /* RCU lock for the relayd pointer */
+       rcu_read_lock();
+
        /* Flag that the current stream if set for network streaming. */
        if (stream->net_seq_idx != -1) {
                relayd = consumer_find_relayd(stream->net_seq_idx);
@@ -189,9 +192,6 @@ ssize_t lttng_kconsumer_on_read_subbuffer_splice(
                }
        }
 
-       /* RCU lock for the relayd pointer */
-       rcu_read_lock();
-
        /* Write metadata stream id before payload */
        if (stream->metadata_flag && relayd) {
                /*
@@ -200,17 +200,17 @@ ssize_t lttng_kconsumer_on_read_subbuffer_splice(
                 */
                pthread_mutex_lock(&relayd->ctrl_sock_mutex);
 
+               metadata_id = htobe64(stream->relayd_stream_id);
                do {
-                       metadata_id = htobe64(stream->relayd_stream_id);
                        ret = write(ctx->consumer_thread_pipe[1],
                                        (void *) &metadata_id,
                                        sizeof(stream->relayd_stream_id));
-                       if (ret < 0) {
-                               PERROR("write metadata stream id");
-                               written = ret;
-                               goto end;
-                       }
-               } while (errno == EINTR);
+               } while (ret < 0 && errno == EINTR);
+               if (ret < 0) {
+                       PERROR("write metadata stream id");
+                       written = ret;
+                       goto end;
+               }
                DBG("Metadata stream id %zu written before data",
                                stream->relayd_stream_id);
        }
@@ -377,6 +377,9 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                return -ENOENT;
        }
 
+       /* relayd needs RCU read-side protection */
+       rcu_read_lock();
+
        switch (msg.cmd_type) {
        case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
        {
@@ -581,8 +584,9 @@ end:
         */
        do {
                ret = write(ctx->consumer_poll_pipe[1], "", 1);
-       } while (ret == -1UL && errno == EINTR);
+       } while (ret < 0 && errno == EINTR);
 end_nosignal:
+       rcu_read_unlock();
        return 0;
 }