1 files changed, 89 insertions, 121 deletions

diff --git a/src/core/init.c b/src/core/init.c
index c7a660b0..6672beb1 100644
--- a/src/core/init.c
+++ b/src/core/init.c
@@ -9,6 +9,8 @@
 //
 
 #include "core/nng_impl.h"
+#include "core/platform.h"
+#include "core/socket.h"
 #include "nng/nng.h"
 
 #include <stdbool.h>
@@ -18,152 +20,120 @@
 extern int  nni_tls_sys_init(void);
 extern void nni_tls_sys_fini(void);
 
-static bool nni_inited = false;
+#ifndef NNG_NUM_EXPIRE_THREADS
+#define NNG_NUM_EXPIRE_THREADS (nni_plat_ncpu())
+#endif
 
-static int
-nni_init_helper(void)
-{
-	int rv;
+#ifndef NNG_NUM_TASKQ_THREADS
+#define NNG_NUM_TASKQ_THREADS (nni_plat_ncpu() * 2)
+#endif
 
-#ifdef NNG_TEST_LIB
-	static bool cleanup = false;
-	if (!cleanup) {
-		atexit(nng_fini);
-		cleanup = true;
-	}
+#ifndef NNG_NUM_POLLER_THREADS
+#define NNG_NUM_POLLER_THREADS (nni_plat_ncpu())
 #endif
 
-	if (((rv = nni_taskq_sys_init()) != 0) ||
-	    ((rv = nni_reap_sys_init()) != 0) ||
-	    ((rv = nni_aio_sys_init()) != 0) ||
-	    ((rv = nni_tls_sys_init()) != 0)) {
-		nni_fini();
-		return (rv);
-	}
+#ifndef NNG_RESOLV_CONCURRENCY
+#define NNG_RESOLV_CONCURRENCY 4
+#endif
 
-	// following never fail
-	nni_sp_tran_sys_init();
+#ifndef NNG_MAX_TASKQ_THREADS
+#define NNG_MAX_TASKQ_THREADS 16
+#endif
 
-	nni_inited = true;
-	nng_log_notice(
-	    "NNG-INIT", "NNG library version %s initialized", nng_version());
+#ifndef NNG_MAX_EXPIRE_THREADS
+#define NNG_MAX_EXPIRE_THREADS 8
+#endif
 
-	return (0);
-}
+static nng_init_params init_params;
+
+unsigned int    init_count;
+nni_atomic_flag init_busy;
 
 int
-nni_init(void)
+nng_init(nng_init_params *params)
 {
-	int rv;
-	if ((rv = nni_plat_init(nni_init_helper)) != 0) {
-		nng_log_err("NNG-INIT",
-		    "NNG library initialization failed: %s", nng_strerror(rv));
-	}
-	return (rv);
-}
-
-// accessing the list of parameters
-typedef struct nni_init_param {
-	nni_list_node      node;
-	nng_init_parameter param;
-	uint64_t           value;
-#ifdef NNG_TEST_LIB
-	uint64_t effective;
-#endif
-} nni_init_param;
+	nng_init_params zero = { 0 };
+	int             rv;
 
-static nni_list nni_init_params =
-    NNI_LIST_INITIALIZER(nni_init_params, nni_init_param, node);
-
-void
-nni_init_set_param(nng_init_parameter p, uint64_t value)
-{
-	if (nni_inited) {
-		// this is paranoia -- if some library code started already
-		// then we cannot safely change parameters, and modifying the
-		// list is not thread safe.
-		return;
+	// cheap spin lock
+	while (nni_atomic_flag_test_and_set(&init_busy)) {
+		continue;
 	}
-	nni_init_param *item;
-	NNI_LIST_FOREACH (&nni_init_params, item) {
-		if (item->param == p) {
-			item->value = value;
-			return;
+	if (init_count > 0) {
+		if (params != NULL) {
+			nni_atomic_flag_reset(&init_busy);
+			return (NNG_EBUSY);
 		}
+		init_count++;
+		nni_atomic_flag_reset(&init_busy);
+		return (0);
 	}
-	if ((item = NNI_ALLOC_STRUCT(item)) != NULL) {
-		item->param = p;
-		item->value = value;
-		nni_list_append(&nni_init_params, item);
+	if (params == NULL) {
+		params = &zero;
 	}
-}
-
-uint64_t
-nni_init_get_param(nng_init_parameter p, uint64_t default_value)
-{
-	nni_init_param *item;
-	NNI_LIST_FOREACH (&nni_init_params, item) {
-		if (item->param == p) {
-			return (item->value);
-		}
+	init_params.num_task_threads     = params->num_task_threads
+	        ? params->num_task_threads
+	        : NNG_NUM_TASKQ_THREADS;
+	init_params.max_task_threads     = params->max_task_threads
+	        ? params->max_task_threads
+	        : NNG_MAX_TASKQ_THREADS;
+	init_params.num_expire_threads   = params->num_expire_threads
+	      ? params->num_expire_threads
+	      : NNG_NUM_EXPIRE_THREADS;
+	init_params.max_expire_threads   = params->max_expire_threads
+	      ? params->max_expire_threads
+	      : NNG_MAX_EXPIRE_THREADS;
+	init_params.num_poller_threads   = params->num_poller_threads
+	      ? params->num_poller_threads
+	      : NNG_NUM_POLLER_THREADS;
+	init_params.max_poller_threads   = params->max_poller_threads
+	      ? params->max_poller_threads
+	      : NNG_MAX_POLLER_THREADS;
+	init_params.num_resolver_threads = params->num_resolver_threads
+	    ? params->num_resolver_threads
+	    : NNG_RESOLV_CONCURRENCY;
+
+	if (((rv = nni_plat_init(&init_params)) != 0) ||
+	    ((rv = nni_taskq_sys_init(&init_params)) != 0) ||
+	    ((rv = nni_reap_sys_init()) != 0) ||
+	    ((rv = nni_aio_sys_init(&init_params)) != 0) ||
+	    ((rv = nni_tls_sys_init()) != 0)) {
+		nni_atomic_flag_reset(&init_busy);
+		nng_fini();
+		return (rv);
 	}
-	return (default_value);
-}
 
-void
-nni_init_set_effective(nng_init_parameter p, uint64_t value)
-{
-#ifdef NNG_TEST_LIB
-	nni_init_param *item;
-	NNI_LIST_FOREACH (&nni_init_params, item) {
-		if (item->param == p) {
-			item->effective = value;
-			return;
-		}
-	}
-	if ((item = NNI_ALLOC_STRUCT(item)) != NULL) {
-		item->param     = p;
-		item->effective = value;
-		nni_list_append(&nni_init_params, item);
-	}
-#else
-	NNI_ARG_UNUSED(p);
-	NNI_ARG_UNUSED(value);
-#endif
+	// following never fails
+	nni_sp_tran_sys_init();
+
+	nng_log_notice(
+	    "NNG-INIT", "NNG library version %s initialized", nng_version());
+	init_count++;
+	nni_atomic_flag_reset(&init_busy);
+	return (rv);
 }
 
+// Undocumented, for test code only
 #ifdef NNG_TEST_LIB
-uint64_t
-nni_init_get_effective(nng_init_parameter p)
+nng_init_params *
+nng_init_get_params(void)
 {
-	nni_init_param *item;
-	NNI_LIST_FOREACH (&nni_init_params, item) {
-		if (item->param == p) {
-			return (item->effective);
-		}
-	}
-	return ((uint64_t) -1);
+	return &init_params;
 }
 #endif
 
-static void
-nni_init_params_fini(void)
-{
-	nni_init_param *item;
-	while ((item = nni_list_first(&nni_init_params)) != NULL) {
-		nni_list_remove(&nni_init_params, item);
-		NNI_FREE_STRUCT(item);
-	}
-}
-
 void
-nni_fini(void)
+nng_fini(void)
 {
-	if (!nni_inited) {
-		// make sure we discard parameters even if we didn't startup
-		nni_init_params_fini();
+	while (nni_atomic_flag_test_and_set(&init_busy)) {
+		continue;
+	}
+	init_count--;
+	if (init_count > 0) {
+		nni_atomic_flag_reset(&init_busy);
 		return;
 	}
+	nni_sock_closeall();
 	nni_sp_tran_sys_fini();
 	nni_tls_sys_fini();
 	nni_reap_drain();
@@ -171,8 +141,6 @@ nni_fini(void)
 	nni_taskq_sys_fini();
 	nni_reap_sys_fini(); // must be before timer and aio (expire)
 	nni_id_map_sys_fini();
-	nni_init_params_fini();
-
 	nni_plat_fini();
-	nni_inited = false;
+	nni_atomic_flag_reset(&init_busy);
 }