Patchwork [04/18] qemu-sockets: add nonblocking connect for Unix sockets

login
register
mail settings
Submitter Paolo Bonzini
Date Oct. 3, 2012, 2:36 p.m.
Message ID <1349275025-5093-5-git-send-email-pbonzini@redhat.com>
Download mbox | patch
Permalink /patch/188806/
State New
Headers show

Comments

Paolo Bonzini - Oct. 3, 2012, 2:36 p.m.
Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 qemu-char.c    |  2 +-
 qemu-sockets.c | 78 +++++++++++++++++++++++++++++++++++++++++++++++++---------
 qemu_socket.h  |  6 ++++-
 3 file modificati, 72 inserzioni(+), 14 rimozioni(-)
Luiz Capitulino - Oct. 4, 2012, 5:38 p.m.
On Wed,  3 Oct 2012 16:36:51 +0200
Paolo Bonzini <pbonzini@redhat.com> wrote:

This looks like a bug fix, because if is_waitconnect==false unix_connect_opts()
shouldn't block. Am I right?

Somme comments below.

> Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
> ---
>  qemu-char.c    |  2 +-
>  qemu-sockets.c | 78 +++++++++++++++++++++++++++++++++++++++++++++++++---------
>  qemu_socket.h  |  6 ++++-
>  3 file modificati, 72 inserzioni(+), 14 rimozioni(-)
> 
> diff --git a/qemu-char.c b/qemu-char.c
> index 3cc6cb5..8ebd582 100644
> --- a/qemu-char.c
> +++ b/qemu-char.c
> @@ -2450,7 +2450,7 @@ static CharDriverState *qemu_chr_open_socket(QemuOpts *opts)
>          if (is_listen) {
>              fd = unix_listen_opts(opts, NULL);
>          } else {
> -            fd = unix_connect_opts(opts, NULL);
> +            fd = unix_connect_opts(opts, NULL, NULL, NULL);
>          }
>      } else {
>          if (is_listen) {
> diff --git a/qemu-sockets.c b/qemu-sockets.c
> index 79c7b66..341ae21 100644
> --- a/qemu-sockets.c
> +++ b/qemu-sockets.c
> @@ -252,16 +252,19 @@ static void wait_for_connect(void *opaque)
>      }
>  
>      /* try to connect to the next address on the list */
> -    while (s->current_addr->ai_next != NULL && s->fd < 0) {
> -        s->current_addr = s->current_addr->ai_next;
> -        s->fd = inet_connect_addr(s->current_addr, &in_progress, s);
> -        /* connect in progress */
> -        if (in_progress) {
> -            return;
> +    if (s->current_addr) {
> +        while (s->current_addr->ai_next != NULL && s->fd < 0) {
> +            s->current_addr = s->current_addr->ai_next;
> +            s->fd = inet_connect_addr(s->current_addr, &in_progress, s);
> +            /* connect in progress */
> +            if (in_progress) {
> +                return;
> +            }
>          }
> +
> +        freeaddrinfo(s->addr_list);
>      }
>  
> -    freeaddrinfo(s->addr_list);
>      if (s->callback) {
>          s->callback(s->fd, s->opaque);
>      }
> @@ -702,32 +705,65 @@ err:
>      return -1;
>  }
>  
> -int unix_connect_opts(QemuOpts *opts, Error **errp)
> +int unix_connect_opts(QemuOpts *opts, Error **errp,
> +                      NonBlockingConnectHandler *callback, void *opaque)

I wonder if having two functions (one non-blocking and another blocking)
would be better, as we have some "if (non-blocking) do; else" kind of tests.

>  {
>      struct sockaddr_un un;
>      const char *path = qemu_opt_get(opts, "path");
> -    int sock;
> +    ConnectState *connect_state = NULL;
> +    int sock, rc;
>  
>      if (NULL == path) {
>          fprintf(stderr, "unix connect: no path specified\n");
>          return -1;
>      }
>  
> +    if (callback != NULL) {
> +        connect_state = g_malloc0(sizeof(*connect_state));
> +        connect_state->callback = callback;
> +        connect_state->opaque = opaque;
> +    }
> +
>      sock = qemu_socket(PF_UNIX, SOCK_STREAM, 0);
>      if (sock < 0) {
>          perror("socket(unix)");
>          return -1;
>      }

Leaks connect_state on error, you could delay its allocation.

> +    if (connect_state != NULL) {
> +        socket_set_nonblock(sock);
> +    }
>  
>      memset(&un, 0, sizeof(un));
>      un.sun_family = AF_UNIX;
>      snprintf(un.sun_path, sizeof(un.sun_path), "%s", path);
> -    if (connect(sock, (struct sockaddr*) &un, sizeof(un)) < 0) {
> +
> +    /* connect to peer */
> +    do {
> +        rc = 0;
> +        if (connect(sock, (struct sockaddr *) &un, sizeof(un)) < 0) {
> +            rc = -socket_error();
> +        }
> +    } while (rc == -EINTR);

Bug fix, woud be nice to comment in the changelog.

> +
> +    if (connect_state != NULL && QEMU_SOCKET_RC_INPROGRESS(rc)) {
> +        connect_state->fd = sock;
> +        qemu_set_fd_handler2(sock, NULL, NULL, wait_for_connect,
> +                             connect_state);
> +        return sock;
> +    } else {
> +        /* non blocking socket immediate success, call callback */
> +        if (callback != NULL) {
> +            callback(sock, opaque);
> +        }
> +    }

Wouldn't this be better written like:

if (connect_state) {
	if (rc == 0) {
		/* immediate success */
		callack(sock, opaque);
	} else if (QEMU_SOCKET_RC_INPROGRESS(rc)) {
		connect_state->fd = sock;
		qemu_set_fd_handler2(...);
		return sock;
	}
}

> +
> +    if (rc < 0) {
>          fprintf(stderr, "connect(unix:%s): %s\n", path, strerror(errno));
>          close(sock);
>  	return -1;
>      }

leaks connect_state.

>  
> +    g_free(connect_state);
>      return sock;
>  }
>  
> @@ -740,7 +776,8 @@ int unix_listen_opts(QemuOpts *opts, Error **errp)
>      return -1;
>  }
>  
> -int unix_connect_opts(QemuOpts *opts, Error **errp)
> +int unix_connect_opts(QemuOpts *opts, Error **errp,
> +                      NonBlockingConnectHandler *callback, void *opaque)
>  {
>      error_setg(errp, "unix sockets are not available on windows");
>      errno = ENOTSUP;
> @@ -785,7 +822,24 @@ int unix_connect(const char *path, Error **errp)
>  
>      opts = qemu_opts_create(&dummy_opts, NULL, 0, NULL);
>      qemu_opt_set(opts, "path", path);
> -    sock = unix_connect_opts(opts, errp);
> +    sock = unix_connect_opts(opts, errp, NULL, NULL);
> +    qemu_opts_del(opts);
> +    return sock;
> +}
> +
> +
> +int unix_nonblocking_connect(const char *path,
> +                             NonBlockingConnectHandler *callback,
> +                             void *opaque, Error **errp)
> +{
> +    QemuOpts *opts;
> +    int sock = -1;
> +
> +    g_assert(callback != NULL);
> +
> +    opts = qemu_opts_create(&dummy_opts, NULL, 0, NULL);
> +    qemu_opt_set(opts, "path", path);
> +    sock = unix_connect_opts(opts, errp, callback, opaque);
>      qemu_opts_del(opts);
>      return sock;
>  }
> diff --git a/qemu_socket.h b/qemu_socket.h
> index ff979b5..89a5feb 100644
> --- a/qemu_socket.h
> +++ b/qemu_socket.h
> @@ -58,8 +58,12 @@ const char *inet_strfamily(int family);
>  
>  int unix_listen_opts(QemuOpts *opts, Error **errp);
>  int unix_listen(const char *path, char *ostr, int olen, Error **errp);
> -int unix_connect_opts(QemuOpts *opts, Error **errp);
> +int unix_connect_opts(QemuOpts *opts, Error **errp,
> +                      NonBlockingConnectHandler *callback, void *opaque);
>  int unix_connect(const char *path, Error **errp);
> +int unix_nonblocking_connect(const char *str,
> +                             NonBlockingConnectHandler *callback,
> +                             void *opaque, Error **errp);
>  
>  /* Old, ipv4 only bits.  Don't use for new code. */
>  int parse_host_port(struct sockaddr_in *saddr, const char *str);
Paolo Bonzini - Oct. 5, 2012, 8:57 a.m.
Il 04/10/2012 19:38, Luiz Capitulino ha scritto:
> On Wed,  3 Oct 2012 16:36:51 +0200
> Paolo Bonzini <pbonzini@redhat.com> wrote:
> 
> This looks like a bug fix, because if is_waitconnect==false unix_connect_opts()
> shouldn't block. Am I right?

It could be a start, but it doesn't fix it yet.  qemu-char always calls
inet_connect_opts() and unix_connect_opts() in blocking mode.

I wonder if Orit's patches have broken wait=off, need to check.

Paolo

Patch

diff --git a/qemu-char.c b/qemu-char.c
index 3cc6cb5..8ebd582 100644
--- a/qemu-char.c
+++ b/qemu-char.c
@@ -2450,7 +2450,7 @@  static CharDriverState *qemu_chr_open_socket(QemuOpts *opts)
         if (is_listen) {
             fd = unix_listen_opts(opts, NULL);
         } else {
-            fd = unix_connect_opts(opts, NULL);
+            fd = unix_connect_opts(opts, NULL, NULL, NULL);
         }
     } else {
         if (is_listen) {
diff --git a/qemu-sockets.c b/qemu-sockets.c
index 79c7b66..341ae21 100644
--- a/qemu-sockets.c
+++ b/qemu-sockets.c
@@ -252,16 +252,19 @@  static void wait_for_connect(void *opaque)
     }
 
     /* try to connect to the next address on the list */
-    while (s->current_addr->ai_next != NULL && s->fd < 0) {
-        s->current_addr = s->current_addr->ai_next;
-        s->fd = inet_connect_addr(s->current_addr, &in_progress, s);
-        /* connect in progress */
-        if (in_progress) {
-            return;
+    if (s->current_addr) {
+        while (s->current_addr->ai_next != NULL && s->fd < 0) {
+            s->current_addr = s->current_addr->ai_next;
+            s->fd = inet_connect_addr(s->current_addr, &in_progress, s);
+            /* connect in progress */
+            if (in_progress) {
+                return;
+            }
         }
+
+        freeaddrinfo(s->addr_list);
     }
 
-    freeaddrinfo(s->addr_list);
     if (s->callback) {
         s->callback(s->fd, s->opaque);
     }
@@ -702,32 +705,65 @@  err:
     return -1;
 }
 
-int unix_connect_opts(QemuOpts *opts, Error **errp)
+int unix_connect_opts(QemuOpts *opts, Error **errp,
+                      NonBlockingConnectHandler *callback, void *opaque)
 {
     struct sockaddr_un un;
     const char *path = qemu_opt_get(opts, "path");
-    int sock;
+    ConnectState *connect_state = NULL;
+    int sock, rc;
 
     if (NULL == path) {
         fprintf(stderr, "unix connect: no path specified\n");
         return -1;
     }
 
+    if (callback != NULL) {
+        connect_state = g_malloc0(sizeof(*connect_state));
+        connect_state->callback = callback;
+        connect_state->opaque = opaque;
+    }
+
     sock = qemu_socket(PF_UNIX, SOCK_STREAM, 0);
     if (sock < 0) {
         perror("socket(unix)");
         return -1;
     }
+    if (connect_state != NULL) {
+        socket_set_nonblock(sock);
+    }
 
     memset(&un, 0, sizeof(un));
     un.sun_family = AF_UNIX;
     snprintf(un.sun_path, sizeof(un.sun_path), "%s", path);
-    if (connect(sock, (struct sockaddr*) &un, sizeof(un)) < 0) {
+
+    /* connect to peer */
+    do {
+        rc = 0;
+        if (connect(sock, (struct sockaddr *) &un, sizeof(un)) < 0) {
+            rc = -socket_error();
+        }
+    } while (rc == -EINTR);
+
+    if (connect_state != NULL && QEMU_SOCKET_RC_INPROGRESS(rc)) {
+        connect_state->fd = sock;
+        qemu_set_fd_handler2(sock, NULL, NULL, wait_for_connect,
+                             connect_state);
+        return sock;
+    } else {
+        /* non blocking socket immediate success, call callback */
+        if (callback != NULL) {
+            callback(sock, opaque);
+        }
+    }
+
+    if (rc < 0) {
         fprintf(stderr, "connect(unix:%s): %s\n", path, strerror(errno));
         close(sock);
 	return -1;
     }
 
+    g_free(connect_state);
     return sock;
 }
 
@@ -740,7 +776,8 @@  int unix_listen_opts(QemuOpts *opts, Error **errp)
     return -1;
 }
 
-int unix_connect_opts(QemuOpts *opts, Error **errp)
+int unix_connect_opts(QemuOpts *opts, Error **errp,
+                      NonBlockingConnectHandler *callback, void *opaque)
 {
     error_setg(errp, "unix sockets are not available on windows");
     errno = ENOTSUP;
@@ -785,7 +822,24 @@  int unix_connect(const char *path, Error **errp)
 
     opts = qemu_opts_create(&dummy_opts, NULL, 0, NULL);
     qemu_opt_set(opts, "path", path);
-    sock = unix_connect_opts(opts, errp);
+    sock = unix_connect_opts(opts, errp, NULL, NULL);
+    qemu_opts_del(opts);
+    return sock;
+}
+
+
+int unix_nonblocking_connect(const char *path,
+                             NonBlockingConnectHandler *callback,
+                             void *opaque, Error **errp)
+{
+    QemuOpts *opts;
+    int sock = -1;
+
+    g_assert(callback != NULL);
+
+    opts = qemu_opts_create(&dummy_opts, NULL, 0, NULL);
+    qemu_opt_set(opts, "path", path);
+    sock = unix_connect_opts(opts, errp, callback, opaque);
     qemu_opts_del(opts);
     return sock;
 }
diff --git a/qemu_socket.h b/qemu_socket.h
index ff979b5..89a5feb 100644
--- a/qemu_socket.h
+++ b/qemu_socket.h
@@ -58,8 +58,12 @@  const char *inet_strfamily(int family);
 
 int unix_listen_opts(QemuOpts *opts, Error **errp);
 int unix_listen(const char *path, char *ostr, int olen, Error **errp);
-int unix_connect_opts(QemuOpts *opts, Error **errp);
+int unix_connect_opts(QemuOpts *opts, Error **errp,
+                      NonBlockingConnectHandler *callback, void *opaque);
 int unix_connect(const char *path, Error **errp);
+int unix_nonblocking_connect(const char *str,
+                             NonBlockingConnectHandler *callback,
+                             void *opaque, Error **errp);
 
 /* Old, ipv4 only bits.  Don't use for new code. */
 int parse_host_port(struct sockaddr_in *saddr, const char *str);