[libvirt] [PATCH 1/3] libxl: Receive migration data in a thread

[Jim Fehlig]

Michal Privoznik wrote:
> On 13.11.2014 03:36, Jim Fehlig wrote:
>> The libxl driver receives migration data within an IO callback invoked
>> by the event loop, effectively disabling the event loop while migration
>> occurs.
>>
>> This patch moves receving of the migration data to a thread.  The
>> incoming connection is still accepted in the IO callback, but control
>> is immediately returned to the event loop after spawning the thread.
>>
>> Signed-off-by: Jim Fehlig <jfehlig at suse.com>
>> ---
>>   src/libxl/libxl_migration.c | 60
>> +++++++++++++++++++++++++++++++--------------
>>   1 file changed, 42 insertions(+), 18 deletions(-)
>>
>> diff --git a/src/libxl/libxl_migration.c b/src/libxl/libxl_migration.c
>> index 0b562f7..1940209 100644
>> --- a/src/libxl/libxl_migration.c
>> +++ b/src/libxl/libxl_migration.c
>> @@ -35,6 +35,7 @@
>>   #include "vircommand.h"
>>   #include "virstring.h"
>>   #include "virobject.h"
>> +#include "virthread.h"
>>   #include "rpc/virnetsocket.h"
>>   #include "libxl_domain.h"
>>   #include "libxl_driver.h"
>> @@ -48,6 +49,7 @@ VIR_LOG_INIT("libxl.libxl_migration");
>>   typedef struct _libxlMigrationDstArgs {
>>       virObject parent;
>>
>> +    int recvfd;
>>       virConnectPtr conn;
>>       virDomainObjPtr vm;
>>       unsigned int flags;
>> @@ -82,31 +84,18 @@ libxlMigrationDstArgsOnceInit(void)
>>   VIR_ONCE_GLOBAL_INIT(libxlMigrationDstArgs)
>>
>>   static void
>> -libxlDoMigrateReceive(virNetSocketPtr sock,
>> -                      int events ATTRIBUTE_UNUSED,
>> -                      void *opaque)
>> +libxlDoMigrateReceive(void *opaque)
>>   {
>>       libxlMigrationDstArgs *args = opaque;
>> -    virConnectPtr conn = args->conn;
>>       virDomainObjPtr vm = args->vm;
>>       virNetSocketPtr *socks = args->socks;
>>       size_t nsocks = args->nsocks;
>>       bool paused = args->flags & VIR_MIGRATE_PAUSED;
>> -    libxlDriverPrivatePtr driver = conn->privateData;
>> -    virNetSocketPtr client_sock;
>> -    int recvfd = -1;
>> +    libxlDriverPrivatePtr driver = args->conn->privateData;
>> +    int recvfd = args->recvfd;
>>       size_t i;
>>       int ret;
>>
>> -    if (virNetSocketAccept(sock, &client_sock) < 0) {
>> -        virReportError(VIR_ERR_OPERATION_INVALID, "%s",
>> -                       _("Fail to accept migration connection"));
>> -        goto cleanup;
>> -    }
>> -    VIR_DEBUG("Accepted migration connection\n");
>> -    recvfd = virNetSocketDupFD(client_sock, true);
>> -    virObjectUnref(client_sock);
>> -
>>       virObjectLock(vm);
>>       ret = libxlDomainStart(driver, vm, paused, recvfd);
>>       virObjectUnlock(vm);
>> @@ -114,7 +103,6 @@ libxlDoMigrateReceive(virNetSocketPtr sock,
>>       if (ret < 0 && !vm->persistent)
>>           virDomainObjListRemove(driver->domains, vm);
>>
>> - cleanup:
>>       /* Remove all listen socks from event handler, and close them. */
>>       for (i = 0; i < nsocks; i++) {
>>           virNetSocketUpdateIOCallback(socks[i], 0);
>> @@ -127,6 +115,42 @@ libxlDoMigrateReceive(virNetSocketPtr sock,
>>       VIR_FORCE_CLOSE(recvfd);
>>   }
>>
>> +
>> +static void
>> +libxlMigrateReceive(virNetSocketPtr sock,
>> +                    int events ATTRIBUTE_UNUSED,
>> +                    void *opaque)
>> +{
>> +    libxlMigrationDstArgs *args = opaque;
>> +    virNetSocketPtr client_sock;
>> +    int recvfd;
>> +    virThread thread;
>> +
>> +    /* Accept migration connection */
>> +    virNetSocketAccept(sock, &client_sock);
>> +    if (client_sock == NULL) {
>> +        virReportError(VIR_ERR_OPERATION_FAILED, "%s",
>> +                       _("Failed to accept migration connection"));
>> +        return;
>> +    }
>> +    VIR_DEBUG("Accepted migration connection."
>> +              "  Spawing thread to process migration data");
>> +    recvfd = virNetSocketDupFD(client_sock, true);
>> +    virObjectUnref(client_sock);
>> +
>> +    /*
>> +     * Avoid blocking the event loop.  Start a thread to receive
>> +     * the migration data
>> +     */
>> +    args->recvfd = recvfd;
>> +    if (virThreadCreate(&thread, false,
>> +                        libxlDoMigrateReceive, args) < 0) {
>> +        virReportError(VIR_ERR_OPERATION_FAILED, "%s",
>> +                       _("Failed to create thread for receiving
>> migration data"));
>
> I believe you want VIR_FORCE_CLOSE(args->recvfd) in case thread
> creation failed.

Ah, yes, thanks for catching that.  In fact, I need to cleanup all the
listening sockets on failure too.

Also, while reordering this series I mistakenly dropped a patch that
fixes a race condition introduced by moving the receive processing to a
thread.

V2 on the way.  Sorry for wasting your time on an incomplete V1.

Regards,
Jim