nbd client refactoring and fixing

"NBDReply *reply" parameter of nbd_co_receive_reply is used only to pass return value for nbd_co_request (reply.error). Remove it and use function return value instead.

Signed-off-by Vladimir Sementsov-Ogievskiy <[email protected]>

---
 block/nbd-client.c | 30 +++++++++++++++---------------
 1 file changed, 15 insertions(+), 15 deletions(-)

Check reply-handle == request-handle in the same place, where

recv coroutine number is calculated from reply->handle and it's correctness checked - in nbd_read_reply_entry.

Also finish nbd_read_reply_entry in case of reply-handle != request-handle in the same way as in case of incorrect reply-handle.

Signed-off-by Vladimir Sementsov-Ogievskiy <[email protected]>

---
 block/nbd-client.h | 1 +
 block/nbd-client.c | 8 ++++++--
 2 files changed, 7 insertions(+), 2 deletions(-)

Check reply-handle == request-handle in the same place, where recv coroutine number is calculated from reply->handle and it's correctness checked - in nbd_read_reply_entry.

Also finish nbd_read_reply_entry in case of reply-handle != request-handle in the same way as in case of incorrect reply-handle.

Signed-off-by Vladimir Sementsov-Ogievskiy <[email protected]>

---
 block/nbd-client.h | 1 +
 block/nbd-client.c | 8 ++++++--
 2 files changed, 7 insertions(+), 2 deletions(-)

Read the whole reply in one place - in nbd_read_reply_entry.

Signed-off-by Vladimir Sementsov-Ogievskiy <[email protected]>

---
 block/nbd-client.h |  1 +
 block/nbd-client.c | 42 ++++++++++++++++++++++++------------------
 2 files changed, 25 insertions(+), 18 deletions(-)

block/nbd-client.h +1 -0

index b1900e6a6d..3f97d31013 100644
@@ -21,6 +21,7 @@ typedef struct {
     Coroutine *coroutine;
     bool receiving;         /* waiting for read_reply_co? */
     NBDRequest *request;
+    QEMUIOVector *qiov;
 } NBDClientRequest;
 
 typedef struct NBDClientSession {

block/nbd-client.c +4 -0

index 5f241ecc22..f7b642f079 100644
@@ -98,6 +98,21 @@ static coroutine_fn void nbd_read_reply_entry(void *opaque)
             break;
         }
 
+        if (s->reply.error == 0 &&
+            s->requests[i].request->type == NBD_CMD_READ)
+        {
+            QEMUIOVector *qiov = s->requests[i].qiov;

block/nbd-client.c +9 -18

         /* We're woken up again by the request itself.  Note that there
          * is no race between yielding and reentering read_reply_co.  This
          * is because:

@@ -118,6 +133,7 @@ static coroutine_fn void nbd_read_reply_entry(void *opaque)
     s->read_reply_co = NULL;
 }
 
+/* qiov should be provided for READ and WRITE */
 static int nbd_co_send_request(BlockDriverState *bs,
                                NBDRequest *request,
                                QEMUIOVector *qiov)

@@ -145,6 +161,7 @@ static int nbd_co_send_request(BlockDriverState *bs,
 
     request->handle = INDEX_TO_HANDLE(s, i);
     s->requests[i].request = request;
+    s->requests[i].qiov = qiov;
 
     if (s->quit) {
         rc = -EIO;

@@ -155,7 +172,8 @@ static int nbd_co_send_request(BlockDriverState *bs,
         goto err;
     }
 
-    if (qiov) {
+    if (s->requests[i].request->type == NBD_CMD_WRITE) {
+        assert(qiov);
         qio_channel_set_cork(s->ioc, true);
         rc = nbd_send_request(s->ioc, request);
         if (rc >= 0 && !s->quit) {

@@ -181,9 +199,7 @@ err:
     return rc;
 }
 
-static int nbd_co_receive_reply(NBDClientSession *s,
-                                NBDRequest *request,
-                                QEMUIOVector *qiov)
+static int nbd_co_receive_reply(NBDClientSession *s, NBDRequest *request)
 {
     int ret;
     int i = HANDLE_TO_INDEX(s, request->handle);

@@ -197,14 +213,6 @@ static int nbd_co_receive_reply(NBDClientSession *s,
     } else {
         assert(s->reply.handle == request->handle);
         ret = -s->reply.error;
-        if (qiov && s->reply.error == 0) {
-            assert(request->len == iov_size(qiov->iov, qiov->niov));
-            if (qio_channel_readv_all(s->ioc, qiov->iov, qiov->niov,
-                                      NULL) < 0) {
-                ret = -EIO;
-                s->quit = true;
-            }
-        }
 
         /* Tell the read handler to read another header.  */
         s->reply.handle = 0;

@@ -232,16 +240,14 @@ static int nbd_co_request(BlockDriverState *bs,
     NBDClientSession *client = nbd_get_client_session(bs);
     int ret;
 
-    assert(!qiov || request->type == NBD_CMD_WRITE ||
-           request->type == NBD_CMD_READ);
-    ret = nbd_co_send_request(bs, request,
-                              request->type == NBD_CMD_WRITE ? qiov : NULL);
+    assert((qiov == NULL) !=
+           (request->type == NBD_CMD_WRITE || request->type == NBD_CMD_READ));
+    ret = nbd_co_send_request(bs, request, qiov);
     if (ret < 0) {
         return ret;
     }
 
-    return nbd_co_receive_reply(client, request,
-                                request->type == NBD_CMD_READ ? qiov : NULL);
+    return nbd_co_receive_reply(client, request);
 }
 
 int nbd_client_co_preadv(BlockDriverState *bs, uint64_t offset,

Quoted 8 lines · Vladimir Sementsov-Ogievskiy

Read the whole reply in one place - in nbd_read_reply_entry.

Signed-off-by Vladimir Sementsov-Ogievskiy <[email protected]>

---
 block/nbd-client.h |  1 +
 block/nbd-client.c | 42 ++++++++++++++++++++++++------------------
 2 files changed, 25 insertions(+), 18 deletions(-)

I am not sure this is an improvement. In principle you could have commands that read replies a bit at a time without using a QEMUIOVector.

Quoted 32 lines · Vladimir Sementsov-Ogievskiy

Read the whole reply in one place - in nbd_read_reply_entry.

Signed-off-by Vladimir Sementsov-Ogievskiy <[email protected]>

---
  block/nbd-client.h |  1 +
  block/nbd-client.c | 42 ++++++++++++++++++++++++------------------
  2 files changed, 25 insertions(+), 18 deletions(-)

block/nbd-client.h +1 -0

index b1900e6a6d..3f97d31013 100644
@@ -21,6 +21,7 @@ typedef struct {
      Coroutine *coroutine;
      bool receiving;         /* waiting for read_reply_co? */
      NBDRequest *request;
+    QEMUIOVector *qiov;
  } NBDClientRequest;
 
  typedef struct NBDClientSession {

block/nbd-client.c +4 -0

index 5f241ecc22..f7b642f079 100644
@@ -98,6 +98,21 @@ static coroutine_fn void nbd_read_reply_entry(void *opaque)
              break;
          }
 
+        if (s->reply.error == 0 &&
+            s->requests[i].request->type == NBD_CMD_READ)
+        {
+            QEMUIOVector *qiov = s->requests[i].qiov;

block/nbd-client.c static coroutine_fn void nbd_read_reply_entry(void *opaque) · line 114

Quoted 10 lines

block/nbd-client.c +10 -0

+            assert(qiov != NULL);
+            assert(s->requests[i].request->len ==
+                   iov_size(qiov->iov, qiov->niov));
+            if (qio_channel_readv_all(s->ioc, qiov->iov, qiov->niov,
+                                      NULL) < 0)
+            {
+                s->reply.error = EIO;
+                break;
+            }
+        }

I am not sure this is an improvement. In principle you could have commands that read replies a bit at a time without using a QEMUIOVector.

Paolo

Quoted 78 lines

block/nbd-client.c +9 -18

          /* We're woken up again by the request itself.  Note that there
           * is no race between yielding and reentering read_reply_co.  This
           * is because:

@@ -118,6 +133,7 @@ static coroutine_fn void nbd_read_reply_entry(void *opaque)
      s->read_reply_co = NULL;
  }
 
+/* qiov should be provided for READ and WRITE */
  static int nbd_co_send_request(BlockDriverState *bs,
                                 NBDRequest *request,
                                 QEMUIOVector *qiov)

@@ -145,6 +161,7 @@ static int nbd_co_send_request(BlockDriverState *bs,
 
      request->handle = INDEX_TO_HANDLE(s, i);
      s->requests[i].request = request;
+    s->requests[i].qiov = qiov;
 
      if (s->quit) {
          rc = -EIO;

@@ -155,7 +172,8 @@ static int nbd_co_send_request(BlockDriverState *bs,
          goto err;
      }
 
-    if (qiov) {
+    if (s->requests[i].request->type == NBD_CMD_WRITE) {
+        assert(qiov);
          qio_channel_set_cork(s->ioc, true);
          rc = nbd_send_request(s->ioc, request);
          if (rc >= 0 && !s->quit) {

@@ -181,9 +199,7 @@ err:
      return rc;
  }
 
-static int nbd_co_receive_reply(NBDClientSession *s,
-                                NBDRequest *request,
-                                QEMUIOVector *qiov)
+static int nbd_co_receive_reply(NBDClientSession *s, NBDRequest *request)
  {
      int ret;
      int i = HANDLE_TO_INDEX(s, request->handle);

@@ -197,14 +213,6 @@ static int nbd_co_receive_reply(NBDClientSession *s,
      } else {
          assert(s->reply.handle == request->handle);
          ret = -s->reply.error;
-        if (qiov && s->reply.error == 0) {
-            assert(request->len == iov_size(qiov->iov, qiov->niov));
-            if (qio_channel_readv_all(s->ioc, qiov->iov, qiov->niov,
-                                      NULL) < 0) {
-                ret = -EIO;
-                s->quit = true;
-            }
-        }
 
          /* Tell the read handler to read another header.  */
          s->reply.handle = 0;

@@ -232,16 +240,14 @@ static int nbd_co_request(BlockDriverState *bs,
      NBDClientSession *client = nbd_get_client_session(bs);
      int ret;
 
-    assert(!qiov || request->type == NBD_CMD_WRITE ||
-           request->type == NBD_CMD_READ);
-    ret = nbd_co_send_request(bs, request,
-                              request->type == NBD_CMD_WRITE ? qiov : NULL);
+    assert((qiov == NULL) !=
+           (request->type == NBD_CMD_WRITE || request->type == NBD_CMD_READ));
+    ret = nbd_co_send_request(bs, request, qiov);
      if (ret < 0) {
          return ret;
      }
 
-    return nbd_co_receive_reply(client, request,
-                                request->type == NBD_CMD_READ ? qiov : NULL);
+    return nbd_co_receive_reply(client, request);
  }
 
  int nbd_client_co_preadv(BlockDriverState *bs, uint64_t offset,

Quoted 45 lines · Paolo Bonzini

Quoted 31 lines · Vladimir Sementsov-Ogievskiy

Read the whole reply in one place - in nbd_read_reply_entry.

Signed-off-by Vladimir Sementsov-Ogievskiy <[email protected]>

---
  block/nbd-client.h |  1 +
  block/nbd-client.c | 42 ++++++++++++++++++++++++------------------
  2 files changed, 25 insertions(+), 18 deletions(-)

block/nbd-client.h +1 -0

index b1900e6a6d..3f97d31013 100644
@@ -21,6 +21,7 @@ typedef struct {
      Coroutine *coroutine;
      bool receiving;         /* waiting for read_reply_co? */
      NBDRequest *request;
+    QEMUIOVector *qiov;
  } NBDClientRequest;
    typedef struct NBDClientSession {

block/nbd-client.c +3 -0

index 5f241ecc22..f7b642f079 100644
@@ -98,6 +98,21 @@ static coroutine_fn void nbd_read_reply_entry(void
 *opaque)
              break;
          }
  +        if (s->reply.error == 0 &&
+            s->requests[i].request->type == NBD_CMD_READ)
+        {
+            QEMUIOVector *qiov = s->requests[i].qiov;

block/nbd-client.c static coroutine_fn void nbd_read_reply_entry(void · line 114

Quoted 10 lines

block/nbd-client.c +10 -0

+            assert(qiov != NULL);
+            assert(s->requests[i].request->len ==
+                   iov_size(qiov->iov, qiov->niov));
+            if (qio_channel_readv_all(s->ioc, qiov->iov, qiov->niov,
+                                      NULL) < 0)
+            {
+                s->reply.error = EIO;
+                break;
+            }
+        }

I am not sure this is an improvement. In principle you could have commands that read replies a bit at a time without using a QEMUIOVector.

Quoted 1 line

Paolo

Hm, what do you mean? In this patch I don't change "how do we read it", I only move the reading code to one coroutine, to make it simple to extend it in future (block status, etc).

nbd_read_reply_enty has all
information it need (s->requests[i].request) to handle the whole reply.. It's an improvement, because it leads to separating reply_entry coroutine - it don't need any synchronization (yield/wake) more with request coroutines.

Quoted 77 lines

Quoted 77 lines

block/nbd-client.c +0 -0

          /* We're woken up again by the request itself.  Note that

there
* is no race between yielding and reentering
read_reply_co. This
* is because:

block/nbd-client.c +8 -14

@@ -118,6 +133,7 @@ static coroutine_fn void
 nbd_read_reply_entry(void *opaque)
      s->read_reply_co = NULL;
  }
  +/* qiov should be provided for READ and WRITE */
  static int nbd_co_send_request(BlockDriverState *bs,
                                 NBDRequest *request,
                                 QEMUIOVector *qiov)

@@ -145,6 +161,7 @@ static int nbd_co_send_request(BlockDriverState *bs,
        request->handle = INDEX_TO_HANDLE(s, i);
      s->requests[i].request = request;
+    s->requests[i].qiov = qiov;
        if (s->quit) {
          rc = -EIO;

@@ -155,7 +172,8 @@ static int nbd_co_send_request(BlockDriverState *bs,
          goto err;
      }
  -    if (qiov) {
+    if (s->requests[i].request->type == NBD_CMD_WRITE) {
+        assert(qiov);
          qio_channel_set_cork(s->ioc, true);
          rc = nbd_send_request(s->ioc, request);
          if (rc >= 0 && !s->quit) {

@@ -181,9 +199,7 @@ err:
      return rc;
  }
  -static int nbd_co_receive_reply(NBDClientSession *s,
-                                NBDRequest *request,
-                                QEMUIOVector *qiov)
+static int nbd_co_receive_reply(NBDClientSession *s, NBDRequest
 *request)
  {
      int ret;
      int i = HANDLE_TO_INDEX(s, request->handle);

@@ -197,14 +213,6 @@ static int nbd_co_receive_reply(NBDClientSession
 *s,
      } else {
          assert(s->reply.handle == request->handle);
          ret = -s->reply.error;
-        if (qiov && s->reply.error == 0) {
-            assert(request->len == iov_size(qiov->iov, qiov->niov));
-            if (qio_channel_readv_all(s->ioc, qiov->iov, qiov->niov,
-                                      NULL) < 0) {
-                ret = -EIO;
-                s->quit = true;
-            }
-        }
            /* Tell the read handler to read another header.  */
          s->reply.handle = 0;

@@ -232,16 +240,14 @@ static int nbd_co_request(BlockDriverState *bs,
      NBDClientSession *client = nbd_get_client_session(bs);
      int ret;
  -    assert(!qiov || request->type == NBD_CMD_WRITE ||
-           request->type == NBD_CMD_READ);
-    ret = nbd_co_send_request(bs, request,
-                              request->type == NBD_CMD_WRITE ? qiov
 : NULL);
+    assert((qiov == NULL) !=
+           (request->type == NBD_CMD_WRITE || request->type ==
 NBD_CMD_READ));
+    ret = nbd_co_send_request(bs, request, qiov);
      if (ret < 0) {
          return ret;
      }
  -    return nbd_co_receive_reply(client, request,
-                                request->type == NBD_CMD_READ ? qiov
 : NULL);
+    return nbd_co_receive_reply(client, request);
  }
    int nbd_client_co_preadv(BlockDriverState *bs, uint64_t offset,

Quoted 46 lines · Vladimir Sementsov-Ogievskiy

Quoted 45 lines · Paolo Bonzini

Quoted 31 lines · Vladimir Sementsov-Ogievskiy

Read the whole reply in one place - in nbd_read_reply_entry.

Signed-off-by Vladimir Sementsov-Ogievskiy <[email protected]>

---
   block/nbd-client.h |  1 +
   block/nbd-client.c | 42 ++++++++++++++++++++++++------------------
   2 files changed, 25 insertions(+), 18 deletions(-)

block/nbd-client.h +1 -0

index b1900e6a6d..3f97d31013 100644
@@ -21,6 +21,7 @@ typedef struct {
       Coroutine *coroutine;
       bool receiving;         /* waiting for read_reply_co? */
       NBDRequest *request;
+    QEMUIOVector *qiov;
   } NBDClientRequest;
     typedef struct NBDClientSession {

block/nbd-client.c +3 -0

index 5f241ecc22..f7b642f079 100644
@@ -98,6 +98,21 @@ static coroutine_fn void nbd_read_reply_entry(void
 *opaque)
               break;
           }
   +        if (s->reply.error == 0 &&
+            s->requests[i].request->type == NBD_CMD_READ)
+        {
+            QEMUIOVector *qiov = s->requests[i].qiov;

block/nbd-client.c static coroutine_fn void nbd_read_reply_entry(void · line 114

Quoted 10 lines

block/nbd-client.c +10 -0

+            assert(qiov != NULL);
+            assert(s->requests[i].request->len ==
+                   iov_size(qiov->iov, qiov->niov));
+            if (qio_channel_readv_all(s->ioc, qiov->iov, qiov->niov,
+                                      NULL) < 0)
+            {
+                s->reply.error = EIO;
+                break;
+            }
+        }

I am not sure this is an improvement. In principle you could have commands that read replies a bit at a time without using a QEMUIOVector.

Quoted 4 lines

Quoted 1 line

Paolo

Hm, what do you mean? In this patch I don't change "how do we read it", I only move the reading code to one coroutine, to make it simple to extend it in future (block status, etc).

I disagree that it is easier to extend it in the future. If some commands in the future need a different "how do we read it" (e.g. for structured reply), nbd_read_reply_entry may not have all the information it needs anymore.

In fact, you're pushing knowledge of the commands from nbd_co_request to nbd_read_reply_entry:

block/nbd-client.c +2 -0

+        if (s->reply.error == 0 &&
+            s->requests[i].request->type == NBD_CMD_READ)

and knowledge of NBD_CMD_READ simply doesn't belong there. So there is an example of that right now, it is not some arbitrary bad thing that could happen in the future.

Paolo

Quoted 84 lines

nbd_read_reply_enty has all
information it need (s->requests[i].request) to handle the whole reply.. It's an improvement, because it leads to separating reply_entry coroutine - it don't need any synchronization (yield/wake) more with request coroutines.

Quoted 76 lines

Quoted 76 lines

block/nbd-client.c +0 -0

           /* We're woken up again by the request itself.  Note that

there
* is no race between yielding and reentering
read_reply_co. This
* is because:

block/nbd-client.c +8 -14

@@ -118,6 +133,7 @@ static coroutine_fn void
 nbd_read_reply_entry(void *opaque)
       s->read_reply_co = NULL;
   }
   +/* qiov should be provided for READ and WRITE */
   static int nbd_co_send_request(BlockDriverState *bs,
                                  NBDRequest *request,
                                  QEMUIOVector *qiov)

@@ -145,6 +161,7 @@ static int nbd_co_send_request(BlockDriverState *bs,
         request->handle = INDEX_TO_HANDLE(s, i);
       s->requests[i].request = request;
+    s->requests[i].qiov = qiov;
         if (s->quit) {
           rc = -EIO;

@@ -155,7 +172,8 @@ static int nbd_co_send_request(BlockDriverState *bs,
           goto err;
       }
   -    if (qiov) {
+    if (s->requests[i].request->type == NBD_CMD_WRITE) {
+        assert(qiov);
           qio_channel_set_cork(s->ioc, true);
           rc = nbd_send_request(s->ioc, request);
           if (rc >= 0 && !s->quit) {

@@ -181,9 +199,7 @@ err:
       return rc;
   }
   -static int nbd_co_receive_reply(NBDClientSession *s,
-                                NBDRequest *request,
-                                QEMUIOVector *qiov)
+static int nbd_co_receive_reply(NBDClientSession *s, NBDRequest
 *request)
   {
       int ret;
       int i = HANDLE_TO_INDEX(s, request->handle);

@@ -197,14 +213,6 @@ static int nbd_co_receive_reply(NBDClientSession
 *s,
       } else {
           assert(s->reply.handle == request->handle);
           ret = -s->reply.error;
-        if (qiov && s->reply.error == 0) {
-            assert(request->len == iov_size(qiov->iov, qiov->niov));
-            if (qio_channel_readv_all(s->ioc, qiov->iov, qiov->niov,
-                                      NULL) < 0) {
-                ret = -EIO;
-                s->quit = true;
-            }
-        }
             /* Tell the read handler to read another header.  */
           s->reply.handle = 0;

@@ -232,16 +240,14 @@ static int nbd_co_request(BlockDriverState *bs,
       NBDClientSession *client = nbd_get_client_session(bs);
       int ret;
   -    assert(!qiov || request->type == NBD_CMD_WRITE ||
-           request->type == NBD_CMD_READ);
-    ret = nbd_co_send_request(bs, request,
-                              request->type == NBD_CMD_WRITE ? qiov
 : NULL);
+    assert((qiov == NULL) !=
+           (request->type == NBD_CMD_WRITE || request->type ==
 NBD_CMD_READ));
+    ret = nbd_co_send_request(bs, request, qiov);
       if (ret < 0) {
           return ret;
       }
   -    return nbd_co_receive_reply(client, request,
-                                request->type == NBD_CMD_READ ? qiov
 : NULL);
+    return nbd_co_receive_reply(client, request);
   }
     int nbd_client_co_preadv(BlockDriverState *bs, uint64_t offset,

Quoted 5 lines

Quoted 1 line

I disagree that it is easier to extend it in the future. If some commands in the future need a different "how do we read it" (e.g. for structured reply), nbd_read_reply_entry may not have all the information it needs anymore.

how is it possible? all information is in requests[i].

Quoted 10 lines

In fact, you're pushing knowledge of the commands from nbd_co_request to nbd_read_reply_entry:

block/nbd-client.c +2 -0

+        if (s->reply.error == 0 &&
+            s->requests[i].request->type == NBD_CMD_READ)

and knowledge of NBD_CMD_READ simply doesn't belong there. So there is an example of that right now, it is not some arbitrary bad thing that could happen in the future.

I can't agree that my point of view is wrong, it's just different. You want commands, reading from socket, each knows what it should read. I want the receiver - one sequential reader, that can read all kinds of replies and just wake requesting coroutines when all reading is done. For me it looks simpler than switching.

Drop 'reply' from NBDClientSession. It's used to only deliver request return code to receiving coroutine. Instead introduce per-request ret variable to simplify the scheme and make further refactoring possible.

Signed-off-by Vladimir Sementsov-Ogievskiy <[email protected]>

---
 block/nbd-client.h |  2 +-
 block/nbd-client.c | 22 +++++++++-------------
 2 files changed, 10 insertions(+), 14 deletions(-)

block/nbd-client.h +0 -0

index 3f97d31013..4bc8fe3993 100644

It's incorrect to return success rc >= 0 if we skip qio_channel_writev_all() call due to s->quit.

Signed-off-by Vladimir Sementsov-Ogievskiy <[email protected]>

---
 block/nbd-client.c | 3 +++
 1 file changed, 3 insertions(+)

block/nbd-client.c +0 -0

index f331f08a9a..280147e6a7 100644

block/nbd-client.c +0 -0

     if (rc < 0) {
         s->quit = true;
         s->requests[i].coroutine = NULL;

Quoted 2 lines · Vladimir Sementsov-Ogievskiy

It's incorrect to return success rc >= 0 if we skip qio_channel_writev_all() call due to s->quit.

Does this need to cc: qemu-stable for 2.10.1 (or put another way, can we come up with some scenario of EAGAIN or other handling that would actually set s->quit in a parallel coroutine when a client sends out multiple requests at once)?

Quoted 2 lines · Vladimir Sementsov-Ogievskiy

It's incorrect to return success rc >= 0 if we skip qio_channel_writev_all() call due to s->quit.

Does this need to cc: qemu-stable for 2.10.1 (or put another way, can we come up with some scenario of EAGAIN or other handling that would actually set s->quit in a parallel coroutine when a client sends out multiple requests at once)?

Quoted 11 lines

Signed-off-by Vladimir Sementsov-Ogievskiy <[email protected]>

---
  block/nbd-client.c | 3 +++
  1 file changed, 3 insertions(+)

block/nbd-client.c +0 -0

index f331f08a9a..280147e6a7 100644

block/nbd-client.c static int nbd_co_send_request(BlockDriverState *bs, · line 194

Quoted 6 lines

block/nbd-client.c +3 -0

@@ -189,6 +189,9 @@ static int nbd_co_send_request(BlockDriverState *bs,
      }
 
  err:
+    if (rc >= 0 && s->quit) {
+        rc = -EIO;
+    }

I'm still not convinced this is in the right place. This fails the send_request regardless of whether we skipped qio_channel_writev_all(); shouldn't the rc be set only in the case that we actually skipped writing the full command because s->quit was detected at that point in time?

Quoted 4 lines

block/nbd-client.c +0 -0

      if (rc < 0) {
          s->quit = true;
          s->requests[i].coroutine = NULL;

Do not continue any operation if s->quit is set in parallel.

Signed-off-by Vladimir Sementsov-Ogievskiy <[email protected]>

---
 block/nbd-client.c | 7 +++----
 1 file changed, 3 insertions(+), 4 deletions(-)

block/nbd-client.c +0 -0

index 280147e6a7..f80a4c5564 100644

Quoted 12 lines · Vladimir Sementsov-Ogievskiy

Do not continue any operation if s->quit is set in parallel.

Signed-off-by Vladimir Sementsov-Ogievskiy <[email protected]>

---
  block/nbd-client.c | 7 +++----
  1 file changed, 3 insertions(+), 4 deletions(-)

block/nbd-client.c +0 -0

index 280147e6a7..f80a4c5564 100644

block/nbd-client.c static coroutine_fn void nbd_read_reply_entry(void *opaque) · line 86

Quoted 7 lines

block/nbd-client.c +1 -1

@@ -81,7 +81,7 @@ static coroutine_fn void nbd_read_reply_entry(void *opaque)
          if (ret < 0) {
              error_report_err(local_err);
          }
-        if (ret <= 0) {
+        if (ret <= 0 || s->quit) {
              break;
          }

I'm still not convinced this helps: either nbd_receive_reply() already returned an error, or we got a successful reply header, at which point either the command is done (no need to abort the command early - it succeeded) or it is a read command (and we should detect at the point where we try to populate the qiov that we don't want to read any more). It depends on your 3/7 patch for the fact that we even try to read the qiov directly in this while loop rather than in the coroutine handler, where Paolo questioned whether we need that change.

block/nbd-client.c static coroutine_fn void nbd_read_reply_entry(void *opaque) · line 112

Quoted 12 lines

block/nbd-client.c +2 -3

@@ -105,9 +105,8 @@ static coroutine_fn void nbd_read_reply_entry(void *opaque)
              assert(qiov != NULL);
              assert(s->requests[i].request->len ==
                     iov_size(qiov->iov, qiov->niov));
-            if (qio_channel_readv_all(s->ioc, qiov->iov, qiov->niov,
-                                      NULL) < 0)
-            {
+            ret = qio_channel_readv_all(s->ioc, qiov->iov, qiov->niov, NULL);
+            if (ret < 0 || s->quit) {
                  s->requests[i].ret = -EIO;
                  break;
              }

The placement here looks odd. Either we should not attempt the read because s->quit was already set (okay, your first addition makes sense in that light), or we succeeded at the read (at which point, why do we need to claim it failed?).

I'm trying to look forward to structured reads, where we will have to deal with more than one server message in reply to a single client request. For read, we just piece together portions of the qiov until the server has sent us all the pieces. But for block query, I really do think we'll want to defer to specialized handlers for doing further reads (the amount of data to be read from the server is not known by the client a priori, so it is a two-part read, one to get the length, and another to read that remaining length); if we need some sort of callback function rather than cramming all the logic here in the main read loop,

then the qio_channel_readv_all for read commands would belong in the callback, and it is more a question of the main read loop checking whether there is an early quit condition before calling into the callback.

Quoted 10 lines

I'm trying to look forward to structured reads, where we will have to deal with more than one server message in reply to a single client request. For read, we just piece together portions of the qiov until the server has sent us all the pieces. But for block query, I really do think we'll want to defer to specialized handlers for doing further reads (the amount of data to be read from the server is not known by the client a priori, so it is a two-part read, one to get the length, and another to read that remaining length); if we need some sort of callback function rather than cramming all the logic here in the main read loop,

by patch 3 I do not mean in any way that I want to have all reading staff in one function, ofcourse it should be refactored
for block-status addition. I just want to have all reading staff in one coroutine. Reading from ioc is sequential, so it's
native to do it sequentially in one coroutine, without switches.

Quoted 9 lines · Vladimir Sementsov-Ogievskiy

Quoted 9 lines

I'm trying to look forward to structured reads, where we will have to deal with more than one server message in reply to a single client request. For read, we just piece together portions of the qiov until the server has sent us all the pieces. But for block query, I really do think we'll want to defer to specialized handlers for doing further reads (the amount of data to be read from the server is not known by the client a priori, so it is a two-part read, one to get the length, and another to read that remaining length); if we need some sort of callback function rather than cramming all the logic here in the main read loop,

Quoted 5 lines

by patch 3 I do not mean in any way that I want to have all reading staff in one function, ofcourse it should be refactored
for block-status addition. I just want to have all reading staff in one coroutine. Reading from ioc is sequential, so it's
native to do it sequentially in one coroutine, without switches.

But why is that a goal? See it as a state machine that goes between the "waiting for header" and "waiting for payload" states. Reading header corresponds to read_reply_co waiting on the socket (and reading when it's ready); reading payload corresponds to the request coroutine waiting on the socket and reading when it's ready.

Paolo

Refactor nbd client to not yield from nbd_read_reply_entry. It's possible now as all reading is done in nbd_read_reply_entry and all request-related data is stored in per-request s->requests[i].

We need here some additional work with s->requests[i].ret and s->quit to not fail requests on normal disconnet and to not report reading errors on normal disconnect.

Signed-off-by Vladimir Sementsov-Ogievskiy <[email protected]>

---
 block/nbd-client.c | 29 ++++++++++-------------------
 1 file changed, 10 insertions(+), 19 deletions(-)

block/nbd-client.c +0 -5

     } else {
         ret = s->requests[i].ret;

@@ -218,11 +212,6 @@ static int nbd_co_receive_reply(NBDClientSession *s, NBDRequest *request)
 
     s->requests[i].coroutine = NULL;
 
-    /* Kick the read_reply_co to get the next reply.  */
-    if (s->read_reply_co) {
-        aio_co_wake(s->read_reply_co);
-    }
-
     qemu_co_mutex_lock(&s->send_mutex);
     s->in_flight--;
     qemu_co_queue_next(&s->free_sema);

Quoted 4 lines · Vladimir Sementsov-Ogievskiy

Refactor nbd client to not yield from nbd_read_reply_entry. It's possible now as all reading is done in nbd_read_reply_entry and all request-related data is stored in per-request s->requests[i].

Quoted 8 lines

We need here some additional work with s->requests[i].ret and s->quit to not fail requests on normal disconnet and to not report reading errors on normal disconnect.

Signed-off-by Vladimir Sementsov-Ogievskiy <[email protected]>

---
  block/nbd-client.c | 29 ++++++++++-------------------
  1 file changed, 10 insertions(+), 19 deletions(-)

The diffstat may have merit, but I'm wondering:

block/nbd-client.c static coroutine_fn void nbd_read_reply_entry(void *opaque) · line 83

Quoted 11 lines

block/nbd-client.c +2 -1

index f80a4c5564..bf20d5d5e6 100644
@@ -79,7 +79,11 @@ static coroutine_fn void nbd_read_reply_entry(void *opaque)
      while (!s->quit) {
          ret = nbd_receive_reply(s->ioc, &reply, &local_err);
          if (ret < 0) {
-            error_report_err(local_err);
+            if (s->quit) {
+                error_free(local_err);

This discards errors on all remaining coroutines after we detect an early exit. Are we sure the coroutine that set s->quit will have reported an appropriate error, and thus explaining why we can discard all secondary errors? A comment in the code would be helpful.

block/nbd-client.c static int nbd_co_receive_reply(NBDClientSession *s, NBDRequest *request) · line 208

Quoted 12 lines

block/nbd-client.c +4 -1

+            } else {
+                error_report_err(local_err);
+            }
          }

@@ -210,7 +204,7 @@ static int nbd_co_receive_reply(NBDClientSession *s, NBDRequest *request)
      s->requests[i].receiving = true;
      qemu_coroutine_yield();
      s->requests[i].receiving = false;
-    if (!s->ioc || s->quit) {
+    if (!s->ioc) {
          ret = -EIO;

Why don't we need to check s->quit any more? That was just added earlier in the series; why the churn?

Quoted 17 lines

block/nbd-client.c +0 -5

      } else {
          ret = s->requests[i].ret;

@@ -218,11 +212,6 @@ static int nbd_co_receive_reply(NBDClientSession *s, NBDRequest *request)
 
      s->requests[i].coroutine = NULL;
 
-    /* Kick the read_reply_co to get the next reply.  */
-    if (s->read_reply_co) {
-        aio_co_wake(s->read_reply_co);
-    }
-
      qemu_co_mutex_lock(&s->send_mutex);
      s->in_flight--;
      qemu_co_queue_next(&s->free_sema);

@@ -364,6 +353,8 @@ void nbd_client_close(BlockDriverState *bs)
 

block/nbd-client.c void nbd_client_close(BlockDriverState *bs) · line 356

Quoted 3 lines

block/nbd-client.c +1 -0

      nbd_send_request(client->ioc, &request);
 
+    client->quit = true;

Previously, client->quit was only set when detecting a broken server, now it is also set for a clean exit. Do we need to change any documentation of the field?

Quoted 9 lines · Eric Blake

Quoted 8 lines · Vladimir Sementsov-Ogievskiy

Refactor nbd client to not yield from nbd_read_reply_entry. It's possible now as all reading is done in nbd_read_reply_entry and all request-related data is stored in per-request s->requests[i].

We need here some additional work with s->requests[i].ret and s->quit to not fail requests on normal disconnet and to not report reading errors on normal disconnect.

Quoted 6 lines

Review void nbd_client_close(BlockDriverState *bs) · line 354

Quoted 3 lines

(quoted diff) +0 -0

@@ -364,6 +353,8 @@ void nbd_client_close(BlockDriverState *bs)
         nbd_send_request(client->ioc, &request);
   +    client->quit = true;

Previously, client->quit was only set when detecting a broken server, now it is also set for a clean exit.  Do we need to change any documentation of the field?

It has documentation?