[Qemu-devel] [PATCH v8 0/3] NBD reconnect

[Qemu-devel] [PATCH v8 0/3] NBD reconnect

[Vladimir Sementsov-Ogievskiy]

Hi all!
Here is NBD reconnect. Previously, if connection failed all current
and future requests will fail. After the series, nbd-client driver
will try to reconnect unlimited times. During first @reconnect-delay
seconds of reconnecting all requests will wait for the connection,
and if it is established requests will be resent. After
@reconnect-delay period all requests will be failed (until successful
reconnect).

v8:
preparations are already merged [thx to Eric], old 07 with SI_* constants
dropped [Peter]
02: - use NANOSECONDS_PER_SECOND
03: - move to tests/qemu-iotests/264
    - limit job speed, otherwise it fails on ramfs as backup finishes too early

v7:
almost all: rebased on merged nbd.c and nbd-client.c (including patch subject)
01-04: add Eric's r-b
04: wording
05: new
06: rewrite to remove timer earlier
07: new
08:
 - rebase on 05 and 07
 - drop "All rights reserved"
 - handle drain
 - improve handling aio context attach
09: move 249 -> 257

Vladimir Sementsov-Ogievskiy (3):
  qemu-coroutine-sleep: introduce qemu_co_sleep_wake
  block/nbd: nbd reconnect
  iotests: test nbd reconnect

 include/qemu/coroutine.h      |  17 +-
 block/nbd.c                   | 335 +++++++++++++++++++++++++++-------
 block/null.c                  |   2 +-
 block/sheepdog.c              |   2 +-
 tests/test-bdrv-drain.c       |   6 +-
 tests/test-block-iothread.c   |   2 +-
 util/qemu-coroutine-sleep.c   |  47 +++--
 tests/qemu-iotests/264        |  65 +++++++
 tests/qemu-iotests/264.out    |  12 ++
 tests/qemu-iotests/group      |   1 +
 tests/qemu-iotests/iotests.py |   4 +
 11 files changed, 408 insertions(+), 85 deletions(-)
 create mode 100755 tests/qemu-iotests/264
 create mode 100644 tests/qemu-iotests/264.out

-- 
2.18.0

[Qemu-devel] [PATCH v8 1/3] qemu-coroutine-sleep: introduce qemu_co_sleep_wake

[Vladimir Sementsov-Ogievskiy]

Introduce a function to gracefully wake a coroutine sleeping in
qemu_co_sleep_ns().

Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
---
 include/qemu/coroutine.h    | 17 ++++++++++++--
 block/null.c                |  2 +-
 block/sheepdog.c            |  2 +-
 tests/test-bdrv-drain.c     |  6 ++---
 tests/test-block-iothread.c |  2 +-
 util/qemu-coroutine-sleep.c | 47 +++++++++++++++++++++++++++----------
 6 files changed, 55 insertions(+), 21 deletions(-)

diff --git a/include/qemu/coroutine.h b/include/qemu/coroutine.h
index 9801e7f5a4..96780a4902 100644
--- a/include/qemu/coroutine.h
+++ b/include/qemu/coroutine.h
@@ -274,9 +274,22 @@ void qemu_co_rwlock_wrlock(CoRwlock *lock);
 void qemu_co_rwlock_unlock(CoRwlock *lock);
 
 /**
- * Yield the coroutine for a given duration
+ * Yield the coroutine for a given duration. During this yield @sleep_state (if
+ * not NULL) is set to opaque pointer, which may be used for
+ * qemu_co_sleep_wake(). Be careful, the pointer is set back to zero when timer
+ * shoots. Don't save obtained value to other variables and don't call
+ * qemu_co_sleep_wake from another aio context.
  */
-void coroutine_fn qemu_co_sleep_ns(QEMUClockType type, int64_t ns);
+void coroutine_fn qemu_co_sleep_ns(QEMUClockType type, int64_t ns,
+                                   void **sleep_state);
+
+/**
+ * Wake a coroutine if it is sleeping by qemu_co_sleep_ns. Timer will be
+ * deleted. @sleep_state must be the variable which address was given to
+ * qemu_co_sleep_ns() and should be checked to be non-NULL before calling
+ * qemu_co_sleep_wake().
+ */
+void qemu_co_sleep_wake(void *sleep_state);
 
 /**
  * Yield until a file descriptor becomes readable
diff --git a/block/null.c b/block/null.c
index 699aa295cb..1e3f26b07e 100644
--- a/block/null.c
+++ b/block/null.c
@@ -109,7 +109,7 @@ static coroutine_fn int null_co_common(BlockDriverState *bs)
     BDRVNullState *s = bs->opaque;
 
     if (s->latency_ns) {
-        qemu_co_sleep_ns(QEMU_CLOCK_REALTIME, s->latency_ns);
+        qemu_co_sleep_ns(QEMU_CLOCK_REALTIME, s->latency_ns, NULL);
     }
     return 0;
 }
diff --git a/block/sheepdog.c b/block/sheepdog.c
index 773dfc6ab1..3a7ef2f209 100644
--- a/block/sheepdog.c
+++ b/block/sheepdog.c
@@ -743,7 +743,7 @@ static coroutine_fn void reconnect_to_sdog(void *opaque)
         if (s->fd < 0) {
             trace_sheepdog_reconnect_to_sdog();
             error_report_err(local_err);
-            qemu_co_sleep_ns(QEMU_CLOCK_REALTIME, 1000000000ULL);
+            qemu_co_sleep_ns(QEMU_CLOCK_REALTIME, 1000000000ULL, NULL);
         }
     };
 
diff --git a/tests/test-bdrv-drain.c b/tests/test-bdrv-drain.c
index 374bef6bb2..2f53a7add5 100644
--- a/tests/test-bdrv-drain.c
+++ b/tests/test-bdrv-drain.c
@@ -43,7 +43,7 @@ static void coroutine_fn bdrv_test_co_drain_begin(BlockDriverState *bs)
     BDRVTestState *s = bs->opaque;
     s->drain_count++;
     if (s->sleep_in_drain_begin) {
-        qemu_co_sleep_ns(QEMU_CLOCK_REALTIME, 100000);
+        qemu_co_sleep_ns(QEMU_CLOCK_REALTIME, 100000, NULL);
     }
 }
 
@@ -74,7 +74,7 @@ static int coroutine_fn bdrv_test_co_preadv(BlockDriverState *bs,
      * it to complete. We need to sleep a while as bdrv_drain_invoke() comes
      * first and polls its result, too, but it shouldn't accidentally complete
      * this request yet. */
-    qemu_co_sleep_ns(QEMU_CLOCK_REALTIME, 100000);
+    qemu_co_sleep_ns(QEMU_CLOCK_REALTIME, 100000, NULL);
 
     if (s->bh_indirection_ctx) {
         aio_bh_schedule_oneshot(s->bh_indirection_ctx, co_reenter_bh,
@@ -829,7 +829,7 @@ static int coroutine_fn test_job_run(Job *job, Error **errp)
         /* Avoid job_sleep_ns() because it marks the job as !busy. We want to
          * emulate some actual activity (probably some I/O) here so that drain
          * has to wait for this activity to stop. */
-        qemu_co_sleep_ns(QEMU_CLOCK_REALTIME, 1000000);
+        qemu_co_sleep_ns(QEMU_CLOCK_REALTIME, 1000000, NULL);
 
         job_pause_point(&s->common.job);
     }
diff --git a/tests/test-block-iothread.c b/tests/test-block-iothread.c
index 926577b1f9..a1ac5efcaa 100644
--- a/tests/test-block-iothread.c
+++ b/tests/test-block-iothread.c
@@ -381,7 +381,7 @@ static int coroutine_fn test_job_run(Job *job, Error **errp)
          * emulate some actual activity (probably some I/O) here so that the
          * drain involved in AioContext switches has to wait for this activity
          * to stop. */
-        qemu_co_sleep_ns(QEMU_CLOCK_REALTIME, 1000000);
+        qemu_co_sleep_ns(QEMU_CLOCK_REALTIME, 1000000, NULL);
 
         job_pause_point(&s->common.job);
     }
diff --git a/util/qemu-coroutine-sleep.c b/util/qemu-coroutine-sleep.c
index 4bfdd30cbf..48a64bb8d8 100644
--- a/util/qemu-coroutine-sleep.c
+++ b/util/qemu-coroutine-sleep.c
@@ -17,31 +17,52 @@
 #include "qemu/timer.h"
 #include "block/aio.h"
 
-static void co_sleep_cb(void *opaque)
-{
-    Coroutine *co = opaque;
+const char *qemu_co_sleep_ns__scheduled = "qemu_co_sleep_ns";
+
+typedef struct QemuCoSleepState {
+    Coroutine *co;
+    QEMUTimer *ts;
+    void **user_state_pointer;
+} QemuCoSleepState;
 
+void qemu_co_sleep_wake(void *sleep_state)
+{
+    QemuCoSleepState *s = (QemuCoSleepState *)sleep_state;
     /* Write of schedule protected by barrier write in aio_co_schedule */
-    atomic_set(&co->scheduled, NULL);
-    aio_co_wake(co);
+    const char *scheduled = atomic_cmpxchg(&s->co->scheduled,
+                                           qemu_co_sleep_ns__scheduled, NULL);
+
+    assert(scheduled == qemu_co_sleep_ns__scheduled);
+    if (s->user_state_pointer) {
+        *s->user_state_pointer = NULL;
+    }
+    timer_del(s->ts);
+    aio_co_wake(s->co);
 }
 
-void coroutine_fn qemu_co_sleep_ns(QEMUClockType type, int64_t ns)
+void coroutine_fn qemu_co_sleep_ns(QEMUClockType type, int64_t ns,
+                                   void **sleep_state)
 {
     AioContext *ctx = qemu_get_current_aio_context();
-    QEMUTimer *ts;
-    Coroutine *co = qemu_coroutine_self();
+    QemuCoSleepState state = {
+        .co = qemu_coroutine_self(),
+        .ts = aio_timer_new(ctx, type, SCALE_NS, qemu_co_sleep_wake, &state),
+        .user_state_pointer = sleep_state,
+    };
 
-    const char *scheduled = atomic_cmpxchg(&co->scheduled, NULL, __func__);
+    const char *scheduled = atomic_cmpxchg(&state.co->scheduled, NULL,
+                                           qemu_co_sleep_ns__scheduled);
     if (scheduled) {
         fprintf(stderr,
                 "%s: Co-routine was already scheduled in '%s'\n",
                 __func__, scheduled);
         abort();
     }
-    ts = aio_timer_new(ctx, type, SCALE_NS, co_sleep_cb, co);
-    timer_mod(ts, qemu_clock_get_ns(type) + ns);
+
+    if (sleep_state) {
+        *sleep_state = &state;
+    }
+    timer_mod(state.ts, qemu_clock_get_ns(type) + ns);
     qemu_coroutine_yield();
-    timer_del(ts);
-    timer_free(ts);
+    timer_free(state.ts);
 }
-- 
2.18.0

Re: [Qemu-devel] [PATCH v8 1/3] qemu-coroutine-sleep: introduce qemu_co_sleep_wake

[Eric Blake]

[-- Attachment #1.1: Type: text/plain, Size: 5067 bytes --]

On 8/21/19 11:52 AM, Vladimir Sementsov-Ogievskiy wrote:
> Introduce a function to gracefully wake a coroutine sleeping in
> qemu_co_sleep_ns().
> 
> Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
> ---

I'd like a second reviewer on this, but I'll at least give it a spin.

>  include/qemu/coroutine.h    | 17 ++++++++++++--
>  block/null.c                |  2 +-
>  block/sheepdog.c            |  2 +-
>  tests/test-bdrv-drain.c     |  6 ++---
>  tests/test-block-iothread.c |  2 +-
>  util/qemu-coroutine-sleep.c | 47 +++++++++++++++++++++++++++----------
>  6 files changed, 55 insertions(+), 21 deletions(-)

This merely updates existing callers to pass in NULL for the new
argument.  I'd feel a lot better if one of the two tests/ changes also
added a test passing a non-NULL sleep_state, and demonstrated its use.

> 
> diff --git a/include/qemu/coroutine.h b/include/qemu/coroutine.h
> index 9801e7f5a4..96780a4902 100644
> --- a/include/qemu/coroutine.h
> +++ b/include/qemu/coroutine.h
> @@ -274,9 +274,22 @@ void qemu_co_rwlock_wrlock(CoRwlock *lock);
>  void qemu_co_rwlock_unlock(CoRwlock *lock);
>  
>  /**
> - * Yield the coroutine for a given duration
> + * Yield the coroutine for a given duration. During this yield @sleep_state (if

s/yield/yield,/

> + * not NULL) is set to opaque pointer, which may be used for

s/to/to an/

> + * qemu_co_sleep_wake(). Be careful, the pointer is set back to zero when timer
> + * shoots. Don't save obtained value to other variables and don't call

s/when timer shoots/when the timer fires/

s/save/save/the/

> + * qemu_co_sleep_wake from another aio context.
>   */
> -void coroutine_fn qemu_co_sleep_ns(QEMUClockType type, int64_t ns);
> +void coroutine_fn qemu_co_sleep_ns(QEMUClockType type, int64_t ns,
> +                                   void **sleep_state);
> +
> +/**
> + * Wake a coroutine if it is sleeping by qemu_co_sleep_ns. Timer will be

s/by/in/
s/Timer/The timer/

> + * deleted. @sleep_state must be the variable which address was given to

s/which/whose/

> + * qemu_co_sleep_ns() and should be checked to be non-NULL before calling
> + * qemu_co_sleep_wake().
> + */
> +void qemu_co_sleep_wake(void *sleep_state);
>  

> +++ b/util/qemu-coroutine-sleep.c
> @@ -17,31 +17,52 @@
>  #include "qemu/timer.h"
>  #include "block/aio.h"
>  
> -static void co_sleep_cb(void *opaque)
> -{
> -    Coroutine *co = opaque;
> +const char *qemu_co_sleep_ns__scheduled = "qemu_co_sleep_ns";

Why is this not marked static?

> +
> +typedef struct QemuCoSleepState {
> +    Coroutine *co;
> +    QEMUTimer *ts;
> +    void **user_state_pointer;
> +} QemuCoSleepState;
>  
> +void qemu_co_sleep_wake(void *sleep_state)
> +{
> +    QemuCoSleepState *s = (QemuCoSleepState *)sleep_state;

This is C; the cast is not necessary.

>      /* Write of schedule protected by barrier write in aio_co_schedule */
> -    atomic_set(&co->scheduled, NULL);
> -    aio_co_wake(co);
> +    const char *scheduled = atomic_cmpxchg(&s->co->scheduled,
> +                                           qemu_co_sleep_ns__scheduled, NULL);
> +
> +    assert(scheduled == qemu_co_sleep_ns__scheduled);
> +    if (s->user_state_pointer) {
> +        *s->user_state_pointer = NULL;
> +    }
> +    timer_del(s->ts);
> +    aio_co_wake(s->co);
>  }
>  
> -void coroutine_fn qemu_co_sleep_ns(QEMUClockType type, int64_t ns)
> +void coroutine_fn qemu_co_sleep_ns(QEMUClockType type, int64_t ns,
> +                                   void **sleep_state)
>  {
>      AioContext *ctx = qemu_get_current_aio_context();
> -    QEMUTimer *ts;
> -    Coroutine *co = qemu_coroutine_self();
> +    QemuCoSleepState state = {
> +        .co = qemu_coroutine_self(),
> +        .ts = aio_timer_new(ctx, type, SCALE_NS, qemu_co_sleep_wake, &state),
> +        .user_state_pointer = sleep_state,
> +    };
>  
> -    const char *scheduled = atomic_cmpxchg(&co->scheduled, NULL, __func__);
> +    const char *scheduled = atomic_cmpxchg(&state.co->scheduled, NULL,
> +                                           qemu_co_sleep_ns__scheduled);
>      if (scheduled) {
>          fprintf(stderr,
>                  "%s: Co-routine was already scheduled in '%s'\n",
>                  __func__, scheduled);
>          abort();
>      }
> -    ts = aio_timer_new(ctx, type, SCALE_NS, co_sleep_cb, co);
> -    timer_mod(ts, qemu_clock_get_ns(type) + ns);
> +
> +    if (sleep_state) {
> +        *sleep_state = &state;
> +    }
> +    timer_mod(state.ts, qemu_clock_get_ns(type) + ns);
>      qemu_coroutine_yield();
> -    timer_del(ts);
> -    timer_free(ts);
> +    timer_free(state.ts);
>  }

Grammar changes are trivial; and while it is not the most trivial of
patches, I at least follow what it is doing.

Reviewed-by: Eric Blake <eblake@redhat.com>

-- 
Eric Blake, Principal Software Engineer
Red Hat, Inc.           +1-919-301-3226
Virtualization:  qemu.org | libvirt.org


[-- Attachment #2: OpenPGP digital signature --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [Qemu-devel] [PATCH v8 1/3] qemu-coroutine-sleep: introduce qemu_co_sleep_wake

[Kevin Wolf]

Am 21.08.2019 um 18:52 hat Vladimir Sementsov-Ogievskiy geschrieben:
> Introduce a function to gracefully wake a coroutine sleeping in
> qemu_co_sleep_ns().
> 
> Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>

Hm, this has become a bit more complex with the new sleep_state, but
it's probably unavoidable even we need to timer_del() from the callback.

>  include/qemu/coroutine.h    | 17 ++++++++++++--
>  block/null.c                |  2 +-
>  block/sheepdog.c            |  2 +-
>  tests/test-bdrv-drain.c     |  6 ++---
>  tests/test-block-iothread.c |  2 +-
>  util/qemu-coroutine-sleep.c | 47 +++++++++++++++++++++++++++----------
>  6 files changed, 55 insertions(+), 21 deletions(-)
> 
> diff --git a/include/qemu/coroutine.h b/include/qemu/coroutine.h
> index 9801e7f5a4..96780a4902 100644
> --- a/include/qemu/coroutine.h
> +++ b/include/qemu/coroutine.h
> @@ -274,9 +274,22 @@ void qemu_co_rwlock_wrlock(CoRwlock *lock);
>  void qemu_co_rwlock_unlock(CoRwlock *lock);
>  
>  /**
> - * Yield the coroutine for a given duration
> + * Yield the coroutine for a given duration. During this yield @sleep_state (if
> + * not NULL) is set to opaque pointer, which may be used for
> + * qemu_co_sleep_wake(). Be careful, the pointer is set back to zero when timer
> + * shoots. Don't save obtained value to other variables and don't call
> + * qemu_co_sleep_wake from another aio context.
>   */
> -void coroutine_fn qemu_co_sleep_ns(QEMUClockType type, int64_t ns);
> +void coroutine_fn qemu_co_sleep_ns(QEMUClockType type, int64_t ns,
> +                                   void **sleep_state);

Let's make the typedef for QemuCoSleepState public so that we don't have
to use a void pointer here.

I wonder if it would make sense to rename this function and keep
qemu_co_sleep_ns() as a wrapper (maybe even just macro) so that most
callers don't have to add a NULL.

> +/**
> + * Wake a coroutine if it is sleeping by qemu_co_sleep_ns. Timer will be
> + * deleted. @sleep_state must be the variable which address was given to
> + * qemu_co_sleep_ns() and should be checked to be non-NULL before calling
> + * qemu_co_sleep_wake().
> + */
> +void qemu_co_sleep_wake(void *sleep_state);
>  
>  /**
>   * Yield until a file descriptor becomes readable

The actual implementation looks right to me, so with a public
QemuCoSleepState and optionally the wrapper:

Reviewed-by: Kevin Wolf <kwolf@redhat.com>

Re: [Qemu-devel] [PATCH v8 1/3] qemu-coroutine-sleep: introduce qemu_co_sleep_wake

[Vladimir Sementsov-Ogievskiy]

17.09.2019 18:27, Kevin Wolf wrote:
> Am 21.08.2019 um 18:52 hat Vladimir Sementsov-Ogievskiy geschrieben:
>> Introduce a function to gracefully wake a coroutine sleeping in
>> qemu_co_sleep_ns().
>>
>> Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
> 
> Hm, this has become a bit more complex with the new sleep_state, but
> it's probably unavoidable even we need to timer_del() from the callback.
> 
>>   include/qemu/coroutine.h    | 17 ++++++++++++--
>>   block/null.c                |  2 +-
>>   block/sheepdog.c            |  2 +-
>>   tests/test-bdrv-drain.c     |  6 ++---
>>   tests/test-block-iothread.c |  2 +-
>>   util/qemu-coroutine-sleep.c | 47 +++++++++++++++++++++++++++----------
>>   6 files changed, 55 insertions(+), 21 deletions(-)
>>
>> diff --git a/include/qemu/coroutine.h b/include/qemu/coroutine.h
>> index 9801e7f5a4..96780a4902 100644
>> --- a/include/qemu/coroutine.h
>> +++ b/include/qemu/coroutine.h
>> @@ -274,9 +274,22 @@ void qemu_co_rwlock_wrlock(CoRwlock *lock);
>>   void qemu_co_rwlock_unlock(CoRwlock *lock);
>>   
>>   /**
>> - * Yield the coroutine for a given duration
>> + * Yield the coroutine for a given duration. During this yield @sleep_state (if
>> + * not NULL) is set to opaque pointer, which may be used for
>> + * qemu_co_sleep_wake(). Be careful, the pointer is set back to zero when timer
>> + * shoots. Don't save obtained value to other variables and don't call
>> + * qemu_co_sleep_wake from another aio context.
>>    */
>> -void coroutine_fn qemu_co_sleep_ns(QEMUClockType type, int64_t ns);
>> +void coroutine_fn qemu_co_sleep_ns(QEMUClockType type, int64_t ns,
>> +                                   void **sleep_state);
> 
> Let's make the typedef for QemuCoSleepState public so that we don't have
> to use a void pointer here.
> 
> I wonder if it would make sense to rename this function and keep
> qemu_co_sleep_ns() as a wrapper (maybe even just macro) so that most
> callers don't have to add a NULL.

Sure. Strange that I didn't go that way and touched a lot of extra files.

> 
>> +/**
>> + * Wake a coroutine if it is sleeping by qemu_co_sleep_ns. Timer will be
>> + * deleted. @sleep_state must be the variable which address was given to
>> + * qemu_co_sleep_ns() and should be checked to be non-NULL before calling
>> + * qemu_co_sleep_wake().
>> + */
>> +void qemu_co_sleep_wake(void *sleep_state);
>>   
>>   /**
>>    * Yield until a file descriptor becomes readable
> 
> The actual implementation looks right to me, so with a public
> QemuCoSleepState and optionally the wrapper:
> 
> Reviewed-by: Kevin Wolf <kwolf@redhat.com>
> 

Will resend with both things, thank you!

-- 
Best regards,
Vladimir

[Qemu-devel] [PATCH v8 2/3] block/nbd: nbd reconnect

[Vladimir Sementsov-Ogievskiy]

Implement reconnect. To achieve this:

1. add new modes:
   connecting-wait: means, that reconnecting is in progress, and there
     were small number of reconnect attempts, so all requests are
     waiting for the connection.
   connecting-nowait: reconnecting is in progress, there were a lot of
     attempts of reconnect, all requests will return errors.

   two old modes are used too:
   connected: normal state
   quit: exiting after fatal error or on close

Possible transitions are:

   * -> quit
   connecting-* -> connected
   connecting-wait -> connecting-nowait (transition is done after
                      reconnect-delay seconds in connecting-wait mode)
   connected -> connecting-wait

2. Implement reconnect in connection_co. So, in connecting-* mode,
    connection_co, tries to reconnect unlimited times.

3. Retry nbd queries on channel error, if we are in connecting-wait
    state.

Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
---
 block/nbd.c | 335 ++++++++++++++++++++++++++++++++++++++++++----------
 1 file changed, 271 insertions(+), 64 deletions(-)

diff --git a/block/nbd.c b/block/nbd.c
index beed46fb34..f272154d4b 100644
--- a/block/nbd.c
+++ b/block/nbd.c
@@ -1,6 +1,7 @@
 /*
  * QEMU Block driver for  NBD
  *
+ * Copyright (c) 2019 Virtuozzo International GmbH.
  * Copyright (C) 2016 Red Hat, Inc.
  * Copyright (C) 2008 Bull S.A.S.
  *     Author: Laurent Vivier <Laurent.Vivier@bull.net>
@@ -55,6 +56,8 @@ typedef struct {
 } NBDClientRequest;
 
 typedef enum NBDClientState {
+    NBD_CLIENT_CONNECTING_WAIT,
+    NBD_CLIENT_CONNECTING_NOWAIT,
     NBD_CLIENT_CONNECTED,
     NBD_CLIENT_QUIT
 } NBDClientState;
@@ -67,8 +70,14 @@ typedef struct BDRVNBDState {
     CoMutex send_mutex;
     CoQueue free_sema;
     Coroutine *connection_co;
+    void *connection_co_sleep_ns_state;
+    bool drained;
+    bool wait_drained_end;
     int in_flight;
     NBDClientState state;
+    int connect_status;
+    Error *connect_err;
+    bool wait_in_flight;
 
     NBDClientRequest requests[MAX_NBD_REQUESTS];
     NBDReply reply;
@@ -83,10 +92,21 @@ typedef struct BDRVNBDState {
     char *x_dirty_bitmap;
 } BDRVNBDState;
 
-/* @ret will be used for reconnect in future */
+static int nbd_client_connect(BlockDriverState *bs, Error **errp);
+
 static void nbd_channel_error(BDRVNBDState *s, int ret)
 {
-    s->state = NBD_CLIENT_QUIT;
+    if (ret == -EIO) {
+        if (s->state == NBD_CLIENT_CONNECTED) {
+            s->state = s->reconnect_delay ? NBD_CLIENT_CONNECTING_WAIT :
+                                            NBD_CLIENT_CONNECTING_NOWAIT;
+        }
+    } else {
+        if (s->state == NBD_CLIENT_CONNECTED) {
+            qio_channel_shutdown(s->ioc, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
+        }
+        s->state = NBD_CLIENT_QUIT;
+    }
 }
 
 static void nbd_recv_coroutines_wake_all(BDRVNBDState *s)
@@ -129,7 +149,13 @@ static void nbd_client_attach_aio_context(BlockDriverState *bs,
 {
     BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
 
-    qio_channel_attach_aio_context(QIO_CHANNEL(s->ioc), new_context);
+    /*
+     * s->connection_co is either yielded from nbd_receive_reply or from
+     * nbd_reconnect_loop()
+     */
+    if (s->state == NBD_CLIENT_CONNECTED) {
+        qio_channel_attach_aio_context(QIO_CHANNEL(s->ioc), new_context);
+    }
 
     bdrv_inc_in_flight(bs);
 
@@ -140,29 +166,157 @@ static void nbd_client_attach_aio_context(BlockDriverState *bs,
     aio_wait_bh_oneshot(new_context, nbd_client_attach_aio_context_bh, bs);
 }
 
+static void coroutine_fn nbd_client_co_drain_begin(BlockDriverState *bs)
+{
+    BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
 
-static void nbd_teardown_connection(BlockDriverState *bs)
+    s->drained = true;
+    if (s->connection_co_sleep_ns_state) {
+        qemu_co_sleep_wake(s->connection_co_sleep_ns_state);
+    }
+}
+
+static void coroutine_fn nbd_client_co_drain_end(BlockDriverState *bs)
 {
     BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
 
-    assert(s->ioc);
+    s->drained = false;
+    if (s->wait_drained_end) {
+        s->wait_drained_end = false;
+        aio_co_wake(s->connection_co);
+    }
+}
+
 
-    /* finish any pending coroutines */
-    qio_channel_shutdown(s->ioc,
-                         QIO_CHANNEL_SHUTDOWN_BOTH,
-                         NULL);
+static void nbd_teardown_connection(BlockDriverState *bs)
+{
+    BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
+
+    if (s->state == NBD_CLIENT_CONNECTED) {
+        /* finish any pending coroutines */
+        assert(s->ioc);
+        qio_channel_shutdown(s->ioc, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
+    }
+    s->state = NBD_CLIENT_QUIT;
+    if (s->connection_co) {
+        if (s->connection_co_sleep_ns_state) {
+            qemu_co_sleep_wake(s->connection_co_sleep_ns_state);
+        }
+    }
     BDRV_POLL_WHILE(bs, s->connection_co);
+}
 
-    nbd_client_detach_aio_context(bs);
-    object_unref(OBJECT(s->sioc));
-    s->sioc = NULL;
-    object_unref(OBJECT(s->ioc));
-    s->ioc = NULL;
+static bool nbd_client_connecting(BDRVNBDState *s)
+{
+    return s->state == NBD_CLIENT_CONNECTING_WAIT ||
+            s->state == NBD_CLIENT_CONNECTING_NOWAIT;
+}
+
+static bool nbd_client_connecting_wait(BDRVNBDState *s)
+{
+    return s->state == NBD_CLIENT_CONNECTING_WAIT;
+}
+
+static coroutine_fn void nbd_reconnect_attempt(BDRVNBDState *s)
+{
+    Error *local_err = NULL;
+
+    if (!nbd_client_connecting(s)) {
+        return;
+    }
+    assert(nbd_client_connecting(s));
+
+    /* Wait for completion of all in-flight requests */
+
+    qemu_co_mutex_lock(&s->send_mutex);
+
+    while (s->in_flight > 0) {
+        qemu_co_mutex_unlock(&s->send_mutex);
+        nbd_recv_coroutines_wake_all(s);
+        s->wait_in_flight = true;
+        qemu_coroutine_yield();
+        s->wait_in_flight = false;
+        qemu_co_mutex_lock(&s->send_mutex);
+    }
+
+    qemu_co_mutex_unlock(&s->send_mutex);
+
+    if (!nbd_client_connecting(s)) {
+        return;
+    }
+
+    /*
+     * Now we are sure that nobody is accessing the channel, and no one will
+     * try until we set the state to CONNECTED.
+     */
+
+    /* Finalize previous connection if any */
+    if (s->ioc) {
+        nbd_client_detach_aio_context(s->bs);
+        object_unref(OBJECT(s->sioc));
+        s->sioc = NULL;
+        object_unref(OBJECT(s->ioc));
+        s->ioc = NULL;
+    }
+
+    s->connect_status = nbd_client_connect(s->bs, &local_err);
+    error_free(s->connect_err);
+    s->connect_err = NULL;
+    error_propagate(&s->connect_err, local_err);
+    local_err = NULL;
+
+    if (s->connect_status < 0) {
+        /* failed attempt */
+        return;
+    }
+
+    /* successfully connected */
+    s->state = NBD_CLIENT_CONNECTED;
+    qemu_co_queue_restart_all(&s->free_sema);
+}
+
+static coroutine_fn void nbd_reconnect_loop(BDRVNBDState *s)
+{
+    uint64_t start_time_ns = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
+    uint64_t delay_ns = s->reconnect_delay * NANOSECONDS_PER_SECOND;
+    uint64_t timeout = 1 * NANOSECONDS_PER_SECOND;
+    uint64_t max_timeout = 16 * NANOSECONDS_PER_SECOND;
+
+    nbd_reconnect_attempt(s);
+
+    while (nbd_client_connecting(s)) {
+        if (s->state == NBD_CLIENT_CONNECTING_WAIT &&
+            qemu_clock_get_ns(QEMU_CLOCK_REALTIME) - start_time_ns > delay_ns)
+        {
+            s->state = NBD_CLIENT_CONNECTING_NOWAIT;
+            qemu_co_queue_restart_all(&s->free_sema);
+        }
+
+        qemu_co_sleep_ns(QEMU_CLOCK_REALTIME, timeout,
+                         &s->connection_co_sleep_ns_state);
+        if (s->drained) {
+            bdrv_dec_in_flight(s->bs);
+            s->wait_drained_end = true;
+            while (s->drained) {
+                /*
+                 * We may be entered once from nbd_client_attach_aio_context_bh
+                 * and then from nbd_client_co_drain_end. So here is a loop.
+                 */
+                qemu_coroutine_yield();
+            }
+            bdrv_inc_in_flight(s->bs);
+        }
+        if (timeout < max_timeout) {
+            timeout *= 2;
+        }
+
+        nbd_reconnect_attempt(s);
+    }
 }
 
 static coroutine_fn void nbd_connection_entry(void *opaque)
 {
-    BDRVNBDState *s = opaque;
+    BDRVNBDState *s = (BDRVNBDState *)opaque;
     uint64_t i;
     int ret = 0;
     Error *local_err = NULL;
@@ -177,16 +331,26 @@ static coroutine_fn void nbd_connection_entry(void *opaque)
          * Therefore we keep an additional in_flight reference all the time and
          * only drop it temporarily here.
          */
+
+        if (nbd_client_connecting(s)) {
+            nbd_reconnect_loop(s);
+        }
+
+        if (s->state != NBD_CLIENT_CONNECTED) {
+            continue;
+        }
+
         assert(s->reply.handle == 0);
         ret = nbd_receive_reply(s->bs, s->ioc, &s->reply, &local_err);
 
         if (local_err) {
             trace_nbd_read_reply_entry_fail(ret, error_get_pretty(local_err));
             error_free(local_err);
+            local_err = NULL;
         }
         if (ret <= 0) {
             nbd_channel_error(s, ret ? ret : -EIO);
-            break;
+            continue;
         }
 
         /*
@@ -201,7 +365,7 @@ static coroutine_fn void nbd_connection_entry(void *opaque)
             (nbd_reply_is_structured(&s->reply) && !s->info.structured_reply))
         {
             nbd_channel_error(s, -EINVAL);
-            break;
+            continue;
         }
 
         /*
@@ -220,10 +384,19 @@ static coroutine_fn void nbd_connection_entry(void *opaque)
         qemu_coroutine_yield();
     }
 
+    qemu_co_queue_restart_all(&s->free_sema);
     nbd_recv_coroutines_wake_all(s);
     bdrv_dec_in_flight(s->bs);
 
     s->connection_co = NULL;
+    if (s->ioc) {
+        nbd_client_detach_aio_context(s->bs);
+        object_unref(OBJECT(s->sioc));
+        s->sioc = NULL;
+        object_unref(OBJECT(s->ioc));
+        s->ioc = NULL;
+    }
+
     aio_wait_kick();
 }
 
@@ -235,7 +408,7 @@ static int nbd_co_send_request(BlockDriverState *bs,
     int rc, i = -1;
 
     qemu_co_mutex_lock(&s->send_mutex);
-    while (s->in_flight == MAX_NBD_REQUESTS) {
+    while (s->in_flight == MAX_NBD_REQUESTS || nbd_client_connecting_wait(s)) {
         qemu_co_queue_wait(&s->free_sema, &s->send_mutex);
     }
 
@@ -286,7 +459,11 @@ err:
             s->requests[i].coroutine = NULL;
             s->in_flight--;
         }
-        qemu_co_queue_next(&s->free_sema);
+        if (s->in_flight == 0 && s->wait_in_flight) {
+            aio_co_wake(s->connection_co);
+        } else {
+            qemu_co_queue_next(&s->free_sema);
+        }
     }
     qemu_co_mutex_unlock(&s->send_mutex);
     return rc;
@@ -666,10 +843,15 @@ static coroutine_fn int nbd_co_receive_one_chunk(
     } else {
         /* For assert at loop start in nbd_connection_entry */
         *reply = s->reply;
-        s->reply.handle = 0;
     }
+    s->reply.handle = 0;
 
-    if (s->connection_co) {
+    if (s->connection_co && !s->wait_in_flight) {
+        /*
+         * We must check s->wait_in_flight, because we may entered by
+         * nbd_recv_coroutines_wake_all(), in this case we should not
+         * wake connection_co here, it will woken by last request.
+         */
         aio_co_wake(s->connection_co);
     }
 
@@ -781,7 +963,11 @@ break_loop:
 
     qemu_co_mutex_lock(&s->send_mutex);
     s->in_flight--;
-    qemu_co_queue_next(&s->free_sema);
+    if (s->in_flight == 0 && s->wait_in_flight) {
+        aio_co_wake(s->connection_co);
+    } else {
+        qemu_co_queue_next(&s->free_sema);
+    }
     qemu_co_mutex_unlock(&s->send_mutex);
 
     return false;
@@ -927,20 +1113,26 @@ static int nbd_co_request(BlockDriverState *bs, NBDRequest *request,
     } else {
         assert(request->type != NBD_CMD_WRITE);
     }
-    ret = nbd_co_send_request(bs, request, write_qiov);
-    if (ret < 0) {
-        return ret;
-    }
 
-    ret = nbd_co_receive_return_code(s, request->handle,
-                                     &request_ret, &local_err);
-    if (local_err) {
-        trace_nbd_co_request_fail(request->from, request->len, request->handle,
-                                  request->flags, request->type,
-                                  nbd_cmd_lookup(request->type),
-                                  ret, error_get_pretty(local_err));
-        error_free(local_err);
-    }
+    do {
+        ret = nbd_co_send_request(bs, request, write_qiov);
+        if (ret < 0) {
+            continue;
+        }
+
+        ret = nbd_co_receive_return_code(s, request->handle,
+                                         &request_ret, &local_err);
+        if (local_err) {
+            trace_nbd_co_request_fail(request->from, request->len,
+                                      request->handle, request->flags,
+                                      request->type,
+                                      nbd_cmd_lookup(request->type),
+                                      ret, error_get_pretty(local_err));
+            error_free(local_err);
+            local_err = NULL;
+        }
+    } while (ret < 0 && nbd_client_connecting_wait(s));
+
     return ret ? ret : request_ret;
 }
 
@@ -981,20 +1173,24 @@ static int nbd_client_co_preadv(BlockDriverState *bs, uint64_t offset,
         request.len -= slop;
     }
 
-    ret = nbd_co_send_request(bs, &request, NULL);
-    if (ret < 0) {
-        return ret;
-    }
+    do {
+        ret = nbd_co_send_request(bs, &request, NULL);
+        if (ret < 0) {
+            continue;
+        }
+
+        ret = nbd_co_receive_cmdread_reply(s, request.handle, offset, qiov,
+                                           &request_ret, &local_err);
+        if (local_err) {
+            trace_nbd_co_request_fail(request.from, request.len, request.handle,
+                                      request.flags, request.type,
+                                      nbd_cmd_lookup(request.type),
+                                      ret, error_get_pretty(local_err));
+            error_free(local_err);
+            local_err = NULL;
+        }
+    } while (ret < 0 && nbd_client_connecting_wait(s));
 
-    ret = nbd_co_receive_cmdread_reply(s, request.handle, offset, qiov,
-                                       &request_ret, &local_err);
-    if (local_err) {
-        trace_nbd_co_request_fail(request.from, request.len, request.handle,
-                                  request.flags, request.type,
-                                  nbd_cmd_lookup(request.type),
-                                  ret, error_get_pretty(local_err));
-        error_free(local_err);
-    }
     return ret ? ret : request_ret;
 }
 
@@ -1127,20 +1323,25 @@ static int coroutine_fn nbd_client_co_block_status(
     if (s->info.min_block) {
         assert(QEMU_IS_ALIGNED(request.len, s->info.min_block));
     }
-    ret = nbd_co_send_request(bs, &request, NULL);
-    if (ret < 0) {
-        return ret;
-    }
+    do {
+        ret = nbd_co_send_request(bs, &request, NULL);
+        if (ret < 0) {
+            continue;
+        }
+
+        ret = nbd_co_receive_blockstatus_reply(s, request.handle, bytes,
+                                               &extent, &request_ret,
+                                               &local_err);
+        if (local_err) {
+            trace_nbd_co_request_fail(request.from, request.len, request.handle,
+                                      request.flags, request.type,
+                                      nbd_cmd_lookup(request.type),
+                                      ret, error_get_pretty(local_err));
+            error_free(local_err);
+            local_err = NULL;
+        }
+    } while (ret < 0 && nbd_client_connecting_wait(s));
 
-    ret = nbd_co_receive_blockstatus_reply(s, request.handle, bytes,
-                                           &extent, &request_ret, &local_err);
-    if (local_err) {
-        trace_nbd_co_request_fail(request.from, request.len, request.handle,
-                                  request.flags, request.type,
-                                  nbd_cmd_lookup(request.type),
-                                  ret, error_get_pretty(local_err));
-        error_free(local_err);
-    }
     if (ret < 0 || request_ret < 0) {
         return ret ? ret : request_ret;
     }
@@ -1159,9 +1360,9 @@ static void nbd_client_close(BlockDriverState *bs)
     BDRVNBDState *s = (BDRVNBDState *)bs->opaque;
     NBDRequest request = { .type = NBD_CMD_DISC };
 
-    assert(s->ioc);
-
-    nbd_send_request(s->ioc, &request);
+    if (s->ioc) {
+        nbd_send_request(s->ioc, &request);
+    }
 
     nbd_teardown_connection(bs);
 }
@@ -1804,6 +2005,8 @@ static BlockDriver bdrv_nbd = {
     .bdrv_getlength             = nbd_getlength,
     .bdrv_detach_aio_context    = nbd_client_detach_aio_context,
     .bdrv_attach_aio_context    = nbd_client_attach_aio_context,
+    .bdrv_co_drain_begin        = nbd_client_co_drain_begin,
+    .bdrv_co_drain_end          = nbd_client_co_drain_end,
     .bdrv_refresh_filename      = nbd_refresh_filename,
     .bdrv_co_block_status       = nbd_client_co_block_status,
     .bdrv_dirname               = nbd_dirname,
@@ -1826,6 +2029,8 @@ static BlockDriver bdrv_nbd_tcp = {
     .bdrv_getlength             = nbd_getlength,
     .bdrv_detach_aio_context    = nbd_client_detach_aio_context,
     .bdrv_attach_aio_context    = nbd_client_attach_aio_context,
+    .bdrv_co_drain_begin        = nbd_client_co_drain_begin,
+    .bdrv_co_drain_end          = nbd_client_co_drain_end,
     .bdrv_refresh_filename      = nbd_refresh_filename,
     .bdrv_co_block_status       = nbd_client_co_block_status,
     .bdrv_dirname               = nbd_dirname,
@@ -1848,6 +2053,8 @@ static BlockDriver bdrv_nbd_unix = {
     .bdrv_getlength             = nbd_getlength,
     .bdrv_detach_aio_context    = nbd_client_detach_aio_context,
     .bdrv_attach_aio_context    = nbd_client_attach_aio_context,
+    .bdrv_co_drain_begin        = nbd_client_co_drain_begin,
+    .bdrv_co_drain_end          = nbd_client_co_drain_end,
     .bdrv_refresh_filename      = nbd_refresh_filename,
     .bdrv_co_block_status       = nbd_client_co_block_status,
     .bdrv_dirname               = nbd_dirname,
-- 
2.18.0

Re: [Qemu-devel] [PATCH v8 2/3] block/nbd: nbd reconnect

[Eric Blake]

[-- Attachment #1.1: Type: text/plain, Size: 8690 bytes --]

On 8/21/19 11:52 AM, Vladimir Sementsov-Ogievskiy wrote:
> Implement reconnect. To achieve this:
> 
> 1. add new modes:
>    connecting-wait: means, that reconnecting is in progress, and there
>      were small number of reconnect attempts, so all requests are
>      waiting for the connection.
>    connecting-nowait: reconnecting is in progress, there were a lot of
>      attempts of reconnect, all requests will return errors.
> 
>    two old modes are used too:
>    connected: normal state
>    quit: exiting after fatal error or on close
> 
> Possible transitions are:
> 
>    * -> quit
>    connecting-* -> connected
>    connecting-wait -> connecting-nowait (transition is done after
>                       reconnect-delay seconds in connecting-wait mode)
>    connected -> connecting-wait
> 
> 2. Implement reconnect in connection_co. So, in connecting-* mode,
>     connection_co, tries to reconnect unlimited times.
> 
> 3. Retry nbd queries on channel error, if we are in connecting-wait
>     state.
> 
> Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
> ---

> +static bool nbd_client_connecting(BDRVNBDState *s)
> +{
> +    return s->state == NBD_CLIENT_CONNECTING_WAIT ||
> +            s->state == NBD_CLIENT_CONNECTING_NOWAIT;


Indentation looks unusual. I might have done:

    return (s->state == NBD_CLIENT_CONNECTING_WAIT ||
            s->state == NBD_CLIENT_CONNECTING_NOWAIT);

Or even exploit the enum encoding:

    return s->state <= NBD_CLIENT_CONNECTING_NOWAIT

Is s->state updated atomically, or do we risk the case where we might
see two different values of s->state across the || sequence point?  Does
that matter?

> +}
> +
> +static bool nbd_client_connecting_wait(BDRVNBDState *s)
> +{
> +    return s->state == NBD_CLIENT_CONNECTING_WAIT;
> +}
> +
> +static coroutine_fn void nbd_reconnect_attempt(BDRVNBDState *s)
> +{
> +    Error *local_err = NULL;
> +
> +    if (!nbd_client_connecting(s)) {
> +        return;
> +    }
> +    assert(nbd_client_connecting(s));

This assert adds nothing given the condition beforehand.

> +
> +    /* Wait for completion of all in-flight requests */
> +
> +    qemu_co_mutex_lock(&s->send_mutex);
> +
> +    while (s->in_flight > 0) {
> +        qemu_co_mutex_unlock(&s->send_mutex);
> +        nbd_recv_coroutines_wake_all(s);
> +        s->wait_in_flight = true;
> +        qemu_coroutine_yield();
> +        s->wait_in_flight = false;
> +        qemu_co_mutex_lock(&s->send_mutex);
> +    }
> +
> +    qemu_co_mutex_unlock(&s->send_mutex);
> +
> +    if (!nbd_client_connecting(s)) {
> +        return;
> +    }
> +
> +    /*
> +     * Now we are sure that nobody is accessing the channel, and no one will
> +     * try until we set the state to CONNECTED.
> +     */
> +
> +    /* Finalize previous connection if any */
> +    if (s->ioc) {
> +        nbd_client_detach_aio_context(s->bs);
> +        object_unref(OBJECT(s->sioc));
> +        s->sioc = NULL;
> +        object_unref(OBJECT(s->ioc));
> +        s->ioc = NULL;
> +    }
> +
> +    s->connect_status = nbd_client_connect(s->bs, &local_err);
> +    error_free(s->connect_err);
> +    s->connect_err = NULL;
> +    error_propagate(&s->connect_err, local_err);
> +    local_err = NULL;
> +
> +    if (s->connect_status < 0) {
> +        /* failed attempt */
> +        return;
> +    }
> +
> +    /* successfully connected */
> +    s->state = NBD_CLIENT_CONNECTED;
> +    qemu_co_queue_restart_all(&s->free_sema);
> +}
> +
> +static coroutine_fn void nbd_reconnect_loop(BDRVNBDState *s)
> +{
> +    uint64_t start_time_ns = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
> +    uint64_t delay_ns = s->reconnect_delay * NANOSECONDS_PER_SECOND;
> +    uint64_t timeout = 1 * NANOSECONDS_PER_SECOND;
> +    uint64_t max_timeout = 16 * NANOSECONDS_PER_SECOND;
> +
> +    nbd_reconnect_attempt(s);
> +
> +    while (nbd_client_connecting(s)) {
> +        if (s->state == NBD_CLIENT_CONNECTING_WAIT &&
> +            qemu_clock_get_ns(QEMU_CLOCK_REALTIME) - start_time_ns > delay_ns)
> +        {
> +            s->state = NBD_CLIENT_CONNECTING_NOWAIT;
> +            qemu_co_queue_restart_all(&s->free_sema);
> +        }
> +
> +        qemu_co_sleep_ns(QEMU_CLOCK_REALTIME, timeout,
> +                         &s->connection_co_sleep_ns_state);
> +        if (s->drained) {
> +            bdrv_dec_in_flight(s->bs);
> +            s->wait_drained_end = true;
> +            while (s->drained) {
> +                /*
> +                 * We may be entered once from nbd_client_attach_aio_context_bh
> +                 * and then from nbd_client_co_drain_end. So here is a loop.
> +                 */
> +                qemu_coroutine_yield();
> +            }
> +            bdrv_inc_in_flight(s->bs);
> +        }
> +        if (timeout < max_timeout) {
> +            timeout *= 2;
> +        }
> +
> +        nbd_reconnect_attempt(s);
> +    }
>  }
>  
>  static coroutine_fn void nbd_connection_entry(void *opaque)
>  {
> -    BDRVNBDState *s = opaque;
> +    BDRVNBDState *s = (BDRVNBDState *)opaque;

The cast is not necessary.

>      uint64_t i;
>      int ret = 0;
>      Error *local_err = NULL;
> @@ -177,16 +331,26 @@ static coroutine_fn void nbd_connection_entry(void *opaque)
>           * Therefore we keep an additional in_flight reference all the time and
>           * only drop it temporarily here.
>           */
> +
> +        if (nbd_client_connecting(s)) {
> +            nbd_reconnect_loop(s);
> +        }
> +
> +        if (s->state != NBD_CLIENT_CONNECTED) {
> +            continue;
> +        }
> +
>          assert(s->reply.handle == 0);
>          ret = nbd_receive_reply(s->bs, s->ioc, &s->reply, &local_err);
>  
>          if (local_err) {
>              trace_nbd_read_reply_entry_fail(ret, error_get_pretty(local_err));
>              error_free(local_err);
> +            local_err = NULL;
>          }
>          if (ret <= 0) {
>              nbd_channel_error(s, ret ? ret : -EIO);
> -            break;
> +            continue;
>          }
>  
>          /*
> @@ -201,7 +365,7 @@ static coroutine_fn void nbd_connection_entry(void *opaque)
>              (nbd_reply_is_structured(&s->reply) && !s->info.structured_reply))
>          {
>              nbd_channel_error(s, -EINVAL);
> -            break;
> +            continue;
>          }
>  

The commit message says you re-attempt the request after reconnection if
you have not yet timed out from the previous connection; but do you also
need to clear out any partial reply received to make sure the new
request isn't operating on stale assumptions left over if the server
died between two structured chunks?


> @@ -927,20 +1113,26 @@ static int nbd_co_request(BlockDriverState *bs, NBDRequest *request,
>      } else {
>          assert(request->type != NBD_CMD_WRITE);
>      }
> -    ret = nbd_co_send_request(bs, request, write_qiov);
> -    if (ret < 0) {
> -        return ret;
> -    }
>  
> -    ret = nbd_co_receive_return_code(s, request->handle,
> -                                     &request_ret, &local_err);
> -    if (local_err) {
> -        trace_nbd_co_request_fail(request->from, request->len, request->handle,
> -                                  request->flags, request->type,
> -                                  nbd_cmd_lookup(request->type),
> -                                  ret, error_get_pretty(local_err));
> -        error_free(local_err);
> -    }
> +    do {
> +        ret = nbd_co_send_request(bs, request, write_qiov);
> +        if (ret < 0) {
> +            continue;
> +        }
> +
> +        ret = nbd_co_receive_return_code(s, request->handle,
> +                                         &request_ret, &local_err);
> +        if (local_err) {
> +            trace_nbd_co_request_fail(request->from, request->len,
> +                                      request->handle, request->flags,
> +                                      request->type,
> +                                      nbd_cmd_lookup(request->type),
> +                                      ret, error_get_pretty(local_err));
> +            error_free(local_err);
> +            local_err = NULL;
> +        }
> +    } while (ret < 0 && nbd_client_connecting_wait(s));

I ask because nothing seems to reset request_ret here in the new loop.

-- 
Eric Blake, Principal Software Engineer
Red Hat, Inc.           +1-919-301-3226
Virtualization:  qemu.org | libvirt.org


[-- Attachment #2: OpenPGP digital signature --]
[-- Type: application/pgp-signature, Size: 488 bytes --]

Re: [Qemu-devel] [PATCH v8 2/3] block/nbd: nbd reconnect

[Vladimir Sementsov-Ogievskiy]

21.08.2019 20:35, Eric Blake wrote:
> On 8/21/19 11:52 AM, Vladimir Sementsov-Ogievskiy wrote:
>> Implement reconnect. To achieve this:
>>
>> 1. add new modes:
>>     connecting-wait: means, that reconnecting is in progress, and there
>>       were small number of reconnect attempts, so all requests are
>>       waiting for the connection.
>>     connecting-nowait: reconnecting is in progress, there were a lot of
>>       attempts of reconnect, all requests will return errors.
>>
>>     two old modes are used too:
>>     connected: normal state
>>     quit: exiting after fatal error or on close
>>
>> Possible transitions are:
>>
>>     * -> quit
>>     connecting-* -> connected
>>     connecting-wait -> connecting-nowait (transition is done after
>>                        reconnect-delay seconds in connecting-wait mode)
>>     connected -> connecting-wait
>>
>> 2. Implement reconnect in connection_co. So, in connecting-* mode,
>>      connection_co, tries to reconnect unlimited times.
>>
>> 3. Retry nbd queries on channel error, if we are in connecting-wait
>>      state.
>>
>> Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
>> ---
> 
>> +static bool nbd_client_connecting(BDRVNBDState *s)
>> +{
>> +    return s->state == NBD_CLIENT_CONNECTING_WAIT ||
>> +            s->state == NBD_CLIENT_CONNECTING_NOWAIT;
> 
> 
> Indentation looks unusual. I might have done:
> 
>      return (s->state == NBD_CLIENT_CONNECTING_WAIT ||
>              s->state == NBD_CLIENT_CONNECTING_NOWAIT);
> 
> Or even exploit the enum encoding:
> 
>      return s->state <= NBD_CLIENT_CONNECTING_NOWAIT
> 
> Is s->state updated atomically, or do we risk the case where we might
> see two different values of s->state across the || sequence point?  Does
> that matter?

I hope it all happens in one aio context so state change should not intersects with this
function as it doesn't yield.

> 
>> +}
>> +
>> +static bool nbd_client_connecting_wait(BDRVNBDState *s)
>> +{
>> +    return s->state == NBD_CLIENT_CONNECTING_WAIT;
>> +}
>> +
>> +static coroutine_fn void nbd_reconnect_attempt(BDRVNBDState *s)
>> +{
>> +    Error *local_err = NULL;
>> +
>> +    if (!nbd_client_connecting(s)) {
>> +        return;
>> +    }
>> +    assert(nbd_client_connecting(s));
> 
> This assert adds nothing given the condition beforehand.
> 
>> +
>> +    /* Wait for completion of all in-flight requests */
>> +
>> +    qemu_co_mutex_lock(&s->send_mutex);
>> +
>> +    while (s->in_flight > 0) {
>> +        qemu_co_mutex_unlock(&s->send_mutex);
>> +        nbd_recv_coroutines_wake_all(s);
>> +        s->wait_in_flight = true;
>> +        qemu_coroutine_yield();
>> +        s->wait_in_flight = false;
>> +        qemu_co_mutex_lock(&s->send_mutex);
>> +    }
>> +
>> +    qemu_co_mutex_unlock(&s->send_mutex);
>> +
>> +    if (!nbd_client_connecting(s)) {
>> +        return;
>> +    }
>> +
>> +    /*
>> +     * Now we are sure that nobody is accessing the channel, and no one will
>> +     * try until we set the state to CONNECTED.
>> +     */
>> +
>> +    /* Finalize previous connection if any */
>> +    if (s->ioc) {
>> +        nbd_client_detach_aio_context(s->bs);
>> +        object_unref(OBJECT(s->sioc));
>> +        s->sioc = NULL;
>> +        object_unref(OBJECT(s->ioc));
>> +        s->ioc = NULL;
>> +    }
>> +
>> +    s->connect_status = nbd_client_connect(s->bs, &local_err);
>> +    error_free(s->connect_err);
>> +    s->connect_err = NULL;
>> +    error_propagate(&s->connect_err, local_err);
>> +    local_err = NULL;
>> +
>> +    if (s->connect_status < 0) {
>> +        /* failed attempt */
>> +        return;
>> +    }
>> +
>> +    /* successfully connected */
>> +    s->state = NBD_CLIENT_CONNECTED;
>> +    qemu_co_queue_restart_all(&s->free_sema);
>> +}
>> +
>> +static coroutine_fn void nbd_reconnect_loop(BDRVNBDState *s)
>> +{
>> +    uint64_t start_time_ns = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
>> +    uint64_t delay_ns = s->reconnect_delay * NANOSECONDS_PER_SECOND;
>> +    uint64_t timeout = 1 * NANOSECONDS_PER_SECOND;
>> +    uint64_t max_timeout = 16 * NANOSECONDS_PER_SECOND;
>> +
>> +    nbd_reconnect_attempt(s);
>> +
>> +    while (nbd_client_connecting(s)) {
>> +        if (s->state == NBD_CLIENT_CONNECTING_WAIT &&
>> +            qemu_clock_get_ns(QEMU_CLOCK_REALTIME) - start_time_ns > delay_ns)
>> +        {
>> +            s->state = NBD_CLIENT_CONNECTING_NOWAIT;
>> +            qemu_co_queue_restart_all(&s->free_sema);
>> +        }
>> +
>> +        qemu_co_sleep_ns(QEMU_CLOCK_REALTIME, timeout,
>> +                         &s->connection_co_sleep_ns_state);
>> +        if (s->drained) {
>> +            bdrv_dec_in_flight(s->bs);
>> +            s->wait_drained_end = true;
>> +            while (s->drained) {
>> +                /*
>> +                 * We may be entered once from nbd_client_attach_aio_context_bh
>> +                 * and then from nbd_client_co_drain_end. So here is a loop.
>> +                 */
>> +                qemu_coroutine_yield();
>> +            }
>> +            bdrv_inc_in_flight(s->bs);
>> +        }
>> +        if (timeout < max_timeout) {
>> +            timeout *= 2;
>> +        }
>> +
>> +        nbd_reconnect_attempt(s);
>> +    }
>>   }
>>   
>>   static coroutine_fn void nbd_connection_entry(void *opaque)
>>   {
>> -    BDRVNBDState *s = opaque;
>> +    BDRVNBDState *s = (BDRVNBDState *)opaque;
> 
> The cast is not necessary.
> 
>>       uint64_t i;
>>       int ret = 0;
>>       Error *local_err = NULL;
>> @@ -177,16 +331,26 @@ static coroutine_fn void nbd_connection_entry(void *opaque)
>>            * Therefore we keep an additional in_flight reference all the time and
>>            * only drop it temporarily here.
>>            */
>> +
>> +        if (nbd_client_connecting(s)) {
>> +            nbd_reconnect_loop(s);
>> +        }
>> +
>> +        if (s->state != NBD_CLIENT_CONNECTED) {
>> +            continue;
>> +        }
>> +
>>           assert(s->reply.handle == 0);
>>           ret = nbd_receive_reply(s->bs, s->ioc, &s->reply, &local_err);
>>   
>>           if (local_err) {
>>               trace_nbd_read_reply_entry_fail(ret, error_get_pretty(local_err));
>>               error_free(local_err);
>> +            local_err = NULL;
>>           }
>>           if (ret <= 0) {
>>               nbd_channel_error(s, ret ? ret : -EIO);
>> -            break;
>> +            continue;
>>           }
>>   
>>           /*
>> @@ -201,7 +365,7 @@ static coroutine_fn void nbd_connection_entry(void *opaque)
>>               (nbd_reply_is_structured(&s->reply) && !s->info.structured_reply))
>>           {
>>               nbd_channel_error(s, -EINVAL);
>> -            break;
>> +            continue;
>>           }
>>   
> 
> The commit message says you re-attempt the request after reconnection if
> you have not yet timed out from the previous connection; but do you also
> need to clear out any partial reply received to make sure the new
> request isn't operating on stale assumptions left over if the server
> died between two structured chunks?


In nbd_reconnect_attempt we "Wait for completion of all in-flight requests", so
all in-flight requests are failed, and no partial progress appears at reconnect point.

> 
> 
>> @@ -927,20 +1113,26 @@ static int nbd_co_request(BlockDriverState *bs, NBDRequest *request,
>>       } else {
>>           assert(request->type != NBD_CMD_WRITE);
>>       }
>> -    ret = nbd_co_send_request(bs, request, write_qiov);
>> -    if (ret < 0) {
>> -        return ret;
>> -    }
>>   
>> -    ret = nbd_co_receive_return_code(s, request->handle,
>> -                                     &request_ret, &local_err);
>> -    if (local_err) {
>> -        trace_nbd_co_request_fail(request->from, request->len, request->handle,
>> -                                  request->flags, request->type,
>> -                                  nbd_cmd_lookup(request->type),
>> -                                  ret, error_get_pretty(local_err));
>> -        error_free(local_err);
>> -    }
>> +    do {
>> +        ret = nbd_co_send_request(bs, request, write_qiov);
>> +        if (ret < 0) {
>> +            continue;
>> +        }
>> +
>> +        ret = nbd_co_receive_return_code(s, request->handle,
>> +                                         &request_ret, &local_err);
>> +        if (local_err) {
>> +            trace_nbd_co_request_fail(request->from, request->len,
>> +                                      request->handle, request->flags,
>> +                                      request->type,
>> +                                      nbd_cmd_lookup(request->type),
>> +                                      ret, error_get_pretty(local_err));
>> +            error_free(local_err);
>> +            local_err = NULL;
>> +        }
>> +    } while (ret < 0 && nbd_client_connecting_wait(s));
> 
> I ask because nothing seems to reset request_ret here in the new loop.
> 

We don't need to reset it. It is set only on the last iterations, as if it is set ret
must be 0.

-- 
Best regards,
Vladimir

[Qemu-devel] [PATCH v8 3/3] iotests: test nbd reconnect

[Vladimir Sementsov-Ogievskiy]

Add test, which starts backup to nbd target and restarts nbd server
during backup.

Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
---
 tests/qemu-iotests/264        | 65 +++++++++++++++++++++++++++++++++++
 tests/qemu-iotests/264.out    | 12 +++++++
 tests/qemu-iotests/group      |  1 +
 tests/qemu-iotests/iotests.py |  4 +++
 4 files changed, 82 insertions(+)
 create mode 100755 tests/qemu-iotests/264
 create mode 100644 tests/qemu-iotests/264.out

diff --git a/tests/qemu-iotests/264 b/tests/qemu-iotests/264
new file mode 100755
index 0000000000..e70f91c5ca
--- /dev/null
+++ b/tests/qemu-iotests/264
@@ -0,0 +1,65 @@
+#!/usr/bin/env python
+#
+# Test nbd reconnect
+#
+# Copyright (c) 2019 Virtuozzo International GmbH. All rights reserved.
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+
+import time
+
+import iotests
+from iotests import qemu_img_create, file_path, qemu_nbd_popen, log
+
+disk_a, disk_b, nbd_sock = file_path('disk_a', 'disk_b', 'nbd-sock')
+
+qemu_img_create('-f', iotests.imgfmt, disk_a, '5M')
+qemu_img_create('-f', iotests.imgfmt, disk_b, '5M')
+srv = qemu_nbd_popen('-k', nbd_sock, '-f', iotests.imgfmt, disk_b)
+time.sleep(1)
+
+vm = iotests.VM().add_drive(disk_a)
+vm.launch()
+vm.hmp_qemu_io('drive0', 'write 0 5M')
+
+vm.qmp_log('blockdev-add', filters=[iotests.filter_qmp_testfiles],
+           **{'node_name': 'backup0',
+              'driver': 'raw',
+              'file': {'driver': 'nbd',
+                       'server': {'type': 'unix', 'path': nbd_sock},
+                       'reconnect-delay': 10}})
+vm.qmp_log('blockdev-backup', device='drive0', sync='full', target='backup0',
+           speed=(1 * 1024 * 1024))
+
+time.sleep(1)
+log('Kill NBD server')
+srv.kill()
+
+jobs = vm.qmp('query-block-jobs')['return']
+if jobs and jobs[0]['offset'] < jobs[0]['len']:
+    log('Backup job is still in progress')
+
+time.sleep(1)
+
+log('Start NBD server')
+srv = qemu_nbd_popen('-k', nbd_sock, '-f', iotests.imgfmt, disk_b)
+
+vm.qmp_log('block-job-set-speed', device='drive0', speed=0)
+e = vm.event_wait('BLOCK_JOB_COMPLETED')
+log('Backup completed: {}'.format(e['data']['offset']))
+
+vm.qmp_log('blockdev-del', node_name='backup0')
+srv.kill()
+vm.shutdown()
diff --git a/tests/qemu-iotests/264.out b/tests/qemu-iotests/264.out
new file mode 100644
index 0000000000..4a2f4aa509
--- /dev/null
+++ b/tests/qemu-iotests/264.out
@@ -0,0 +1,12 @@
+{"execute": "blockdev-add", "arguments": {"driver": "raw", "file": {"driver": "nbd", "reconnect-delay": 10, "server": {"path": "TEST_DIR/PID-nbd-sock", "type": "unix"}}, "node-name": "backup0"}}
+{"return": {}}
+{"execute": "blockdev-backup", "arguments": {"device": "drive0", "speed": 1048576, "sync": "full", "target": "backup0"}}
+{"return": {}}
+Kill NBD server
+Backup job is still in progress
+Start NBD server
+{"execute": "block-job-set-speed", "arguments": {"device": "drive0", "speed": 0}}
+{"return": {}}
+Backup completed: 5242880
+{"execute": "blockdev-del", "arguments": {"node-name": "backup0"}}
+{"return": {}}
diff --git a/tests/qemu-iotests/group b/tests/qemu-iotests/group
index d95d556414..34c2b89108 100644
--- a/tests/qemu-iotests/group
+++ b/tests/qemu-iotests/group
@@ -274,3 +274,4 @@
 257 rw
 258 rw quick
 262 rw quick migration
+264 rw quick
diff --git a/tests/qemu-iotests/iotests.py b/tests/qemu-iotests/iotests.py
index 84438e837c..ab2f8f7d1b 100644
--- a/tests/qemu-iotests/iotests.py
+++ b/tests/qemu-iotests/iotests.py
@@ -229,6 +229,10 @@ def qemu_nbd_early_pipe(*args):
     else:
         return exitcode, subp.communicate()[0]
 
+def qemu_nbd_popen(*args):
+    '''Run qemu-nbd in daemon mode and return the parent's exit code'''
+    return subprocess.Popen(qemu_nbd_args + ['--persistent'] + list(args))
+
 def compare_images(img1, img2, fmt1=imgfmt, fmt2=imgfmt):
     '''Return True if two image files are identical'''
     return qemu_img('compare', '-f', fmt1,
-- 
2.18.0

Re: [Qemu-devel] [PATCH v8 0/3] NBD reconnect

[Vladimir Sementsov-Ogievskiy]

ping

21.08.2019 19:52, Vladimir Sementsov-Ogievskiy wrote:
> Hi all!
> Here is NBD reconnect. Previously, if connection failed all current
> and future requests will fail. After the series, nbd-client driver
> will try to reconnect unlimited times. During first @reconnect-delay
> seconds of reconnecting all requests will wait for the connection,
> and if it is established requests will be resent. After
> @reconnect-delay period all requests will be failed (until successful
> reconnect).
> 
> v8:
> preparations are already merged [thx to Eric], old 07 with SI_* constants
> dropped [Peter]
> 02: - use NANOSECONDS_PER_SECOND
> 03: - move to tests/qemu-iotests/264
>      - limit job speed, otherwise it fails on ramfs as backup finishes too early
> 
> v7:
> almost all: rebased on merged nbd.c and nbd-client.c (including patch subject)
> 01-04: add Eric's r-b
> 04: wording
> 05: new
> 06: rewrite to remove timer earlier
> 07: new
> 08:
>   - rebase on 05 and 07
>   - drop "All rights reserved"
>   - handle drain
>   - improve handling aio context attach
> 09: move 249 -> 257
> 
> Vladimir Sementsov-Ogievskiy (3):
>    qemu-coroutine-sleep: introduce qemu_co_sleep_wake
>    block/nbd: nbd reconnect
>    iotests: test nbd reconnect
> 
>   include/qemu/coroutine.h      |  17 +-
>   block/nbd.c                   | 335 +++++++++++++++++++++++++++-------
>   block/null.c                  |   2 +-
>   block/sheepdog.c              |   2 +-
>   tests/test-bdrv-drain.c       |   6 +-
>   tests/test-block-iothread.c   |   2 +-
>   util/qemu-coroutine-sleep.c   |  47 +++--
>   tests/qemu-iotests/264        |  65 +++++++
>   tests/qemu-iotests/264.out    |  12 ++
>   tests/qemu-iotests/group      |   1 +
>   tests/qemu-iotests/iotests.py |   4 +
>   11 files changed, 408 insertions(+), 85 deletions(-)
>   create mode 100755 tests/qemu-iotests/264
>   create mode 100644 tests/qemu-iotests/264.out
> 


-- 
Best regards,
Vladimir

[Qemu-devel] ping Re: [PATCH v8 0/3] NBD reconnect

[Vladimir Sementsov-Ogievskiy]

ping

21.08.2019 19:52, Vladimir Sementsov-Ogievskiy wrote:
> Hi all!
> Here is NBD reconnect. Previously, if connection failed all current
> and future requests will fail. After the series, nbd-client driver
> will try to reconnect unlimited times. During first @reconnect-delay
> seconds of reconnecting all requests will wait for the connection,
> and if it is established requests will be resent. After
> @reconnect-delay period all requests will be failed (until successful
> reconnect).
> 
> v8:
> preparations are already merged [thx to Eric], old 07 with SI_* constants
> dropped [Peter]
> 02: - use NANOSECONDS_PER_SECOND
> 03: - move to tests/qemu-iotests/264
>      - limit job speed, otherwise it fails on ramfs as backup finishes too early
> 
> v7:
> almost all: rebased on merged nbd.c and nbd-client.c (including patch subject)
> 01-04: add Eric's r-b
> 04: wording
> 05: new
> 06: rewrite to remove timer earlier
> 07: new
> 08:
>   - rebase on 05 and 07
>   - drop "All rights reserved"
>   - handle drain
>   - improve handling aio context attach
> 09: move 249 -> 257
> 
> Vladimir Sementsov-Ogievskiy (3):
>    qemu-coroutine-sleep: introduce qemu_co_sleep_wake
>    block/nbd: nbd reconnect
>    iotests: test nbd reconnect
> 
>   include/qemu/coroutine.h      |  17 +-
>   block/nbd.c                   | 335 +++++++++++++++++++++++++++-------
>   block/null.c                  |   2 +-
>   block/sheepdog.c              |   2 +-
>   tests/test-bdrv-drain.c       |   6 +-
>   tests/test-block-iothread.c   |   2 +-
>   util/qemu-coroutine-sleep.c   |  47 +++--
>   tests/qemu-iotests/264        |  65 +++++++
>   tests/qemu-iotests/264.out    |  12 ++
>   tests/qemu-iotests/group      |   1 +
>   tests/qemu-iotests/iotests.py |   4 +
>   11 files changed, 408 insertions(+), 85 deletions(-)
>   create mode 100755 tests/qemu-iotests/264
>   create mode 100644 tests/qemu-iotests/264.out
> 


-- 
Best regards,
Vladimir