qemu: Use domain condition for synchronous block jobs

By switching block jobs to use domain conditions, we can drop some
pretty complicated code in NBD storage migration.

Signed-off-by: Jiri Denemark <jdenemar@redhat.com>

po/POTFILES.in

@@ -114,7 +114,6 @@ src/vz/vz_utils.h
 src/vz/vz_storage.c
 src/phyp/phyp_driver.c
 src/qemu/qemu_agent.c
-src/qemu/qemu_blockjob.c
 src/qemu/qemu_capabilities.c
 src/qemu/qemu_cgroup.c
 src/qemu/qemu_command.c

src/qemu/qemu_blockjob.c

@@ -214,19 +214,17 @@ qemuBlockJobEventProcess(virQEMUDriverPtr driver,
  *
  * During a synchronous block job, a block job event for @disk
  * will not be processed asynchronously. Instead, it will be
- * processed only when qemuBlockJobSyncWait* or
- * qemuBlockJobSyncEnd is called.
+ * processed only when qemuBlockJobUpdate or qemuBlockJobSyncEnd
+ * is called.
  */
 void
 qemuBlockJobSyncBegin(virDomainDiskDefPtr disk)
 {
     qemuDomainDiskPrivatePtr diskPriv = QEMU_DOMAIN_DISK_PRIVATE(disk);
 
-    if (diskPriv->blockJobSync)
-        VIR_WARN("Disk %s already has synchronous block job",
-                 disk->dst);
-
+    VIR_DEBUG("disk=%s", disk->dst);
     diskPriv->blockJobSync = true;
+    diskPriv->blockJobStatus = -1;
 }
 
 
@@ -235,135 +233,16 @@ qemuBlockJobSyncBegin(virDomainDiskDefPtr disk)
  * @driver: qemu driver
  * @vm: domain
  * @disk: domain disk
- * @ret_status: pointer to virConnectDomainEventBlockJobStatus
  *
  * End a synchronous block job for @disk. Any pending block job event
- * for the disk is processed, and its status is recorded in the
- * virConnectDomainEventBlockJobStatus field pointed to by
- * @ret_status.
+ * for the disk is processed.
  */
 void
 qemuBlockJobSyncEnd(virQEMUDriverPtr driver,
                     virDomainObjPtr vm,
-                    virDomainDiskDefPtr disk,
-                    virConnectDomainEventBlockJobStatus *ret_status)
+                    virDomainDiskDefPtr disk)
 {
-    qemuDomainDiskPrivatePtr diskPriv = QEMU_DOMAIN_DISK_PRIVATE(disk);
-
-    if (diskPriv->blockJobSync && diskPriv->blockJobStatus != -1) {
-        if (ret_status)
-            *ret_status = diskPriv->blockJobStatus;
-        qemuBlockJobUpdate(driver, vm, disk);
-        diskPriv->blockJobStatus = -1;
-    }
-    diskPriv->blockJobSync = false;
-}
-
-
-/**
- * qemuBlockJobSyncWaitWithTimeout:
- * @driver: qemu driver
- * @vm: domain
- * @disk: domain disk
- * @timeout: timeout in milliseconds
- * @ret_status: pointer to virConnectDomainEventBlockJobStatus
- *
- * Wait up to @timeout milliseconds for a block job event for @disk.
- * If an event is received it is processed, and its status is recorded
- * in the virConnectDomainEventBlockJobStatus field pointed to by
- * @ret_status.
- *
- * If @timeout is not 0, @vm will be unlocked while waiting for the event.
- *
- * Returns 0 if an event was received or the timeout expired,
- *        -1 otherwise.
- */
-int
-qemuBlockJobSyncWaitWithTimeout(virQEMUDriverPtr driver,
-                                virDomainObjPtr vm,
-                                virDomainDiskDefPtr disk,
-                                unsigned long long timeout,
-                                virConnectDomainEventBlockJobStatus *ret_status)
-{
-    qemuDomainDiskPrivatePtr diskPriv = QEMU_DOMAIN_DISK_PRIVATE(disk);
-
-    if (!diskPriv->blockJobSync) {
-        virReportError(VIR_ERR_INTERNAL_ERROR, "%s",
-                       _("No current synchronous block job"));
-        return -1;
-    }
-
-    while (diskPriv->blockJobSync && diskPriv->blockJobStatus == -1) {
-        int r;
-
-        if (!virDomainObjIsActive(vm)) {
-            virReportError(VIR_ERR_INTERNAL_ERROR, "%s",
-                           _("guest unexpectedly quit"));
-            diskPriv->blockJobSync = false;
-            return -1;
-        }
-
-        if (timeout == (unsigned long long)-1) {
-            r = virCondWait(&diskPriv->blockJobSyncCond, &vm->parent.lock);
-        } else if (timeout) {
-            unsigned long long now;
-            if (virTimeMillisNow(&now) < 0) {
-                virReportSystemError(errno, "%s",
-                                     _("Unable to get current time"));
-                return -1;
-            }
-            r = virCondWaitUntil(&diskPriv->blockJobSyncCond,
-                                 &vm->parent.lock,
-                                 now + timeout);
-            if (r < 0 && errno == ETIMEDOUT)
-                return 0;
-        } else {
-            errno = ETIMEDOUT;
-            return 0;
-        }
-
-        if (r < 0) {
-            diskPriv->blockJobSync = false;
-            virReportSystemError(errno, "%s",
-                                 _("Unable to wait on block job sync "
-                                   "condition"));
-            return -1;
-        }
-    }
-
-    if (ret_status)
-        *ret_status = diskPriv->blockJobStatus;
+    VIR_DEBUG("disk=%s", disk->dst);
     qemuBlockJobUpdate(driver, vm, disk);
-    diskPriv->blockJobStatus = -1;
-
-    return 0;
-}
-
-
-/**
- * qemuBlockJobSyncWait:
- * @driver: qemu driver
- * @vm: domain
- * @disk: domain disk
- * @ret_status: pointer to virConnectDomainEventBlockJobStatus
- *
- * Wait for a block job event for @disk. If an event is received it
- * is processed, and its status is recorded in the
- * virConnectDomainEventBlockJobStatus field pointed to by
- * @ret_status.
- *
- * @vm will be unlocked while waiting for the event.
- *
- * Returns 0 if an event was received,
- *        -1 otherwise.
- */
-int
-qemuBlockJobSyncWait(virQEMUDriverPtr driver,
-                     virDomainObjPtr vm,
-                     virDomainDiskDefPtr disk,
-                     virConnectDomainEventBlockJobStatus *ret_status)
-{
-    return qemuBlockJobSyncWaitWithTimeout(driver, vm, disk,
-                                           (unsigned long long)-1,
-                                           ret_status);
+    QEMU_DOMAIN_DISK_PRIVATE(disk)->blockJobSync = false;
 }

src/qemu/qemu_blockjob.h

@@ -37,16 +37,6 @@ void qemuBlockJobEventProcess(virQEMUDriverPtr driver,
 void qemuBlockJobSyncBegin(virDomainDiskDefPtr disk);
 void qemuBlockJobSyncEnd(virQEMUDriverPtr driver,
                          virDomainObjPtr vm,
-                         virDomainDiskDefPtr disk,
-                         virConnectDomainEventBlockJobStatus *ret_status);
-int qemuBlockJobSyncWaitWithTimeout(virQEMUDriverPtr driver,
-                                    virDomainObjPtr vm,
-                                    virDomainDiskDefPtr disk,
-                                    unsigned long long timeout,
-                                    virConnectDomainEventBlockJobStatus *ret_status);
-int qemuBlockJobSyncWait(virQEMUDriverPtr driver,
-                         virDomainObjPtr vm,
-                         virDomainDiskDefPtr disk,
-                         virConnectDomainEventBlockJobStatus *ret_status);
+                         virDomainDiskDefPtr disk);
 
 #endif /* __QEMU_BLOCKJOB_H__ */

src/qemu/qemu_domain.c

@@ -413,7 +413,6 @@ qemuDomainJobInfoToParams(qemuDomainJobInfoPtr jobInfo,
 
 
 static virClassPtr qemuDomainDiskPrivateClass;
-static void qemuDomainDiskPrivateDispose(void *obj);
 
 static int
 qemuDomainDiskPrivateOnceInit(void)
@@ -421,7 +420,7 @@ qemuDomainDiskPrivateOnceInit(void)
     qemuDomainDiskPrivateClass = virClassNew(virClassForObject(),
                                              "qemuDomainDiskPrivate",
                                              sizeof(qemuDomainDiskPrivate),
-                                             qemuDomainDiskPrivateDispose);
+                                             NULL);
     if (!qemuDomainDiskPrivateClass)
         return -1;
     else
@@ -441,23 +440,9 @@ qemuDomainDiskPrivateNew(void)
     if (!(priv = virObjectNew(qemuDomainDiskPrivateClass)))
         return NULL;
 
-    if (virCondInit(&priv->blockJobSyncCond) < 0) {
-        virReportSystemError(errno, "%s", _("Failed to initialize condition"));
-        virObjectUnref(priv);
-        return NULL;
-    }
-
     return (virObjectPtr) priv;
 }
 
-static void
-qemuDomainDiskPrivateDispose(void *obj)
-{
-    qemuDomainDiskPrivatePtr priv = obj;
-
-    virCondDestroy(&priv->blockJobSyncCond);
-}
-
 
 static void *
 qemuDomainObjPrivateAlloc(void)

src/qemu/qemu_domain.h

@@ -214,7 +214,6 @@ struct _qemuDomainDiskPrivate {
     bool blockjob;
 
     /* for some synchronous block jobs, we need to notify the owner */
-    virCond blockJobSyncCond;
     int blockJobType;   /* type of the block job from the event */
     int blockJobStatus; /* status of the finished block job */
     bool blockJobSync; /* the block job needs synchronized termination */

src/qemu/qemu_driver.c

@@ -16355,10 +16355,8 @@ qemuDomainBlockJobAbort(virDomainPtr dom,
         goto endjob;
     }
 
-    if (modern && !async) {
-        /* prepare state for event delivery */
+    if (modern && !async)
         qemuBlockJobSyncBegin(disk);
-    }
 
     if (pivot) {
         if ((ret = qemuDomainBlockPivot(driver, vm, device, disk)) < 0)
@@ -16406,21 +16404,21 @@ qemuDomainBlockJobAbort(virDomainPtr dom,
                                      VIR_DOMAIN_BLOCK_JOB_TYPE_PULL,
                                      VIR_DOMAIN_BLOCK_JOB_CANCELED);
         } else {
-            virConnectDomainEventBlockJobStatus status = -1;
-            if (qemuBlockJobSyncWait(driver, vm, disk, &status) < 0) {
-                ret = -1;
-            } else if (status == VIR_DOMAIN_BLOCK_JOB_FAILED) {
-                virReportError(VIR_ERR_OPERATION_FAILED,
-                               _("failed to terminate block job on disk '%s'"),
-                               disk->dst);
-                ret = -1;
+            qemuDomainDiskPrivatePtr diskPriv = QEMU_DOMAIN_DISK_PRIVATE(disk);
+            qemuBlockJobUpdate(driver, vm, disk);
+            while (diskPriv->blockjob) {
+                if (virDomainObjWait(vm) < 0) {
+                    ret = -1;
+                    goto endjob;
+                }
+                qemuBlockJobUpdate(driver, vm, disk);
             }
         }
     }
 
  endjob:
-    if (disk && QEMU_DOMAIN_DISK_PRIVATE(disk)->blockJobSync)
-        qemuBlockJobSyncEnd(driver, vm, disk, NULL);
+    if (disk)
+        qemuBlockJobSyncEnd(driver, vm, disk);
     qemuDomainObjEndJob(driver, vm);
 
  cleanup:

src/qemu/qemu_migration.c

@@ -1744,7 +1744,7 @@ qemuMigrationStopNBDServer(virQEMUDriverPtr driver,
 
 
 /**
- * qemuMigrationCheckDriveMirror:
+ * qemuMigrationDriveMirrorReady:
  * @driver: qemu driver
  * @vm: domain
  *
@@ -1757,37 +1757,39 @@ qemuMigrationStopNBDServer(virQEMUDriverPtr driver,
  *        -1 on error.
  */
 static int
-qemuMigrationCheckDriveMirror(virQEMUDriverPtr driver,
+qemuMigrationDriveMirrorReady(virQEMUDriverPtr driver,
                               virDomainObjPtr vm)
 {
     size_t i;
-    int ret = 1;
+    size_t notReady = 0;
+    int status;
 
     for (i = 0; i < vm->def->ndisks; i++) {
         virDomainDiskDefPtr disk = vm->def->disks[i];
         qemuDomainDiskPrivatePtr diskPriv = QEMU_DOMAIN_DISK_PRIVATE(disk);
 
-        if (!diskPriv->migrating || !diskPriv->blockJobSync)
+        if (!diskPriv->migrating)
             continue;
 
-        /* process any pending event */
-        if (qemuBlockJobSyncWaitWithTimeout(driver, vm, disk,
-                                            0ull, NULL) < 0)
-            return -1;
-
-        switch (disk->mirrorState) {
-        case VIR_DOMAIN_DISK_MIRROR_STATE_NONE:
-            ret = 0;
-            break;
-        case VIR_DOMAIN_DISK_MIRROR_STATE_ABORT:
+        status = qemuBlockJobUpdate(driver, vm, disk);
+        if (status == VIR_DOMAIN_BLOCK_JOB_FAILED) {
             virReportError(VIR_ERR_OPERATION_FAILED,
                            _("migration of disk %s failed"),
                            disk->dst);
             return -1;
         }
+
+        if (disk->mirrorState != VIR_DOMAIN_DISK_MIRROR_STATE_READY)
+            notReady++;
     }
 
-    return ret;
+    if (notReady) {
+        VIR_DEBUG("Waiting for %zu disk mirrors to get ready", notReady);
+        return 0;
+    } else {
+        VIR_DEBUG("All disk mirrors are ready");
+        return 1;
+    }
 }
 
 
@@ -1847,18 +1849,17 @@ qemuMigrationCancelOneDriveMirror(virQEMUDriverPtr driver,
 
         /* Mirror may become ready before cancellation takes
          * effect; loop if we get that event first */
-        do {
-            ret = qemuBlockJobSyncWait(driver, vm, disk, &status);
-            if (ret < 0) {
-                VIR_WARN("Unable to wait for block job on %s to cancel",
-                         diskAlias);
+        while (1) {
+            status = qemuBlockJobUpdate(driver, vm, disk);
+            if (status != -1 && status != VIR_DOMAIN_BLOCK_JOB_READY)
+                break;
+            if ((ret = virDomainObjWait(vm)) < 0)
                 goto endjob;
-            }
-        } while (status == VIR_DOMAIN_BLOCK_JOB_READY);
+        }
     }
 
  endjob:
-    qemuBlockJobSyncEnd(driver, vm, disk, NULL);
+    qemuBlockJobSyncEnd(driver, vm, disk);
 
     if (disk->mirrorState == VIR_DOMAIN_DISK_MIRROR_STATE_ABORT)
         disk->mirrorState = VIR_DOMAIN_DISK_MIRROR_STATE_NONE;
@@ -1950,6 +1951,9 @@ qemuMigrationDriveMirror(virQEMUDriverPtr driver,
     char *nbd_dest = NULL;
     char *hoststr = NULL;
     unsigned int mirror_flags = VIR_DOMAIN_BLOCK_REBASE_REUSE_EXT;
+    int rv;
+
+    VIR_DEBUG("Starting drive mirrors for domain %s", vm->def->name);
 
     /* steal NBD port and thus prevent its propagation back to destination */
     port = mig->nbd->port;
@@ -1975,61 +1979,47 @@ qemuMigrationDriveMirror(virQEMUDriverPtr driver,
         if (!qemuMigrateDisk(disk, nmigrate_disks, migrate_disks))
             continue;
 
-        VIR_FREE(diskAlias);
-        VIR_FREE(nbd_dest);
         if ((virAsprintf(&diskAlias, "%s%s",
                          QEMU_DRIVE_HOST_PREFIX, disk->info.alias) < 0) ||
             (virAsprintf(&nbd_dest, "nbd:%s:%d:exportname=%s",
                          hoststr, port, diskAlias) < 0))
             goto cleanup;
 
-        qemuBlockJobSyncBegin(disk);
-
         if (qemuDomainObjEnterMonitorAsync(driver, vm,
-                                           QEMU_ASYNC_JOB_MIGRATION_OUT) < 0) {
-            qemuBlockJobSyncEnd(driver, vm, disk, NULL);
+                                           QEMU_ASYNC_JOB_MIGRATION_OUT) < 0)
             goto cleanup;
-        }
 
+        qemuBlockJobSyncBegin(disk);
         /* Force "raw" format for NBD export */
         mon_ret = qemuMonitorDriveMirror(priv->mon, diskAlias, nbd_dest,
                                          "raw", speed, 0, 0, mirror_flags);
+        VIR_FREE(diskAlias);
+        VIR_FREE(nbd_dest);
 
         if (qemuDomainObjExitMonitor(driver, vm) < 0 || mon_ret < 0) {
-            qemuBlockJobSyncEnd(driver, vm, disk, NULL);
+            qemuBlockJobSyncEnd(driver, vm, disk);
             goto cleanup;
         }
         diskPriv->migrating = true;
     }
 
-    /* Wait for each disk to become ready in turn, but check the status
-     * for *all* mirrors to determine if any have aborted. */
-    for (i = 0; i < vm->def->ndisks; i++) {
-        virDomainDiskDefPtr disk = vm->def->disks[i];
-        qemuDomainDiskPrivatePtr diskPriv = QEMU_DOMAIN_DISK_PRIVATE(disk);
+    while ((rv = qemuMigrationDriveMirrorReady(driver, vm)) != 1) {
+        unsigned long long now;
 
-        if (!diskPriv->migrating)
-            continue;
+        if (rv < 0)
+            goto cleanup;
 
-        while (disk->mirrorState != VIR_DOMAIN_DISK_MIRROR_STATE_READY) {
-            /* The following check should be race free as long as the variable
-             * is set only with domain object locked. And here we have the
-             * domain object locked too. */
-            if (priv->job.asyncAbort) {
-                priv->job.current->type = VIR_DOMAIN_JOB_CANCELLED;
-                virReportError(VIR_ERR_OPERATION_ABORTED, _("%s: %s"),
-                               qemuDomainAsyncJobTypeToString(priv->job.asyncJob),
-                               _("canceled by client"));
-                goto cleanup;
-            }
-
-            if (qemuBlockJobSyncWaitWithTimeout(driver, vm, disk,
-                                                500ull, NULL) < 0)
-                goto cleanup;
-
-            if (qemuMigrationCheckDriveMirror(driver, vm) < 0)
-                goto cleanup;
+        if (priv->job.asyncAbort) {
+            priv->job.current->type = VIR_DOMAIN_JOB_CANCELLED;
+            virReportError(VIR_ERR_OPERATION_ABORTED, _("%s: %s"),
+                           qemuDomainAsyncJobTypeToString(priv->job.asyncJob),
+                           _("canceled by client"));
+            goto cleanup;
         }
+
+        if (virTimeMillisNow(&now) < 0 ||
+            virDomainObjWaitUntil(vm, now + 500) < 0)
+            goto cleanup;
     }
 
     /* Okay, all disks are ready. Modify migrate_flags */
@@ -4177,7 +4167,7 @@ qemuMigrationRun(virQEMUDriverPtr driver,
 
     /* Confirm state of drive mirrors */
     if (mig->nbd) {
-        if (qemuMigrationCheckDriveMirror(driver, vm) != 1) {
+        if (qemuMigrationDriveMirrorReady(driver, vm) != 1) {
             ret = -1;
             goto cancel;
         }

src/qemu/qemu_process.c

@@ -1001,10 +1001,10 @@ qemuProcessHandleBlockJob(qemuMonitorPtr mon ATTRIBUTE_UNUSED,
     diskPriv = QEMU_DOMAIN_DISK_PRIVATE(disk);
 
     if (diskPriv->blockJobSync) {
+        /* We have a SYNC API waiting for this event, dispatch it back */
         diskPriv->blockJobType = type;
         diskPriv->blockJobStatus = status;
-        /* We have an SYNC API waiting for this event, dispatch it back */
-        virCondSignal(&diskPriv->blockJobSyncCond);
+        virDomainObjSignal(vm);
     } else {
         /* there is no waiting SYNC API, dispatch the update to a thread */
         if (VIR_ALLOC(processEvent) < 0)
@@ -5055,13 +5055,8 @@ void qemuProcessStop(virQEMUDriverPtr driver,
     if (virAtomicIntDecAndTest(&driver->nactive) && driver->inhibitCallback)
         driver->inhibitCallback(false, driver->inhibitOpaque);
 
-    /* Wake up anything waiting on synchronous block jobs */
-    for (i = 0; i < vm->def->ndisks; i++) {
-        qemuDomainDiskPrivatePtr diskPriv =
-            QEMU_DOMAIN_DISK_PRIVATE(vm->def->disks[i]);
-        if (diskPriv->blockJobSync && diskPriv->blockJobStatus == -1)
-            virCondSignal(&diskPriv->blockJobSyncCond);
-    }
+    /* Wake up anything waiting on domain condition */
+    virDomainObjBroadcast(vm);
 
     if ((logfile = qemuDomainCreateLog(driver, vm, true)) < 0) {
         /* To not break the normal domain shutdown process, skip the