Re: [PATCH 2/2] nbd/server: Quiesce coroutines on context switch

[Eric Blake]

On 12/4/20 10:53 AM, Sergio Lopez wrote:
> When switching between AIO contexts we need to me make sure that both
> recv_coroutine and send_coroutine are not scheduled to run. Otherwise,
> QEMU may crash while attaching the new context with an error like
> this one:
> 
> aio_co_schedule: Co-routine was already scheduled in 'aio_co_schedule'
> 
> To achieve this we need a local implementation of
> 'qio_channel_readv_all_eof' named 'nbd_read_eof' (a trick already done
> by 'nbd/client.c') that allows us to interrupt the operation and to
> know when recv_coroutine is yielding.
> 
> With this in place, we delegate detaching the AIO context to the
> owning context with a BH ('nbd_aio_detach_bh') scheduled using
> 'aio_wait_bh_oneshot'. This BH signals that we need to quiesce the
> channel by setting 'client->quiescing' to 'true', and either waits for
> the coroutine to finish using AIO_WAIT_WHILE or, if it's yielding in
> 'nbd_read_eof', actively enters the coroutine to interrupt it.
> 
> RHBZ: https://bugzilla.redhat.com/show_bug.cgi?id=1900326
> Signed-off-by: Sergio Lopez <slp@redhat.com>
> ---
>  nbd/server.c | 120 +++++++++++++++++++++++++++++++++++++++++++++------
>  1 file changed, 106 insertions(+), 14 deletions(-)

A complex patch, so I'd appreciate a second set of eyes.

> 
> diff --git a/nbd/server.c b/nbd/server.c
> index 613ed2634a..7229f487d2 100644
> --- a/nbd/server.c
> +++ b/nbd/server.c
> @@ -132,6 +132,9 @@ struct NBDClient {
>      CoMutex send_lock;
>      Coroutine *send_coroutine;
>  
> +    bool read_yielding;
> +    bool quiescing;

Will either of these fields need to be accessed atomically once the
'yank' code is added, or are we still safe with direct access because
coroutines are not multithreaded?

> +
>      QTAILQ_ENTRY(NBDClient) next;
>      int nb_requests;
>      bool closing;
> @@ -1352,14 +1355,60 @@ static coroutine_fn int nbd_negotiate(NBDClient 
> *client, Error **errp)
>      return 0;
>  }
>  
> -static int nbd_receive_request(QIOChannel *ioc, NBDRequest *request,
> +/* nbd_read_eof
> + * Tries to read @size bytes from @ioc. This is a local implementation of
> + * qio_channel_readv_all_eof. We have it here because we need it to be
> + * interruptible and to know when the coroutine is yielding.
> + * Returns 1 on success
> + *         0 on eof, when no data was read (errp is not set)
> + *         negative errno on failure (errp is set)
> + */
> +static inline int coroutine_fn
> +nbd_read_eof(NBDClient *client, void *buffer, size_t size, Error **errp)
> +{
> +    bool partial = false;
> +
> +    assert(size);
> +    while (size > 0) {
> +        struct iovec iov = { .iov_base = buffer, .iov_len = size };
> +        ssize_t len;
> +
> +        len = qio_channel_readv(client->ioc, &iov, 1, errp);
> +        if (len == QIO_CHANNEL_ERR_BLOCK) {
> +            client->read_yielding = true;
> +            qio_channel_yield(client->ioc, G_IO_IN);
> +            client->read_yielding = false;

nbd/client.c:nbd_read_eof() uses bdrv_dec/inc_in_flight instead of
read_yielding...

> +            if (client->quiescing) {
> +                return -EAGAIN;
> +            }

and the quiescing check is new; otherwise, these two functions look
identical.  Having two static functions with the same name makes gdb a
bit more annoying (which one of the two did you want your breakpoint
on?).  Is there any way we could write this code only once in
nbd/common.c for reuse by both client and server?  But I can live with
it as written.

> @@ -2151,20 +2223,23 @@ static int nbd_co_send_bitmap(NBDClient *client, 
> uint64_t handle,
>  
>  /* nbd_co_receive_request
>   * Collect a client request. Return 0 if request looks valid, -EIO to drop
> - * connection right away, and any other negative value to report an error to
> - * the client (although the caller may still need to disconnect after 
> reporting
> - * the error).
> + * connection right away, -EAGAIN to indicate we were interrupted and the
> + * channel should be quiesced, and any other negative value to report an 
> error
> + * to the client (although the caller may still need to disconnect after
> + * reporting the error).
>   */
>  static int nbd_co_receive_request(NBDRequestData *req, NBDRequest *request,
>                                    Error **errp)
>  {
>      NBDClient *client = req->client;
>      int valid_flags;
> +    int ret;
>  
>      g_assert(qemu_in_coroutine());
>      assert(client->recv_coroutine == qemu_coroutine_self());
> -    if (nbd_receive_request(client->ioc, request, errp) < 0) {
> -        return -EIO;
> +    ret = nbd_receive_request(client, request, errp);
> +    if (ret < 0) {
> +        return  ret;

Why the double space?

The old code slams to EIO, you preserve errors.  Is that going to bite
us by causing us to see a different errno leaked through?

>      }
>  
>      trace_nbd_co_receive_request_decode_type(request->handle, request->type,
> @@ -2507,6 +2582,17 @@ static coroutine_fn void nbd_trip(void *opaque)
>          return;
>      }
>  
> +    if (client->quiescing) {
> +        /*
> +         * We're switching between AIO contexts. Don't attempt to receive a 
> new
> +         * request and kick the main context which may be waiting for us.

s/request/request,/

> +         */
> +        nbd_client_put(client);
> +        client->recv_coroutine = NULL;
> +        aio_wait_kick();
> +        return;
> +    }
> +
>      req = nbd_request_get(client);
>      ret = nbd_co_receive_request(req, &request, &local_err);
>      client->recv_coroutine = NULL;
> @@ -2519,6 +2605,11 @@ static coroutine_fn void nbd_trip(void *opaque)
>          goto done;
>      }
>  
> +    if (ret == -EAGAIN) {
> +        assert(client->quiescing);
> +        goto done;
> +    }
> +
>      nbd_client_receive_next_request(client);
>      if (ret == -EIO) {
>          goto disconnect;
> @@ -2565,7 +2656,8 @@ disconnect:
>  
>  static void nbd_client_receive_next_request(NBDClient *client)
>  {
> -    if (!client->recv_coroutine && client->nb_requests < MAX_NBD_REQUESTS) {
> +    if (!client->recv_coroutine && client->nb_requests < MAX_NBD_REQUESTS &&
> +        !client->quiescing) {
>          nbd_client_get(client);
>          client->recv_coroutine = qemu_coroutine_create(nbd_trip, client);
>          aio_co_schedule(client->exp->common.ctx, client->recv_coroutine);
> 

Overall looks okay to me,
Reviewed-by: Eric Blake <eblake@redhat.com>

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