[PATCH for-5.1? 0/4] non-blocking connect

[PATCH for-5.1? 0/4] non-blocking connect

[Vladimir Sementsov-Ogievskiy]

Hi! This fixes real problem (see 04). On the other hand it may be too
much for 5.1, and it's not a degradation. So, up to you.

It's based on "[PATCH for-5.1? 0/3] Fix nbd reconnect dead-locks", or
in other words
Based-on: <20200720090024.18186-1-vsementsov@virtuozzo.com>

Vladimir Sementsov-Ogievskiy (4):
  qemu-sockets: refactor inet_connect_addr
  qemu-sockets: implement non-blocking connect interface
  io/channel-socket: implement non-blocking connect
  block/nbd: use non-blocking connect: fix vm hang on connect()

 include/io/channel-socket.h | 14 +++++++
 include/qemu/sockets.h      |  6 +++
 block/nbd.c                 | 11 +++---
 io/channel-socket.c         | 74 ++++++++++++++++++++++++++++++++++++
 util/qemu-sockets.c         | 76 ++++++++++++++++++++++++++-----------
 5 files changed, 153 insertions(+), 28 deletions(-)

-- 
2.21.0

[PATCH 1/4] qemu-sockets: refactor inet_connect_addr

[Vladimir Sementsov-Ogievskiy]

We are going to publish inet_connect_addr to be used in separate. Let's
move keep_alive handling to it. Pass the whole InetSocketAddress
pointer, not only keep_alive, so that future external callers will not
care about internals of InetSocketAddress.

While being here, remove redundant inet_connect_addr() declaration.

Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
---
 util/qemu-sockets.c | 37 ++++++++++++++++---------------------
 1 file changed, 16 insertions(+), 21 deletions(-)

diff --git a/util/qemu-sockets.c b/util/qemu-sockets.c
index b37d288866..8ccf4088c2 100644
--- a/util/qemu-sockets.c
+++ b/util/qemu-sockets.c
@@ -354,9 +354,8 @@ listen_ok:
     ((rc) == -EINPROGRESS)
 #endif
 
-static int inet_connect_addr(struct addrinfo *addr, Error **errp);
-
-static int inet_connect_addr(struct addrinfo *addr, Error **errp)
+static int inet_connect_addr(InetSocketAddress *saddr,
+                             struct addrinfo *addr, Error **errp)
 {
     int sock, rc;
 
@@ -381,6 +380,18 @@ static int inet_connect_addr(struct addrinfo *addr, Error **errp)
         return -1;
     }
 
+    if (saddr->keep_alive) {
+        int val = 1;
+        int ret = qemu_setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE,
+                                  &val, sizeof(val));
+
+        if (ret < 0) {
+            error_setg_errno(errp, errno, "Unable to set KEEPALIVE");
+            closesocket(sock);
+            return -1;
+        }
+    }
+
     return sock;
 }
 
@@ -455,7 +466,7 @@ int inet_connect_saddr(InetSocketAddress *saddr, Error **errp)
     for (e = res; e != NULL; e = e->ai_next) {
         error_free(local_err);
         local_err = NULL;
-        sock = inet_connect_addr(e, &local_err);
+        sock = inet_connect_addr(saddr, e, &local_err);
         if (sock >= 0) {
             break;
         }
@@ -463,23 +474,7 @@ int inet_connect_saddr(InetSocketAddress *saddr, Error **errp)
 
     freeaddrinfo(res);
 
-    if (sock < 0) {
-        error_propagate(errp, local_err);
-        return sock;
-    }
-
-    if (saddr->keep_alive) {
-        int val = 1;
-        int ret = qemu_setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE,
-                                  &val, sizeof(val));
-
-        if (ret < 0) {
-            error_setg_errno(errp, errno, "Unable to set KEEPALIVE");
-            close(sock);
-            return -1;
-        }
-    }
-
+    error_propagate(errp, local_err);
     return sock;
 }
 
-- 
2.21.0

[PATCH 2/4] qemu-sockets: implement non-blocking connect interface

[Vladimir Sementsov-Ogievskiy]

We are going to implement non-blocking connect in io/channel-socket.

non-blocking connect includes three phases:

    1. connect() call
    2. wait until socket is ready
    3. check result

io/channel-socket has wait-on-socket API (qio_channel_yield(),
qio_channel_wait()), so it's a good place for [2].

Still, the whole thing is not simple, because socket connect in case of
inet socket includes several connect() calls, as SocketAddress may be
parsed into a list of inet addresses. And after each non-blocking
connect() upper layer should have a possibility to wait on the socket.

We may try to implement a kind of abstract list or iterator for
"sub" addresses of SocketAddress, but all this appears to be too
complex and not worth doing (as actually, only inet sockets has such a
"multiple" SocketAddress).

So, let's instead make public API for inet sockets themselves, to be
handled in separate in upper layer, if it needs non-blocking connect.

Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
---
 include/qemu/sockets.h |  6 ++++++
 util/qemu-sockets.c    | 45 +++++++++++++++++++++++++++++++++++++-----
 2 files changed, 46 insertions(+), 5 deletions(-)

diff --git a/include/qemu/sockets.h b/include/qemu/sockets.h
index 7d1f813576..7389d6be55 100644
--- a/include/qemu/sockets.h
+++ b/include/qemu/sockets.h
@@ -34,6 +34,12 @@ int inet_ai_family_from_address(InetSocketAddress *addr,
 int inet_parse(InetSocketAddress *addr, const char *str, Error **errp);
 int inet_connect(const char *str, Error **errp);
 int inet_connect_saddr(InetSocketAddress *saddr, Error **errp);
+int inet_connect_addr(InetSocketAddress *saddr, struct addrinfo *addr,
+                      bool blocking, bool *in_progress, Error **errp);
+struct addrinfo *inet_parse_connect_saddr(InetSocketAddress *saddr,
+                                          Error **errp);
+
+int socket_check(int fd, Error **errp);
 
 NetworkAddressFamily inet_netfamily(int family);
 
diff --git a/util/qemu-sockets.c b/util/qemu-sockets.c
index 8ccf4088c2..a02d00f342 100644
--- a/util/qemu-sockets.c
+++ b/util/qemu-sockets.c
@@ -354,11 +354,17 @@ listen_ok:
     ((rc) == -EINPROGRESS)
 #endif
 
-static int inet_connect_addr(InetSocketAddress *saddr,
-                             struct addrinfo *addr, Error **errp)
+int inet_connect_addr(InetSocketAddress *saddr, struct addrinfo *addr,
+                      bool blocking, bool *in_progress, Error **errp)
 {
     int sock, rc;
 
+    assert(blocking == !in_progress);
+
+    if (in_progress) {
+        *in_progress = false;
+    }
+
     sock = qemu_socket(addr->ai_family, addr->ai_socktype, addr->ai_protocol);
     if (sock < 0) {
         error_setg_errno(errp, errno, "Failed to create socket");
@@ -366,6 +372,10 @@ static int inet_connect_addr(InetSocketAddress *saddr,
     }
     socket_set_fast_reuse(sock);
 
+    if (!blocking) {
+        qemu_set_nonblock(sock);
+    }
+
     /* connect to peer */
     do {
         rc = 0;
@@ -374,6 +384,13 @@ static int inet_connect_addr(InetSocketAddress *saddr,
         }
     } while (rc == -EINTR);
 
+    if (!blocking && rc == -EINPROGRESS) {
+        if (in_progress) {
+            *in_progress = true;
+        }
+        return sock;
+    }
+
     if (rc < 0) {
         error_setg_errno(errp, errno, "Failed to connect socket");
         closesocket(sock);
@@ -395,8 +412,26 @@ static int inet_connect_addr(InetSocketAddress *saddr,
     return sock;
 }
 
-static struct addrinfo *inet_parse_connect_saddr(InetSocketAddress *saddr,
-                                                 Error **errp)
+int socket_check(int fd, Error **errp)
+{
+    int optval;
+    socklen_t optlen = sizeof(optval);
+    if (qemu_getsockopt(fd, SOL_SOCKET, SO_ERROR, &optval, &optlen) < 0) {
+        error_setg_errno(errp, errno, "Unable to check connection");
+        return -1;
+    }
+
+    if (optval != 0) {
+        error_setg_errno(errp, errno, "Connection failed");
+        return -1;
+    }
+
+    return 0;
+}
+
+
+struct addrinfo *inet_parse_connect_saddr(InetSocketAddress *saddr,
+                                          Error **errp)
 {
     struct addrinfo ai, *res;
     int rc;
@@ -466,7 +501,7 @@ int inet_connect_saddr(InetSocketAddress *saddr, Error **errp)
     for (e = res; e != NULL; e = e->ai_next) {
         error_free(local_err);
         local_err = NULL;
-        sock = inet_connect_addr(saddr, e, &local_err);
+        sock = inet_connect_addr(saddr, e, true, NULL, &local_err);
         if (sock >= 0) {
             break;
         }
-- 
2.21.0

[PATCH 3/4] io/channel-socket: implement non-blocking connect

[Vladimir Sementsov-Ogievskiy]

Utilize new socket API to make a non-blocking connect for inet sockets.

Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
---
 include/io/channel-socket.h | 14 +++++++
 io/channel-socket.c         | 74 +++++++++++++++++++++++++++++++++++++
 2 files changed, 88 insertions(+)

diff --git a/include/io/channel-socket.h b/include/io/channel-socket.h
index 777ff5954e..82e868bc02 100644
--- a/include/io/channel-socket.h
+++ b/include/io/channel-socket.h
@@ -94,6 +94,20 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
                                     SocketAddress *addr,
                                     Error **errp);
 
+/**
+ * qio_channel_socket_connect_non_blocking_sync:
+ * @ioc: the socket channel object
+ * @addr: the address to connect to
+ * @errp: pointer to a NULL-initialized error object
+ *
+ * Attempt to connect to the address @addr using non-blocking mode of
+ * the socket. Function is synchronous, but being called from
+ * coroutine context will yield during connect operation.
+ */
+int qio_channel_socket_connect_non_blocking_sync(QIOChannelSocket *ioc,
+                                                 SocketAddress *addr,
+                                                 Error **errp);
+
 /**
  * qio_channel_socket_connect_async:
  * @ioc: the socket channel object
diff --git a/io/channel-socket.c b/io/channel-socket.c
index e1b4667087..076de7578a 100644
--- a/io/channel-socket.c
+++ b/io/channel-socket.c
@@ -22,6 +22,7 @@
 #include "qapi/error.h"
 #include "qapi/qapi-visit-sockets.h"
 #include "qemu/module.h"
+#include "qemu/sockets.h"
 #include "io/channel-socket.h"
 #include "io/channel-watch.h"
 #include "trace.h"
@@ -29,6 +30,8 @@
 
 #define SOCKET_MAX_FDS 16
 
+static int qio_channel_socket_close(QIOChannel *ioc, Error **errp);
+
 SocketAddress *
 qio_channel_socket_get_local_address(QIOChannelSocket *ioc,
                                      Error **errp)
@@ -157,6 +160,77 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
     return 0;
 }
 
+static int qio_channel_inet_connect_non_blocking_sync(QIOChannelSocket *ioc,
+        InetSocketAddress *addr, Error **errp)
+{
+    Error *local_err = NULL;
+    struct addrinfo *infos, *info;
+    int sock = -1;
+
+    infos = inet_parse_connect_saddr(addr, errp);
+    if (!infos) {
+        return -1;
+    }
+
+    for (info = infos; info != NULL; info = info->ai_next) {
+        bool in_progress;
+
+        error_free(local_err);
+        local_err = NULL;
+
+        sock = inet_connect_addr(addr, info, false, &in_progress, &local_err);
+        if (sock < 0) {
+            continue;
+        }
+
+        if (qio_channel_socket_set_fd(ioc, sock, &local_err) < 0) {
+            close(sock);
+            continue;
+        }
+
+        if (in_progress) {
+            if (qemu_in_coroutine()) {
+                qio_channel_yield(QIO_CHANNEL(ioc), G_IO_OUT);
+            } else {
+                qio_channel_wait(QIO_CHANNEL(ioc), G_IO_OUT);
+            }
+            if (socket_check(sock, &local_err) < 0) {
+                qio_channel_socket_close(QIO_CHANNEL(ioc), NULL);
+                continue;
+            }
+        }
+
+        break;
+    }
+
+    freeaddrinfo(infos);
+
+    error_propagate(errp, local_err);
+    return sock;
+}
+
+int qio_channel_socket_connect_non_blocking_sync(QIOChannelSocket *ioc,
+                                                 SocketAddress *addr,
+                                                 Error **errp)
+{
+    if (addr->type == SOCKET_ADDRESS_TYPE_INET) {
+        return qio_channel_inet_connect_non_blocking_sync(ioc, &addr->u.inet,
+                                                          errp);
+    } else {
+        /*
+         * TODO: implement non-blocking connect for other socket types.
+         * For now just use blocking connect, and then make socket non-blocking
+         * for consistancy.
+         */
+        int ret = qio_channel_socket_connect_sync(ioc, addr, errp);
+
+        if (ret < 0) {
+            return ret;
+        }
+
+        return qio_channel_set_blocking(QIO_CHANNEL(ioc), false, errp);
+    }
+}
 
 static void qio_channel_socket_connect_worker(QIOTask *task,
                                               gpointer opaque)
-- 
2.21.0

Re: [PATCH 3/4] io/channel-socket: implement non-blocking connect

[Daniel P. Berrangé]

On Mon, Jul 20, 2020 at 09:07:14PM +0300, Vladimir Sementsov-Ogievskiy wrote:
> Utilize new socket API to make a non-blocking connect for inet sockets.
> 
> Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
> ---
>  include/io/channel-socket.h | 14 +++++++
>  io/channel-socket.c         | 74 +++++++++++++++++++++++++++++++++++++
>  2 files changed, 88 insertions(+)
> 
> diff --git a/include/io/channel-socket.h b/include/io/channel-socket.h
> index 777ff5954e..82e868bc02 100644
> --- a/include/io/channel-socket.h
> +++ b/include/io/channel-socket.h
> @@ -94,6 +94,20 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
>                                      SocketAddress *addr,
>                                      Error **errp);
>  
> +/**
> + * qio_channel_socket_connect_non_blocking_sync:
> + * @ioc: the socket channel object
> + * @addr: the address to connect to
> + * @errp: pointer to a NULL-initialized error object
> + *
> + * Attempt to connect to the address @addr using non-blocking mode of
> + * the socket. Function is synchronous, but being called from
> + * coroutine context will yield during connect operation.
> + */
> +int qio_channel_socket_connect_non_blocking_sync(QIOChannelSocket *ioc,
> +                                                 SocketAddress *addr,
> +                                                 Error **errp);
> +
>  /**
>   * qio_channel_socket_connect_async:
>   * @ioc: the socket channel object
> diff --git a/io/channel-socket.c b/io/channel-socket.c
> index e1b4667087..076de7578a 100644
> --- a/io/channel-socket.c
> +++ b/io/channel-socket.c
> @@ -22,6 +22,7 @@
>  #include "qapi/error.h"
>  #include "qapi/qapi-visit-sockets.h"
>  #include "qemu/module.h"
> +#include "qemu/sockets.h"
>  #include "io/channel-socket.h"
>  #include "io/channel-watch.h"
>  #include "trace.h"
> @@ -29,6 +30,8 @@
>  
>  #define SOCKET_MAX_FDS 16
>  
> +static int qio_channel_socket_close(QIOChannel *ioc, Error **errp);
> +
>  SocketAddress *
>  qio_channel_socket_get_local_address(QIOChannelSocket *ioc,
>                                       Error **errp)
> @@ -157,6 +160,77 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
>      return 0;
>  }
>  
> +static int qio_channel_inet_connect_non_blocking_sync(QIOChannelSocket *ioc,
> +        InetSocketAddress *addr, Error **errp)
> +{
> +    Error *local_err = NULL;
> +    struct addrinfo *infos, *info;
> +    int sock = -1;
> +
> +    infos = inet_parse_connect_saddr(addr, errp);
> +    if (!infos) {
> +        return -1;
> +    }

This call is blocking since it calls getaddrinfo whose design
offers no ability todo non-blocking DNS lookups. Given this
call, ...

> +
> +    for (info = infos; info != NULL; info = info->ai_next) {
> +        bool in_progress;
> +
> +        error_free(local_err);
> +        local_err = NULL;
> +
> +        sock = inet_connect_addr(addr, info, false, &in_progress, &local_err);
> +        if (sock < 0) {
> +            continue;
> +        }
> +
> +        if (qio_channel_socket_set_fd(ioc, sock, &local_err) < 0) {
> +            close(sock);
> +            continue;
> +        }
> +
> +        if (in_progress) {
> +            if (qemu_in_coroutine()) {
> +                qio_channel_yield(QIO_CHANNEL(ioc), G_IO_OUT);
> +            } else {
> +                qio_channel_wait(QIO_CHANNEL(ioc), G_IO_OUT);
> +            }

...this is offering false assurances of being non-blocking.

If we don't want the current thread to be blocked then we
need to be using the existing qio_channel_socket_connect_async
method or similar. It uses a throw away background thread to
run the connection attempt, and then reports completion back
later, thus avoiding the getaddrinfo design flaw for the callers.

I explicitly didn't want to add an method like the impl in this
patch, because getaddrinfo dooms it and we already had bugs in
the pre-QIOChannel code where QEMU thought it was non-blocking
but wasn't due to getaddrinfo lookups.


IIUC, the main appeal of this method is that the non-blocking
nature is hidden from the caller who can continue to treat it
as a synchronous call and have the coroutine magic happen in
behind the scenes.

IOW, What's needed is a simple way to run the operation in a
thread, and sleep for completion while having the coroutine
yield.

I think this could likely be achieved with QIOTask with an
alternate impl of the qio_task_wait_thread() method that is
friendly to coroutines instead of being based on pthread
condition variable waits.


> +            if (socket_check(sock, &local_err) < 0) {
> +                qio_channel_socket_close(QIO_CHANNEL(ioc), NULL);
> +                continue;
> +            }
> +        }
> +
> +        break;
> +    }
> +
> +    freeaddrinfo(infos);
> +
> +    error_propagate(errp, local_err);
> +    return sock;
> +}

Regards,
Daniel
-- 
|: https://berrange.com      -o-    https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org         -o-            https://fstop138.berrange.com :|
|: https://entangle-photo.org    -o-    https://www.instagram.com/dberrange :|

Re: [PATCH 3/4] io/channel-socket: implement non-blocking connect

[Vladimir Sementsov-Ogievskiy]

20.07.2020 21:29, Daniel P. Berrangé wrote:
> On Mon, Jul 20, 2020 at 09:07:14PM +0300, Vladimir Sementsov-Ogievskiy wrote:
>> Utilize new socket API to make a non-blocking connect for inet sockets.
>>
>> Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
>> ---
>>   include/io/channel-socket.h | 14 +++++++
>>   io/channel-socket.c         | 74 +++++++++++++++++++++++++++++++++++++
>>   2 files changed, 88 insertions(+)
>>
>> diff --git a/include/io/channel-socket.h b/include/io/channel-socket.h
>> index 777ff5954e..82e868bc02 100644
>> --- a/include/io/channel-socket.h
>> +++ b/include/io/channel-socket.h
>> @@ -94,6 +94,20 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
>>                                       SocketAddress *addr,
>>                                       Error **errp);
>>   
>> +/**
>> + * qio_channel_socket_connect_non_blocking_sync:
>> + * @ioc: the socket channel object
>> + * @addr: the address to connect to
>> + * @errp: pointer to a NULL-initialized error object
>> + *
>> + * Attempt to connect to the address @addr using non-blocking mode of
>> + * the socket. Function is synchronous, but being called from
>> + * coroutine context will yield during connect operation.
>> + */
>> +int qio_channel_socket_connect_non_blocking_sync(QIOChannelSocket *ioc,
>> +                                                 SocketAddress *addr,
>> +                                                 Error **errp);
>> +
>>   /**
>>    * qio_channel_socket_connect_async:
>>    * @ioc: the socket channel object
>> diff --git a/io/channel-socket.c b/io/channel-socket.c
>> index e1b4667087..076de7578a 100644
>> --- a/io/channel-socket.c
>> +++ b/io/channel-socket.c
>> @@ -22,6 +22,7 @@
>>   #include "qapi/error.h"
>>   #include "qapi/qapi-visit-sockets.h"
>>   #include "qemu/module.h"
>> +#include "qemu/sockets.h"
>>   #include "io/channel-socket.h"
>>   #include "io/channel-watch.h"
>>   #include "trace.h"
>> @@ -29,6 +30,8 @@
>>   
>>   #define SOCKET_MAX_FDS 16
>>   
>> +static int qio_channel_socket_close(QIOChannel *ioc, Error **errp);
>> +
>>   SocketAddress *
>>   qio_channel_socket_get_local_address(QIOChannelSocket *ioc,
>>                                        Error **errp)
>> @@ -157,6 +160,77 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
>>       return 0;
>>   }
>>   
>> +static int qio_channel_inet_connect_non_blocking_sync(QIOChannelSocket *ioc,
>> +        InetSocketAddress *addr, Error **errp)
>> +{
>> +    Error *local_err = NULL;
>> +    struct addrinfo *infos, *info;
>> +    int sock = -1;
>> +
>> +    infos = inet_parse_connect_saddr(addr, errp);
>> +    if (!infos) {
>> +        return -1;
>> +    }
> 
> This call is blocking since it calls getaddrinfo whose design
> offers no ability todo non-blocking DNS lookups. Given this
> call, ...

Oh, that's bad, thanks for taking a look on that early stage!

> 
>> +
>> +    for (info = infos; info != NULL; info = info->ai_next) {
>> +        bool in_progress;
>> +
>> +        error_free(local_err);
>> +        local_err = NULL;
>> +
>> +        sock = inet_connect_addr(addr, info, false, &in_progress, &local_err);
>> +        if (sock < 0) {
>> +            continue;
>> +        }
>> +
>> +        if (qio_channel_socket_set_fd(ioc, sock, &local_err) < 0) {
>> +            close(sock);
>> +            continue;
>> +        }
>> +
>> +        if (in_progress) {
>> +            if (qemu_in_coroutine()) {
>> +                qio_channel_yield(QIO_CHANNEL(ioc), G_IO_OUT);
>> +            } else {
>> +                qio_channel_wait(QIO_CHANNEL(ioc), G_IO_OUT);
>> +            }
> 
> ...this is offering false assurances of being non-blocking.
> 
> If we don't want the current thread to be blocked then we
> need to be using the existing qio_channel_socket_connect_async
> method or similar. It uses a throw away background thread to
> run the connection attempt, and then reports completion back
> later, thus avoiding the getaddrinfo design flaw for the callers.
> 
> I explicitly didn't want to add an method like the impl in this
> patch, because getaddrinfo dooms it and we already had bugs in
> the pre-QIOChannel code where QEMU thought it was non-blocking
> but wasn't due to getaddrinfo lookups.
> 
> 
> IIUC, the main appeal of this method is that the non-blocking
> nature is hidden from the caller who can continue to treat it
> as a synchronous call and have the coroutine magic happen in
> behind the scenes.
> 
> IOW, What's needed is a simple way to run the operation in a
> thread, and sleep for completion while having the coroutine
> yield.
> 
> I think this could likely be achieved with QIOTask with an
> alternate impl of the qio_task_wait_thread() method that is
> friendly to coroutines instead of being based on pthread
> condition variable waits.

The most simple thing is just run qio_channel_socket_connect_sync in
a thread with help of thread_pool_submit_co() which is coroutine-friendly.
And this don't need any changes in io/channel.

Actually, I've started with such design, but decided that better use
non-blocking connect to not deal with cancelling the connecting thread
on shutdown.

I think, I'll resend based on thread_pool_submit_co().

===

Hmm, there is async getaddrinfo_a function.. What do you think of it?
But seems simpler to use a thread than move to async interfaces everywhere.

> 
> 
>> +            if (socket_check(sock, &local_err) < 0) {
>> +                qio_channel_socket_close(QIO_CHANNEL(ioc), NULL);
>> +                continue;
>> +            }
>> +        }
>> +
>> +        break;
>> +    }
>> +
>> +    freeaddrinfo(infos);
>> +
>> +    error_propagate(errp, local_err);
>> +    return sock;
>> +}
> 
> Regards,
> Daniel
> 


-- 
Best regards,
Vladimir

Re: [PATCH 3/4] io/channel-socket: implement non-blocking connect

[Daniel P. Berrangé]

On Wed, Jul 22, 2020 at 02:00:25PM +0300, Vladimir Sementsov-Ogievskiy wrote:
> 20.07.2020 21:29, Daniel P. Berrangé wrote:
> > On Mon, Jul 20, 2020 at 09:07:14PM +0300, Vladimir Sementsov-Ogievskiy wrote:
> > > Utilize new socket API to make a non-blocking connect for inet sockets.
> > > 
> > > Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
> > > ---
> > >   include/io/channel-socket.h | 14 +++++++
> > >   io/channel-socket.c         | 74 +++++++++++++++++++++++++++++++++++++
> > >   2 files changed, 88 insertions(+)
> > > 
> > > diff --git a/include/io/channel-socket.h b/include/io/channel-socket.h
> > > index 777ff5954e..82e868bc02 100644
> > > --- a/include/io/channel-socket.h
> > > +++ b/include/io/channel-socket.h
> > > @@ -94,6 +94,20 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
> > >                                       SocketAddress *addr,
> > >                                       Error **errp);
> > > +/**
> > > + * qio_channel_socket_connect_non_blocking_sync:
> > > + * @ioc: the socket channel object
> > > + * @addr: the address to connect to
> > > + * @errp: pointer to a NULL-initialized error object
> > > + *
> > > + * Attempt to connect to the address @addr using non-blocking mode of
> > > + * the socket. Function is synchronous, but being called from
> > > + * coroutine context will yield during connect operation.
> > > + */
> > > +int qio_channel_socket_connect_non_blocking_sync(QIOChannelSocket *ioc,
> > > +                                                 SocketAddress *addr,
> > > +                                                 Error **errp);
> > > +
> > >   /**
> > >    * qio_channel_socket_connect_async:
> > >    * @ioc: the socket channel object
> > > diff --git a/io/channel-socket.c b/io/channel-socket.c
> > > index e1b4667087..076de7578a 100644
> > > --- a/io/channel-socket.c
> > > +++ b/io/channel-socket.c
> > > @@ -22,6 +22,7 @@
> > >   #include "qapi/error.h"
> > >   #include "qapi/qapi-visit-sockets.h"
> > >   #include "qemu/module.h"
> > > +#include "qemu/sockets.h"
> > >   #include "io/channel-socket.h"
> > >   #include "io/channel-watch.h"
> > >   #include "trace.h"
> > > @@ -29,6 +30,8 @@
> > >   #define SOCKET_MAX_FDS 16
> > > +static int qio_channel_socket_close(QIOChannel *ioc, Error **errp);
> > > +
> > >   SocketAddress *
> > >   qio_channel_socket_get_local_address(QIOChannelSocket *ioc,
> > >                                        Error **errp)
> > > @@ -157,6 +160,77 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
> > >       return 0;
> > >   }
> > > +static int qio_channel_inet_connect_non_blocking_sync(QIOChannelSocket *ioc,
> > > +        InetSocketAddress *addr, Error **errp)
> > > +{
> > > +    Error *local_err = NULL;
> > > +    struct addrinfo *infos, *info;
> > > +    int sock = -1;
> > > +
> > > +    infos = inet_parse_connect_saddr(addr, errp);
> > > +    if (!infos) {
> > > +        return -1;
> > > +    }
> > 
> > This call is blocking since it calls getaddrinfo whose design
> > offers no ability todo non-blocking DNS lookups. Given this
> > call, ...
> 
> Oh, that's bad, thanks for taking a look on that early stage!
> 
> > 
> > > +
> > > +    for (info = infos; info != NULL; info = info->ai_next) {
> > > +        bool in_progress;
> > > +
> > > +        error_free(local_err);
> > > +        local_err = NULL;
> > > +
> > > +        sock = inet_connect_addr(addr, info, false, &in_progress, &local_err);
> > > +        if (sock < 0) {
> > > +            continue;
> > > +        }
> > > +
> > > +        if (qio_channel_socket_set_fd(ioc, sock, &local_err) < 0) {
> > > +            close(sock);
> > > +            continue;
> > > +        }
> > > +
> > > +        if (in_progress) {
> > > +            if (qemu_in_coroutine()) {
> > > +                qio_channel_yield(QIO_CHANNEL(ioc), G_IO_OUT);
> > > +            } else {
> > > +                qio_channel_wait(QIO_CHANNEL(ioc), G_IO_OUT);
> > > +            }
> > 
> > ...this is offering false assurances of being non-blocking.
> > 
> > If we don't want the current thread to be blocked then we
> > need to be using the existing qio_channel_socket_connect_async
> > method or similar. It uses a throw away background thread to
> > run the connection attempt, and then reports completion back
> > later, thus avoiding the getaddrinfo design flaw for the callers.
> > 
> > I explicitly didn't want to add an method like the impl in this
> > patch, because getaddrinfo dooms it and we already had bugs in
> > the pre-QIOChannel code where QEMU thought it was non-blocking
> > but wasn't due to getaddrinfo lookups.
> > 
> > 
> > IIUC, the main appeal of this method is that the non-blocking
> > nature is hidden from the caller who can continue to treat it
> > as a synchronous call and have the coroutine magic happen in
> > behind the scenes.
> > 
> > IOW, What's needed is a simple way to run the operation in a
> > thread, and sleep for completion while having the coroutine
> > yield.
> > 
> > I think this could likely be achieved with QIOTask with an
> > alternate impl of the qio_task_wait_thread() method that is
> > friendly to coroutines instead of being based on pthread
> > condition variable waits.
> 
> The most simple thing is just run qio_channel_socket_connect_sync in
> a thread with help of thread_pool_submit_co() which is coroutine-friendly.
> And this don't need any changes in io/channel.
> 
> Actually, I've started with such design, but decided that better use
> non-blocking connect to not deal with cancelling the connecting thread
> on shutdown.
> 
> I think, I'll resend based on thread_pool_submit_co().
> 
> ===
> 
> Hmm, there is async getaddrinfo_a function.. What do you think of it?

It isn't portable, glibc only.

> But seems simpler to use a thread than move to async interfaces everywhere.


Regards,
Daniel
-- 
|: https://berrange.com      -o-    https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org         -o-            https://fstop138.berrange.com :|
|: https://entangle-photo.org    -o-    https://www.instagram.com/dberrange :|

Re: [PATCH 3/4] io/channel-socket: implement non-blocking connect

[Vladimir Sementsov-Ogievskiy]

22.07.2020 14:21, Daniel P. Berrangé wrote:
> On Wed, Jul 22, 2020 at 02:00:25PM +0300, Vladimir Sementsov-Ogievskiy wrote:
>> 20.07.2020 21:29, Daniel P. Berrangé wrote:
>>> On Mon, Jul 20, 2020 at 09:07:14PM +0300, Vladimir Sementsov-Ogievskiy wrote:
>>>> Utilize new socket API to make a non-blocking connect for inet sockets.
>>>>
>>>> Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
>>>> ---
>>>>    include/io/channel-socket.h | 14 +++++++
>>>>    io/channel-socket.c         | 74 +++++++++++++++++++++++++++++++++++++
>>>>    2 files changed, 88 insertions(+)
>>>>
>>>> diff --git a/include/io/channel-socket.h b/include/io/channel-socket.h
>>>> index 777ff5954e..82e868bc02 100644
>>>> --- a/include/io/channel-socket.h
>>>> +++ b/include/io/channel-socket.h
>>>> @@ -94,6 +94,20 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
>>>>                                        SocketAddress *addr,
>>>>                                        Error **errp);
>>>> +/**
>>>> + * qio_channel_socket_connect_non_blocking_sync:
>>>> + * @ioc: the socket channel object
>>>> + * @addr: the address to connect to
>>>> + * @errp: pointer to a NULL-initialized error object
>>>> + *
>>>> + * Attempt to connect to the address @addr using non-blocking mode of
>>>> + * the socket. Function is synchronous, but being called from
>>>> + * coroutine context will yield during connect operation.
>>>> + */
>>>> +int qio_channel_socket_connect_non_blocking_sync(QIOChannelSocket *ioc,
>>>> +                                                 SocketAddress *addr,
>>>> +                                                 Error **errp);
>>>> +
>>>>    /**
>>>>     * qio_channel_socket_connect_async:
>>>>     * @ioc: the socket channel object
>>>> diff --git a/io/channel-socket.c b/io/channel-socket.c
>>>> index e1b4667087..076de7578a 100644
>>>> --- a/io/channel-socket.c
>>>> +++ b/io/channel-socket.c
>>>> @@ -22,6 +22,7 @@
>>>>    #include "qapi/error.h"
>>>>    #include "qapi/qapi-visit-sockets.h"
>>>>    #include "qemu/module.h"
>>>> +#include "qemu/sockets.h"
>>>>    #include "io/channel-socket.h"
>>>>    #include "io/channel-watch.h"
>>>>    #include "trace.h"
>>>> @@ -29,6 +30,8 @@
>>>>    #define SOCKET_MAX_FDS 16
>>>> +static int qio_channel_socket_close(QIOChannel *ioc, Error **errp);
>>>> +
>>>>    SocketAddress *
>>>>    qio_channel_socket_get_local_address(QIOChannelSocket *ioc,
>>>>                                         Error **errp)
>>>> @@ -157,6 +160,77 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
>>>>        return 0;
>>>>    }
>>>> +static int qio_channel_inet_connect_non_blocking_sync(QIOChannelSocket *ioc,
>>>> +        InetSocketAddress *addr, Error **errp)
>>>> +{
>>>> +    Error *local_err = NULL;
>>>> +    struct addrinfo *infos, *info;
>>>> +    int sock = -1;
>>>> +
>>>> +    infos = inet_parse_connect_saddr(addr, errp);
>>>> +    if (!infos) {
>>>> +        return -1;
>>>> +    }
>>>
>>> This call is blocking since it calls getaddrinfo whose design
>>> offers no ability todo non-blocking DNS lookups. Given this
>>> call, ...
>>
>> Oh, that's bad, thanks for taking a look on that early stage!
>>
>>>
>>>> +
>>>> +    for (info = infos; info != NULL; info = info->ai_next) {
>>>> +        bool in_progress;
>>>> +
>>>> +        error_free(local_err);
>>>> +        local_err = NULL;
>>>> +
>>>> +        sock = inet_connect_addr(addr, info, false, &in_progress, &local_err);
>>>> +        if (sock < 0) {
>>>> +            continue;
>>>> +        }
>>>> +
>>>> +        if (qio_channel_socket_set_fd(ioc, sock, &local_err) < 0) {
>>>> +            close(sock);
>>>> +            continue;
>>>> +        }
>>>> +
>>>> +        if (in_progress) {
>>>> +            if (qemu_in_coroutine()) {
>>>> +                qio_channel_yield(QIO_CHANNEL(ioc), G_IO_OUT);
>>>> +            } else {
>>>> +                qio_channel_wait(QIO_CHANNEL(ioc), G_IO_OUT);
>>>> +            }
>>>
>>> ...this is offering false assurances of being non-blocking.
>>>
>>> If we don't want the current thread to be blocked then we
>>> need to be using the existing qio_channel_socket_connect_async
>>> method or similar. It uses a throw away background thread to
>>> run the connection attempt, and then reports completion back
>>> later, thus avoiding the getaddrinfo design flaw for the callers.
>>>
>>> I explicitly didn't want to add an method like the impl in this
>>> patch, because getaddrinfo dooms it and we already had bugs in
>>> the pre-QIOChannel code where QEMU thought it was non-blocking
>>> but wasn't due to getaddrinfo lookups.
>>>
>>>
>>> IIUC, the main appeal of this method is that the non-blocking
>>> nature is hidden from the caller who can continue to treat it
>>> as a synchronous call and have the coroutine magic happen in
>>> behind the scenes.
>>>
>>> IOW, What's needed is a simple way to run the operation in a
>>> thread, and sleep for completion while having the coroutine
>>> yield.
>>>
>>> I think this could likely be achieved with QIOTask with an
>>> alternate impl of the qio_task_wait_thread() method that is
>>> friendly to coroutines instead of being based on pthread
>>> condition variable waits.
>>
>> The most simple thing is just run qio_channel_socket_connect_sync in
>> a thread with help of thread_pool_submit_co() which is coroutine-friendly.
>> And this don't need any changes in io/channel.
>>
>> Actually, I've started with such design, but decided that better use
>> non-blocking connect to not deal with cancelling the connecting thread
>> on shutdown.
>>
>> I think, I'll resend based on thread_pool_submit_co().
>>
>> ===
>>
>> Hmm, there is async getaddrinfo_a function.. What do you think of it?
> 
> It isn't portable, glibc only.
> 
>> But seems simpler to use a thread than move to async interfaces everywhere.
> 
> 

Hmm.. Still, on shutdown, how to cancel this connect and getaddrinfo ? I'm not sure
how much time may getaddrinfo take, but connect can take about a minute. It's not really
good to wait for it on shutdown.

So, it may make sense to split connect, and keep it true-async. And call getaddrinfo in a thread
(or just keep it synchronous).



-- 
Best regards,
Vladimir

Re: [PATCH 3/4] io/channel-socket: implement non-blocking connect

[Daniel P. Berrangé]

On Wed, Jul 22, 2020 at 03:43:54PM +0300, Vladimir Sementsov-Ogievskiy wrote:
> 22.07.2020 14:21, Daniel P. Berrangé wrote:
> > On Wed, Jul 22, 2020 at 02:00:25PM +0300, Vladimir Sementsov-Ogievskiy wrote:
> > > 20.07.2020 21:29, Daniel P. Berrangé wrote:
> > > > On Mon, Jul 20, 2020 at 09:07:14PM +0300, Vladimir Sementsov-Ogievskiy wrote:
> > > > > Utilize new socket API to make a non-blocking connect for inet sockets.
> > > > > 
> > > > > Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
> > > > > ---
> > > > >    include/io/channel-socket.h | 14 +++++++
> > > > >    io/channel-socket.c         | 74 +++++++++++++++++++++++++++++++++++++
> > > > >    2 files changed, 88 insertions(+)
> > > > > 
> > > > > diff --git a/include/io/channel-socket.h b/include/io/channel-socket.h
> > > > > index 777ff5954e..82e868bc02 100644
> > > > > --- a/include/io/channel-socket.h
> > > > > +++ b/include/io/channel-socket.h
> > > > > @@ -94,6 +94,20 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
> > > > >                                        SocketAddress *addr,
> > > > >                                        Error **errp);
> > > > > +/**
> > > > > + * qio_channel_socket_connect_non_blocking_sync:
> > > > > + * @ioc: the socket channel object
> > > > > + * @addr: the address to connect to
> > > > > + * @errp: pointer to a NULL-initialized error object
> > > > > + *
> > > > > + * Attempt to connect to the address @addr using non-blocking mode of
> > > > > + * the socket. Function is synchronous, but being called from
> > > > > + * coroutine context will yield during connect operation.
> > > > > + */
> > > > > +int qio_channel_socket_connect_non_blocking_sync(QIOChannelSocket *ioc,
> > > > > +                                                 SocketAddress *addr,
> > > > > +                                                 Error **errp);
> > > > > +
> > > > >    /**
> > > > >     * qio_channel_socket_connect_async:
> > > > >     * @ioc: the socket channel object
> > > > > diff --git a/io/channel-socket.c b/io/channel-socket.c
> > > > > index e1b4667087..076de7578a 100644
> > > > > --- a/io/channel-socket.c
> > > > > +++ b/io/channel-socket.c
> > > > > @@ -22,6 +22,7 @@
> > > > >    #include "qapi/error.h"
> > > > >    #include "qapi/qapi-visit-sockets.h"
> > > > >    #include "qemu/module.h"
> > > > > +#include "qemu/sockets.h"
> > > > >    #include "io/channel-socket.h"
> > > > >    #include "io/channel-watch.h"
> > > > >    #include "trace.h"
> > > > > @@ -29,6 +30,8 @@
> > > > >    #define SOCKET_MAX_FDS 16
> > > > > +static int qio_channel_socket_close(QIOChannel *ioc, Error **errp);
> > > > > +
> > > > >    SocketAddress *
> > > > >    qio_channel_socket_get_local_address(QIOChannelSocket *ioc,
> > > > >                                         Error **errp)
> > > > > @@ -157,6 +160,77 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
> > > > >        return 0;
> > > > >    }
> > > > > +static int qio_channel_inet_connect_non_blocking_sync(QIOChannelSocket *ioc,
> > > > > +        InetSocketAddress *addr, Error **errp)
> > > > > +{
> > > > > +    Error *local_err = NULL;
> > > > > +    struct addrinfo *infos, *info;
> > > > > +    int sock = -1;
> > > > > +
> > > > > +    infos = inet_parse_connect_saddr(addr, errp);
> > > > > +    if (!infos) {
> > > > > +        return -1;
> > > > > +    }
> > > > 
> > > > This call is blocking since it calls getaddrinfo whose design
> > > > offers no ability todo non-blocking DNS lookups. Given this
> > > > call, ...
> > > 
> > > Oh, that's bad, thanks for taking a look on that early stage!
> > > 
> > > > 
> > > > > +
> > > > > +    for (info = infos; info != NULL; info = info->ai_next) {
> > > > > +        bool in_progress;
> > > > > +
> > > > > +        error_free(local_err);
> > > > > +        local_err = NULL;
> > > > > +
> > > > > +        sock = inet_connect_addr(addr, info, false, &in_progress, &local_err);
> > > > > +        if (sock < 0) {
> > > > > +            continue;
> > > > > +        }
> > > > > +
> > > > > +        if (qio_channel_socket_set_fd(ioc, sock, &local_err) < 0) {
> > > > > +            close(sock);
> > > > > +            continue;
> > > > > +        }
> > > > > +
> > > > > +        if (in_progress) {
> > > > > +            if (qemu_in_coroutine()) {
> > > > > +                qio_channel_yield(QIO_CHANNEL(ioc), G_IO_OUT);
> > > > > +            } else {
> > > > > +                qio_channel_wait(QIO_CHANNEL(ioc), G_IO_OUT);
> > > > > +            }
> > > > 
> > > > ...this is offering false assurances of being non-blocking.
> > > > 
> > > > If we don't want the current thread to be blocked then we
> > > > need to be using the existing qio_channel_socket_connect_async
> > > > method or similar. It uses a throw away background thread to
> > > > run the connection attempt, and then reports completion back
> > > > later, thus avoiding the getaddrinfo design flaw for the callers.
> > > > 
> > > > I explicitly didn't want to add an method like the impl in this
> > > > patch, because getaddrinfo dooms it and we already had bugs in
> > > > the pre-QIOChannel code where QEMU thought it was non-blocking
> > > > but wasn't due to getaddrinfo lookups.
> > > > 
> > > > 
> > > > IIUC, the main appeal of this method is that the non-blocking
> > > > nature is hidden from the caller who can continue to treat it
> > > > as a synchronous call and have the coroutine magic happen in
> > > > behind the scenes.
> > > > 
> > > > IOW, What's needed is a simple way to run the operation in a
> > > > thread, and sleep for completion while having the coroutine
> > > > yield.
> > > > 
> > > > I think this could likely be achieved with QIOTask with an
> > > > alternate impl of the qio_task_wait_thread() method that is
> > > > friendly to coroutines instead of being based on pthread
> > > > condition variable waits.
> > > 
> > > The most simple thing is just run qio_channel_socket_connect_sync in
> > > a thread with help of thread_pool_submit_co() which is coroutine-friendly.
> > > And this don't need any changes in io/channel.
> > > 
> > > Actually, I've started with such design, but decided that better use
> > > non-blocking connect to not deal with cancelling the connecting thread
> > > on shutdown.
> > > 
> > > I think, I'll resend based on thread_pool_submit_co().
> > > 
> > > ===
> > > 
> > > Hmm, there is async getaddrinfo_a function.. What do you think of it?
> > 
> > It isn't portable, glibc only.
> > 
> > > But seems simpler to use a thread than move to async interfaces everywhere.
> > 
> > 
> 
> Hmm.. Still, on shutdown, how to cancel this connect and getaddrinfo ? I'm not sure
> how much time may getaddrinfo take, but connect can take about a minute. It's not really
> good to wait for it on shutdown.

The intention was that if you don't want to carry on waiting for the
async operation to complete you just give and pretend it no longer
exists. Eventually it will fail or complete and the thread will exit.
The only important thing there is making sure that the callback you
are passing to the _async() method can cope with the cleanup when the
work eventually completes, even if you've given up.

Regards,
Daniel
-- 
|: https://berrange.com      -o-    https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org         -o-            https://fstop138.berrange.com :|
|: https://entangle-photo.org    -o-    https://www.instagram.com/dberrange :|

Re: [PATCH 3/4] io/channel-socket: implement non-blocking connect

[Vladimir Sementsov-Ogievskiy]

22.07.2020 15:53, Daniel P. Berrangé wrote:
> On Wed, Jul 22, 2020 at 03:43:54PM +0300, Vladimir Sementsov-Ogievskiy wrote:
>> 22.07.2020 14:21, Daniel P. Berrangé wrote:
>>> On Wed, Jul 22, 2020 at 02:00:25PM +0300, Vladimir Sementsov-Ogievskiy wrote:
>>>> 20.07.2020 21:29, Daniel P. Berrangé wrote:
>>>>> On Mon, Jul 20, 2020 at 09:07:14PM +0300, Vladimir Sementsov-Ogievskiy wrote:
>>>>>> Utilize new socket API to make a non-blocking connect for inet sockets.
>>>>>>
>>>>>> Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
>>>>>> ---
>>>>>>     include/io/channel-socket.h | 14 +++++++
>>>>>>     io/channel-socket.c         | 74 +++++++++++++++++++++++++++++++++++++
>>>>>>     2 files changed, 88 insertions(+)
>>>>>>
>>>>>> diff --git a/include/io/channel-socket.h b/include/io/channel-socket.h
>>>>>> index 777ff5954e..82e868bc02 100644
>>>>>> --- a/include/io/channel-socket.h
>>>>>> +++ b/include/io/channel-socket.h
>>>>>> @@ -94,6 +94,20 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
>>>>>>                                         SocketAddress *addr,
>>>>>>                                         Error **errp);
>>>>>> +/**
>>>>>> + * qio_channel_socket_connect_non_blocking_sync:
>>>>>> + * @ioc: the socket channel object
>>>>>> + * @addr: the address to connect to
>>>>>> + * @errp: pointer to a NULL-initialized error object
>>>>>> + *
>>>>>> + * Attempt to connect to the address @addr using non-blocking mode of
>>>>>> + * the socket. Function is synchronous, but being called from
>>>>>> + * coroutine context will yield during connect operation.
>>>>>> + */
>>>>>> +int qio_channel_socket_connect_non_blocking_sync(QIOChannelSocket *ioc,
>>>>>> +                                                 SocketAddress *addr,
>>>>>> +                                                 Error **errp);
>>>>>> +
>>>>>>     /**
>>>>>>      * qio_channel_socket_connect_async:
>>>>>>      * @ioc: the socket channel object
>>>>>> diff --git a/io/channel-socket.c b/io/channel-socket.c
>>>>>> index e1b4667087..076de7578a 100644
>>>>>> --- a/io/channel-socket.c
>>>>>> +++ b/io/channel-socket.c
>>>>>> @@ -22,6 +22,7 @@
>>>>>>     #include "qapi/error.h"
>>>>>>     #include "qapi/qapi-visit-sockets.h"
>>>>>>     #include "qemu/module.h"
>>>>>> +#include "qemu/sockets.h"
>>>>>>     #include "io/channel-socket.h"
>>>>>>     #include "io/channel-watch.h"
>>>>>>     #include "trace.h"
>>>>>> @@ -29,6 +30,8 @@
>>>>>>     #define SOCKET_MAX_FDS 16
>>>>>> +static int qio_channel_socket_close(QIOChannel *ioc, Error **errp);
>>>>>> +
>>>>>>     SocketAddress *
>>>>>>     qio_channel_socket_get_local_address(QIOChannelSocket *ioc,
>>>>>>                                          Error **errp)
>>>>>> @@ -157,6 +160,77 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
>>>>>>         return 0;
>>>>>>     }
>>>>>> +static int qio_channel_inet_connect_non_blocking_sync(QIOChannelSocket *ioc,
>>>>>> +        InetSocketAddress *addr, Error **errp)
>>>>>> +{
>>>>>> +    Error *local_err = NULL;
>>>>>> +    struct addrinfo *infos, *info;
>>>>>> +    int sock = -1;
>>>>>> +
>>>>>> +    infos = inet_parse_connect_saddr(addr, errp);
>>>>>> +    if (!infos) {
>>>>>> +        return -1;
>>>>>> +    }
>>>>>
>>>>> This call is blocking since it calls getaddrinfo whose design
>>>>> offers no ability todo non-blocking DNS lookups. Given this
>>>>> call, ...
>>>>
>>>> Oh, that's bad, thanks for taking a look on that early stage!
>>>>
>>>>>
>>>>>> +
>>>>>> +    for (info = infos; info != NULL; info = info->ai_next) {
>>>>>> +        bool in_progress;
>>>>>> +
>>>>>> +        error_free(local_err);
>>>>>> +        local_err = NULL;
>>>>>> +
>>>>>> +        sock = inet_connect_addr(addr, info, false, &in_progress, &local_err);
>>>>>> +        if (sock < 0) {
>>>>>> +            continue;
>>>>>> +        }
>>>>>> +
>>>>>> +        if (qio_channel_socket_set_fd(ioc, sock, &local_err) < 0) {
>>>>>> +            close(sock);
>>>>>> +            continue;
>>>>>> +        }
>>>>>> +
>>>>>> +        if (in_progress) {
>>>>>> +            if (qemu_in_coroutine()) {
>>>>>> +                qio_channel_yield(QIO_CHANNEL(ioc), G_IO_OUT);
>>>>>> +            } else {
>>>>>> +                qio_channel_wait(QIO_CHANNEL(ioc), G_IO_OUT);
>>>>>> +            }
>>>>>
>>>>> ...this is offering false assurances of being non-blocking.
>>>>>
>>>>> If we don't want the current thread to be blocked then we
>>>>> need to be using the existing qio_channel_socket_connect_async
>>>>> method or similar. It uses a throw away background thread to
>>>>> run the connection attempt, and then reports completion back
>>>>> later, thus avoiding the getaddrinfo design flaw for the callers.
>>>>>
>>>>> I explicitly didn't want to add an method like the impl in this
>>>>> patch, because getaddrinfo dooms it and we already had bugs in
>>>>> the pre-QIOChannel code where QEMU thought it was non-blocking
>>>>> but wasn't due to getaddrinfo lookups.
>>>>>
>>>>>
>>>>> IIUC, the main appeal of this method is that the non-blocking
>>>>> nature is hidden from the caller who can continue to treat it
>>>>> as a synchronous call and have the coroutine magic happen in
>>>>> behind the scenes.
>>>>>
>>>>> IOW, What's needed is a simple way to run the operation in a
>>>>> thread, and sleep for completion while having the coroutine
>>>>> yield.
>>>>>
>>>>> I think this could likely be achieved with QIOTask with an
>>>>> alternate impl of the qio_task_wait_thread() method that is
>>>>> friendly to coroutines instead of being based on pthread
>>>>> condition variable waits.
>>>>
>>>> The most simple thing is just run qio_channel_socket_connect_sync in
>>>> a thread with help of thread_pool_submit_co() which is coroutine-friendly.
>>>> And this don't need any changes in io/channel.
>>>>
>>>> Actually, I've started with such design, but decided that better use
>>>> non-blocking connect to not deal with cancelling the connecting thread
>>>> on shutdown.
>>>>
>>>> I think, I'll resend based on thread_pool_submit_co().
>>>>
>>>> ===
>>>>
>>>> Hmm, there is async getaddrinfo_a function.. What do you think of it?
>>>
>>> It isn't portable, glibc only.
>>>
>>>> But seems simpler to use a thread than move to async interfaces everywhere.
>>>
>>>
>>
>> Hmm.. Still, on shutdown, how to cancel this connect and getaddrinfo ? I'm not sure
>> how much time may getaddrinfo take, but connect can take about a minute. It's not really
>> good to wait for it on shutdown.
> 
> The intention was that if you don't want to carry on waiting for the
> async operation to complete you just give and pretend it no longer
> exists. Eventually it will fail or complete and the thread will exit.
> The only important thing there is making sure that the callback you
> are passing to the _async() method can cope with the cleanup when the
> work eventually completes, even if you've given up.
> 

At least it's not possible with thread_pool_submit_co as I wanted, because underlying
thread pool waits for all its threads to complete on exit.


-- 
Best regards,
Vladimir

Re: [PATCH 3/4] io/channel-socket: implement non-blocking connect

[Vladimir Sementsov-Ogievskiy]

22.07.2020 16:47, Vladimir Sementsov-Ogievskiy wrote:
> 22.07.2020 15:53, Daniel P. Berrangé wrote:
>> On Wed, Jul 22, 2020 at 03:43:54PM +0300, Vladimir Sementsov-Ogievskiy wrote:
>>> 22.07.2020 14:21, Daniel P. Berrangé wrote:
>>>> On Wed, Jul 22, 2020 at 02:00:25PM +0300, Vladimir Sementsov-Ogievskiy wrote:
>>>>> 20.07.2020 21:29, Daniel P. Berrangé wrote:
>>>>>> On Mon, Jul 20, 2020 at 09:07:14PM +0300, Vladimir Sementsov-Ogievskiy wrote:
>>>>>>> Utilize new socket API to make a non-blocking connect for inet sockets.
>>>>>>>
>>>>>>> Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
>>>>>>> ---
>>>>>>>     include/io/channel-socket.h | 14 +++++++
>>>>>>>     io/channel-socket.c         | 74 +++++++++++++++++++++++++++++++++++++
>>>>>>>     2 files changed, 88 insertions(+)
>>>>>>>
>>>>>>> diff --git a/include/io/channel-socket.h b/include/io/channel-socket.h
>>>>>>> index 777ff5954e..82e868bc02 100644
>>>>>>> --- a/include/io/channel-socket.h
>>>>>>> +++ b/include/io/channel-socket.h
>>>>>>> @@ -94,6 +94,20 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
>>>>>>>                                         SocketAddress *addr,
>>>>>>>                                         Error **errp);
>>>>>>> +/**
>>>>>>> + * qio_channel_socket_connect_non_blocking_sync:
>>>>>>> + * @ioc: the socket channel object
>>>>>>> + * @addr: the address to connect to
>>>>>>> + * @errp: pointer to a NULL-initialized error object
>>>>>>> + *
>>>>>>> + * Attempt to connect to the address @addr using non-blocking mode of
>>>>>>> + * the socket. Function is synchronous, but being called from
>>>>>>> + * coroutine context will yield during connect operation.
>>>>>>> + */
>>>>>>> +int qio_channel_socket_connect_non_blocking_sync(QIOChannelSocket *ioc,
>>>>>>> +                                                 SocketAddress *addr,
>>>>>>> +                                                 Error **errp);
>>>>>>> +
>>>>>>>     /**
>>>>>>>      * qio_channel_socket_connect_async:
>>>>>>>      * @ioc: the socket channel object
>>>>>>> diff --git a/io/channel-socket.c b/io/channel-socket.c
>>>>>>> index e1b4667087..076de7578a 100644
>>>>>>> --- a/io/channel-socket.c
>>>>>>> +++ b/io/channel-socket.c
>>>>>>> @@ -22,6 +22,7 @@
>>>>>>>     #include "qapi/error.h"
>>>>>>>     #include "qapi/qapi-visit-sockets.h"
>>>>>>>     #include "qemu/module.h"
>>>>>>> +#include "qemu/sockets.h"
>>>>>>>     #include "io/channel-socket.h"
>>>>>>>     #include "io/channel-watch.h"
>>>>>>>     #include "trace.h"
>>>>>>> @@ -29,6 +30,8 @@
>>>>>>>     #define SOCKET_MAX_FDS 16
>>>>>>> +static int qio_channel_socket_close(QIOChannel *ioc, Error **errp);
>>>>>>> +
>>>>>>>     SocketAddress *
>>>>>>>     qio_channel_socket_get_local_address(QIOChannelSocket *ioc,
>>>>>>>                                          Error **errp)
>>>>>>> @@ -157,6 +160,77 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
>>>>>>>         return 0;
>>>>>>>     }
>>>>>>> +static int qio_channel_inet_connect_non_blocking_sync(QIOChannelSocket *ioc,
>>>>>>> +        InetSocketAddress *addr, Error **errp)
>>>>>>> +{
>>>>>>> +    Error *local_err = NULL;
>>>>>>> +    struct addrinfo *infos, *info;
>>>>>>> +    int sock = -1;
>>>>>>> +
>>>>>>> +    infos = inet_parse_connect_saddr(addr, errp);
>>>>>>> +    if (!infos) {
>>>>>>> +        return -1;
>>>>>>> +    }
>>>>>>
>>>>>> This call is blocking since it calls getaddrinfo whose design
>>>>>> offers no ability todo non-blocking DNS lookups. Given this
>>>>>> call, ...
>>>>>
>>>>> Oh, that's bad, thanks for taking a look on that early stage!
>>>>>
>>>>>>
>>>>>>> +
>>>>>>> +    for (info = infos; info != NULL; info = info->ai_next) {
>>>>>>> +        bool in_progress;
>>>>>>> +
>>>>>>> +        error_free(local_err);
>>>>>>> +        local_err = NULL;
>>>>>>> +
>>>>>>> +        sock = inet_connect_addr(addr, info, false, &in_progress, &local_err);
>>>>>>> +        if (sock < 0) {
>>>>>>> +            continue;
>>>>>>> +        }
>>>>>>> +
>>>>>>> +        if (qio_channel_socket_set_fd(ioc, sock, &local_err) < 0) {
>>>>>>> +            close(sock);
>>>>>>> +            continue;
>>>>>>> +        }
>>>>>>> +
>>>>>>> +        if (in_progress) {
>>>>>>> +            if (qemu_in_coroutine()) {
>>>>>>> +                qio_channel_yield(QIO_CHANNEL(ioc), G_IO_OUT);
>>>>>>> +            } else {
>>>>>>> +                qio_channel_wait(QIO_CHANNEL(ioc), G_IO_OUT);
>>>>>>> +            }
>>>>>>
>>>>>> ...this is offering false assurances of being non-blocking.
>>>>>>
>>>>>> If we don't want the current thread to be blocked then we
>>>>>> need to be using the existing qio_channel_socket_connect_async
>>>>>> method or similar. It uses a throw away background thread to
>>>>>> run the connection attempt, and then reports completion back
>>>>>> later, thus avoiding the getaddrinfo design flaw for the callers.
>>>>>>
>>>>>> I explicitly didn't want to add an method like the impl in this
>>>>>> patch, because getaddrinfo dooms it and we already had bugs in
>>>>>> the pre-QIOChannel code where QEMU thought it was non-blocking
>>>>>> but wasn't due to getaddrinfo lookups.
>>>>>>
>>>>>>
>>>>>> IIUC, the main appeal of this method is that the non-blocking
>>>>>> nature is hidden from the caller who can continue to treat it
>>>>>> as a synchronous call and have the coroutine magic happen in
>>>>>> behind the scenes.
>>>>>>
>>>>>> IOW, What's needed is a simple way to run the operation in a
>>>>>> thread, and sleep for completion while having the coroutine
>>>>>> yield.
>>>>>>
>>>>>> I think this could likely be achieved with QIOTask with an
>>>>>> alternate impl of the qio_task_wait_thread() method that is
>>>>>> friendly to coroutines instead of being based on pthread
>>>>>> condition variable waits.
>>>>>
>>>>> The most simple thing is just run qio_channel_socket_connect_sync in
>>>>> a thread with help of thread_pool_submit_co() which is coroutine-friendly.
>>>>> And this don't need any changes in io/channel.
>>>>>
>>>>> Actually, I've started with such design, but decided that better use
>>>>> non-blocking connect to not deal with cancelling the connecting thread
>>>>> on shutdown.
>>>>>
>>>>> I think, I'll resend based on thread_pool_submit_co().
>>>>>
>>>>> ===
>>>>>
>>>>> Hmm, there is async getaddrinfo_a function.. What do you think of it?
>>>>
>>>> It isn't portable, glibc only.
>>>>
>>>>> But seems simpler to use a thread than move to async interfaces everywhere.
>>>>
>>>>
>>>
>>> Hmm.. Still, on shutdown, how to cancel this connect and getaddrinfo ? I'm not sure
>>> how much time may getaddrinfo take, but connect can take about a minute. It's not really
>>> good to wait for it on shutdown.
>>
>> The intention was that if you don't want to carry on waiting for the
>> async operation to complete you just give and pretend it no longer
>> exists. Eventually it will fail or complete and the thread will exit.
>> The only important thing there is making sure that the callback you
>> are passing to the _async() method can cope with the cleanup when the
>> work eventually completes, even if you've given up.
>>
> 
> At least it's not possible with thread_pool_submit_co as I wanted, because underlying
> thread pool waits for all its threads to complete on exit.
> 
> 


I'm trying to use qio_channel_socket_connect_async().. But callback is not called.

How to make it be executed? In tests/test-io-channel-socket.c it's done by g_main_loop_new .. g_main_loop_run. But I need to yield. socket_start_outgoing_migration uses qio_channel_socket_connect_async as well, but is not doing any magic with g_main_loop. But it works. How?


-- 
Best regards,
Vladimir

Re: [PATCH 3/4] io/channel-socket: implement non-blocking connect

[Daniel P. Berrangé]

On Wed, Jul 22, 2020 at 06:04:53PM +0300, Vladimir Sementsov-Ogievskiy wrote:
> 22.07.2020 16:47, Vladimir Sementsov-Ogievskiy wrote:
> > 22.07.2020 15:53, Daniel P. Berrangé wrote:
> > > On Wed, Jul 22, 2020 at 03:43:54PM +0300, Vladimir Sementsov-Ogievskiy wrote:
> > > > 22.07.2020 14:21, Daniel P. Berrangé wrote:
> > > > > On Wed, Jul 22, 2020 at 02:00:25PM +0300, Vladimir Sementsov-Ogievskiy wrote:
> > > > > > 20.07.2020 21:29, Daniel P. Berrangé wrote:
> > > > > > > On Mon, Jul 20, 2020 at 09:07:14PM +0300, Vladimir Sementsov-Ogievskiy wrote:
> > > > > > > > Utilize new socket API to make a non-blocking connect for inet sockets.
> > > > > > > > 
> > > > > > > > Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
> > > > > > > > ---
> > > > > > > >     include/io/channel-socket.h | 14 +++++++
> > > > > > > >     io/channel-socket.c         | 74 +++++++++++++++++++++++++++++++++++++
> > > > > > > >     2 files changed, 88 insertions(+)
> > > > > > > > 
> > > > > > > > diff --git a/include/io/channel-socket.h b/include/io/channel-socket.h
> > > > > > > > index 777ff5954e..82e868bc02 100644
> > > > > > > > --- a/include/io/channel-socket.h
> > > > > > > > +++ b/include/io/channel-socket.h
> > > > > > > > @@ -94,6 +94,20 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
> > > > > > > >                                         SocketAddress *addr,
> > > > > > > >                                         Error **errp);
> > > > > > > > +/**
> > > > > > > > + * qio_channel_socket_connect_non_blocking_sync:
> > > > > > > > + * @ioc: the socket channel object
> > > > > > > > + * @addr: the address to connect to
> > > > > > > > + * @errp: pointer to a NULL-initialized error object
> > > > > > > > + *
> > > > > > > > + * Attempt to connect to the address @addr using non-blocking mode of
> > > > > > > > + * the socket. Function is synchronous, but being called from
> > > > > > > > + * coroutine context will yield during connect operation.
> > > > > > > > + */
> > > > > > > > +int qio_channel_socket_connect_non_blocking_sync(QIOChannelSocket *ioc,
> > > > > > > > +                                                 SocketAddress *addr,
> > > > > > > > +                                                 Error **errp);
> > > > > > > > +
> > > > > > > >     /**
> > > > > > > >      * qio_channel_socket_connect_async:
> > > > > > > >      * @ioc: the socket channel object
> > > > > > > > diff --git a/io/channel-socket.c b/io/channel-socket.c
> > > > > > > > index e1b4667087..076de7578a 100644
> > > > > > > > --- a/io/channel-socket.c
> > > > > > > > +++ b/io/channel-socket.c
> > > > > > > > @@ -22,6 +22,7 @@
> > > > > > > >     #include "qapi/error.h"
> > > > > > > >     #include "qapi/qapi-visit-sockets.h"
> > > > > > > >     #include "qemu/module.h"
> > > > > > > > +#include "qemu/sockets.h"
> > > > > > > >     #include "io/channel-socket.h"
> > > > > > > >     #include "io/channel-watch.h"
> > > > > > > >     #include "trace.h"
> > > > > > > > @@ -29,6 +30,8 @@
> > > > > > > >     #define SOCKET_MAX_FDS 16
> > > > > > > > +static int qio_channel_socket_close(QIOChannel *ioc, Error **errp);
> > > > > > > > +
> > > > > > > >     SocketAddress *
> > > > > > > >     qio_channel_socket_get_local_address(QIOChannelSocket *ioc,
> > > > > > > >                                          Error **errp)
> > > > > > > > @@ -157,6 +160,77 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
> > > > > > > >         return 0;
> > > > > > > >     }
> > > > > > > > +static int qio_channel_inet_connect_non_blocking_sync(QIOChannelSocket *ioc,
> > > > > > > > +        InetSocketAddress *addr, Error **errp)
> > > > > > > > +{
> > > > > > > > +    Error *local_err = NULL;
> > > > > > > > +    struct addrinfo *infos, *info;
> > > > > > > > +    int sock = -1;
> > > > > > > > +
> > > > > > > > +    infos = inet_parse_connect_saddr(addr, errp);
> > > > > > > > +    if (!infos) {
> > > > > > > > +        return -1;
> > > > > > > > +    }
> > > > > > > 
> > > > > > > This call is blocking since it calls getaddrinfo whose design
> > > > > > > offers no ability todo non-blocking DNS lookups. Given this
> > > > > > > call, ...
> > > > > > 
> > > > > > Oh, that's bad, thanks for taking a look on that early stage!
> > > > > > 
> > > > > > > 
> > > > > > > > +
> > > > > > > > +    for (info = infos; info != NULL; info = info->ai_next) {
> > > > > > > > +        bool in_progress;
> > > > > > > > +
> > > > > > > > +        error_free(local_err);
> > > > > > > > +        local_err = NULL;
> > > > > > > > +
> > > > > > > > +        sock = inet_connect_addr(addr, info, false, &in_progress, &local_err);
> > > > > > > > +        if (sock < 0) {
> > > > > > > > +            continue;
> > > > > > > > +        }
> > > > > > > > +
> > > > > > > > +        if (qio_channel_socket_set_fd(ioc, sock, &local_err) < 0) {
> > > > > > > > +            close(sock);
> > > > > > > > +            continue;
> > > > > > > > +        }
> > > > > > > > +
> > > > > > > > +        if (in_progress) {
> > > > > > > > +            if (qemu_in_coroutine()) {
> > > > > > > > +                qio_channel_yield(QIO_CHANNEL(ioc), G_IO_OUT);
> > > > > > > > +            } else {
> > > > > > > > +                qio_channel_wait(QIO_CHANNEL(ioc), G_IO_OUT);
> > > > > > > > +            }
> > > > > > > 
> > > > > > > ...this is offering false assurances of being non-blocking.
> > > > > > > 
> > > > > > > If we don't want the current thread to be blocked then we
> > > > > > > need to be using the existing qio_channel_socket_connect_async
> > > > > > > method or similar. It uses a throw away background thread to
> > > > > > > run the connection attempt, and then reports completion back
> > > > > > > later, thus avoiding the getaddrinfo design flaw for the callers.
> > > > > > > 
> > > > > > > I explicitly didn't want to add an method like the impl in this
> > > > > > > patch, because getaddrinfo dooms it and we already had bugs in
> > > > > > > the pre-QIOChannel code where QEMU thought it was non-blocking
> > > > > > > but wasn't due to getaddrinfo lookups.
> > > > > > > 
> > > > > > > 
> > > > > > > IIUC, the main appeal of this method is that the non-blocking
> > > > > > > nature is hidden from the caller who can continue to treat it
> > > > > > > as a synchronous call and have the coroutine magic happen in
> > > > > > > behind the scenes.
> > > > > > > 
> > > > > > > IOW, What's needed is a simple way to run the operation in a
> > > > > > > thread, and sleep for completion while having the coroutine
> > > > > > > yield.
> > > > > > > 
> > > > > > > I think this could likely be achieved with QIOTask with an
> > > > > > > alternate impl of the qio_task_wait_thread() method that is
> > > > > > > friendly to coroutines instead of being based on pthread
> > > > > > > condition variable waits.
> > > > > > 
> > > > > > The most simple thing is just run qio_channel_socket_connect_sync in
> > > > > > a thread with help of thread_pool_submit_co() which is coroutine-friendly.
> > > > > > And this don't need any changes in io/channel.
> > > > > > 
> > > > > > Actually, I've started with such design, but decided that better use
> > > > > > non-blocking connect to not deal with cancelling the connecting thread
> > > > > > on shutdown.
> > > > > > 
> > > > > > I think, I'll resend based on thread_pool_submit_co().
> > > > > > 
> > > > > > ===
> > > > > > 
> > > > > > Hmm, there is async getaddrinfo_a function.. What do you think of it?
> > > > > 
> > > > > It isn't portable, glibc only.
> > > > > 
> > > > > > But seems simpler to use a thread than move to async interfaces everywhere.
> > > > > 
> > > > > 
> > > > 
> > > > Hmm.. Still, on shutdown, how to cancel this connect and getaddrinfo ? I'm not sure
> > > > how much time may getaddrinfo take, but connect can take about a minute. It's not really
> > > > good to wait for it on shutdown.
> > > 
> > > The intention was that if you don't want to carry on waiting for the
> > > async operation to complete you just give and pretend it no longer
> > > exists. Eventually it will fail or complete and the thread will exit.
> > > The only important thing there is making sure that the callback you
> > > are passing to the _async() method can cope with the cleanup when the
> > > work eventually completes, even if you've given up.
> > > 
> > 
> > At least it's not possible with thread_pool_submit_co as I wanted, because underlying
> > thread pool waits for all its threads to complete on exit.
> > 
> > 
> 
> 
> I'm trying to use qio_channel_socket_connect_async().. But callback
> is not called.
> 
> How to make it be executed? In tests/test-io-channel-socket.c it's
> done by g_main_loop_new .. g_main_loop_run. But I need to yield.
> socket_start_outgoing_migration uses qio_channel_socket_connect_async
> as well, but is not doing any magic with g_main_loop. But it works. How?

The _async() impls uses  qio_task_run_in_thread to spawn the background
thread. When the thread finishes, it uses g_idle_add to invoke the
callback so that it runs in the context of the main thread, not the
background thread. So something needs to be running the main loop
in QEMU.

If that isn't viable for the scenario you're in, then you can create
your own async version by spawning the thread yourself and running
the _sync() function - the _async() function is ajust a convenient
wrapper combining QIOTask + _sync(), but doesn't always do what 
everyone might need.   In the char-socket.c for example, I do the
async bit manually instead of using teh _async() wrapper.

Regards,
Daniel
-- 
|: https://berrange.com      -o-    https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org         -o-            https://fstop138.berrange.com :|
|: https://entangle-photo.org    -o-    https://www.instagram.com/dberrange :|

Re: [PATCH 3/4] io/channel-socket: implement non-blocking connect

[Vladimir Sementsov-Ogievskiy]

22.07.2020 18:21, Daniel P. Berrangé wrote:
> On Wed, Jul 22, 2020 at 06:04:53PM +0300, Vladimir Sementsov-Ogievskiy wrote:
>> 22.07.2020 16:47, Vladimir Sementsov-Ogievskiy wrote:
>>> 22.07.2020 15:53, Daniel P. Berrangé wrote:
>>>> On Wed, Jul 22, 2020 at 03:43:54PM +0300, Vladimir Sementsov-Ogievskiy wrote:
>>>>> 22.07.2020 14:21, Daniel P. Berrangé wrote:
>>>>>> On Wed, Jul 22, 2020 at 02:00:25PM +0300, Vladimir Sementsov-Ogievskiy wrote:
>>>>>>> 20.07.2020 21:29, Daniel P. Berrangé wrote:
>>>>>>>> On Mon, Jul 20, 2020 at 09:07:14PM +0300, Vladimir Sementsov-Ogievskiy wrote:
>>>>>>>>> Utilize new socket API to make a non-blocking connect for inet sockets.
>>>>>>>>>
>>>>>>>>> Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
>>>>>>>>> ---
>>>>>>>>>      include/io/channel-socket.h | 14 +++++++
>>>>>>>>>      io/channel-socket.c         | 74 +++++++++++++++++++++++++++++++++++++
>>>>>>>>>      2 files changed, 88 insertions(+)
>>>>>>>>>
>>>>>>>>> diff --git a/include/io/channel-socket.h b/include/io/channel-socket.h
>>>>>>>>> index 777ff5954e..82e868bc02 100644
>>>>>>>>> --- a/include/io/channel-socket.h
>>>>>>>>> +++ b/include/io/channel-socket.h
>>>>>>>>> @@ -94,6 +94,20 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
>>>>>>>>>                                          SocketAddress *addr,
>>>>>>>>>                                          Error **errp);
>>>>>>>>> +/**
>>>>>>>>> + * qio_channel_socket_connect_non_blocking_sync:
>>>>>>>>> + * @ioc: the socket channel object
>>>>>>>>> + * @addr: the address to connect to
>>>>>>>>> + * @errp: pointer to a NULL-initialized error object
>>>>>>>>> + *
>>>>>>>>> + * Attempt to connect to the address @addr using non-blocking mode of
>>>>>>>>> + * the socket. Function is synchronous, but being called from
>>>>>>>>> + * coroutine context will yield during connect operation.
>>>>>>>>> + */
>>>>>>>>> +int qio_channel_socket_connect_non_blocking_sync(QIOChannelSocket *ioc,
>>>>>>>>> +                                                 SocketAddress *addr,
>>>>>>>>> +                                                 Error **errp);
>>>>>>>>> +
>>>>>>>>>      /**
>>>>>>>>>       * qio_channel_socket_connect_async:
>>>>>>>>>       * @ioc: the socket channel object
>>>>>>>>> diff --git a/io/channel-socket.c b/io/channel-socket.c
>>>>>>>>> index e1b4667087..076de7578a 100644
>>>>>>>>> --- a/io/channel-socket.c
>>>>>>>>> +++ b/io/channel-socket.c
>>>>>>>>> @@ -22,6 +22,7 @@
>>>>>>>>>      #include "qapi/error.h"
>>>>>>>>>      #include "qapi/qapi-visit-sockets.h"
>>>>>>>>>      #include "qemu/module.h"
>>>>>>>>> +#include "qemu/sockets.h"
>>>>>>>>>      #include "io/channel-socket.h"
>>>>>>>>>      #include "io/channel-watch.h"
>>>>>>>>>      #include "trace.h"
>>>>>>>>> @@ -29,6 +30,8 @@
>>>>>>>>>      #define SOCKET_MAX_FDS 16
>>>>>>>>> +static int qio_channel_socket_close(QIOChannel *ioc, Error **errp);
>>>>>>>>> +
>>>>>>>>>      SocketAddress *
>>>>>>>>>      qio_channel_socket_get_local_address(QIOChannelSocket *ioc,
>>>>>>>>>                                           Error **errp)
>>>>>>>>> @@ -157,6 +160,77 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
>>>>>>>>>          return 0;
>>>>>>>>>      }
>>>>>>>>> +static int qio_channel_inet_connect_non_blocking_sync(QIOChannelSocket *ioc,
>>>>>>>>> +        InetSocketAddress *addr, Error **errp)
>>>>>>>>> +{
>>>>>>>>> +    Error *local_err = NULL;
>>>>>>>>> +    struct addrinfo *infos, *info;
>>>>>>>>> +    int sock = -1;
>>>>>>>>> +
>>>>>>>>> +    infos = inet_parse_connect_saddr(addr, errp);
>>>>>>>>> +    if (!infos) {
>>>>>>>>> +        return -1;
>>>>>>>>> +    }
>>>>>>>>
>>>>>>>> This call is blocking since it calls getaddrinfo whose design
>>>>>>>> offers no ability todo non-blocking DNS lookups. Given this
>>>>>>>> call, ...
>>>>>>>
>>>>>>> Oh, that's bad, thanks for taking a look on that early stage!
>>>>>>>
>>>>>>>>
>>>>>>>>> +
>>>>>>>>> +    for (info = infos; info != NULL; info = info->ai_next) {
>>>>>>>>> +        bool in_progress;
>>>>>>>>> +
>>>>>>>>> +        error_free(local_err);
>>>>>>>>> +        local_err = NULL;
>>>>>>>>> +
>>>>>>>>> +        sock = inet_connect_addr(addr, info, false, &in_progress, &local_err);
>>>>>>>>> +        if (sock < 0) {
>>>>>>>>> +            continue;
>>>>>>>>> +        }
>>>>>>>>> +
>>>>>>>>> +        if (qio_channel_socket_set_fd(ioc, sock, &local_err) < 0) {
>>>>>>>>> +            close(sock);
>>>>>>>>> +            continue;
>>>>>>>>> +        }
>>>>>>>>> +
>>>>>>>>> +        if (in_progress) {
>>>>>>>>> +            if (qemu_in_coroutine()) {
>>>>>>>>> +                qio_channel_yield(QIO_CHANNEL(ioc), G_IO_OUT);
>>>>>>>>> +            } else {
>>>>>>>>> +                qio_channel_wait(QIO_CHANNEL(ioc), G_IO_OUT);
>>>>>>>>> +            }
>>>>>>>>
>>>>>>>> ...this is offering false assurances of being non-blocking.
>>>>>>>>
>>>>>>>> If we don't want the current thread to be blocked then we
>>>>>>>> need to be using the existing qio_channel_socket_connect_async
>>>>>>>> method or similar. It uses a throw away background thread to
>>>>>>>> run the connection attempt, and then reports completion back
>>>>>>>> later, thus avoiding the getaddrinfo design flaw for the callers.
>>>>>>>>
>>>>>>>> I explicitly didn't want to add an method like the impl in this
>>>>>>>> patch, because getaddrinfo dooms it and we already had bugs in
>>>>>>>> the pre-QIOChannel code where QEMU thought it was non-blocking
>>>>>>>> but wasn't due to getaddrinfo lookups.
>>>>>>>>
>>>>>>>>
>>>>>>>> IIUC, the main appeal of this method is that the non-blocking
>>>>>>>> nature is hidden from the caller who can continue to treat it
>>>>>>>> as a synchronous call and have the coroutine magic happen in
>>>>>>>> behind the scenes.
>>>>>>>>
>>>>>>>> IOW, What's needed is a simple way to run the operation in a
>>>>>>>> thread, and sleep for completion while having the coroutine
>>>>>>>> yield.
>>>>>>>>
>>>>>>>> I think this could likely be achieved with QIOTask with an
>>>>>>>> alternate impl of the qio_task_wait_thread() method that is
>>>>>>>> friendly to coroutines instead of being based on pthread
>>>>>>>> condition variable waits.
>>>>>>>
>>>>>>> The most simple thing is just run qio_channel_socket_connect_sync in
>>>>>>> a thread with help of thread_pool_submit_co() which is coroutine-friendly.
>>>>>>> And this don't need any changes in io/channel.
>>>>>>>
>>>>>>> Actually, I've started with such design, but decided that better use
>>>>>>> non-blocking connect to not deal with cancelling the connecting thread
>>>>>>> on shutdown.
>>>>>>>
>>>>>>> I think, I'll resend based on thread_pool_submit_co().
>>>>>>>
>>>>>>> ===
>>>>>>>
>>>>>>> Hmm, there is async getaddrinfo_a function.. What do you think of it?
>>>>>>
>>>>>> It isn't portable, glibc only.
>>>>>>
>>>>>>> But seems simpler to use a thread than move to async interfaces everywhere.
>>>>>>
>>>>>>
>>>>>
>>>>> Hmm.. Still, on shutdown, how to cancel this connect and getaddrinfo ? I'm not sure
>>>>> how much time may getaddrinfo take, but connect can take about a minute. It's not really
>>>>> good to wait for it on shutdown.
>>>>
>>>> The intention was that if you don't want to carry on waiting for the
>>>> async operation to complete you just give and pretend it no longer
>>>> exists. Eventually it will fail or complete and the thread will exit.
>>>> The only important thing there is making sure that the callback you
>>>> are passing to the _async() method can cope with the cleanup when the
>>>> work eventually completes, even if you've given up.
>>>>
>>>
>>> At least it's not possible with thread_pool_submit_co as I wanted, because underlying
>>> thread pool waits for all its threads to complete on exit.
>>>
>>>
>>
>>
>> I'm trying to use qio_channel_socket_connect_async().. But callback
>> is not called.
>>
>> How to make it be executed? In tests/test-io-channel-socket.c it's
>> done by g_main_loop_new .. g_main_loop_run. But I need to yield.
>> socket_start_outgoing_migration uses qio_channel_socket_connect_async
>> as well, but is not doing any magic with g_main_loop. But it works. How?
> 
> The _async() impls uses  qio_task_run_in_thread to spawn the background
> thread. When the thread finishes, it uses g_idle_add to invoke the
> callback so that it runs in the context of the main thread, not the
> background thread. So something needs to be running the main loop
> in QEMU.

I came to same idea. But still, I don't see where g_main_loop is run inside qemu_main_loop(). Only iothread_run() does it. But what if we don't have iothreads?

OK, it's obvious that when qemu parses -drive options it doesn't run any main-loop. But on reconnect attempts it should
run qemu_main_loop, but callback still is not called for me..

I understand, that I can create custom wrapper on sync connect, to run it in a thread. But I just try to understand:

  - how qemu_main_loop and all these g_main_loop* things are connected?
  - if they are not, how qio_channel_socket_connect_async() works in socket_start_outgoing_migration(), where nobody run the g_main_loop_run() ?

> 
> If that isn't viable for the scenario you're in, then you can create
> your own async version by spawning the thread yourself and running
> the _sync() function - the _async() function is ajust a convenient
> wrapper combining QIOTask + _sync(), but doesn't always do what
> everyone might need.   In the char-socket.c for example, I do the
> async bit manually instead of using teh _async() wrapper.
> 
> Regards,
> Daniel
> 


-- 
Best regards,
Vladimir

Re: [PATCH 3/4] io/channel-socket: implement non-blocking connect

[Daniel P. Berrangé]

On Wed, Jul 22, 2020 at 06:40:10PM +0300, Vladimir Sementsov-Ogievskiy wrote:
> 22.07.2020 18:21, Daniel P. Berrangé wrote:
> > On Wed, Jul 22, 2020 at 06:04:53PM +0300, Vladimir Sementsov-Ogievskiy wrote:
> > > 22.07.2020 16:47, Vladimir Sementsov-Ogievskiy wrote:
> > > > 22.07.2020 15:53, Daniel P. Berrangé wrote:
> > > > > On Wed, Jul 22, 2020 at 03:43:54PM +0300, Vladimir Sementsov-Ogievskiy wrote:
> > > > > > 22.07.2020 14:21, Daniel P. Berrangé wrote:
> > > > > > > On Wed, Jul 22, 2020 at 02:00:25PM +0300, Vladimir Sementsov-Ogievskiy wrote:
> > > > > > > > 20.07.2020 21:29, Daniel P. Berrangé wrote:
> > > > > > > > > On Mon, Jul 20, 2020 at 09:07:14PM +0300, Vladimir Sementsov-Ogievskiy wrote:
> > > > > > > > > > Utilize new socket API to make a non-blocking connect for inet sockets.
> > > > > > > > > > 
> > > > > > > > > > Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
> > > > > > > > > > ---
> > > > > > > > > >      include/io/channel-socket.h | 14 +++++++
> > > > > > > > > >      io/channel-socket.c         | 74 +++++++++++++++++++++++++++++++++++++
> > > > > > > > > >      2 files changed, 88 insertions(+)
> > > > > > > > > > 
> > > > > > > > > > diff --git a/include/io/channel-socket.h b/include/io/channel-socket.h
> > > > > > > > > > index 777ff5954e..82e868bc02 100644
> > > > > > > > > > --- a/include/io/channel-socket.h
> > > > > > > > > > +++ b/include/io/channel-socket.h
> > > > > > > > > > @@ -94,6 +94,20 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
> > > > > > > > > >                                          SocketAddress *addr,
> > > > > > > > > >                                          Error **errp);
> > > > > > > > > > +/**
> > > > > > > > > > + * qio_channel_socket_connect_non_blocking_sync:
> > > > > > > > > > + * @ioc: the socket channel object
> > > > > > > > > > + * @addr: the address to connect to
> > > > > > > > > > + * @errp: pointer to a NULL-initialized error object
> > > > > > > > > > + *
> > > > > > > > > > + * Attempt to connect to the address @addr using non-blocking mode of
> > > > > > > > > > + * the socket. Function is synchronous, but being called from
> > > > > > > > > > + * coroutine context will yield during connect operation.
> > > > > > > > > > + */
> > > > > > > > > > +int qio_channel_socket_connect_non_blocking_sync(QIOChannelSocket *ioc,
> > > > > > > > > > +                                                 SocketAddress *addr,
> > > > > > > > > > +                                                 Error **errp);
> > > > > > > > > > +
> > > > > > > > > >      /**
> > > > > > > > > >       * qio_channel_socket_connect_async:
> > > > > > > > > >       * @ioc: the socket channel object
> > > > > > > > > > diff --git a/io/channel-socket.c b/io/channel-socket.c
> > > > > > > > > > index e1b4667087..076de7578a 100644
> > > > > > > > > > --- a/io/channel-socket.c
> > > > > > > > > > +++ b/io/channel-socket.c
> > > > > > > > > > @@ -22,6 +22,7 @@
> > > > > > > > > >      #include "qapi/error.h"
> > > > > > > > > >      #include "qapi/qapi-visit-sockets.h"
> > > > > > > > > >      #include "qemu/module.h"
> > > > > > > > > > +#include "qemu/sockets.h"
> > > > > > > > > >      #include "io/channel-socket.h"
> > > > > > > > > >      #include "io/channel-watch.h"
> > > > > > > > > >      #include "trace.h"
> > > > > > > > > > @@ -29,6 +30,8 @@
> > > > > > > > > >      #define SOCKET_MAX_FDS 16
> > > > > > > > > > +static int qio_channel_socket_close(QIOChannel *ioc, Error **errp);
> > > > > > > > > > +
> > > > > > > > > >      SocketAddress *
> > > > > > > > > >      qio_channel_socket_get_local_address(QIOChannelSocket *ioc,
> > > > > > > > > >                                           Error **errp)
> > > > > > > > > > @@ -157,6 +160,77 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
> > > > > > > > > >          return 0;
> > > > > > > > > >      }
> > > > > > > > > > +static int qio_channel_inet_connect_non_blocking_sync(QIOChannelSocket *ioc,
> > > > > > > > > > +        InetSocketAddress *addr, Error **errp)
> > > > > > > > > > +{
> > > > > > > > > > +    Error *local_err = NULL;
> > > > > > > > > > +    struct addrinfo *infos, *info;
> > > > > > > > > > +    int sock = -1;
> > > > > > > > > > +
> > > > > > > > > > +    infos = inet_parse_connect_saddr(addr, errp);
> > > > > > > > > > +    if (!infos) {
> > > > > > > > > > +        return -1;
> > > > > > > > > > +    }
> > > > > > > > > 
> > > > > > > > > This call is blocking since it calls getaddrinfo whose design
> > > > > > > > > offers no ability todo non-blocking DNS lookups. Given this
> > > > > > > > > call, ...
> > > > > > > > 
> > > > > > > > Oh, that's bad, thanks for taking a look on that early stage!
> > > > > > > > 
> > > > > > > > > 
> > > > > > > > > > +
> > > > > > > > > > +    for (info = infos; info != NULL; info = info->ai_next) {
> > > > > > > > > > +        bool in_progress;
> > > > > > > > > > +
> > > > > > > > > > +        error_free(local_err);
> > > > > > > > > > +        local_err = NULL;
> > > > > > > > > > +
> > > > > > > > > > +        sock = inet_connect_addr(addr, info, false, &in_progress, &local_err);
> > > > > > > > > > +        if (sock < 0) {
> > > > > > > > > > +            continue;
> > > > > > > > > > +        }
> > > > > > > > > > +
> > > > > > > > > > +        if (qio_channel_socket_set_fd(ioc, sock, &local_err) < 0) {
> > > > > > > > > > +            close(sock);
> > > > > > > > > > +            continue;
> > > > > > > > > > +        }
> > > > > > > > > > +
> > > > > > > > > > +        if (in_progress) {
> > > > > > > > > > +            if (qemu_in_coroutine()) {
> > > > > > > > > > +                qio_channel_yield(QIO_CHANNEL(ioc), G_IO_OUT);
> > > > > > > > > > +            } else {
> > > > > > > > > > +                qio_channel_wait(QIO_CHANNEL(ioc), G_IO_OUT);
> > > > > > > > > > +            }
> > > > > > > > > 
> > > > > > > > > ...this is offering false assurances of being non-blocking.
> > > > > > > > > 
> > > > > > > > > If we don't want the current thread to be blocked then we
> > > > > > > > > need to be using the existing qio_channel_socket_connect_async
> > > > > > > > > method or similar. It uses a throw away background thread to
> > > > > > > > > run the connection attempt, and then reports completion back
> > > > > > > > > later, thus avoiding the getaddrinfo design flaw for the callers.
> > > > > > > > > 
> > > > > > > > > I explicitly didn't want to add an method like the impl in this
> > > > > > > > > patch, because getaddrinfo dooms it and we already had bugs in
> > > > > > > > > the pre-QIOChannel code where QEMU thought it was non-blocking
> > > > > > > > > but wasn't due to getaddrinfo lookups.
> > > > > > > > > 
> > > > > > > > > 
> > > > > > > > > IIUC, the main appeal of this method is that the non-blocking
> > > > > > > > > nature is hidden from the caller who can continue to treat it
> > > > > > > > > as a synchronous call and have the coroutine magic happen in
> > > > > > > > > behind the scenes.
> > > > > > > > > 
> > > > > > > > > IOW, What's needed is a simple way to run the operation in a
> > > > > > > > > thread, and sleep for completion while having the coroutine
> > > > > > > > > yield.
> > > > > > > > > 
> > > > > > > > > I think this could likely be achieved with QIOTask with an
> > > > > > > > > alternate impl of the qio_task_wait_thread() method that is
> > > > > > > > > friendly to coroutines instead of being based on pthread
> > > > > > > > > condition variable waits.
> > > > > > > > 
> > > > > > > > The most simple thing is just run qio_channel_socket_connect_sync in
> > > > > > > > a thread with help of thread_pool_submit_co() which is coroutine-friendly.
> > > > > > > > And this don't need any changes in io/channel.
> > > > > > > > 
> > > > > > > > Actually, I've started with such design, but decided that better use
> > > > > > > > non-blocking connect to not deal with cancelling the connecting thread
> > > > > > > > on shutdown.
> > > > > > > > 
> > > > > > > > I think, I'll resend based on thread_pool_submit_co().
> > > > > > > > 
> > > > > > > > ===
> > > > > > > > 
> > > > > > > > Hmm, there is async getaddrinfo_a function.. What do you think of it?
> > > > > > > 
> > > > > > > It isn't portable, glibc only.
> > > > > > > 
> > > > > > > > But seems simpler to use a thread than move to async interfaces everywhere.
> > > > > > > 
> > > > > > > 
> > > > > > 
> > > > > > Hmm.. Still, on shutdown, how to cancel this connect and getaddrinfo ? I'm not sure
> > > > > > how much time may getaddrinfo take, but connect can take about a minute. It's not really
> > > > > > good to wait for it on shutdown.
> > > > > 
> > > > > The intention was that if you don't want to carry on waiting for the
> > > > > async operation to complete you just give and pretend it no longer
> > > > > exists. Eventually it will fail or complete and the thread will exit.
> > > > > The only important thing there is making sure that the callback you
> > > > > are passing to the _async() method can cope with the cleanup when the
> > > > > work eventually completes, even if you've given up.
> > > > > 
> > > > 
> > > > At least it's not possible with thread_pool_submit_co as I wanted, because underlying
> > > > thread pool waits for all its threads to complete on exit.
> > > > 
> > > > 
> > > 
> > > 
> > > I'm trying to use qio_channel_socket_connect_async().. But callback
> > > is not called.
> > > 
> > > How to make it be executed? In tests/test-io-channel-socket.c it's
> > > done by g_main_loop_new .. g_main_loop_run. But I need to yield.
> > > socket_start_outgoing_migration uses qio_channel_socket_connect_async
> > > as well, but is not doing any magic with g_main_loop. But it works. How?
> > 
> > The _async() impls uses  qio_task_run_in_thread to spawn the background
> > thread. When the thread finishes, it uses g_idle_add to invoke the
> > callback so that it runs in the context of the main thread, not the
> > background thread. So something needs to be running the main loop
> > in QEMU.
> 
> I came to same idea. But still, I don't see where g_main_loop is
> run inside qemu_main_loop(). Only iothread_run() does it. But
> what if we don't have iothreads?

There's no requirement to use g_main_loop, what matters is actually
that something runs the default GMainContext.  qemu_main_loop
satisfies this.

If you have a different GMainLoop that you want to use, then you
can pass its GMainContext into the _async() functions, and the
result will get dispatched from whatever thread runs that
GMainContext/GMainLoop. So you could use this to get the callback
to be invoked in your iothread context if that's desirable. If
a NULL GMainContext is passed to _async(), then the callback is
dispatched from qemu_main_loop() thread.

Regards,
Daniel
-- 
|: https://berrange.com      -o-    https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org         -o-            https://fstop138.berrange.com :|
|: https://entangle-photo.org    -o-    https://www.instagram.com/dberrange :|

Re: [PATCH 3/4] io/channel-socket: implement non-blocking connect

[Vladimir Sementsov-Ogievskiy]

22.07.2020 18:43, Daniel P. Berrangé wrote:
> On Wed, Jul 22, 2020 at 06:40:10PM +0300, Vladimir Sementsov-Ogievskiy wrote:
>> 22.07.2020 18:21, Daniel P. Berrangé wrote:
>>> On Wed, Jul 22, 2020 at 06:04:53PM +0300, Vladimir Sementsov-Ogievskiy wrote:
>>>> 22.07.2020 16:47, Vladimir Sementsov-Ogievskiy wrote:
>>>>> 22.07.2020 15:53, Daniel P. Berrangé wrote:
>>>>>> On Wed, Jul 22, 2020 at 03:43:54PM +0300, Vladimir Sementsov-Ogievskiy wrote:
>>>>>>> 22.07.2020 14:21, Daniel P. Berrangé wrote:
>>>>>>>> On Wed, Jul 22, 2020 at 02:00:25PM +0300, Vladimir Sementsov-Ogievskiy wrote:
>>>>>>>>> 20.07.2020 21:29, Daniel P. Berrangé wrote:
>>>>>>>>>> On Mon, Jul 20, 2020 at 09:07:14PM +0300, Vladimir Sementsov-Ogievskiy wrote:
>>>>>>>>>>> Utilize new socket API to make a non-blocking connect for inet sockets.
>>>>>>>>>>>
>>>>>>>>>>> Signed-off-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com>
>>>>>>>>>>> ---
>>>>>>>>>>>       include/io/channel-socket.h | 14 +++++++
>>>>>>>>>>>       io/channel-socket.c         | 74 +++++++++++++++++++++++++++++++++++++
>>>>>>>>>>>       2 files changed, 88 insertions(+)
>>>>>>>>>>>
>>>>>>>>>>> diff --git a/include/io/channel-socket.h b/include/io/channel-socket.h
>>>>>>>>>>> index 777ff5954e..82e868bc02 100644
>>>>>>>>>>> --- a/include/io/channel-socket.h
>>>>>>>>>>> +++ b/include/io/channel-socket.h
>>>>>>>>>>> @@ -94,6 +94,20 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
>>>>>>>>>>>                                           SocketAddress *addr,
>>>>>>>>>>>                                           Error **errp);
>>>>>>>>>>> +/**
>>>>>>>>>>> + * qio_channel_socket_connect_non_blocking_sync:
>>>>>>>>>>> + * @ioc: the socket channel object
>>>>>>>>>>> + * @addr: the address to connect to
>>>>>>>>>>> + * @errp: pointer to a NULL-initialized error object
>>>>>>>>>>> + *
>>>>>>>>>>> + * Attempt to connect to the address @addr using non-blocking mode of
>>>>>>>>>>> + * the socket. Function is synchronous, but being called from
>>>>>>>>>>> + * coroutine context will yield during connect operation.
>>>>>>>>>>> + */
>>>>>>>>>>> +int qio_channel_socket_connect_non_blocking_sync(QIOChannelSocket *ioc,
>>>>>>>>>>> +                                                 SocketAddress *addr,
>>>>>>>>>>> +                                                 Error **errp);
>>>>>>>>>>> +
>>>>>>>>>>>       /**
>>>>>>>>>>>        * qio_channel_socket_connect_async:
>>>>>>>>>>>        * @ioc: the socket channel object
>>>>>>>>>>> diff --git a/io/channel-socket.c b/io/channel-socket.c
>>>>>>>>>>> index e1b4667087..076de7578a 100644
>>>>>>>>>>> --- a/io/channel-socket.c
>>>>>>>>>>> +++ b/io/channel-socket.c
>>>>>>>>>>> @@ -22,6 +22,7 @@
>>>>>>>>>>>       #include "qapi/error.h"
>>>>>>>>>>>       #include "qapi/qapi-visit-sockets.h"
>>>>>>>>>>>       #include "qemu/module.h"
>>>>>>>>>>> +#include "qemu/sockets.h"
>>>>>>>>>>>       #include "io/channel-socket.h"
>>>>>>>>>>>       #include "io/channel-watch.h"
>>>>>>>>>>>       #include "trace.h"
>>>>>>>>>>> @@ -29,6 +30,8 @@
>>>>>>>>>>>       #define SOCKET_MAX_FDS 16
>>>>>>>>>>> +static int qio_channel_socket_close(QIOChannel *ioc, Error **errp);
>>>>>>>>>>> +
>>>>>>>>>>>       SocketAddress *
>>>>>>>>>>>       qio_channel_socket_get_local_address(QIOChannelSocket *ioc,
>>>>>>>>>>>                                            Error **errp)
>>>>>>>>>>> @@ -157,6 +160,77 @@ int qio_channel_socket_connect_sync(QIOChannelSocket *ioc,
>>>>>>>>>>>           return 0;
>>>>>>>>>>>       }
>>>>>>>>>>> +static int qio_channel_inet_connect_non_blocking_sync(QIOChannelSocket *ioc,
>>>>>>>>>>> +        InetSocketAddress *addr, Error **errp)
>>>>>>>>>>> +{
>>>>>>>>>>> +    Error *local_err = NULL;
>>>>>>>>>>> +    struct addrinfo *infos, *info;
>>>>>>>>>>> +    int sock = -1;
>>>>>>>>>>> +
>>>>>>>>>>> +    infos = inet_parse_connect_saddr(addr, errp);
>>>>>>>>>>> +    if (!infos) {
>>>>>>>>>>> +        return -1;
>>>>>>>>>>> +    }
>>>>>>>>>>
>>>>>>>>>> This call is blocking since it calls getaddrinfo whose design
>>>>>>>>>> offers no ability todo non-blocking DNS lookups. Given this
>>>>>>>>>> call, ...
>>>>>>>>>
>>>>>>>>> Oh, that's bad, thanks for taking a look on that early stage!
>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>>> +
>>>>>>>>>>> +    for (info = infos; info != NULL; info = info->ai_next) {
>>>>>>>>>>> +        bool in_progress;
>>>>>>>>>>> +
>>>>>>>>>>> +        error_free(local_err);
>>>>>>>>>>> +        local_err = NULL;
>>>>>>>>>>> +
>>>>>>>>>>> +        sock = inet_connect_addr(addr, info, false, &in_progress, &local_err);
>>>>>>>>>>> +        if (sock < 0) {
>>>>>>>>>>> +            continue;
>>>>>>>>>>> +        }
>>>>>>>>>>> +
>>>>>>>>>>> +        if (qio_channel_socket_set_fd(ioc, sock, &local_err) < 0) {
>>>>>>>>>>> +            close(sock);
>>>>>>>>>>> +            continue;
>>>>>>>>>>> +        }
>>>>>>>>>>> +
>>>>>>>>>>> +        if (in_progress) {
>>>>>>>>>>> +            if (qemu_in_coroutine()) {
>>>>>>>>>>> +                qio_channel_yield(QIO_CHANNEL(ioc), G_IO_OUT);
>>>>>>>>>>> +            } else {
>>>>>>>>>>> +                qio_channel_wait(QIO_CHANNEL(ioc), G_IO_OUT);
>>>>>>>>>>> +            }
>>>>>>>>>>
>>>>>>>>>> ...this is offering false assurances of being non-blocking.
>>>>>>>>>>
>>>>>>>>>> If we don't want the current thread to be blocked then we
>>>>>>>>>> need to be using the existing qio_channel_socket_connect_async
>>>>>>>>>> method or similar. It uses a throw away background thread to
>>>>>>>>>> run the connection attempt, and then reports completion back
>>>>>>>>>> later, thus avoiding the getaddrinfo design flaw for the callers.
>>>>>>>>>>
>>>>>>>>>> I explicitly didn't want to add an method like the impl in this
>>>>>>>>>> patch, because getaddrinfo dooms it and we already had bugs in
>>>>>>>>>> the pre-QIOChannel code where QEMU thought it was non-blocking
>>>>>>>>>> but wasn't due to getaddrinfo lookups.
>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>> IIUC, the main appeal of this method is that the non-blocking
>>>>>>>>>> nature is hidden from the caller who can continue to treat it
>>>>>>>>>> as a synchronous call and have the coroutine magic happen in
>>>>>>>>>> behind the scenes.
>>>>>>>>>>
>>>>>>>>>> IOW, What's needed is a simple way to run the operation in a
>>>>>>>>>> thread, and sleep for completion while having the coroutine
>>>>>>>>>> yield.
>>>>>>>>>>
>>>>>>>>>> I think this could likely be achieved with QIOTask with an
>>>>>>>>>> alternate impl of the qio_task_wait_thread() method that is
>>>>>>>>>> friendly to coroutines instead of being based on pthread
>>>>>>>>>> condition variable waits.
>>>>>>>>>
>>>>>>>>> The most simple thing is just run qio_channel_socket_connect_sync in
>>>>>>>>> a thread with help of thread_pool_submit_co() which is coroutine-friendly.
>>>>>>>>> And this don't need any changes in io/channel.
>>>>>>>>>
>>>>>>>>> Actually, I've started with such design, but decided that better use
>>>>>>>>> non-blocking connect to not deal with cancelling the connecting thread
>>>>>>>>> on shutdown.
>>>>>>>>>
>>>>>>>>> I think, I'll resend based on thread_pool_submit_co().
>>>>>>>>>
>>>>>>>>> ===
>>>>>>>>>
>>>>>>>>> Hmm, there is async getaddrinfo_a function.. What do you think of it?
>>>>>>>>
>>>>>>>> It isn't portable, glibc only.
>>>>>>>>
>>>>>>>>> But seems simpler to use a thread than move to async interfaces everywhere.
>>>>>>>>
>>>>>>>>
>>>>>>>
>>>>>>> Hmm.. Still, on shutdown, how to cancel this connect and getaddrinfo ? I'm not sure
>>>>>>> how much time may getaddrinfo take, but connect can take about a minute. It's not really
>>>>>>> good to wait for it on shutdown.
>>>>>>
>>>>>> The intention was that if you don't want to carry on waiting for the
>>>>>> async operation to complete you just give and pretend it no longer
>>>>>> exists. Eventually it will fail or complete and the thread will exit.
>>>>>> The only important thing there is making sure that the callback you
>>>>>> are passing to the _async() method can cope with the cleanup when the
>>>>>> work eventually completes, even if you've given up.
>>>>>>
>>>>>
>>>>> At least it's not possible with thread_pool_submit_co as I wanted, because underlying
>>>>> thread pool waits for all its threads to complete on exit.
>>>>>
>>>>>
>>>>
>>>>
>>>> I'm trying to use qio_channel_socket_connect_async().. But callback
>>>> is not called.
>>>>
>>>> How to make it be executed? In tests/test-io-channel-socket.c it's
>>>> done by g_main_loop_new .. g_main_loop_run. But I need to yield.
>>>> socket_start_outgoing_migration uses qio_channel_socket_connect_async
>>>> as well, but is not doing any magic with g_main_loop. But it works. How?
>>>
>>> The _async() impls uses  qio_task_run_in_thread to spawn the background
>>> thread. When the thread finishes, it uses g_idle_add to invoke the
>>> callback so that it runs in the context of the main thread, not the
>>> background thread. So something needs to be running the main loop
>>> in QEMU.
>>
>> I came to same idea. But still, I don't see where g_main_loop is
>> run inside qemu_main_loop(). Only iothread_run() does it. But
>> what if we don't have iothreads?
> 
> There's no requirement to use g_main_loop, what matters is actually
> that something runs the default GMainContext. 

O, I see now, g_main_context_dispatch() in glib_pollfds_poll() does it, yes?

>qemu_main_loop
> satisfies this.
> 
> If you have a different GMainLoop that you want to use, then you
> can pass its GMainContext into the _async() functions, and the
> result will get dispatched from whatever thread runs that
> GMainContext/GMainLoop. So you could use this to get the callback
> to be invoked in your iothread context if that's desirable. If
> a NULL GMainContext is passed to _async(), then the callback is
> dispatched from qemu_main_loop() thread.
> 

Thanks for your help!

-- 
Best regards,
Vladimir

[PATCH 4/4] block/nbd: use non-blocking connect: fix vm hang on connect()

[Vladimir Sementsov-Ogievskiy]

This make nbd connection_co to yield during reconnects, so that
reconnect doesn't hang up the main thread. This is very important in
case of unavailable nbd server host: connect() call may take a long
time, blocking the main thread (and due to reconnect, it will hang
again and again with small gaps of working time during pauses between
connection attempts).

How to reproduce the bug:

1. Create an image on node1:
   qemu-img create -f qcow2 xx 100M

2. Start NBD server on node1:
   qemu-nbd xx

3. Start vm with second nbd disk on node2, like this:

  ./x86_64-softmmu/qemu-system-x86_64 -nodefaults -drive \
    file=/work/images/cent7.qcow2 -drive file=nbd+tcp://192.168.100.2 \
    -vnc :0 -qmp stdio -m 2G -enable-kvm -vga std

4. Access the vm through vnc (or some other way?), and check that NBD
   drive works:

   dd if=/dev/sdb of=/dev/null bs=1M count=10

   - the command should succeed.

5. Now, let's trigger nbd-reconnect loop in Qemu process. For this:

5.1 Kill NBD server on node1

5.2 run "dd if=/dev/sdb of=/dev/null bs=1M count=10" in the guest
    again. The command should fail and a lot of error messages about
    failing disk may appear as well.

    Now NBD client driver in Qemu tries to reconnect.
    Still, VM works well.

6. Make node1 unavailable on NBD port, so connect() from node2 will
   last for a long time:

   On node1 (Note, that 10809 is just a default NBD port):

   sudo iptables -A INPUT -p tcp --dport 10809 -j DROP

   After some time the guest hangs, and you may check in gdb that Qemu
   hangs in connect() call, issued from the main thread. This is the
   BUG.

7. Don't forget to drop iptables rule from your node1:

   sudo iptables -D INPUT -p tcp --dport 10809 -j DROP

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

diff --git a/block/nbd.c b/block/nbd.c
index d9cde30457..931eadbe6f 100644
--- a/block/nbd.c
+++ b/block/nbd.c
@@ -1421,16 +1421,19 @@ static void nbd_client_close(BlockDriverState *bs)
     nbd_teardown_connection(bs);
 }
 
-static QIOChannelSocket *nbd_establish_connection(SocketAddress *saddr,
+static QIOChannelSocket *nbd_establish_connection(BlockDriverState *bs,
+                                                  SocketAddress *saddr,
                                                   Error **errp)
 {
     ERRP_GUARD();
     QIOChannelSocket *sioc;
+    AioContext *aio_context = bdrv_get_aio_context(bs);
 
     sioc = qio_channel_socket_new();
     qio_channel_set_name(QIO_CHANNEL(sioc), "nbd-client");
 
-    qio_channel_socket_connect_sync(sioc, saddr, errp);
+    qio_channel_attach_aio_context(QIO_CHANNEL(sioc), aio_context);
+    qio_channel_socket_connect_non_blocking_sync(sioc, saddr, errp);
     if (*errp) {
         object_unref(OBJECT(sioc));
         return NULL;
@@ -1451,7 +1454,7 @@ static int nbd_client_connect(BlockDriverState *bs, Error **errp)
      * establish TCP connection, return error if it fails
      * TODO: Configurable retry-until-timeout behaviour.
      */
-    QIOChannelSocket *sioc = nbd_establish_connection(s->saddr, errp);
+    QIOChannelSocket *sioc = nbd_establish_connection(bs, s->saddr, errp);
 
     if (!sioc) {
         return -ECONNREFUSED;
@@ -1461,8 +1464,6 @@ static int nbd_client_connect(BlockDriverState *bs, Error **errp)
 
     /* NBD handshake */
     trace_nbd_client_connect(s->export);
-    qio_channel_set_blocking(QIO_CHANNEL(sioc), false, NULL);
-    qio_channel_attach_aio_context(QIO_CHANNEL(sioc), aio_context);
 
     s->info.request_sizes = true;
     s->info.structured_reply = true;
-- 
2.21.0

Re: [PATCH for-5.1? 0/4] non-blocking connect

[Eric Blake]

On 7/20/20 1:07 PM, Vladimir Sementsov-Ogievskiy wrote:
> Hi! This fixes real problem (see 04). On the other hand it may be too
> much for 5.1, and it's not a degradation. So, up to you.

Given the concerns raised on 3, I think I'll wait for v2 of the series, 
and defer it to 5.2.

> 
> It's based on "[PATCH for-5.1? 0/3] Fix nbd reconnect dead-locks", or
> in other words
> Based-on: <20200720090024.18186-1-vsementsov@virtuozzo.com>
> 
> Vladimir Sementsov-Ogievskiy (4):
>    qemu-sockets: refactor inet_connect_addr
>    qemu-sockets: implement non-blocking connect interface
>    io/channel-socket: implement non-blocking connect
>    block/nbd: use non-blocking connect: fix vm hang on connect()
> 
>   include/io/channel-socket.h | 14 +++++++
>   include/qemu/sockets.h      |  6 +++
>   block/nbd.c                 | 11 +++---
>   io/channel-socket.c         | 74 ++++++++++++++++++++++++++++++++++++
>   util/qemu-sockets.c         | 76 ++++++++++++++++++++++++++-----------
>   5 files changed, 153 insertions(+), 28 deletions(-)
> 

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