Fix segfault in blockjob race condition

blockjob.c +0 -0

@@ -291,10 +291,10 @@ void block_job_start(BlockJob *job)
 {

blockjob.c +1 -1

@@ -797,11 +797,14 @@ void block_job_sleep_ns(BlockJob *job, QEMUClockType type, int64_t ns)
         return;
     }
 
-    job->busy = false;
+    /* We need to leave job->busy set here, because when we have

Quoted 10 lines · Jeff Cody

blockjob.c +0 -1

@@ -291,10 +291,10 @@ void block_job_start(BlockJob *job)
 {
     assert(job && !block_job_started(job) && job->paused &&
            job->driver && job->driver->start);
-    job->co = qemu_coroutine_create(block_job_co_entry, job);
     job->pause_count--;
     job->busy = true;
     job->paused = false;

blockjob.c void block_job_start(BlockJob *job) · line 300

Quoted 4 lines

blockjob.c +1 -0

+    job->co = qemu_coroutine_create(block_job_co_entry, job);
     bdrv_coroutine_enter(blk_bs(job->blk), job->co);
 }
 

This hunk makes no difference. The coroutine is only entered by bdrv_coroutine_enter() so the order of job field initialization doesn't matter.

Quoted 6 lines

blockjob.c +1 -1

@@ -797,11 +797,14 @@ void block_job_sleep_ns(BlockJob *job, QEMUClockType type, int64_t ns)
         return;
     }
 
-    job->busy = false;
+    /* We need to leave job->busy set here, because when we have

blockjob.c void block_job_sleep_ns(BlockJob *job, QEMUClockType type, int64_t ns) · line 809

Quoted 10 lines

blockjob.c +4 -1

+     * put a coroutine to 'sleep', we have scheduled it to run in
+     * the future.  We cannot enter that same coroutine again before
+     * it wakes and runs, otherwise we risk double-entry or entry after
+     * completion. */
     if (!block_job_should_pause(job)) {
         co_aio_sleep_ns(blk_get_aio_context(job->blk), type, ns);
     }
-    job->busy = true;
 
     block_job_pause_point(job);

This leaves a stale doc comment in include/block/blockjob_int.h:

/**
* block_job_sleep_ns:
* @job: The job that calls the function.
* @clock: The clock to sleep on.
* @ns: How many nanoseconds to stop for.
*
* Put the job to sleep (assuming that it wasn't canceled) for @ns
* nanoseconds. Canceling the job will interrupt the wait immediately.

                   ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

*/

  void block_job_sleep_ns(BlockJob *job, QEMUClockType type, int64_t ns);

This raises questions about the ability to cancel sleep:

1. Does something depend on cancelling sleep?

2. Did cancellation work properly in commit 4513eafe928ff47486f4167c28d364c72b5ff7e3 ("block: add block_job_sleep_ns") and was it broken afterwards?

It is possible to fix the recursive coroutine entry without losing sleep cancellation. Whether it's worth the trouble depends on the answers to the above questions.

Quoted 11 lines · Stefan Hajnoczi

Quoted 10 lines · Jeff Cody

blockjob.c +0 -1

@@ -291,10 +291,10 @@ void block_job_start(BlockJob *job)
 {
     assert(job && !block_job_started(job) && job->paused &&
            job->driver && job->driver->start);
-    job->co = qemu_coroutine_create(block_job_co_entry, job);
     job->pause_count--;
     job->busy = true;
     job->paused = false;

Quoted 9 lines

blockjob.c void block_job_start(BlockJob *job) · line 300

Quoted 4 lines

blockjob.c +1 -0

+    job->co = qemu_coroutine_create(block_job_co_entry, job);
     bdrv_coroutine_enter(blk_bs(job->blk), job->co);
 }
 

This hunk makes no difference. The coroutine is only entered by bdrv_coroutine_enter() so the order of job field initialization doesn't matter.

It likely makes no difference with the current code (unless there is a latent bug). However I made the change to protect against the following scenario - which, perhaps to your point, would be a bug in any case:

1. job->co = qemu_coroutine_create()

* Now block_job_started() returns true, as it just checks for job->co

2. Another thread calls block_job_enter(), before we call bdrv_coroutine_enter().

* block_job_enter() checks job->busy and block_job_started() to determine if coroutine entry is allowed. Without this change, these checks could pass and coroutine entry could occur.

* I don't think this can happen in the current code, but the above hunk change is still correct, and would protect against such an occurrence.

I guess the question is, "is it worth doing?", to try and prevent that sort of buggy behavior. My thought was "yes" because:

A) there is no penalty in doing it this way

B) while a bug, double entry like this can lead to memory and/or data corruption, and the checks for co->caller et al. might not catch it. This is particularly true if the coroutine exits (COROUTINE_TERMINATE) before the re-entry.

But maybe if we are concerned about that we should figure out a way to abort() instead. Of course, that makes allowing recursive coroutines more difficult in the future.

Quoted 29 lines

Quoted 8 lines

blockjob.c +1 -1

@@ -797,11 +797,14 @@ void block_job_sleep_ns(BlockJob *job, QEMUClockType type, int64_t ns)
         return;
     }
 
-    job->busy = false;
+    /* We need to leave job->busy set here, because when we have

blockjob.c void block_job_sleep_ns(BlockJob *job, QEMUClockType type, int64_t ns) · line 809

Quoted 10 lines

blockjob.c +4 -1

+     * put a coroutine to 'sleep', we have scheduled it to run in
+     * the future.  We cannot enter that same coroutine again before
+     * it wakes and runs, otherwise we risk double-entry or entry after
+     * completion. */
     if (!block_job_should_pause(job)) {
         co_aio_sleep_ns(blk_get_aio_context(job->blk), type, ns);
     }
-    job->busy = true;
 
     block_job_pause_point(job);

This leaves a stale doc comment in include/block/blockjob_int.h:

/**
* block_job_sleep_ns:
* @job: The job that calls the function.
* @clock: The clock to sleep on.
* @ns: How many nanoseconds to stop for.
*
* Put the job to sleep (assuming that it wasn't canceled) for @ns
* nanoseconds. Canceling the job will interrupt the wait immediately.

Quoted 2 lines

                   ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

*/

I didn't catch the doc, that should be changed as well.

Quoted 6 lines

  void block_job_sleep_ns(BlockJob *job, QEMUClockType type, int64_t ns);

This raises questions about the ability to cancel sleep:

1. Does something depend on cancelling sleep?

Not that I can tell. The advantage is that you don't have to wait for the timer, so something like qmp_block_job_cancel() will cancel sooner.

But it is obviously broken with the current coroutine implementation to try to do that.

Quoted 4 lines

2. Did cancellation work properly in commit 4513eafe928ff47486f4167c28d364c72b5ff7e3 ("block: add block_job_sleep_ns") and was it broken afterwards?

With iothreads, the answer is complicated. It was broken for a while for other reasons.

It broke after using aio_co_wake() in the sleep timer cb (commmit 2f47da5f7f), which added the ability to schedule a coroutine if the timer callback was called from the wrong AioContext.

Prior to that it "worked" in that the segfault was not present.

But even to bisect back to 2f47da5f7f was not straightforward, because attempting them stream/cancel with iothreads would not even work until c324fd0 (so I only bisected back as far as c324fd0 would cleanly apply).

And it is tricky to say if it "works" or not, because it is racy. What may have appeared to work may be more attributed to luck and timing.

If the coroutine is going to run at a future time, we cannot enter it beforehand. We risk the coroutine not even existing when the timer does run the sleeping coroutine. At the very least, early entry with the current code would require a way to delete the associated timer.

Quoted 4 lines

It is possible to fix the recursive coroutine entry without losing sleep cancellation. Whether it's worth the trouble depends on the answers to the above questions.

I contemplated the same thing.

At least for 2.11, fixing recursive coroutine entry is probably more than we want to do.

Long term, my opinion is that we should fix it, because preventing it becomes more difficult. It is easy to miss something that might cause a recursive entry in code reviews, and since it can be racy, casual testing may often miss it as well.

This raises questions about the ability to cancel sleep:

1. Does something depend on cancelling sleep?

Quoted 3 lines

This raises questions about the ability to cancel sleep:

1. Does something depend on cancelling sleep?

block_job_cancel does, but in practice the sleep time is so small (smaller than SLICE_TIME, which is 100 ms) that we probably don't care.

I agree with Jeff that canceling the sleep by force-entering the coroutine seemed clever but is probably a very bad idea.

Paolo

util/qemu-coroutine.c +0 -0

index d6095c1..2edab63 100644

Quoted 6 lines · Jeff Cody

include/qemu/coroutine_int.h +0 -0

index cb98892..931cdc9 100644
@@ -53,6 +53,9 @@ struct Coroutine {
 

include/qemu/coroutine_int.h struct Coroutine { · line 57

Quoted 4 lines

include/qemu/coroutine_int.h +2 -0

     /* Only used when the coroutine has yielded.  */
     AioContext *ctx;
+
+    int scheduled;

s/int/bool/

Quoted 5 lines

util/qemu-coroutine.c +0 -0

index d6095c1..2edab63 100644

util/qemu-coroutine.c void qemu_aio_coroutine_enter(AioContext *ctx, Coroutine *co) · line 117

Quoted 9 lines

util/qemu-coroutine.c +6 -0

@@ -109,6 +109,15 @@ void qemu_aio_coroutine_enter(AioContext *ctx, Coroutine *co)
 
     trace_qemu_aio_coroutine_enter(ctx, self, co, co->entry_arg);
 
+    /* if the Coroutine has already been scheduled, entering it again will
+     * cause us to enter it twice, potentially even after the coroutine has
+     * been deleted */
+    if (co->scheduled == 1) {
+        fprintf(stderr, "Cannot enter a co-routine that has already "
+                        "been scheduled to run in a different AioContext\n");

This error message is too specific, the AioContext doesn't have to be different from the current one:

block/blkdebug.c: aio_co_schedule(qemu_get_current_aio_context(), qemu_coroutine_self());

If something calls qemu_aio_coroutine_enter() on the coroutine it might be from the same AioContext - but still an error condition worth failing loudly on.

I suggest simplifying the error message:

fprintf(stderr, "Cannot enter a co-routine that has already "

                  "been scheduled\n");

Quoted 6 lines · Jeff Cody

include/qemu/coroutine_int.h +0 -0

index 931cdc9..b071217 100644
@@ -56,6 +56,8 @@ struct Coroutine {
 

include/qemu/coroutine_int.h struct Coroutine { · line 59

Quoted 3 lines

include/qemu/coroutine_int.h +1 -0

     int scheduled;
 
+    int sleeping;

s/int/bool/

BTW an alternative to adding individual bools is to implement a finite state machine for the entire coroutine lifecycle. A single function can validate all state transitions:

  void check_state_transition(CoState old, CoState new,
                              const char *action)
  {
      const char *errmsg = fsm[old][new];
      if (!errmsg) {
          return; /* valid transition! */
      }

      fprintf(stderr, "Cannot %s coroutine from %s state\n",
              action, state_name[old]);
      abort();

}

Specifying fsm[][] forces us to think through all possible state transitions. This approach is proactive whereas adding bool flags is reactive since it only covers a subset of states that were encountered after crashes. I'm not sure if it's worth it though :).

Quoted 18 lines · Stefan Hajnoczi

On Sun, Nov 19, 2017 at 09:46:44PM -0500, Jeff Cody wrote:
BTW an alternative to adding individual bools is to implement a finite state machine for the entire coroutine lifecycle. A single function can validate all state transitions:

  void check_state_transition(CoState old, CoState new,
                              const char *action)
  {
      const char *errmsg = fsm[old][new];
      if (!errmsg) {
          return; /* valid transition! */
      }

      fprintf(stderr, "Cannot %s coroutine from %s state\n",
              action, state_name[old]);
      abort();

}

Quoted 4 lines

Specifying fsm[][] forces us to think through all possible state transitions. This approach is proactive whereas adding bool flags is reactive since it only covers a subset of states that were encountered after crashes. I'm not sure if it's worth it though :).

Interesting idea; maybe more for 2.12 instead of 2.11, though?

Once a coroutine is "sleeping", the timer callback will either enter the coroutine, or schedule it for the next AioContext if using iothreads.

It is illegal to enter that coroutine while waiting for this timer event and subsequent callback. This patch will catch such an attempt, and abort QEMU with an error.

Like with the previous patch, we cannot rely solely on the co->caller check for recursive entry. The prematurely entered coroutine may exit with COROUTINE_TERMINATE before the timer expires, making co->caller no longer valid.

We can clear co->sleeping in in co_sleep_cb(), because any doubly entry attempt after point should be caught by either the co->scheduled or co->caller checks.

Signed-off-by Jeff Cody <[email protected]>

---
 include/qemu/coroutine_int.h | 2 ++
 util/qemu-coroutine-sleep.c  | 3 +++
 util/qemu-coroutine.c        | 5 +++++
 3 files changed, 10 insertions(+)

include/qemu/coroutine_int.h +0 -0

index 931cdc9..b071217 100644

include/qemu/coroutine_int.h +1 -0

+
     QSIMPLEQ_ENTRY(Coroutine) co_queue_next;
     QSLIST_ENTRY(Coroutine) co_scheduled_next;
 };

util/qemu-coroutine-sleep.c +3 -0

index 9c56550..11ae95a 100644
@@ -13,6 +13,7 @@
 
 #include "qemu/osdep.h"
 #include "qemu/coroutine.h"
+#include "qemu/coroutine_int.h"
 #include "qemu/timer.h"
 #include "block/aio.h"
 

@@ -25,6 +26,7 @@ static void co_sleep_cb(void *opaque)
 {
     CoSleepCB *sleep_cb = opaque;
 
+    sleep_cb->co->sleeping = 0;
     aio_co_wake(sleep_cb->co);
 }
 

@@ -34,6 +36,7 @@ void coroutine_fn co_aio_sleep_ns(AioContext *ctx, QEMUClockType type,
     CoSleepCB sleep_cb = {
         .co = qemu_coroutine_self(),
     };
+    sleep_cb.co->sleeping = 1;
     sleep_cb.ts = aio_timer_new(ctx, type, SCALE_NS, co_sleep_cb, &sleep_cb);
     timer_mod(sleep_cb.ts, qemu_clock_get_ns(type) + ns);
     qemu_coroutine_yield();

util/qemu-coroutine.c +5 -0

index 2edab63..1d9f93d 100644
@@ -118,6 +118,11 @@ void qemu_aio_coroutine_enter(AioContext *ctx, Coroutine *co)
         abort();
     }
 
+    if (co->sleeping == 1) {
+        fprintf(stderr, "Cannot enter a co-routine that is still sleeping\n");
+        abort();
+    }
+
     if (co->caller) {
         fprintf(stderr, "Co-routine re-entered recursively\n");
         abort();

Quoted 29 lines · Jeff Cody

Once a coroutine is "sleeping", the timer callback will either enter the coroutine, or schedule it for the next AioContext if using iothreads.

It is illegal to enter that coroutine while waiting for this timer event and subsequent callback. This patch will catch such an attempt, and abort QEMU with an error.

Like with the previous patch, we cannot rely solely on the co->caller check for recursive entry. The prematurely entered coroutine may exit with COROUTINE_TERMINATE before the timer expires, making co->caller no longer valid.

We can clear co->sleeping in in co_sleep_cb(), because any doubly entry attempt after point should be caught by either the co->scheduled or co->caller checks.

Signed-off-by Jeff Cody <[email protected]>

---
 include/qemu/coroutine_int.h | 2 ++
 util/qemu-coroutine-sleep.c  | 3 +++
 util/qemu-coroutine.c        | 5 +++++
 3 files changed, 10 insertions(+)

include/qemu/coroutine_int.h +0 -0

index 931cdc9..b071217 100644
@@ -56,6 +56,8 @@ struct Coroutine {
 

include/qemu/coroutine_int.h struct Coroutine { · line 59

Quoted 3 lines

include/qemu/coroutine_int.h +1 -0

     int scheduled;
 
+    int sleeping;

Is this a different "state" (in Stefan's parlance) than scheduled? In practice both means that someone may call qemu_(aio_)coroutine_enter concurrently, so you'd better not do it yourself.

Quoted 50 lines

include/qemu/coroutine_int.h +1 -0

+
     QSIMPLEQ_ENTRY(Coroutine) co_queue_next;
     QSLIST_ENTRY(Coroutine) co_scheduled_next;
 };

util/qemu-coroutine-sleep.c +3 -0

index 9c56550..11ae95a 100644
@@ -13,6 +13,7 @@
 
 #include "qemu/osdep.h"
 #include "qemu/coroutine.h"
+#include "qemu/coroutine_int.h"
 #include "qemu/timer.h"
 #include "block/aio.h"
 

@@ -25,6 +26,7 @@ static void co_sleep_cb(void *opaque)
 {
     CoSleepCB *sleep_cb = opaque;
 
+    sleep_cb->co->sleeping = 0;
     aio_co_wake(sleep_cb->co);
 }
 

@@ -34,6 +36,7 @@ void coroutine_fn co_aio_sleep_ns(AioContext *ctx, QEMUClockType type,
     CoSleepCB sleep_cb = {
         .co = qemu_coroutine_self(),
     };
+    sleep_cb.co->sleeping = 1;
     sleep_cb.ts = aio_timer_new(ctx, type, SCALE_NS, co_sleep_cb, &sleep_cb);
     timer_mod(sleep_cb.ts, qemu_clock_get_ns(type) + ns);
     qemu_coroutine_yield();

util/qemu-coroutine.c +5 -0

index 2edab63..1d9f93d 100644
@@ -118,6 +118,11 @@ void qemu_aio_coroutine_enter(AioContext *ctx, Coroutine *co)
         abort();
     }
 
+    if (co->sleeping == 1) {
+        fprintf(stderr, "Cannot enter a co-routine that is still sleeping\n");
+        abort();
+    }
+
     if (co->caller) {
         fprintf(stderr, "Co-routine re-entered recursively\n");
         abort();

Quoted 4 lines

Is this a different "state" (in Stefan's parlance) than scheduled? In practice both means that someone may call qemu_(aio_)coroutine_enter concurrently, so you'd better not do it yourself.

It is slightly different; it is from sleeping with a timer via co_aio_sleep_ns and waking via co_sleep_cb. Whereas the 'co->scheduled' is specifically from being scheduled for a specific AioContext, via aio_co_schedule().

In practice, 'co->schedule' and 'co->sleeping' certainly rhyme, at the very least.

But having them separate will put the abort closer to where the problem lies, so it should make debugging a bit easier if we hit it.

Quoted 8 lines · Jeff Cody

Quoted 4 lines

Is this a different "state" (in Stefan's parlance) than scheduled? In practice both means that someone may call qemu_(aio_)coroutine_enter concurrently, so you'd better not do it yourself.

It is slightly different; it is from sleeping with a timer via co_aio_sleep_ns and waking via co_sleep_cb. Whereas the 'co->scheduled' is specifically from being scheduled for a specific AioContext, via aio_co_schedule().

Right; however, that would only make a difference if we allowed canceling a co_aio_sleep_ns. Since we don't want that, they have the same transitions.

Quoted 5 lines

In practice, 'co->schedule' and 'co->sleeping' certainly rhyme, at the very least.

But having them separate will put the abort closer to where the problem lies, so it should make debugging a bit easier if we hit it.

What do you mean by closer? It would print a slightly more informative message, but the message is in qemu_aio_coroutine_for both cases.

In fact, unifying co->scheduled and co->sleeping means that you can easily abort when co_aio_sleep_ns is called on a scheduled coroutine, like

    /* This is valid. */
    aio_co_schedule(qemu_get_current_aio_context(),
                    qemu_coroutine_self());

    /* But only if there was a qemu_coroutine_yield here.  */
    co_aio_sleep_ns(qemu_get_current_aio_context(), 1000);

util/qemu-coroutine-sleep.c void coroutine_fn co_aio_sleep_ns(AioContext *ctx, QEMUClockType type, · line 42

Quoted 13 lines · Jeff Cody

util/qemu-coroutine-sleep.c +6 -0

@@ -34,6 +36,7 @@ void coroutine_fn co_aio_sleep_ns(AioContext *ctx, QEMUClockType type,
     CoSleepCB sleep_cb = {
         .co = qemu_coroutine_self(),
     };
+    if (sleep_cb.co->sleeping == 1 || sleep_cb.co->scheduled == 1) {
+       fprintf(stderr, "Cannot sleep a co-routine that is already sleeping "
+                       " or scheduled\n");
+       abort();
+    }
+    sleep_cb.co->sleeping = 1;
     sleep_cb.ts = aio_timer_new(ctx, type, SCALE_NS, co_sleep_cb, &sleep_cb);
     timer_mod(sleep_cb.ts, qemu_clock_get_ns(type) + ns);
     qemu_coroutine_yield();

I understand that this was just an example and not the actual patch, but I'll still point out that this loses the benefit (better error message) of keeping the flags separate.

What do you think about making "scheduled" a const char * and assigning __func__ to it (i.e. either "aio_co_schedule" or "co_aio_sleep_ns")?