make a pull request for issue 1188

Author: frimpler

src/session_client.c

@@ -898,65 +898,107 @@ nc_connect_inout(int fdin, int fdout, struct ly_ctx *ctx)
     return NULL;
 }
 
+/* A given timeout value limits the time how long the function blocks. If it has to block
+   only for some seconds, a socket connection might not yet have been fully established.
+   Therefore the active (pending) socket will be stored in *sock_pending, but the return
+   value will be -1. In such a case a subsequent invokation is required, by providing the
+   stored sock_pending, again.
+   In general, if this function returns -1, when a timeout has been given, this function
+   has to be invoked, until it returns a valid socket.
+ */
 int
-nc_sock_connect(const char* host, uint16_t port)
+nc_sock_connect(const char* host, uint16_t port, int timeout, int* sock_pending)
 {
-    int i, sock = -1, flags;
+    int i, flags, ret=0;
+    int sock = sock_pending?*sock_pending:-1;
+    fd_set  wset;
     struct addrinfo hints, *res_list, *res;
     char port_s[6]; /* length of string representation of short int */
+    struct timeval  ts;
 
-    snprintf(port_s, 6, "%u", port);
+    ts.tv_sec = timeout;
+    ts.tv_usec = 0;
 
-    /* Connect to a server */
-    memset(&hints, 0, sizeof hints);
-    hints.ai_family = AF_UNSPEC;
-    hints.ai_socktype = SOCK_STREAM;
-    hints.ai_protocol = IPPROTO_TCP;
-    i = getaddrinfo(host, port_s, &hints, &res_list);
-    if (i != 0) {
-        ERR("Unable to translate the host address (%s).", gai_strerror(i));
-        return -1;
-    }
+    VRB("nc_sock_connect(%s, %u, %d, %d)", host, port, timeout, sock);
 
-    for (res = res_list; res != NULL; res = res->ai_next) {
-        sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
-        if (sock == -1) {
-            /* socket was not created, try another resource */
-            continue;
+    /* no pending socket */
+    if (sock == -1) {
+        /* Connect to a server */
+        snprintf(port_s, 6, "%u", port);
+        memset(&hints, 0, sizeof hints);
+        hints.ai_family = AF_UNSPEC;
+        hints.ai_socktype = SOCK_STREAM;
+        hints.ai_protocol = IPPROTO_TCP;
+        i = getaddrinfo(host, port_s, &hints, &res_list);
+        if (i != 0) {
+            ERR("Unable to translate the host address (%s).", gai_strerror(i));
+            return -1;
         }
 
-        if (connect(sock, res->ai_addr, res->ai_addrlen) == -1) {
-            /* network connection failed, try another resource */
-            close(sock);
-            sock = -1;
-            continue;
+        for (res = res_list; res != NULL; res = res->ai_next) {
+            sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
+            if (sock == -1) {
+                /* socket was not created, try another resource */
+                continue;
+            }
+            /* make the socket non-blocking */
+            if (((flags = fcntl(sock, F_GETFL)) == -1) || (fcntl(sock, F_SETFL, flags | O_NONBLOCK) == -1)) {
+                ERR("Fcntl failed (%s).", strerror(errno));
+                close(sock);
+                freeaddrinfo(res_list);
+                return -1;
+            }
+            /* enable keep-alive */
+            i = 1;
+            if (setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE, &i, sizeof i) == -1) {
+                ERR("Setsockopt failed (%s).", strerror(errno));
+                close(sock);
+                freeaddrinfo(res_list);
+                return -1;
+            }
+            /* non-blocking connect! */
+            if (connect(sock, res->ai_addr, res->ai_addrlen) < 0) {
+                if (errno != EINPROGRESS) {
+                    /* network connection failed, try another resource */
+                    VRB("connect failed: (%s).", strerror(errno));
+                    close(sock);
+                    sock = -1;
+                    continue;
+                }
+            }
         }
+        freeaddrinfo(res_list);
+    }
+    /* new socket or pending socket */
+    if (sock != -1) {
 
-        /* make the socket non-blocking */
-        if (((flags = fcntl(sock, F_GETFL)) == -1) || (fcntl(sock, F_SETFL, flags | O_NONBLOCK) == -1)) {
-            ERR("Fcntl failed (%s).", strerror(errno));
+        FD_ZERO(&wset);
+        FD_SET(sock, &wset);
+
+        if ((ret = select(sock + 1, NULL, &wset, NULL, (timeout != -1) ? &ts : NULL)) < 0) {
+            ERR("select failed: (%s).", strerror(errno));
             close(sock);
-            freeaddrinfo(res_list);
             return -1;
         }
 
-        /* enable keep-alive */
-        i = 1;
-        if (setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE, &i, sizeof i) == -1) {
-            ERR("Setsockopt failed (%s).", strerror(errno));
-            close(sock);
-            freeaddrinfo(res_list);
+        if (ret == 0) {   //we had a timeout
+            VRB("timed out after %ds (%s).", timeout, strerror(errno));
+            /* in that case we need to store it as pending for another attempt */
+            if (sock_pending) {
+                *sock_pending = sock;
+            } else {
+                close(sock);
+            }
             return -1;
         }
 
-        /* we're done, network connection established */
-        break;
-    }
-
-    if (sock != -1) {
-        VRB("Successfully connected to %s:%s over %s.", host, port_s, (res->ai_family == AF_INET6) ? "IPv6" : "IPv4");
+        if (!FD_ISSET(sock, &wset)) {
+            ERR("FD_ISSET failed: (%s).", strerror(errno));
+            close(sock);
+            return -1;
+        }
+        VRB("Successfully connected to %s:%s.", host, port_s);
     }
-    freeaddrinfo(res_list);
 
     return sock;
 }

src/session_client_ssh.c

@@ -1514,7 +1514,7 @@ _nc_connect_libssh(ssh_session ssh_session, struct ly_ctx *ctx, struct nc_client
         ssh_options_set(session->ti.libssh.session, SSH_OPTIONS_HOST, host);
 
         /* create and connect socket */
-        sock = nc_sock_connect(host, port);
+        sock = nc_sock_connect(host, port, -1, NULL);
         if (sock == -1) {
             ERR("Unable to connect to %s:%u (%s).", host, port, strerror(errno));
             goto fail;
@@ -1663,7 +1663,7 @@ nc_connect_ssh(const char *host, uint16_t port, struct ly_ctx *ctx)
     }
 
     /* create and assign communication socket */
-    sock = nc_sock_connect(host, port);
+    sock = nc_sock_connect(host, port, -1, NULL);
     if (sock == -1) {
         ERR("Unable to connect to %s:%u (%s).", host, port, strerror(errno));
         goto fail;

src/session_client_tls.c

@@ -617,7 +617,7 @@ nc_connect_tls(const char *host, unsigned short port, struct ly_ctx *ctx)
     }
 
     /* create and assign socket */
-    sock = nc_sock_connect(host, port);
+    sock = nc_sock_connect(host, port, -1, NULL);
     if (sock == -1) {
         ERR("Unable to connect to %s:%u (%s).", host, port, strerror(errno));
         goto fail;

src/session_p.h

@@ -230,6 +230,7 @@ struct nc_server_opts {
             const char *name;
             const char *address;
             uint16_t port;
+            int sock_pending;
         } *ch_endpts;
         uint16_t ch_endpt_count;
         union {
@@ -516,9 +517,11 @@ NC_MSG_TYPE nc_handshake_io(struct nc_session *session);
  *
  * @param[in] host Hostname to connect to.
  * @param[in] port Port to connect on.
+ * @param[in] timeout for blocking the connect+select call (-1 for infinite).
+ * @param[in] sock_pending for exchanging the pending socket, if the blocking timeout was != -1
  * @return Connected socket or -1 on error.
  */
-int nc_sock_connect(const char *host, uint16_t port);
+int nc_sock_connect(const char *host, uint16_t port, int timeout, int* sock_pending);
 
 /**
  * @brief Accept a new socket connection.

src/session_server.c

@@ -467,6 +467,7 @@ API int
 nc_server_init(struct ly_ctx *ctx)
 {
     const struct lys_node *rpc;
+    pthread_rwlockattr_t  attr;
 
     if (!ctx) {
         ERRARG("ctx");
@@ -492,6 +493,23 @@ nc_server_init(struct ly_ctx *ctx)
     server_opts.new_session_id = 1;
     pthread_spin_init(&server_opts.sid_lock, PTHREAD_PROCESS_PRIVATE);
 
+    errno=0;
+
+    if (pthread_rwlockattr_init(&attr) == 0) {
+        if (pthread_rwlockattr_setkind_np(&attr, PTHREAD_RWLOCK_PREFER_WRITER_NONRECURSIVE_NP) == 0) {
+            if (pthread_rwlock_init(&server_opts.endpt_lock, &attr) != 0) {
+                ERR("%s: failed to init rwlock(%s).", __FUNCTION__, strerror(errno));
+            }
+            if (pthread_rwlock_init(&server_opts.ch_client_lock, &attr) != 0) {
+                ERR("%s: failed to init rwlock(%s).", __FUNCTION__, strerror(errno));
+            }
+        } else {
+            ERR("%s: failed set attribute (%s).", __FUNCTION__, strerror(errno));
+        }
+        pthread_rwlockattr_destroy(&attr);
+    } else {
+        ERR("%s: failed init attribute (%s).", __FUNCTION__, strerror(errno));
+    }
     return 0;
 }
 
@@ -2239,6 +2257,7 @@ nc_server_ch_client_add_endpt(const char *client_name, const char *endpt_name)
     client->ch_endpts[client->ch_endpt_count - 1].name = lydict_insert(server_opts.ctx, endpt_name, 0);
     client->ch_endpts[client->ch_endpt_count - 1].address = NULL;
     client->ch_endpts[client->ch_endpt_count - 1].port = 0;
+    client->ch_endpts[client->ch_endpt_count - 1].sock_pending = -1;
 
     /* UNLOCK */
     nc_server_ch_client_unlock(client);
@@ -2269,6 +2288,9 @@ nc_server_ch_client_del_endpt(const char *client_name, const char *endpt_name)
         for (i = 0; i < client->ch_endpt_count; ++i) {
             lydict_remove(server_opts.ctx, client->ch_endpts[i].name);
             lydict_remove(server_opts.ctx, client->ch_endpts[i].address);
+            if (client->ch_endpts[i].sock_pending != -1) {
+                close(client->ch_endpts[i].sock_pending);
+            }
         }
         free(client->ch_endpts);
         client->ch_endpts = NULL;
@@ -2654,10 +2676,12 @@ nc_connect_ch_client_endpt(struct nc_ch_client *client, struct nc_ch_endpt *endp
     int sock, ret;
     struct timespec ts_cur;
 
-    sock = nc_sock_connect(endpt->address, endpt->port);
+    sock = nc_sock_connect(endpt->address, endpt->port, 5, &endpt->sock_pending);
     if (sock < 0) {
         return NC_MSG_ERROR;
     }
+    /* no need to store the socket as pending any longer */
+    endpt->sock_pending = -1;
 
     *session = nc_new_session(NC_SERVER, 0);
     if (!(*session)) {