Cleanup mmap consuming

Only one mmap call per buffer, once the first mmap is done, we use the
offset to know where to start consuming the data.

When we receive the FD and its output mode is mmap, we mmap the buffer,
when it is closing, we munmap it.

Removed the manual padding handling because the kernel already exposes a
padded subbuffer.

Also updated the check of the return code of
kconsumerd_consumerd_recv_fd to be consistent with the rest of the code,
we are in error when it's < 0 not <= 0.

Acked-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
Signed-off-by: Julien Desfossez <julien.desfossez@polymtl.ca>

diff --git a/include/lttng-sessiond-comm.h b/include/lttng-sessiond-comm.h
index be903a6ce3f9f66c909864f1cfbaf2a75743e2a7..f57b0cd5f51bad2de8701b506b013dbe3d791406 100644 (file)
--- a/include/lttng-sessiond-comm.h
+++ b/include/lttng-sessiond-comm.h
@@ -193,6 +193,7 @@ struct lttcomm_kconsumerd_msg {
        int fd;
        uint32_t state;    /* enum lttcomm_kconsumerd_fd_state */
        unsigned long max_sb_size; /* the subbuffer size for this channel */
        int fd;
        uint32_t state;    /* enum lttcomm_kconsumerd_fd_state */
        unsigned long max_sb_size; /* the subbuffer size for this channel */

+       enum lttng_event_output output; /* use splice or mmap to consume this fd */

 };
 
 extern int lttcomm_create_unix_sock(const char *pathname);
 };
 
 extern int lttcomm_create_unix_sock(const char *pathname);

diff --git a/liblttkconsumerd/lttkconsumerd.c b/liblttkconsumerd/lttkconsumerd.c
index d4908d16c9c43bd01023698d91e76efab34ebb88..29f1f95d566e7befe7bcc06e5dde1fbcd9f01117 100644 (file)
--- a/liblttkconsumerd/lttkconsumerd.c
+++ b/liblttkconsumerd/lttkconsumerd.c
@@ -127,11 +127,18 @@ static int kconsumerd_find_session_fd(int fd)
  */
 static void kconsumerd_del_fd(struct kconsumerd_fd *lcf)
 {
  */
 static void kconsumerd_del_fd(struct kconsumerd_fd *lcf)
 {

+       int ret;

        pthread_mutex_lock(&kconsumerd_data.lock);
        cds_list_del(&lcf->list);
        if (kconsumerd_data.fds_count > 0) {
                kconsumerd_data.fds_count--;
                if (lcf != NULL) {
        pthread_mutex_lock(&kconsumerd_data.lock);
        cds_list_del(&lcf->list);
        if (kconsumerd_data.fds_count > 0) {
                kconsumerd_data.fds_count--;
                if (lcf != NULL) {

+                       if (lcf->mmap_base != NULL) {
+                               ret = munmap(lcf->mmap_base, lcf->mmap_len);
+                               if (ret != 0) {
+                                       perror("munmap");
+                               }
+                       }

                        if (lcf->out_fd != 0) {
                                close(lcf->out_fd);
                        }
                        if (lcf->out_fd != 0) {
                                close(lcf->out_fd);
                        }


@@ -168,6 +175,9 @@ static int kconsumerd_add_fd(struct lttcomm_kconsumerd_msg *buf, int consumerd_f
        tmp_fd->max_sb_size = buf->max_sb_size;
        tmp_fd->out_fd = 0;
        tmp_fd->out_fd_offset = 0;
        tmp_fd->max_sb_size = buf->max_sb_size;
        tmp_fd->out_fd = 0;
        tmp_fd->out_fd_offset = 0;

+       tmp_fd->mmap_len = 0;
+       tmp_fd->mmap_base = NULL;
+       tmp_fd->output = buf->output;

        strncpy(tmp_fd->path_name, buf->path_name, PATH_MAX);
        tmp_fd->path_name[PATH_MAX - 1] = '\0';
 
        strncpy(tmp_fd->path_name, buf->path_name, PATH_MAX);
        tmp_fd->path_name[PATH_MAX - 1] = '\0';
 


@@ -185,6 +195,24 @@ static int kconsumerd_add_fd(struct lttcomm_kconsumerd_msg *buf, int consumerd_f
                                tmp_fd->sessiond_fd, tmp_fd->consumerd_fd, tmp_fd->out_fd);
        }
 
                                tmp_fd->sessiond_fd, tmp_fd->consumerd_fd, tmp_fd->out_fd);
        }
 

+       if (tmp_fd->output == LTTNG_EVENT_MMAP) {
+               /* get the len of the mmap region */
+               ret = kernctl_get_mmap_len(tmp_fd->consumerd_fd, &tmp_fd->mmap_len);
+               if (ret != 0) {
+                       ret = errno;
+                       perror("kernctl_get_mmap_len");
+                       goto end;
+               }
+
+               tmp_fd->mmap_base = mmap(NULL, tmp_fd->mmap_len,
+                               PROT_READ, MAP_PRIVATE, tmp_fd->consumerd_fd, 0);
+               if (tmp_fd->mmap_base == MAP_FAILED) {
+                       perror("Error mmaping");
+                       ret = -1;
+                       goto end;
+               }
+       }
+

        cds_list_add(&tmp_fd->list, &kconsumerd_data.fd_list.head);
        kconsumerd_data.fds_count++;
        kconsumerd_data.need_update = 1;
        cds_list_add(&tmp_fd->list, &kconsumerd_data.fd_list.head);
        kconsumerd_data.fds_count++;
        kconsumerd_data.need_update = 1;


@@ -259,33 +287,13 @@ static int kconsumerd_update_poll_array(struct kconsumerd_local_data *ctx,
 int kconsumerd_on_read_subbuffer_mmap(struct kconsumerd_local_data *ctx,
                struct kconsumerd_fd *kconsumerd_fd, unsigned long len)
 {
 int kconsumerd_on_read_subbuffer_mmap(struct kconsumerd_local_data *ctx,
                struct kconsumerd_fd *kconsumerd_fd, unsigned long len)
 {

-       unsigned long mmap_len, mmap_offset, padded_len, padding_len;
-       char *mmap_base;
+       unsigned long mmap_offset;

        char *padding = NULL;
        long ret = 0;
        off_t orig_offset = kconsumerd_fd->out_fd_offset;
        int fd = kconsumerd_fd->consumerd_fd;
        int outfd = kconsumerd_fd->out_fd;
 
        char *padding = NULL;
        long ret = 0;
        off_t orig_offset = kconsumerd_fd->out_fd_offset;
        int fd = kconsumerd_fd->consumerd_fd;
        int outfd = kconsumerd_fd->out_fd;
 

-       /* get the padded subbuffer size to know the padding required */
-       ret = kernctl_get_padded_subbuf_size(fd, &padded_len);
-       if (ret != 0) {
-               ret = errno;
-               perror("kernctl_get_padded_subbuf_size");
-               goto end;
-       }
-       padding_len = padded_len - len;
-       padding = malloc(padding_len * sizeof(char));
-       memset(padding, '\0', padding_len);
-
-       /* get the len of the mmap region */
-       ret = kernctl_get_mmap_len(fd, &mmap_len);
-       if (ret != 0) {
-               ret = errno;
-               perror("kernctl_get_mmap_len");
-               goto end;
-       }
-

        /* get the offset inside the fd to mmap */
        ret = kernctl_get_mmap_read_offset(fd, &mmap_offset);
        if (ret != 0) {
        /* get the offset inside the fd to mmap */
        ret = kernctl_get_mmap_read_offset(fd, &mmap_offset);
        if (ret != 0) {


@@ -294,15 +302,8 @@ int kconsumerd_on_read_subbuffer_mmap(struct kconsumerd_local_data *ctx,
                goto end;
        }
 
                goto end;
        }
 

-       mmap_base = mmap(NULL, mmap_len, PROT_READ, MAP_PRIVATE, fd, mmap_offset);
-       if (mmap_base == MAP_FAILED) {
-               perror("Error mmaping");
-               ret = -1;
-               goto end;
-       }
-

        while (len > 0) {
        while (len > 0) {

-               ret = write(outfd, mmap_base, len);
+               ret = write(outfd, kconsumerd_fd->mmap_base + mmap_offset, len);

                if (ret >= len) {
                        len = 0;
                } else if (ret < 0) {
                if (ret >= len) {
                        len = 0;
                } else if (ret < 0) {


@@ -316,14 +317,6 @@ int kconsumerd_on_read_subbuffer_mmap(struct kconsumerd_local_data *ctx,
                kconsumerd_fd->out_fd_offset += ret;
        }
 
                kconsumerd_fd->out_fd_offset += ret;
        }
 

-       /* once all the data is written, write the padding to disk */
-       ret = write(outfd, padding, padding_len);
-       if (ret < 0) {
-               ret = errno;
-               perror("Error writing padding to file");
-               goto end;
-       }
-

        /*
         * This does a blocking write-and-wait on any page that belongs to the
         * subbuffer prior to the one we just wrote.
        /*
         * This does a blocking write-and-wait on any page that belongs to the
         * subbuffer prior to the one we just wrote.


@@ -914,7 +907,7 @@ void *kconsumerd_thread_receive_fds(void *data)
                /* we received a command to add or update fds */
                ret = kconsumerd_consumerd_recv_fd(ctx, sock, kconsumerd_sockpoll,
                                tmp.payload_size, tmp.cmd_type);
                /* we received a command to add or update fds */
                ret = kconsumerd_consumerd_recv_fd(ctx, sock, kconsumerd_sockpoll,
                                tmp.payload_size, tmp.cmd_type);

-               if (ret <= 0) {
+               if (ret < 0) {

                        ERR("Receiving the FD, exiting");
                        goto end;
                }
                        ERR("Receiving the FD, exiting");
                        goto end;
                }

diff --git a/liblttkconsumerd/lttkconsumerd.h b/liblttkconsumerd/lttkconsumerd.h
index 10e4a5504e19148ac7fe69d10783342090019bc4..1dd85d509e8dcf8f476e67ac03076134a812cba8 100644 (file)
--- a/liblttkconsumerd/lttkconsumerd.h
+++ b/liblttkconsumerd/lttkconsumerd.h
@@ -55,6 +55,9 @@ struct kconsumerd_fd {
        char path_name[PATH_MAX]; /* tracefile name */
        enum kconsumerd_fd_state state;
        unsigned long max_sb_size; /* the subbuffer size for this channel */
        char path_name[PATH_MAX]; /* tracefile name */
        enum kconsumerd_fd_state state;
        unsigned long max_sb_size; /* the subbuffer size for this channel */

+       void *mmap_base;
+       size_t mmap_len;
+       enum lttng_event_output output; /* splice or mmap */

 };
 
 struct kconsumerd_local_data {
 };
 
 struct kconsumerd_local_data {

diff --git a/ltt-kconsumerd/ltt-kconsumerd.c b/ltt-kconsumerd/ltt-kconsumerd.c
index 64c5ccfae908499c20eeb59a5cdb305eae342a2c..ceaee716c8e1a8cfc2f331cc0e1c6369f17702f3 100644 (file)
--- a/ltt-kconsumerd/ltt-kconsumerd.c
+++ b/ltt-kconsumerd/ltt-kconsumerd.c
@@ -237,7 +237,7 @@ static int read_subbuffer(struct kconsumerd_fd *kconsumerd_fd)
                        break;
                case LTTNG_EVENT_MMAP:
                        /* read the used subbuffer size */
                        break;
                case LTTNG_EVENT_MMAP:
                        /* read the used subbuffer size */

-                       err = kernctl_get_subbuf_size(infd, &len);
+                       err = kernctl_get_padded_subbuf_size(infd, &len);

                        if (err != 0) {
                                ret = errno;
                                perror("Getting sub-buffer len failed.");
                        if (err != 0) {
                                ret = errno;
                                perror("Getting sub-buffer len failed.");

diff --git a/ltt-sessiond/main.c b/ltt-sessiond/main.c
index 891436fb0ce8576b10033d2d93408d90db1671fe..5a73b68dfed46fb8c1878725b3727bb2c16e2787 100644 (file)
--- a/ltt-sessiond/main.c
+++ b/ltt-sessiond/main.c
@@ -288,6 +288,7 @@ static int send_kconsumerd_channel_fds(int sock, struct ltt_kernel_channel *chan
                        lkm.fd = stream->fd;
                        lkm.state = stream->state;
                        lkm.max_sb_size = channel->channel->attr.subbuf_size;
                        lkm.fd = stream->fd;
                        lkm.state = stream->state;
                        lkm.max_sb_size = channel->channel->attr.subbuf_size;

+                       lkm.output = DEFAULT_KERNEL_CHANNEL_OUTPUT;

                        strncpy(lkm.path_name, stream->pathname, PATH_MAX);
                        lkm.path_name[PATH_MAX - 1] = '\0';
 
                        strncpy(lkm.path_name, stream->pathname, PATH_MAX);
                        lkm.path_name[PATH_MAX - 1] = '\0';
 


@@ -338,6 +339,7 @@ static int send_kconsumerd_fds(int sock, struct ltt_kernel_session *session)
                lkm.fd = session->metadata_stream_fd;
                lkm.state = ACTIVE_FD;
                lkm.max_sb_size = session->metadata->conf->attr.subbuf_size;
                lkm.fd = session->metadata_stream_fd;
                lkm.state = ACTIVE_FD;
                lkm.max_sb_size = session->metadata->conf->attr.subbuf_size;

+               lkm.output = DEFAULT_KERNEL_CHANNEL_OUTPUT;

                strncpy(lkm.path_name, session->metadata->pathname, PATH_MAX);
                lkm.path_name[PATH_MAX - 1] = '\0';
 
                strncpy(lkm.path_name, session->metadata->pathname, PATH_MAX);
                lkm.path_name[PATH_MAX - 1] = '\0';