Merge pull request #64 from frimpler/persistent_timeout_fix_pr

PR for Issue 1188 (persistent timeout in call home scenarios)

Author: michalvasko

src/session_client.c

@@ -898,65 +898,139 @@ nc_connect_inout(int fdin, int fdout, struct ly_ctx *ctx)
     return NULL;
 }
 
-int
-nc_sock_connect(const char* host, uint16_t port)
+/*
+   Helper for a non-blocking connect (which is required because of the locking
+   concept for e.g. call home settings). For more details see nc_sock_connect().
+ */
+static int
+_non_blocking_connect(int timeout, int* sock_pending, struct addrinfo *res)
 {
-    int i, sock = -1, flags;
-    struct addrinfo hints, *res_list, *res;
-    char port_s[6]; /* length of string representation of short int */
-
-    snprintf(port_s, 6, "%u", port);
-
-    /* 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;
-    }
-
-    for (res = res_list; res != NULL; res = res->ai_next) {
+    int flags, ret=0;
+    int sock = -1;
+    fd_set  wset;
+    struct timeval ts;
+    int error = 0;
+    socklen_t len = sizeof(int);
+
+    if (sock_pending && *sock_pending != -1) {
+        VRB("Trying to connect the pending socket=%d.", *sock_pending );
+        sock = *sock_pending;
+    } else {
+        assert(res);
+        VRB("Trying to connect via %s.", (res->ai_family == AF_INET6) ? "IPv6" : "IPv4");
+        /* Connect to a server */
         sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
         if (sock == -1) {
-            /* socket was not created, try another resource */
-            continue;
+            ERR("socket couldn't be created.", strerror(errno));
+            return -1;
         }
-
-        if (connect(sock, res->ai_addr, res->ai_addrlen) == -1) {
-            /* network connection failed, try another resource */
-            close(sock);
-            sock = -1;
-            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));
-            goto error;
+            goto cleanup;
+        }
+        /* non-blocking connect! */
+        if (connect(sock, res->ai_addr, res->ai_addrlen) < 0) {
+            if (errno != EINPROGRESS) {
+                /* network connection failed, try another resource */
+                ERR("connect failed: (%s).", strerror(errno));
+                goto cleanup;
+            }
+        }
+        /* check the usability of the socket */
+        if (getsockopt(sock, SOL_SOCKET, SO_ERROR, &error, &len) < 0) {
+            ERR("getsockopt failed: (%s).", strerror(errno));
+            goto cleanup;
+        }
+        if (error == ECONNREFUSED) {
+            /* network connection failed, try another resource */
+            VRB("getsockopt error: (%s).", strerror(error));
+            goto cleanup;
         }
-
-        /* we're done, network connection established */
-        break;
     }
+    ts.tv_sec = timeout;
+    ts.tv_usec = 0;
 
-    if (sock != -1) {
-        VRB("Successfully connected to %s:%s over %s.", host, port_s, (res->ai_family == AF_INET6) ? "IPv6" : "IPv4");
+    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));
+        goto cleanup;
     }
-    freeaddrinfo(res_list);
 
+    if (ret == 0) {   //we had a timeout
+        VRB("timed out after %ds (%s).", timeout, strerror(errno));
+        if (sock_pending) {
+            /* no sock-close, we'll try it again */
+            *sock_pending = sock;
+        } else {
+            close(sock);
+        }
+        return -1;
+    }
     return sock;
 
-error:
-    if (sock > -1) {
-        close(sock);
+cleanup:
+    if (sock_pending) {
+        *sock_pending = -1;
     }
-    freeaddrinfo(res_list);
+    close(sock);
     return -1;
 }
 
+/* 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, int timeout, int* sock_pending)
+{
+    int i;
+    int sock = sock_pending?*sock_pending:-1;
+    struct addrinfo hints, *res_list, *res;
+    char port_s[6]; /* length of string representation of short int */
+
+    VRB("nc_sock_connect(%s, %u, %d, %d)", host, port, timeout, sock);
+
+    /* 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;
+        }
+
+        for (res = res_list; res != NULL; res = res->ai_next) {
+            sock = _non_blocking_connect(timeout, sock_pending, res);
+            if (sock == -1 && (!sock_pending || *sock_pending == -1)) {
+                /* try the next resource */
+                continue;
+            }
+            VRB("Successfully connected to %s:%s over %s.", host, port_s, (res->ai_family == AF_INET6) ? "IPv6" : "IPv4");
+            break;
+        }
+        freeaddrinfo(res_list);
+
+    } else {
+        /* try to get a connection with the pending socket */
+        assert(sock_pending);
+        sock = _non_blocking_connect(timeout, sock_pending, NULL);
+    }
+
+    return sock;
+}
+
 static NC_MSG_TYPE
 get_msg(struct nc_session *session, int timeout, uint64_t msgid, struct lyxml_elem **msg)
 {

src/session_client_ssh.c

@@ -1528,7 +1528,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;
@@ -1677,7 +1677,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

@@ -238,6 +238,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 {
@@ -524,9 +525,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

@@ -491,6 +491,7 @@ API int
 nc_server_init(struct ly_ctx *ctx)
 {
     const struct lys_node *rpc;
+    pthread_rwlockattr_t  attr;
 
     if (!ctx) {
         ERRARG("ctx");
@@ -516,6 +517,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;
 }
 
@@ -2263,6 +2281,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);
@@ -2293,6 +2312,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;
@@ -2678,10 +2700,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)) {