Re: [Qemu-devel] [PATCH v3 07/27] migration: introduce a new QEMUFile impl based on QIOChannel

[Dr. David Alan Gilbert]

* Daniel P. Berrange (address@hidden) wrote:
> Introduce a new QEMUFile implementation that is based on
> the QIOChannel objects. This impl is different from existing
> impls in that there is no file descriptor that can be made
> available, as some channels may be based on higher level
> protocols such as TLS.
> 
> Although the QIOChannel based implementation can trivially
> provide a bi-directional stream, initially we have separate
> functions for opening input & output directions to fit with
> the expectation of the current QEMUFile interface.
> 
> Signed-off-by: Daniel P. Berrange <address@hidden>
> ---
>  include/migration/qemu-file.h |   4 +
>  migration/Makefile.objs       |   1 +
>  migration/qemu-file-channel.c | 204 
> ++++++++++++++++++++++++++++++++++++++++++
>  3 files changed, 209 insertions(+)
>  create mode 100644 migration/qemu-file-channel.c
> 
> diff --git a/include/migration/qemu-file.h b/include/migration/qemu-file.h
> index 2dea81f..0329ccc 100644
> --- a/include/migration/qemu-file.h
> +++ b/include/migration/qemu-file.h
> @@ -23,7 +23,9 @@
>   */
>  #ifndef QEMU_FILE_H
>  #define QEMU_FILE_H 1
> +#include "qemu-common.h"
>  #include "exec/cpu-common.h"
> +#include "io/channel.h"
>  
>  
>  /* This function writes a chunk of data to a file at the given position.
> @@ -135,6 +137,8 @@ QEMUFile *qemu_fopen_ops(void *opaque, const QEMUFileOps 
> *ops);
>  QEMUFile *qemu_fopen(const char *filename, const char *mode);
>  QEMUFile *qemu_fdopen(int fd, const char *mode);
>  QEMUFile *qemu_fopen_socket(int fd, const char *mode);
> +QEMUFile *qemu_fopen_channel_input(QIOChannel *ioc);
> +QEMUFile *qemu_fopen_channel_output(QIOChannel *ioc);
>  QEMUFile *qemu_popen_cmd(const char *command, const char *mode);
>  QEMUFile *qemu_bufopen(const char *mode, QEMUSizedBuffer *input);
>  void qemu_file_set_hooks(QEMUFile *f, const QEMUFileHooks *hooks);
> diff --git a/migration/Makefile.objs b/migration/Makefile.objs
> index 0cac6d7..b357e2f 100644
> --- a/migration/Makefile.objs
> +++ b/migration/Makefile.objs
> @@ -1,6 +1,7 @@
>  common-obj-y += migration.o tcp.o
>  common-obj-y += vmstate.o
>  common-obj-y += qemu-file.o qemu-file-buf.o qemu-file-unix.o 
> qemu-file-stdio.o
> +common-obj-y += qemu-file-channel.o
>  common-obj-y += xbzrle.o postcopy-ram.o
>  
>  common-obj-$(CONFIG_RDMA) += rdma.o
> diff --git a/migration/qemu-file-channel.c b/migration/qemu-file-channel.c
> new file mode 100644
> index 0000000..10990e8
> --- /dev/null
> +++ b/migration/qemu-file-channel.c
> @@ -0,0 +1,204 @@
> +/*
> + * QEMUFile backend for QIOChannel objects
> + *
> + * Copyright (c) 2015-2016 Red Hat, Inc
> + *
> + * Permission is hereby granted, free of charge, to any person obtaining a 
> copy
> + * of this software and associated documentation files (the "Software"), to 
> deal
> + * in the Software without restriction, including without limitation the 
> rights
> + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
> + * copies of the Software, and to permit persons to whom the Software is
> + * furnished to do so, subject to the following conditions:
> + *
> + * The above copyright notice and this permission notice shall be included in
> + * all copies or substantial portions of the Software.
> + *
> + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
> + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
> + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
> + * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
> + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING 
> FROM,
> + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
> + * THE SOFTWARE.
> + */
> +
> +#include "qemu/osdep.h"
> +#include "migration/qemu-file.h"
> +#include "io/channel-socket.h"
> +#include "qemu/iov.h"
> +
> +
> +/**
> + * @skip: number of bytes to advance head of @iov
> + * @iov: pointer to iov array, updated on success
> + * @iovcnt: number of elements in @iov, updated on success
> + * @oldiov: pointer single element to hold old info from @iov
> + *
> + * This will update @iov so that its head is advanced
> + * by @skip bytes. To do this, zero or more complete
> + * elements of @iov will be skipped over. The new head
> + * of @iov will then have its base & len updated to
> + * skip the remaining number of bytes. @oldiov will
> + * hold the original data from the new head of @iov.
> + */
> +static void channel_skip_iov(size_t skip,
> +                             struct iovec **iov,
> +                             int *iovcnt,
> +                             struct iovec *oldiov)
> +{
> +    ssize_t done = 0;
> +    size_t i;
> +    for (i = 0; i < *iovcnt; i++) {
> +        if ((*iov)[i].iov_len <= skip) {
> +            done += (*iov)[i].iov_len;
> +            skip -= (*iov)[i].iov_len;
> +        } else {
> +            done += skip;
> +            oldiov->iov_base = (*iov)[i].iov_base;
> +            oldiov->iov_len = (*iov)[i].iov_len;
> +            (*iov)[i].iov_len -= skip;
> +            (*iov)[i].iov_base += skip;
> +            *iov = *iov + i;
> +            *iovcnt = *iovcnt - i;
> +            break;

If 'skip' equals the size of data in the original iov,
and that data is held in *iovcnt elements, then I don't
think the else side of the if would be executed, so
you'd never update iov/iovcnt.
I think putting the *iov =   and *iovcnt =   
after the loop would fix that.
(Does it make sense having 'i' as a size_t if *iovcnt is int ?)

However, I don't think you ever hit that case in your use of it
below, so fix it if you need to resend it, but otherwise:

Reviewed-by: Dr. David Alan Gilbert <address@hidden>

Dave

> +        }
> +    }
> +}
> +
> +static ssize_t channel_writev_buffer(void *opaque,
> +                                     struct iovec *iov,
> +                                     int iovcnt,
> +                                     int64_t pos)
> +{
> +    QIOChannel *ioc = QIO_CHANNEL(opaque);
> +    ssize_t done = 0;
> +    ssize_t want = iov_size(iov, iovcnt);
> +    struct iovec oldiov = { NULL, 0 };
> +
> +    while (done < want) {
> +        ssize_t len;
> +        struct iovec *cur = iov;
> +        int curcnt = iovcnt;
> +
> +        channel_skip_iov(done, &cur, &curcnt, &oldiov);
> +
> +        len = qio_channel_writev(ioc, cur, curcnt, NULL);
> +        if (oldiov.iov_base) {
> +            /* Restore original caller's info in @iov */
> +            cur[0].iov_base = oldiov.iov_base;
> +            cur[0].iov_len = oldiov.iov_len;
> +            oldiov.iov_base = NULL;
> +            oldiov.iov_len = 0;
> +        }
> +        if (len == QIO_CHANNEL_ERR_BLOCK) {
> +            qio_channel_wait(ioc, G_IO_OUT);
> +            continue;
> +        }
> +        if (len < 0) {
> +            /* XXX handle Error objects */
> +            return -EIO;
> +        }
> +
> +        done += len;
> +    }
> +    return done;
> +}
> +
> +
> +static ssize_t channel_get_buffer(void *opaque,
> +                                  uint8_t *buf,
> +                                  int64_t pos,
> +                                  size_t size)
> +{
> +    QIOChannel *ioc = QIO_CHANNEL(opaque);
> +    ssize_t ret;
> +
> +    do {
> +        ret = qio_channel_read(ioc, (char *)buf, size, NULL);
> +        if (ret < 0) {
> +            if (ret == QIO_CHANNEL_ERR_BLOCK) {
> +                qio_channel_yield(ioc, G_IO_IN);
> +            } else {
> +                /* XXX handle Error * object */
> +                return -EIO;
> +            }
> +        }
> +    } while (ret == QIO_CHANNEL_ERR_BLOCK);
> +
> +    return ret;
> +}
> +
> +
> +static int channel_close(void *opaque)
> +{
> +    QIOChannel *ioc = QIO_CHANNEL(opaque);
> +    qio_channel_close(ioc, NULL);
> +    object_unref(OBJECT(ioc));
> +    return 0;
> +}
> +
> +
> +static int channel_shutdown(void *opaque,
> +                            bool rd,
> +                            bool wr)
> +{
> +    QIOChannel *ioc = QIO_CHANNEL(opaque);
> +
> +    if (qio_channel_has_feature(ioc,
> +                                QIO_CHANNEL_FEATURE_SHUTDOWN)) {
> +        QIOChannelShutdown mode;
> +        if (rd && wr) {
> +            mode = QIO_CHANNEL_SHUTDOWN_BOTH;
> +        } else if (rd) {
> +            mode = QIO_CHANNEL_SHUTDOWN_READ;
> +        } else {
> +            mode = QIO_CHANNEL_SHUTDOWN_WRITE;
> +        }
> +        if (qio_channel_shutdown(ioc, mode, NULL) < 0) {
> +            /* XXX handler Error * object */
> +            return -EIO;
> +        }
> +    }
> +    return 0;
> +}
> +
> +
> +static int channel_set_blocking(void *opaque,
> +                                bool enabled)
> +{
> +    QIOChannel *ioc = QIO_CHANNEL(opaque);
> +
> +    if (qio_channel_set_blocking(ioc, enabled, NULL) < 0) {
> +        return -1;
> +    }
> +    return 0;
> +}
> +
> +
> +static const QEMUFileOps channel_input_ops = {
> +    .get_buffer = channel_get_buffer,
> +    .close = channel_close,
> +    .shut_down = channel_shutdown,
> +    .set_blocking = channel_set_blocking,
> +};
> +
> +
> +static const QEMUFileOps channel_output_ops = {
> +    .writev_buffer = channel_writev_buffer,
> +    .close = channel_close,
> +    .shut_down = channel_shutdown,
> +    .set_blocking = channel_set_blocking,
> +};
> +
> +
> +QEMUFile *qemu_fopen_channel_input(QIOChannel *ioc)
> +{
> +    object_ref(OBJECT(ioc));
> +    return qemu_fopen_ops(ioc, &channel_input_ops);
> +}
> +
> +QEMUFile *qemu_fopen_channel_output(QIOChannel *ioc)
> +{
> +    object_ref(OBJECT(ioc));
> +    return qemu_fopen_ops(ioc, &channel_output_ops);
> +}
> -- 
> 2.5.0
> 
--
Dr. David Alan Gilbert / address@hidden / Manchester, UK