[libvirt] [PATCH 13/21] qemu: migration: src: qemuNBDTunnelAcceptAndPipe

[John Ferlan]

On 11/18/2015 01:13 PM, Pavel Boldin wrote:
> Add qemuNBDTunnelAcceptAndPipe function that is called to handle POLLIN
> on the UNIX socket connection from the QEMU's NBD server.
> 
> The function creates a pipe of a remote stream connected to the QEMU
> NBD Unix socket on destination and a local stream connected to
> the incoming connection from the source QEMU's NBD.
> 
> Signed-off-by: Pavel Boldin <pboldin at mirantis.com>
> ---
>  src/qemu/qemu_migration.c | 134 +++++++++++++++++++++++++++++++++++++++++++++-
>  1 file changed, 132 insertions(+), 2 deletions(-)
> 
> diff --git a/src/qemu/qemu_migration.c b/src/qemu/qemu_migration.c
> index 0682fd8..0f35c13 100644
> --- a/src/qemu/qemu_migration.c
> +++ b/src/qemu/qemu_migration.c
> @@ -3987,6 +3987,9 @@ struct _qemuMigrationSpec {
>  
>  #define TUNNEL_SEND_BUF_SIZE 65536
>  
> +typedef struct _qemuMigrationPipe qemuMigrationPipe;
> +typedef qemuMigrationPipe *qemuMigrationPipePtr;
> +
>  typedef struct _qemuMigrationIOThread qemuMigrationIOThread;
>  typedef qemuMigrationIOThread *qemuMigrationIOThreadPtr;
>  struct _qemuMigrationIOThread {
> @@ -3997,9 +4000,124 @@ struct _qemuMigrationIOThread {
>      virError err;
>      int wakeupRecvFD;
>      int wakeupSendFD;
> +    qemuMigrationPipePtr pipes;
> +    virConnectPtr dconn;
> +    unsigned char uuid[VIR_UUID_BUFLEN];
> +};
> +
> +struct _qemuMigrationPipe {
> +    qemuMigrationPipePtr next;
> +    qemuMigrationIOThreadPtr data;
> +    virStreamPtr local;
> +    virStreamPtr remote;
>  };
>  
>  static void
> +qemuMigrationPipeClose(qemuMigrationPipePtr pipe, bool abort)
> +{
> +    virStreamEventUpdateCallback(pipe->local, 0);
> +    virStreamEventUpdateCallback(pipe->remote, 0);
> +
> +    if (abort) {
> +        virStreamAbort(pipe->local);
> +        virStreamAbort(pipe->remote);
> +    } else {
> +        virStreamFinish(pipe->local);
> +        virStreamFinish(pipe->remote);
> +    }
> +
> +    virObjectUnref(pipe->local);
> +    virObjectUnref(pipe->remote);
> +}
> +

Norm seems to be 2 blank lines between functions. There's only one here.

> +static qemuMigrationPipePtr
> +qemuMigrationPipeCreate(virStreamPtr local, virStreamPtr remote)
> +{
> +    qemuMigrationPipePtr pipe = NULL;
> +
> +    if (VIR_ALLOC(pipe) < 0)
> +        goto error;
> +
> +    pipe->local = local;
> +    pipe->remote = remote;
> +
> +    return pipe;
> +
> + error:
> +    virStreamEventRemoveCallback(local);
> +    virStreamEventRemoveCallback(remote);
> +    VIR_FREE(pipe);
> +    return NULL;
> +}
> +

but two here...

> +
> +static int
> +qemuNBDTunnelAcceptAndPipe(qemuMigrationIOThreadPtr data)
> +{
> +    int fd, ret;
> +    virStreamPtr local = NULL, remote = NULL;
> +    qemuMigrationPipePtr pipe = NULL;
> +
> +    while ((fd = accept(data->unixSock, NULL, NULL)) < 0) {
> +        if (errno == EAGAIN || errno == EINTR)
> +            continue;
> +        virReportSystemError(
> +            errno, "%s", _("failed to accept connection from qemu"));
> +        goto abrt;
> +    }
> +
> +    if (!(local = virStreamNew(data->dconn, VIR_STREAM_NONBLOCK)))
> +        goto abrt;
> +
> +    if (!(remote = virStreamNew(data->dconn, VIR_STREAM_NONBLOCK)))
> +        goto abrt;
> +
> +    ret = virDomainMigrateOpenTunnel(data->dconn,
> +                                     remote,
> +                                     data->uuid,
> +                                     VIR_MIGRATE_TUNNEL_NBD);
> +
> +    if (ret < 0)
> +        goto abrt;
> +
> +    if (virFDStreamOpen(local, fd) < 0)
> +        goto abrt;
> +
> +    if (!(pipe = qemuMigrationPipeCreate(local, remote)))
> +        goto abrt;
> +
> +    pipe->data = data;
> +    pipe->next = data->pipes;
> +    data->pipes = pipe;

Didn't think too long about this insertion code, but how many times can
this happen?  It's the removal code which removes them all at one time
that's just has me wondering.

> +
> +    return 0;
> +
> + abrt:
> +    VIR_FORCE_CLOSE(fd);
> +    virStreamAbort(local);
> +    virStreamAbort(remote);
> +
> +    virObjectUnref(local);
> +    virObjectUnref(remote);
> +    return -1;
> +}
> +

back to just one blank line.

> +static void
> +qemuMigrationPipesStop(qemuMigrationPipePtr pipe, bool abort)
> +{
> +    qemuMigrationPipePtr tmp;
> +
> +    while (pipe) {
> +        tmp = pipe->next;
> +
> +        qemuMigrationPipeClose(pipe, abort);
> +        VIR_FREE(pipe);
> +
> +        pipe = tmp;
> +    }
> +}
> +

And again only 1 line...

> +static void
>  qemuMigrationIOFunc(void *arg)
>  {
>      qemuMigrationIOThreadPtr data = arg;
> @@ -4081,9 +4199,14 @@ qemuMigrationIOFunc(void *arg)
>                  break;
>              }
>          }
> +
> +        if (fds[2].revents & (POLLIN | POLLERR | POLLHUP) &&
> +            qemuNBDTunnelAcceptAndPipe(data) < 0)
> +            goto abrt;

This would only seem to be necessary if we have a dconn && uuid set...

>      }
>  
>      virStreamFinish(data->qemuStream);
> +    qemuMigrationPipesStop(data->pipes, false);
>  
>      VIR_FORCE_CLOSE(data->qemuSock);
>      VIR_FREE(buffer);
> @@ -4097,6 +4220,7 @@ qemuMigrationIOFunc(void *arg)
>          err = NULL;
>      }
>      virStreamAbort(data->qemuStream);
> +    qemuMigrationPipesStop(data->pipes, true);
>      if (err) {
>          virSetError(err);
>          virFreeError(err);

Looks like we can get to error: from within the polling loop, but we
don't use qemuMigrationPipesStop there

> @@ -4114,7 +4238,9 @@ qemuMigrationIOFunc(void *arg)
>  
>  
>  static qemuMigrationIOThreadPtr
> -qemuMigrationStartTunnel(virStreamPtr qemuStream)
> +qemuMigrationStartTunnel(virStreamPtr qemuStream,
> +                         virConnectPtr dconn,
> +                         unsigned char uuid[VIR_UUID_BUFLEN])

Again the "const unsigned char *uuid"

So what happens when we don't have nbd disks to migrate?  Do we really
need to have dconn and uuid?  Seems that is the assumption...

>  {
>      qemuMigrationIOThreadPtr io = NULL;
>      int wakeupFD[2] = { -1, -1 };
> @@ -4132,6 +4258,8 @@ qemuMigrationStartTunnel(virStreamPtr qemuStream)
>      io->qemuSock = io->unixSock = -1;
>      io->wakeupRecvFD = wakeupFD[0];
>      io->wakeupSendFD = wakeupFD[1];
> +    io->dconn = dconn;
> +    memcpy(io->uuid, uuid, VIR_UUID_BUFLEN);
>  
>      if (virThreadCreate(&io->thread, true,
>                          qemuMigrationIOFunc,
> @@ -4337,7 +4465,9 @@ qemuMigrationRun(virQEMUDriverPtr driver,
>          VIR_WARN("unable to provide data for graphics client relocation");
>  
>      if (spec->fwdType != MIGRATION_FWD_DIRECT) {
> -        if (!(iothread = qemuMigrationStartTunnel(spec->fwd.stream)))
> +        if (!(iothread = qemuMigrationStartTunnel(spec->fwd.stream,
> +                                                  dconn,
> +                                                  mig->uuid)))

So this will only matter if we have mig->nbd, migrate_flags &
(*DISK|*INC), true? An assumption that's not always true I would think.
And thus the 'need' to start things up with that enabled is unnecessary.

Seems you'd want to keep qemuMigrationStartTunnel as is, then if/when we
known we're going to have this functionality have a "set" function for
dconn && uuid... prior to of course setting the nbd_tunnel* in iothread.

John

FWIW: I'm going to stop here. I've got a backlog of other things to look
at right now and it seems the next one makes things even a bit more
complicated...  Plus as I said in my .0 response - starting at patch 16
I wasn't able to apply the changes.

>              goto cancel;
>  
>          if (nmigrate_disks &&
>