session server UPDATE non-blocking accept

Allow accepting new sessions from multiple threads. Made server sockets
non-blocking.

Author: Roytak

src/server_config.c

@@ -407,7 +407,6 @@ nc_server_config_free(struct nc_server_config *config)
                 }
             }
             free(endpt->binds[j].address);
-            pthread_mutex_destroy(&endpt->bind_lock);
         }
         LY_ARRAY_FREE(endpt->binds);
 
@@ -3325,7 +3324,6 @@ config_endpoint(const struct lyd_node *node, enum nc_operation parent_op,
     struct nc_endpt *endpt = NULL;
     const char *name;
     LY_ARRAY_COUNT_TYPE i = 0;
-    int r;
 
     NC_NODE_GET_OP(node, parent_op, &op);
 
@@ -3346,12 +3344,6 @@ config_endpoint(const struct lyd_node *node, enum nc_operation parent_op,
     } else if (op == NC_OP_CREATE) {
         /* create a new endpoint */
         LY_ARRAY_NEW_RET(LYD_CTX(node), config->endpts, endpt, 1);
-
-        /* init the new endpoint */
-        if ((r = pthread_mutex_init(&endpt->bind_lock, NULL))) {
-            ERR(NULL, "Mutex init failed (%s).", strerror(r));
-            return 1;
-        }
     }
 
     /* config name */
@@ -3376,7 +3368,6 @@ config_endpoint(const struct lyd_node *node, enum nc_operation parent_op,
 
     /* all children processed, we can now delete the endpoint */
     if (op == NC_OP_DELETE) {
-        pthread_mutex_destroy(&endpt->bind_lock);
         if (i < LY_ARRAY_COUNT(config->endpts) - 1) {
             config->endpts[i] = config->endpts[LY_ARRAY_COUNT(config->endpts) - 1];
         }
@@ -6007,7 +5998,6 @@ nc_server_config_dup(const struct nc_server_config *src, struct nc_server_config
             dst_endpt->binds[j].sock = -1;
             LY_ARRAY_INCREMENT(dst_endpt->binds);
         }
-        pthread_mutex_init(&dst_endpt->bind_lock, NULL);
 
         dst_endpt->ka = src_endpt->ka;
 

src/session_client.c

@@ -1846,7 +1846,6 @@ nc_client_ch_add_bind_listen(const char *address, uint16_t port, const char *hos
     client_opts.ch_binds[client_opts.ch_bind_count - 1].address = strdup(address);
     client_opts.ch_binds[client_opts.ch_bind_count - 1].port = port;
     client_opts.ch_binds[client_opts.ch_bind_count - 1].sock = sock;
-    client_opts.ch_binds[client_opts.ch_bind_count - 1].pollin = 0;
 
     return 0;
 }
@@ -1909,7 +1908,7 @@ nc_accept_callhome(int timeout, struct ly_ctx *ctx, struct nc_session **session)
 {
     int ret, sock;
     char *host = NULL;
-    uint16_t port, idx;
+    uint16_t port, bind_idx = 0;
 
     NC_CHECK_ARG_RET(NULL, session, -1);
 
@@ -1918,8 +1917,8 @@ nc_accept_callhome(int timeout, struct ly_ctx *ctx, struct nc_session **session)
         return -1;
     }
 
-    ret = nc_sock_accept_binds(NULL, client_opts.ch_binds, client_opts.ch_bind_count, &client_opts.ch_bind_lock, timeout,
-            &host, &port, &idx, &sock);
+    ret = nc_server_ch_accept_binds(client_opts.ch_binds, client_opts.ch_bind_count, timeout,
+            &host, &port, &bind_idx, &sock);
     if (ret < 1) {
         free(host);
         return ret;
@@ -1932,11 +1931,11 @@ nc_accept_callhome(int timeout, struct ly_ctx *ctx, struct nc_session **session)
         return -1;
     }
 
-    if (client_opts.ch_binds_aux[idx].ti == NC_TI_SSH) {
+    if (client_opts.ch_binds_aux[bind_idx].ti == NC_TI_SSH) {
         *session = nc_accept_callhome_ssh_sock(sock, host, port, ctx, NC_TRANSPORT_TIMEOUT);
-    } else if (client_opts.ch_binds_aux[idx].ti == NC_TI_TLS) {
+    } else if (client_opts.ch_binds_aux[bind_idx].ti == NC_TI_TLS) {
         *session = nc_accept_callhome_tls_sock(sock, host, port, ctx, NC_TRANSPORT_TIMEOUT,
-                client_opts.ch_binds_aux[idx].hostname);
+                client_opts.ch_binds_aux[bind_idx].hostname);
     } else {
         close(sock);
         *session = NULL;
@@ -1994,13 +1993,6 @@ nc_session_ntf_thread_running(const struct nc_session *session)
 API int
 nc_client_init(void)
 {
-    int r;
-
-    if ((r = pthread_mutex_init(&client_opts.ch_bind_lock, NULL))) {
-        ERR(NULL, "%s: failed to init bind lock(%s).", __func__, strerror(r));
-        return -1;
-    }
-
 #ifdef NC_ENABLED_SSH_TLS
     if (nc_tls_backend_init_wrap()) {
         ERR(NULL, "%s: failed to init the SSL library backend.", __func__);
@@ -2018,7 +2010,6 @@ nc_client_init(void)
 API void
 nc_client_destroy(void)
 {
-    pthread_mutex_destroy(&client_opts.ch_bind_lock);
     nc_client_set_schema_searchpath(NULL);
     nc_client_unix_set_username(NULL);
 #ifdef NC_ENABLED_SSH_TLS

src/session_p.h

@@ -499,7 +499,6 @@ struct nc_bind {
     char *address;  /**< Either IPv4/IPv6 address or path to UNIX socket. */
     uint16_t port;  /**< Either port number or 0 for UNIX socket. */
     int sock;       /**< Socket file descriptor, -1 if not created yet. */
-    int pollin;     /**< Specifies, which sockets to poll on. */
 };
 
 struct nc_client_unix_opts {
@@ -583,7 +582,6 @@ struct nc_client_opts {
     uint32_t capabilities_count;    /**< Count of capabilities. */
 
     struct nc_bind *ch_binds;
-    pthread_mutex_t ch_bind_lock;   /**< To avoid concurrent calls of poll and accept on the bound sockets **/
 
     struct {
         NC_TRANSPORT_IMPL ti;
@@ -702,9 +700,7 @@ struct nc_server_config {
     struct nc_endpt {
         char *name;                 /**< Identifier of the endpoint. */
 
-        /* ACCESS locked - bind_lock */
         struct nc_bind *binds;      /**< Listening binds of the endpoint (sized-array, see libyang docs). */
-        pthread_mutex_t bind_lock;  /**< To avoid concurrent calls of poll and accept on the bound sockets. **/
 
         struct nc_keepalives ka;    /**< TCP keepalives configuration. */
 
@@ -1302,23 +1298,19 @@ int nc_sock_accept(int sock, int timeout, char **peer_host, uint16_t *peer_port)
 int nc_sock_listen_inet(const char *address, uint16_t port);
 
 /**
- * @brief Accept a new connection on a listening socket.
+ * @brief Accept a new connection on any of the given Call Home binds.
  *
- * @param[in] endpt Optional endpoint the binds belong to (only for logging purposes).
- * @param[in] binds Structure with the listening sockets.
+ * @param[in] binds Call Home binds to accept on.
  * @param[in] bind_count Number of @p binds.
- * @param[in] bind_lock Lock for avoiding concurrent poll/accept on a single bind.
  * @param[in] timeout Timeout for accepting.
  * @param[out] host Host of the remote peer. Can be NULL.
  * @param[out] port Port of the new connection. Can be NULL.
- * @param[out] idx Index of the bind that was accepted. Can be NULL.
+ * @param[out] bind_idx Index of the bind that was accepted. Can be NULL.
  * @param[out] sock Accepted socket, if any.
- * @return -1 on error.
- * @return 0 on timeout.
- * @return 1 if a socket was accepted.
+ * @return -1 on error, 0 on timeout, 1 if a socket was accepted.
  */
-int nc_sock_accept_binds(struct nc_endpt *endpt, struct nc_bind *binds, uint16_t bind_count,
-        pthread_mutex_t *bind_lock, int timeout, char **host, uint16_t *port, uint16_t *idx, int *sock);
+int nc_server_ch_accept_binds(struct nc_bind *binds, uint16_t bind_count, int timeout, char **host,
+        uint16_t *port, uint16_t *bind_idx, int *sock);
 
 /**
  * @brief Establish a UNIX transport session.