(no subject)

From: Stefan Priebe <[email protected]>

This one fixes a race qemu also had in iscsi block driver between cancellation and io completition.

qemu_rbd_aio_cancel was not synchronously waiting for the end of the command.

It also removes the useless cancelled flag and introduces instead a status flag with EINPROGRESS like iscsi block driver.

Signed-off-by Stefan Priebe <[email protected]>

---
  block/rbd.c |   19 ++++++++++++-------
  1 file changed, 12 insertions(+), 7 deletions(-)

block/rbd.c +1 -1

index 5a0f79f..7b3bcbb 100644
@@ -76,7 +76,7 @@ typedef struct RBDAIOCB {
      int64_t sector_num;
      int error;
      struct BDRVRBDState *s;
-    int cancelled;
+    int status;
  } RBDAIOCB;
 
  typedef struct RADOSCB {

block/rbd.c +7 -1

@@ -406,9 +404,12 @@ static void qemu_rbd_complete_aio(RADOSCB *rcb)
              acb->ret = r;
          }
      }
+    acb->status = acb->ret;
+
      /* Note that acb->bh can be NULL in case where the aio was cancelled */
      acb->bh = qemu_bh_new(rbd_aio_bh_cb, acb);
      qemu_bh_schedule(acb->bh);
+
  done:
      g_free(rcb);
  }

@@ -573,7 +574,10 @@ static void qemu_rbd_close(BlockDriverState *bs)
  static void qemu_rbd_aio_cancel(BlockDriverAIOCB *blockacb)
  {
      RBDAIOCB *acb = (RBDAIOCB *) blockacb;
-    acb->cancelled = 1;
+
+    while (acb->status == -EINPROGRESS) {
+        qemu_aio_wait();
+    }
  }
 
  static AIOPool rbd_aio_pool = {

block/rbd.c +1 -1

      qemu_aio_release(acb);
  }
 

@@ -689,8 +694,8 @@ static BlockDriverAIOCB *rbd_start_aio(BlockDriverState *bs,
      acb->ret = 0;
      acb->error = 0;
      acb->s = s;
-    acb->cancelled = 0;
      acb->bh = NULL;
+    acb->status = -EINPROGRESS;
 
      if (cmd == RBD_AIO_WRITE) {
          qemu_iovec_to_buf(acb->qiov, 0, acb->bounce, qiov->size);

block/rbd.c static void qemu_rbd_complete_aio(RADOSCB *rcb) · line 381

Quoted 10 lines

block/rbd.c +1 -3

@@ -376,9 +376,7 @@ static void qemu_rbd_complete_aio(RADOSCB *rcb)
      RBDAIOCB *acb = rcb->acb;
      int64_t r;
 
-    if (acb->cancelled) {
-        qemu_vfree(acb->bounce);
-        qemu_aio_release(acb);
+    if (acb->bh) {
          goto done;
      }

When is acb->bh != NULL here?

block/rbd.c static void qemu_rbd_complete_aio(RADOSCB *rcb) · line 407

Quoted 6 lines

block/rbd.c +1 -0

@@ -406,9 +404,12 @@ static void qemu_rbd_complete_aio(RADOSCB *rcb)
              acb->ret = r;
          }
      }
+    acb->status = acb->ret;

How about initializing acb->ret with -EINPROGRESS and using it instead of adding a new status field?

block/rbd.c static void qemu_rbd_close(BlockDriverState *bs) · line 581

Quoted 10 lines

block/rbd.c +4 -1

@@ -573,7 +574,10 @@ static void qemu_rbd_close(BlockDriverState *bs)
  static void qemu_rbd_aio_cancel(BlockDriverAIOCB *blockacb)
  {
      RBDAIOCB *acb = (RBDAIOCB *) blockacb;
-    acb->cancelled = 1;
+
+    while (acb->status == -EINPROGRESS) {
+        qemu_aio_wait();
+    }
  }

Need to take care that acb is still valid (not yet released!) when the while loop iterates.

One way of doing this is to mark the acb as cancelled so the completion handler won't release it. Then the cancellation function can release the acb - it's the last piece of code that needs a reference to acb. In this case the acb->cancelled field is useful.

block/rbd.c static void rbd_aio_bh_cb(void *opaque) · line 656

Quoted 13 lines

block/rbd.c +2 -1

@@ -642,10 +646,11 @@ static void rbd_aio_bh_cb(void *opaque)
          qemu_iovec_from_buf(acb->qiov, 0, acb->bounce, acb->qiov->size);
      }
      qemu_vfree(acb->bounce);
-    acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret));
      qemu_bh_delete(acb->bh);
      acb->bh = NULL;
 
+    acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret));
+
      qemu_aio_release(acb);
  }
 

What does this hunk do?

Hello Stefan,
hello Paolo,

most of the ideas and removing the whole cancellation stuff came from Paolo. Maybe he can comment also?

block/rbd.c static void qemu_rbd_complete_aio(RADOSCB *rcb) · line 381

Quoted 10 lines

block/rbd.c +1 -3

@@ -376,9 +376,7 @@ static void qemu_rbd_complete_aio(RADOSCB *rcb)
      RBDAIOCB *acb = rcb->acb;
      int64_t r;
 
-    if (acb->cancelled) {
-        qemu_vfree(acb->bounce);
-        qemu_aio_release(acb);
+    if (acb->bh) {
          goto done;
      }

When is acb->bh != NULL here?

block/rbd.c static void qemu_rbd_complete_aio(RADOSCB *rcb) · line 407

Quoted 6 lines

block/rbd.c +1 -0

@@ -406,9 +404,12 @@ static void qemu_rbd_complete_aio(RADOSCB *rcb)
              acb->ret = r;
          }
      }
+    acb->status = acb->ret;

How about initializing acb->ret with -EINPROGRESS and using it instead of adding a new status field?

block/rbd.c static void qemu_rbd_close(BlockDriverState *bs) · line 581

Quoted 10 lines

block/rbd.c +4 -1

@@ -573,7 +574,10 @@ static void qemu_rbd_close(BlockDriverState *bs)
  static void qemu_rbd_aio_cancel(BlockDriverAIOCB *blockacb)
  {
      RBDAIOCB *acb = (RBDAIOCB *) blockacb;
-    acb->cancelled = 1;
+
+    while (acb->status == -EINPROGRESS) {
+        qemu_aio_wait();
+    }
  }

Need to take care that acb is still valid (not yet released!) when the while loop iterates.

One way of doing this is to mark the acb as cancelled so the completion handler won't release it. Then the cancellation function can release the acb - it's the last piece of code that needs a reference to acb. In this case the acb->cancelled field is useful.

block/rbd.c static void rbd_aio_bh_cb(void *opaque) · line 656

Quoted 13 lines

block/rbd.c +2 -1

@@ -642,10 +646,11 @@ static void rbd_aio_bh_cb(void *opaque)
          qemu_iovec_from_buf(acb->qiov, 0, acb->bounce, acb->qiov->size);
      }
      qemu_vfree(acb->bounce);
-    acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret));
      qemu_bh_delete(acb->bh);
      acb->bh = NULL;
 
+    acb->common.cb(acb->common.opaque, (acb->ret > 0 ? 0 : acb->ret));
+
      qemu_aio_release(acb);
  }
 

What does this hunk do?