diff options
Diffstat (limited to 'src')
35 files changed, 308 insertions, 549 deletions
diff --git a/src/core/aio.c b/src/core/aio.c index 1c7e41d1..e1e9c42f 100644 --- a/src/core/aio.c +++ b/src/core/aio.c @@ -823,24 +823,13 @@ nni_aio_sys_fini(void) } int -nni_aio_sys_init(void) +nni_aio_sys_init(nng_init_params *params) { - int num_thr; - int max_thr; + int16_t num_thr; + int16_t max_thr; -#ifndef NNG_MAX_EXPIRE_THREADS -#define NNG_MAX_EXPIRE_THREADS 8 -#endif - -#ifndef NNG_NUM_EXPIRE_THREADS -#define NNG_NUM_EXPIRE_THREADS (nni_plat_ncpu()) -#endif - - max_thr = (int) nni_init_get_param( - NNG_INIT_MAX_EXPIRE_THREADS, NNG_MAX_EXPIRE_THREADS); - - num_thr = (int) nni_init_get_param( - NNG_INIT_NUM_EXPIRE_THREADS, NNG_NUM_EXPIRE_THREADS); + max_thr = params->max_expire_threads; + num_thr = params->num_expire_threads; if ((max_thr > 0) && (num_thr > max_thr)) { num_thr = max_thr; @@ -848,7 +837,7 @@ nni_aio_sys_init(void) if (num_thr < 1) { num_thr = 1; } - nni_init_set_effective(NNG_INIT_NUM_EXPIRE_THREADS, num_thr); + params->num_expire_threads = num_thr; nni_aio_expire_q_list = nni_zalloc(sizeof(nni_aio_expire_q *) * num_thr); nni_aio_expire_q_cnt = num_thr; diff --git a/src/core/aio.h b/src/core/aio.h index cae8610f..50cb266b 100644 --- a/src/core/aio.h +++ b/src/core/aio.h @@ -192,7 +192,7 @@ extern void nni_aio_completions_run(nni_aio_completions *); extern void nni_aio_completions_add( nni_aio_completions *, nni_aio *, int, size_t); -extern int nni_aio_sys_init(void); +extern int nni_aio_sys_init(nng_init_params *); extern void nni_aio_sys_fini(void); typedef struct nni_aio_expire_q nni_aio_expire_q; diff --git a/src/core/dialer.c b/src/core/dialer.c index 0ee2d361..907da99d 100644 --- a/src/core/dialer.c +++ b/src/core/dialer.c @@ -285,13 +285,8 @@ nni_dialer_create(nni_dialer **dp, nni_sock *s, const char *url_str) int nni_dialer_find(nni_dialer **dp, uint32_t id) { - int rv; nni_dialer *d; - if ((rv = nni_init()) != 0) { - return (rv); - } - nni_mtx_lock(&dialers_lk); if ((d = nni_id_get(&dialers, id)) != NULL) { d->d_ref++; 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); } diff --git a/src/core/init.h b/src/core/init.h index d20cf046..2826870b 100644 --- a/src/core/init.h +++ b/src/core/init.h @@ -11,26 +11,9 @@ #ifndef CORE_INIT_H #define CORE_INIT_H -#include "core/nng_impl.h" - -// nni_init is called each time the user enters the library. It ensures that -// the library is initialized properly, and also deals with checks such as -// whether the process has forked since last initialization. -int nni_init(void); - -// nni_fini tears everything down. In the future it may be used to ensure -// that all resources used by the library are released back to the system. -void nni_fini(void); - -// nni_init_param is used by applications (via nng_init_param) to configure -// some tunable settings at runtime. It must be called before any other NNG -// functions are called, in order to have any effect at all. -void nni_init_set_param(nng_init_parameter, uint64_t value); +#include "nng/nng.h" // subsystems can call this to obtain a parameter value. -uint64_t nni_init_get_param(nng_init_parameter parameter, uint64_t default_value); - -// subsystems can set this to facilitate tests (only used in test code) -void nni_init_set_effective(nng_init_parameter p, uint64_t value); +nng_init_params *nni_init_get_params(void); #endif // CORE_INIT_H diff --git a/src/core/init_test.c b/src/core/init_test.c index 9b9e26b4..0ef13543 100644 --- a/src/core/init_test.c +++ b/src/core/init_test.c @@ -7,93 +7,97 @@ // found online at https://opensource.org/licenses/MIT. // +#include "nng/nng.h" #include <nuts.h> -uint64_t nni_init_get_param( - nng_init_parameter parameter, uint64_t default_value); -uint64_t nni_init_get_effective(nng_init_parameter p); -void nni_init_set_effective(nng_init_parameter p, uint64_t); +nng_init_params *nng_init_get_params(void); void test_init_param(void) { - NUTS_ASSERT(nni_init_get_param(NNG_INIT_PARAMETER_NONE, 456) == 456); - nng_init_set_parameter(NNG_INIT_PARAMETER_NONE, 123); - NUTS_ASSERT(nni_init_get_param(NNG_INIT_PARAMETER_NONE, 567) == 123); - nni_init_set_effective(NNG_INIT_PARAMETER_NONE, 124); - NUTS_ASSERT(nni_init_get_effective(NNG_INIT_PARAMETER_NONE) == 124); - NUTS_ASSERT(nni_init_get_param(NNG_INIT_PARAMETER_NONE, 567) == 123); - nng_fini(); - NUTS_ASSERT(nni_init_get_param(NNG_INIT_PARAMETER_NONE, 567) == 567); -} - -void -test_set_effective(void) -{ - nni_init_set_effective(NNG_INIT_PARAMETER_NONE, 999); - NUTS_ASSERT(nni_init_get_param(NNG_INIT_PARAMETER_NONE, 0) == 0); - NUTS_ASSERT(nni_init_get_effective(NNG_INIT_PARAMETER_NONE) == 999); - nng_fini(); + nng_init_params *p; + p = nng_init_get_params(); + NUTS_ASSERT(p != NULL); } void test_init_zero_resolvers(void) { - nng_socket s; - nng_init_set_parameter(NNG_INIT_NUM_RESOLVER_THREADS, 0); - NUTS_OPEN(s); - NUTS_CLOSE(s); - NUTS_ASSERT( - nni_init_get_effective(NNG_INIT_NUM_RESOLVER_THREADS) == 1); + nng_init_params *pp; + nng_init_params p = { 0 }; + p.num_resolver_threads = 0; + nng_fini(); + NUTS_PASS(nng_init(&p)); + pp = nng_init_get_params(); + NUTS_ASSERT(pp->num_resolver_threads > 0); nng_fini(); } void test_init_one_task_thread(void) { - nng_socket s; - nng_init_set_parameter(NNG_INIT_NUM_TASK_THREADS, 0); - nng_init_set_parameter(NNG_INIT_MAX_TASK_THREADS, 1); - NUTS_OPEN(s); - NUTS_CLOSE(s); - NUTS_ASSERT(nni_init_get_effective(NNG_INIT_NUM_TASK_THREADS) == 2); + nng_init_params *pp; + nng_init_params p = { 0 }; + nng_fini(); + p.max_task_threads = 1; + NUTS_PASS(nng_init(&p)); + pp = nng_init_get_params(); + NUTS_ASSERT(pp->max_task_threads == 1); } void test_init_too_many_task_threads(void) { - nng_socket s; - nng_init_set_parameter(NNG_INIT_NUM_TASK_THREADS, 256); - nng_init_set_parameter(NNG_INIT_MAX_TASK_THREADS, 4); + nng_socket s; + nng_init_params *pp; + nng_init_params p = { 0 }; + + p.num_task_threads = 256; + p.max_task_threads = 4; + + nng_fini(); + NUTS_PASS(nng_init(&p)); NUTS_OPEN(s); NUTS_CLOSE(s); - NUTS_ASSERT(nni_init_get_effective(NNG_INIT_NUM_TASK_THREADS) == 4); - nng_fini(); + pp = nng_init_get_params(); + NUTS_TRUE(pp->num_task_threads == 4); } void test_init_no_expire_thread(void) { - nng_socket s; - nng_init_set_parameter(NNG_INIT_NUM_EXPIRE_THREADS, 0); - nng_init_set_parameter(NNG_INIT_MAX_EXPIRE_THREADS, 0); + nng_socket s; + nng_init_params *pp; + nng_init_params p = { 0 }; + + nng_fini(); + p.num_expire_threads = 0; + p.max_expire_threads = 0; + NUTS_PASS(nng_init(&p)); NUTS_OPEN(s); NUTS_CLOSE(s); - NUTS_ASSERT(nni_init_get_effective(NNG_INIT_NUM_EXPIRE_THREADS) == 1); - nng_fini(); + pp = nng_init_get_params(); + NUTS_TRUE(pp->num_expire_threads > 0); + NUTS_MSG("Got %d expire threads", pp->num_expire_threads); } void test_init_too_many_expire_threads(void) { - nng_socket s; - nng_init_set_parameter(NNG_INIT_NUM_EXPIRE_THREADS, 256); - nng_init_set_parameter(NNG_INIT_MAX_EXPIRE_THREADS, 2); + nng_socket s; + nng_init_params *pp; + nng_init_params p = { 0 }; + + nng_fini(); + p.num_expire_threads = 256; + p.max_expire_threads = 2; + NUTS_PASS(nng_init(&p)); NUTS_OPEN(s); NUTS_CLOSE(s); - NUTS_ASSERT(nni_init_get_effective(NNG_INIT_NUM_EXPIRE_THREADS) == 2); - nng_fini(); + pp = nng_init_get_params(); + NUTS_TRUE(pp->num_expire_threads == 2); + NUTS_MSG("Got %d expire threads", pp->num_expire_threads); } // poller tuning only supported on Windows right now @@ -101,31 +105,42 @@ test_init_too_many_expire_threads(void) void test_init_poller_no_threads(void) { - nng_socket s; - nng_init_set_parameter(NNG_INIT_NUM_POLLER_THREADS, 0); - nng_init_set_parameter(NNG_INIT_MAX_POLLER_THREADS, 0); + nng_socket s; + nng_init_params *pp; + nng_init_params p = { 0 }; + + nng_fini(); + p.num_poller_threads = 0; + p.max_poller_threads = 0; + NUTS_PASS(nng_init(&p)); NUTS_OPEN(s); NUTS_CLOSE(s); - NUTS_ASSERT(nni_init_get_effective(NNG_INIT_NUM_POLLER_THREADS) == 1); - nng_fini(); + pp = nng_init_get_params(); + NUTS_TRUE(pp->num_poller_threads > 0); + NUTS_MSG("Got %d poller threads", pp->num_expire_threads); } void test_init_too_many_poller_threads(void) { - nng_socket s; - nng_init_set_parameter(NNG_INIT_NUM_POLLER_THREADS, 256); - nng_init_set_parameter(NNG_INIT_MAX_POLLER_THREADS, 2); + nng_socket s; + nng_init_params *pp; + nng_init_params p = { 0 }; + + nng_fini(); + p.num_poller_threads = 256; + p.max_poller_threads = 2; + NUTS_PASS(nng_init(&p)); NUTS_OPEN(s); NUTS_CLOSE(s); - NUTS_ASSERT(nni_init_get_effective(NNG_INIT_NUM_POLLER_THREADS) == 2); - nng_fini(); + pp = nng_init_get_params(); + NUTS_TRUE(pp->num_poller_threads == 2); + NUTS_MSG("Got %d poller threads", pp->num_expire_threads); } #endif NUTS_TESTS = { { "init parameter", test_init_param }, - { "init set effective", test_set_effective }, { "init zero resolvers", test_init_zero_resolvers }, { "init one task thread", test_init_one_task_thread }, { "init too many task threads", test_init_too_many_task_threads }, @@ -137,4 +152,4 @@ NUTS_TESTS = { #endif { NULL, NULL }, -};
\ No newline at end of file +}; diff --git a/src/core/listener.c b/src/core/listener.c index 38a7d323..88729f56 100644 --- a/src/core/listener.c +++ b/src/core/listener.c @@ -272,13 +272,8 @@ nni_listener_create(nni_listener **lp, nni_sock *s, const char *url_str) int nni_listener_find(nni_listener **lp, uint32_t id) { - int rv; nni_listener *l; - if ((rv = nni_init()) != 0) { - return (rv); - } - nni_mtx_lock(&listeners_lk); if ((l = nni_id_get(&listeners, id)) != NULL) { l->l_ref++; diff --git a/src/core/platform.h b/src/core/platform.h index d7a88238..62a321bf 100644 --- a/src/core/platform.h +++ b/src/core/platform.h @@ -47,11 +47,6 @@ // for abnormal programs on the platform, such as calling abort(). extern void nni_plat_abort(void); -// nni_plat_println is used to emit debug messages. Typically this is used -// during core debugging, or to emit panic messages. Message content will -// not contain newlines, but the output will add them. -extern void nni_plat_println(const char *); - // nni_plat_printf is like printf. It should conform to C99 standard printf, // but is a function to allow platform ports to redirect. It should go to // the same place that nni_plat_println does. @@ -263,18 +258,10 @@ extern void nni_msleep(nni_duration); uint32_t nni_random(void); // nni_plat_init is called to allow the platform the chance to -// do any necessary initialization. This routine MUST be idempotent, -// and thread-safe, and will be called before any other API calls, and -// may be called at any point thereafter. It is permitted to return -// an error if some critical failure initializing the platform occurs, -// but once this succeeds, all future calls must succeed as well, unless -// nni_plat_fini has been called. -// -// The function argument should be called if the platform has not initialized -// (i.e. exactly once), and its result passed back to the caller. If it -// does not return 0 (success), then it may be called again to try to -// initialize the platform again at a later date. -extern int nni_plat_init(int (*)(void)); +// do any necessary initialization. This will be called before any other API +// calls. It is permitted to return an error if some critical failure +// initializing the platform occurs. +extern int nni_plat_init(nng_init_params *); // nni_plat_fini is called to clean up resources. It is intended to // be called as the last thing executed in the library, and no other functions diff --git a/src/core/socket.c b/src/core/socket.c index 2b94e573..c92a1c3b 100644 --- a/src/core/socket.c +++ b/src/core/socket.c @@ -364,12 +364,9 @@ nni_sock_recvq(nni_sock *s) int nni_sock_find(nni_sock **sockp, uint32_t id) { - int rv; + int rv = 0; nni_sock *s; - if ((rv = nni_init()) != 0) { - return (rv); - } nni_mtx_lock(&sock_lk); if ((s = nni_id_get(&sock_ids, id)) != NULL) { if (s->s_closed) { @@ -621,8 +618,7 @@ nni_sock_open(nni_sock **sockp, const nni_proto *proto) return (NNG_ENOTSUP); } - if (((rv = nni_init()) != 0) || - ((rv = nni_sock_create(&s, proto)) != 0)) { + if ((rv = nni_sock_create(&s, proto)) != 0) { return (rv); } @@ -1076,12 +1072,9 @@ nni_sock_set_pipe_cb(nni_sock *s, int ev, nng_pipe_cb cb, void *arg) int nni_ctx_find(nni_ctx **cp, uint32_t id, bool closing) { - int rv; + int rv = 0; nni_ctx *ctx; - if ((rv = nni_init()) != 0) { - return (rv); - } nni_mtx_lock(&sock_lk); if ((ctx = nni_id_get(&ctx_ids, id)) != NULL) { // We refuse a reference if either the socket is diff --git a/src/core/stats.c b/src/core/stats.c index c049d611..f01f642e 100644 --- a/src/core/stats.c +++ b/src/core/stats.c @@ -385,10 +385,6 @@ int nng_stats_get(nng_stat **statp) { #ifdef NNG_ENABLE_STATS - int rv; - if ((rv = nni_init()) != 0) { - return (rv); - } return (nni_stat_snapshot(statp, &stats_root)); #else NNI_ARG_UNUSED(statp); diff --git a/src/core/stream.c b/src/core/stream.c index 78029ddc..d900329a 100644 --- a/src/core/stream.c +++ b/src/core/stream.c @@ -169,12 +169,6 @@ nng_stream_dialer_dial(nng_stream_dialer *d, nng_aio *aio) int nng_stream_dialer_alloc_url(nng_stream_dialer **dp, const nng_url *url) { - int rv; - - if ((rv = nni_init()) != 0) { - return (rv); - } - for (int i = 0; stream_drivers[i].scheme != NULL; i++) { if (strcmp(stream_drivers[i].scheme, url->u_scheme) == 0) { return (stream_drivers[i].dialer_alloc(dp, url)); @@ -189,9 +183,6 @@ nng_stream_dialer_alloc(nng_stream_dialer **dp, const char *uri) nng_url *url; int rv; - if ((rv = nni_init()) != 0) { - return (rv); - } if ((rv = nng_url_parse(&url, uri)) != 0) { return (rv); } @@ -291,12 +282,6 @@ nni_stream_listener_set_tls(nng_stream_listener *l, nng_tls_config *cfg) int nng_stream_listener_alloc_url(nng_stream_listener **lp, const nng_url *url) { - int rv; - - if ((rv = nni_init()) != 0) { - return (rv); - } - for (int i = 0; stream_drivers[i].scheme != NULL; i++) { if (strcmp(stream_drivers[i].scheme, url->u_scheme) == 0) { return (stream_drivers[i].listener_alloc(lp, url)); @@ -311,10 +296,6 @@ nng_stream_listener_alloc(nng_stream_listener **lp, const char *uri) nng_url *url; int rv; - if ((rv = nni_init()) != 0) { - return (rv); - } - if ((rv = nng_url_parse(&url, uri)) != 0) { return (rv); } diff --git a/src/core/taskq.c b/src/core/taskq.c index 09886596..496c2fab 100644 --- a/src/core/taskq.c +++ b/src/core/taskq.c @@ -9,6 +9,7 @@ // #include "core/nng_impl.h" +#include "nng/nng.h" typedef struct nni_taskq_thr nni_taskq_thr; struct nni_taskq_thr { @@ -243,24 +244,13 @@ nni_task_fini(nni_task *task) } int -nni_taskq_sys_init(void) +nni_taskq_sys_init(nng_init_params *params) { - int num_thr; - int max_thr; + int16_t num_thr; + int16_t max_thr; -#ifndef NNG_NUM_TASKQ_THREADS -#define NNG_NUM_TASKQ_THREADS (nni_plat_ncpu() * 2) -#endif - -#ifndef NNG_MAX_TASKQ_THREADS -#define NNG_MAX_TASKQ_THREADS 16 -#endif - - max_thr = (int) nni_init_get_param( - NNG_INIT_MAX_TASK_THREADS, NNG_MAX_TASKQ_THREADS); - - num_thr = (int) nni_init_get_param( - NNG_INIT_NUM_TASK_THREADS, NNG_NUM_TASKQ_THREADS); + max_thr = params->max_task_threads; + num_thr = params->num_task_threads; if ((max_thr > 0) && (num_thr > max_thr)) { num_thr = max_thr; @@ -268,9 +258,9 @@ nni_taskq_sys_init(void) if (num_thr < 2) { num_thr = 2; } - nni_init_set_effective(NNG_INIT_NUM_TASK_THREADS, num_thr); + params->num_task_threads = num_thr; - return (nni_taskq_init(&nni_taskq_systq, num_thr)); + return (nni_taskq_init(&nni_taskq_systq, (int) num_thr)); } void diff --git a/src/core/taskq.h b/src/core/taskq.h index ccc54f61..498b4f37 100644 --- a/src/core/taskq.h +++ b/src/core/taskq.h @@ -1,5 +1,5 @@ // -// Copyright 2022 Staysail Systems, Inc. <info@staysail.tech> +// Copyright 2024 Staysail Systems, Inc. <info@staysail.tech> // Copyright 2018 Capitar IT Group BV <info@capitar.com> // // This software is supplied under the terms of the MIT License, a @@ -14,6 +14,7 @@ #include "core/defs.h" #include "core/list.h" #include "core/platform.h" +#include "nng/nng.h" typedef struct nni_taskq nni_taskq; typedef struct nni_task nni_task; @@ -60,7 +61,7 @@ extern void nni_task_init(nni_task *, nni_taskq *, nni_cb, void *); // it reschedules the task.) extern void nni_task_fini(nni_task *); -extern int nni_taskq_sys_init(void); +extern int nni_taskq_sys_init(nng_init_params *); extern void nni_taskq_sys_fini(void); // nni_task implementation details are not to be used except by the diff --git a/src/core/tcp.c b/src/core/tcp.c index e54cdee7..7fb67228 100644 --- a/src/core/tcp.c +++ b/src/core/tcp.c @@ -369,9 +369,6 @@ tcp_listener_alloc_addr(nng_stream_listener **lp, const nng_sockaddr *sa) tcp_listener *l; int rv; - if ((rv = nni_init()) != 0) { - return (rv); - } if ((l = NNI_ALLOC_STRUCT(l)) == NULL) { return (NNG_ENOMEM); } @@ -398,9 +395,6 @@ nni_tcp_listener_alloc(nng_stream_listener **lp, const nng_url *url) int rv; nng_sockaddr sa; - if ((rv = nni_init()) != 0) { - return (rv); - } if ((rv = nni_url_to_address(&sa, url)) != 0) { return (rv); } @@ -25,13 +25,6 @@ #include <stdio.h> #include <string.h> -void -nng_fini(void) -{ - nni_sock_closeall(); - nni_fini(); -} - int nng_close(nng_socket s) { @@ -410,8 +403,7 @@ ctx_get(nng_ctx id, const char *n, void *v, size_t *szp, nni_type t) nni_ctx *ctx; int rv; - if (((rv = nni_init()) != 0) || - ((rv = nni_ctx_find(&ctx, id.id, false)) != 0)) { + if ((rv = nni_ctx_find(&ctx, id.id, false)) != 0) { return (rv); } rv = nni_ctx_getopt(ctx, n, v, szp, t); @@ -467,8 +459,7 @@ ctx_set(nng_ctx id, const char *n, const void *v, size_t sz, nni_type t) nni_ctx *ctx; int rv; - if (((rv = nni_init()) != 0) || - ((rv = nni_ctx_find(&ctx, id.id, false)) != 0)) { + if ((rv = nni_ctx_find(&ctx, id.id, false)) != 0) { return (rv); } rv = nni_ctx_setopt(ctx, n, v, sz, t); @@ -662,9 +653,6 @@ dialer_set(nng_dialer id, const char *n, const void *v, size_t sz, nni_type t) nni_dialer *d; int rv; - if ((rv = nni_init()) != 0) { - return (rv); - } if ((rv = nni_dialer_find(&d, id.id)) != 0) { return (rv); } @@ -728,9 +716,6 @@ dialer_get(nng_dialer id, const char *n, void *v, size_t *szp, nni_type t) nni_dialer *d; int rv; - if ((rv = nni_init()) != 0) { - return (rv); - } if ((rv = nni_dialer_find(&d, id.id)) != 0) { return (rv); } @@ -792,8 +777,7 @@ nng_dialer_get_tls(nng_dialer id, nng_tls_config **cfgp) { int rv; nni_dialer *d; - if (((rv = nni_init()) != 0) || - ((rv = nni_dialer_find(&d, id.id)) != 0)) { + if ((rv = nni_dialer_find(&d, id.id)) != 0) { return (rv); } rv = nni_dialer_get_tls(d, cfgp); @@ -806,8 +790,7 @@ nng_dialer_set_tls(nng_dialer id, nng_tls_config *cfg) { int rv; nni_dialer *d; - if (((rv = nni_init()) != 0) || - ((rv = nni_dialer_find(&d, id.id)) != 0)) { + if ((rv = nni_dialer_find(&d, id.id)) != 0) { return (rv); } rv = nni_dialer_set_tls(d, cfg); @@ -822,9 +805,6 @@ listener_set( nni_listener *l; int rv; - if ((rv = nni_init()) != 0) { - return (rv); - } if ((rv = nni_listener_find(&l, lid.id)) != 0) { return (rv); } @@ -889,9 +869,6 @@ listener_get( nni_listener *l; int rv; - if ((rv = nni_init()) != 0) { - return (rv); - } if ((rv = nni_listener_find(&l, lid.id)) != 0) { return (rv); } @@ -953,8 +930,7 @@ nng_listener_get_tls(nng_listener id, nng_tls_config **cfgp) { int rv; nni_listener *l; - if (((rv = nni_init()) != 0) || - ((rv = nni_listener_find(&l, id.id)) != 0)) { + if ((rv = nni_listener_find(&l, id.id)) != 0) { return (rv); } rv = nni_listener_get_tls(l, cfgp); @@ -967,8 +943,7 @@ nng_listener_set_tls(nng_listener id, nng_tls_config *cfg) { int rv; nni_listener *l; - if (((rv = nni_init()) != 0) || - ((rv = nni_listener_find(&l, id.id)) != 0)) { + if ((rv = nni_listener_find(&l, id.id)) != 0) { return (rv); } rv = nni_listener_set_tls(l, cfg); @@ -1008,9 +983,6 @@ socket_set( nni_sock *sock; int rv; - if ((rv = nni_init()) != 0) { - return (rv); - } if ((rv = nni_sock_find(&sock, s.id)) != 0) { return (rv); } @@ -1068,9 +1040,6 @@ socket_get(nng_socket s, const char *name, void *val, size_t *szp, nni_type t) nni_sock *sock; int rv; - if ((rv = nni_init()) != 0) { - return (rv); - } if ((rv = nni_sock_find(&sock, s.id)) != 0) { return (rv); } @@ -1127,8 +1096,7 @@ nng_socket_get_recv_poll_fd(nng_socket id, int *fdp) int rv; nni_sock *sock; - if (((rv = nni_init()) != 0) || - ((rv = nni_sock_find(&sock, id.id)) != 0)) { + if ((rv = nni_sock_find(&sock, id.id)) != 0) { return (rv); } @@ -1143,8 +1111,7 @@ nng_socket_get_send_poll_fd(nng_socket id, int *fdp) int rv; nni_sock *sock; - if (((rv = nni_init()) != 0) || - ((rv = nni_sock_find(&sock, id.id)) != 0)) { + if ((rv = nni_sock_find(&sock, id.id)) != 0) { return (rv); } @@ -1159,8 +1126,7 @@ nng_socket_proto_id(nng_socket id, uint16_t *idp) int rv; nni_sock *sock; - if (((rv = nni_init()) != 0) || - ((rv = nni_sock_find(&sock, id.id)) != 0)) { + if ((rv = nni_sock_find(&sock, id.id)) != 0) { return (rv); } @@ -1175,8 +1141,7 @@ nng_socket_peer_id(nng_socket id, uint16_t *idp) int rv; nni_sock *sock; - if (((rv = nni_init()) != 0) || - ((rv = nni_sock_find(&sock, id.id)) != 0)) { + if ((rv = nni_sock_find(&sock, id.id)) != 0) { return (rv); } @@ -1191,8 +1156,7 @@ nng_socket_proto_name(nng_socket id, const char **name) int rv; nni_sock *sock; - if (((rv = nni_init()) != 0) || - ((rv = nni_sock_find(&sock, id.id)) != 0)) { + if ((rv = nni_sock_find(&sock, id.id)) != 0) { return (rv); } @@ -1207,8 +1171,7 @@ nng_socket_peer_name(nng_socket id, const char **name) int rv; nni_sock *sock; - if (((rv = nni_init()) != 0) || - ((rv = nni_sock_find(&sock, id.id)) != 0)) { + if ((rv = nni_sock_find(&sock, id.id)) != 0) { return (rv); } @@ -1223,8 +1186,7 @@ nng_socket_raw(nng_socket id, bool *rawp) int rv; nni_sock *sock; - if (((rv = nni_init()) != 0) || - ((rv = nni_sock_find(&sock, id.id)) != 0)) { + if ((rv = nni_sock_find(&sock, id.id)) != 0) { return (rv); } *rawp = nni_sock_raw(sock); @@ -1238,9 +1200,6 @@ nng_pipe_notify(nng_socket s, nng_pipe_ev ev, nng_pipe_cb cb, void *arg) int rv; nni_sock *sock; - if ((rv = nni_init()) != 0) { - return (rv); - } if ((rv = nni_sock_find(&sock, s.id)) != 0) { return (rv); } @@ -1289,9 +1248,6 @@ nng_device(nng_socket s1, nng_socket s2) { nni_aio aio; int rv; - if ((rv = nni_init()) != 0) { - return (rv); - } nni_aio_init(&aio, NULL, NULL); nng_device_aio(&aio, s1, s2); nni_aio_wait(&aio); @@ -1375,9 +1331,6 @@ pipe_get(nng_pipe p, const char *name, void *val, size_t *szp, nni_type t) int rv; nni_pipe *pipe; - if ((rv = nni_init()) < 0) { - return (rv); - } if ((rv = nni_pipe_find(&pipe, p.id)) != 0) { return (rv); } @@ -1440,7 +1393,7 @@ nng_pipe_socket(nng_pipe p) nng_socket s = NNG_SOCKET_INITIALIZER; nni_pipe *pipe; - if ((nni_init() == 0) && (nni_pipe_find(&pipe, p.id) == 0)) { + if (nni_pipe_find(&pipe, p.id) == 0) { s.id = nni_pipe_sock_id(pipe); nni_pipe_rele(pipe); } @@ -1452,7 +1405,7 @@ nng_pipe_dialer(nng_pipe p) { nng_dialer d = NNG_DIALER_INITIALIZER; nni_pipe *pipe; - if ((nni_init() == 0) && (nni_pipe_find(&pipe, p.id) == 0)) { + if (nni_pipe_find(&pipe, p.id) == 0) { d.id = nni_pipe_dialer_id(pipe); nni_pipe_rele(pipe); } @@ -1464,7 +1417,7 @@ nng_pipe_listener(nng_pipe p) { nng_listener l = NNG_LISTENER_INITIALIZER; nni_pipe *pipe; - if ((nni_init() == 0) && (nni_pipe_find(&pipe, p.id) == 0)) { + if (nni_pipe_find(&pipe, p.id) == 0) { l.id = nni_pipe_listener_id(pipe); nni_pipe_rele(pipe); } @@ -1917,9 +1870,6 @@ nng_aio_alloc(nng_aio **app, void (*cb)(void *), void *arg) nng_aio *aio; int rv; - if ((rv = nni_init()) != 0) { - return (rv); - } if ((rv = nni_aio_alloc(&aio, (nni_cb) cb, arg)) == 0) { nng_aio_set_timeout(aio, NNG_DURATION_DEFAULT); *app = aio; @@ -2098,16 +2048,9 @@ nng_version(void) NNG_PATCH_VERSION) NNG_RELEASE_SUFFIX); } -void -nng_init_set_parameter(nng_init_parameter p, uint64_t value) -{ - nni_init_set_param(p, value); -} - nng_time nng_clock(void) { - (void) nni_init(); return (nni_clock()); } @@ -2115,7 +2058,6 @@ nng_clock(void) void nng_msleep(nng_duration dur) { - (void) nni_init(); nni_msleep(dur); } @@ -2129,8 +2071,6 @@ nng_thread_create(nng_thread **thrp, void (*func)(void *), void *arg) nni_thr *thr; int rv; - (void) nni_init(); - if ((thr = NNI_ALLOC_STRUCT(thr)) == NULL) { return (NNG_ENOMEM); } @@ -2167,8 +2107,6 @@ nng_mtx_alloc(nng_mtx **mpp) { nng_mtx *mp; - (void) nni_init(); - if ((mp = NNI_ALLOC_STRUCT(mp)) == NULL) { return (NNG_ENOMEM); } @@ -2251,7 +2189,6 @@ nng_cv_wake1(nng_cv *cv) uint32_t nng_random(void) { - (void) nni_init(); return (nni_random()); } @@ -2264,7 +2201,6 @@ nng_socket_pair(int fds[2]) int nng_udp_open(nng_udp **udp, nng_sockaddr *sa) { - (void) nni_init(); return (nni_plat_udp_open((nni_plat_udp **) udp, sa)); } diff --git a/src/platform/posix/posix_impl.h b/src/platform/posix/posix_impl.h index afbdce97..f8d8965e 100644 --- a/src/platform/posix/posix_impl.h +++ b/src/platform/posix/posix_impl.h @@ -147,9 +147,9 @@ struct nni_atomic_ptr { #endif -extern int nni_posix_pollq_sysinit(void); +extern int nni_posix_pollq_sysinit(nng_init_params *); extern void nni_posix_pollq_sysfini(void); -extern int nni_posix_resolv_sysinit(void); +extern int nni_posix_resolv_sysinit(nng_init_params *); extern void nni_posix_resolv_sysfini(void); #endif // PLATFORM_POSIX_IMPL_H diff --git a/src/platform/posix/posix_pollq_epoll.c b/src/platform/posix/posix_pollq_epoll.c index dde331d5..a5c8d8c9 100644 --- a/src/platform/posix/posix_pollq_epoll.c +++ b/src/platform/posix/posix_pollq_epoll.c @@ -1,5 +1,5 @@ // -// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech> +// Copyright 2024 Staysail Systems, Inc. <info@staysail.tech> // Copyright 2018 Capitar IT Group BV <info@capitar.com> // Copyright 2018 Liam Staskawicz <liam@stask.net> // @@ -65,7 +65,7 @@ struct nni_posix_pfd { nni_posix_pollq *pq; int fd; nni_posix_pfd_cb cb; - void * arg; + void *arg; bool closed; bool closing; bool reap; @@ -80,8 +80,8 @@ static nni_posix_pollq nni_posix_global_pollq; int nni_posix_pfd_init(nni_posix_pfd **pfdp, int fd) { - nni_posix_pfd * pfd; - nni_posix_pollq * pq; + nni_posix_pfd *pfd; + nni_posix_pollq *pq; struct epoll_event ev; int rv; @@ -173,7 +173,7 @@ nni_posix_pfd_close(nni_posix_pfd *pfd) { nni_mtx_lock(&pfd->mtx); if (!pfd->closing) { - nni_posix_pollq * pq = pfd->pq; + nni_posix_pollq *pq = pfd->pq; struct epoll_event ev; // Not actually used. pfd->closing = true; @@ -243,7 +243,7 @@ nni_posix_pollq_reap(nni_posix_pollq *pq) static void nni_posix_poll_thr(void *arg) { - nni_posix_pollq * pq = arg; + nni_posix_pollq *pq = arg; struct epoll_event events[NNI_MAX_EPOLL_EVENTS]; for (;;) { @@ -271,9 +271,9 @@ nni_posix_poll_thr(void *arg) } reap = true; } else { - nni_posix_pfd * pfd = ev->data.ptr; + nni_posix_pfd *pfd = ev->data.ptr; nni_posix_pfd_cb cb; - void * cbarg; + void *cbarg; unsigned mask; mask = ev->events & @@ -395,8 +395,9 @@ nni_posix_pollq_create(nni_posix_pollq *pq) } int -nni_posix_pollq_sysinit(void) +nni_posix_pollq_sysinit(nng_init_params *params) { + NNI_ARG_UNUSED(params); return (nni_posix_pollq_create(&nni_posix_global_pollq)); } diff --git a/src/platform/posix/posix_pollq_kqueue.c b/src/platform/posix/posix_pollq_kqueue.c index dda81be4..ae1b2f47 100644 --- a/src/platform/posix/posix_pollq_kqueue.c +++ b/src/platform/posix/posix_pollq_kqueue.c @@ -1,5 +1,5 @@ // -// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech> +// Copyright 2024 Staysail Systems, Inc. <info@staysail.tech> // Copyright 2018 Capitar IT Group BV <info@capitar.com> // Copyright 2018 Liam Staskawicz <liam@stask.net> // @@ -9,6 +9,7 @@ // found online at https://opensource.org/licenses/MIT. // +#include "core/defs.h" #ifdef NNG_HAVE_KQUEUE #include <errno.h> @@ -41,7 +42,7 @@ struct nni_posix_pfd { nni_list_node node; // linkage into the reap list nni_posix_pollq *pq; // associated pollq int fd; // file descriptor to poll - void * arg; // user data + void *arg; // user data nni_posix_pfd_cb cb; // user callback on event bool closed; unsigned events; @@ -57,7 +58,7 @@ static nni_posix_pollq nni_posix_global_pollq; int nni_posix_pfd_init(nni_posix_pfd **pfdp, int fd) { - nni_posix_pfd * pf; + nni_posix_pfd *pf; nni_posix_pollq *pq; struct kevent ev[2]; unsigned flags = EV_ADD | EV_DISABLE | EV_CLEAR; @@ -234,9 +235,9 @@ nni_posix_poll_thr(void *arg) for (;;) { int n; struct kevent evs[NNI_MAX_KQUEUE_EVENTS]; - nni_posix_pfd * pf; + nni_posix_pfd *pf; nni_posix_pfd_cb cb; - void * cbarg; + void *cbarg; unsigned revents; nni_mtx_lock(&pq->mtx); @@ -335,8 +336,9 @@ nni_posix_pollq_create(nni_posix_pollq *pq) } int -nni_posix_pollq_sysinit(void) +nni_posix_pollq_sysinit(nng_init_params *params) { + NNI_ARG_UNUSED(params); return (nni_posix_pollq_create(&nni_posix_global_pollq)); } diff --git a/src/platform/posix/posix_pollq_poll.c b/src/platform/posix/posix_pollq_poll.c index f6f81703..302f97ca 100644 --- a/src/platform/posix/posix_pollq_poll.c +++ b/src/platform/posix/posix_pollq_poll.c @@ -1,5 +1,5 @@ // -// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech> +// Copyright 2024 Staysail Systems, Inc. <info@staysail.tech> // Copyright 2018 Capitar IT Group BV <info@capitar.com> // // This software is supplied under the terms of the MIT License, a @@ -8,6 +8,7 @@ // found online at https://opensource.org/licenses/MIT. // +#include "core/defs.h" #include "core/nng_impl.h" #include "platform/posix/posix_pollq.h" @@ -63,7 +64,7 @@ struct nni_posix_pfd { nni_mtx mtx; unsigned events; nni_posix_pfd_cb cb; - void * arg; + void *arg; }; static nni_posix_pollq nni_posix_global_pollq; @@ -71,7 +72,7 @@ static nni_posix_pollq nni_posix_global_pollq; int nni_posix_pfd_init(nni_posix_pfd **pfdp, int fd) { - nni_posix_pfd * pfd; + nni_posix_pfd *pfd; nni_posix_pollq *pq = &nni_posix_global_pollq; // Set this is as soon as possible (narrow the close-exec race as @@ -186,8 +187,8 @@ nni_posix_poll_thr(void *arg) { nni_posix_pollq *pq = arg; int nalloc = 0; - struct pollfd * fds = NULL; - nni_posix_pfd ** pfds = NULL; + struct pollfd *fds = NULL; + nni_posix_pfd **pfds = NULL; for (;;) { int nfds; @@ -277,7 +278,7 @@ nni_posix_poll_thr(void *arg) for (int i = 1; i < nfds; i++) { if ((events = fds[i].revents) != 0) { nni_posix_pfd_cb cb; - void * arg; + void *arg; pfd = pfds[i]; @@ -336,8 +337,9 @@ nni_posix_pollq_create(nni_posix_pollq *pq) } int -nni_posix_pollq_sysinit(void) +nni_posix_pollq_sysinit(nng_init_params *params) { + NNI_ARG_UNUSED(params); return (nni_posix_pollq_create(&nni_posix_global_pollq)); } diff --git a/src/platform/posix/posix_pollq_port.c b/src/platform/posix/posix_pollq_port.c index 7ce96f68..23b4f074 100644 --- a/src/platform/posix/posix_pollq_port.c +++ b/src/platform/posix/posix_pollq_port.c @@ -42,7 +42,7 @@ struct nni_posix_pfd { bool closed; bool closing; nni_posix_pfd_cb cb; - void * data; + void *data; }; // single global instance for now @@ -52,7 +52,7 @@ int nni_posix_pfd_init(nni_posix_pfd **pfdp, int fd) { nni_posix_pollq *pq; - nni_posix_pfd * pfd; + nni_posix_pfd *pfd; pq = &nni_posix_global_pollq; @@ -153,10 +153,10 @@ nni_posix_poll_thr(void *arg) for (;;) { nni_posix_pollq *pq = arg; port_event_t ev[NNI_MAX_PORTEV]; - nni_posix_pfd * pfd; + nni_posix_pfd *pfd; unsigned events; nni_posix_pfd_cb cb; - void * arg; + void *arg; unsigned n; n = 1; // wake us even on just one event @@ -246,8 +246,9 @@ nni_posix_pfd_set_cb(nni_posix_pfd *pfd, nni_posix_pfd_cb cb, void *arg) } int -nni_posix_pollq_sysinit(void) +nni_posix_pollq_sysinit(nng_init_params *params) { + NNI_ARG_UNUSED(params); return (nni_posix_pollq_create(&nni_posix_global_pollq)); } diff --git a/src/platform/posix/posix_resolv_gai.c b/src/platform/posix/posix_resolv_gai.c index f522499e..f85c27c1 100644 --- a/src/platform/posix/posix_resolv_gai.c +++ b/src/platform/posix/posix_resolv_gai.c @@ -8,6 +8,7 @@ // found online at https://opensource.org/licenses/MIT. // +#include "core/init.h" #include "core/nng_impl.h" #ifdef NNG_USE_POSIX_RESOLV_GAI @@ -42,7 +43,7 @@ static nni_cv resolv_cv = NNI_CV_INITIALIZER(&resolv_mtx); static bool resolv_fini = false; static nni_list resolv_aios; static nni_thr *resolv_thrs; -static int resolv_num_thr; +static int16_t resolv_num_thr; typedef struct resolv_item resolv_item; struct resolv_item { @@ -477,22 +478,17 @@ nni_parse_ip_port(const char *addr, nni_sockaddr *sa) } int -nni_posix_resolv_sysinit(void) +nni_posix_resolv_sysinit(nng_init_params *params) { resolv_fini = false; nni_aio_list_init(&resolv_aios); -#ifndef NNG_RESOLV_CONCURRENCY -#define NNG_RESOLV_CONCURRENCY 4 -#endif - - resolv_num_thr = (int) nni_init_get_param( - NNG_INIT_NUM_RESOLVER_THREADS, NNG_RESOLV_CONCURRENCY); + resolv_num_thr = params->num_resolver_threads; if (resolv_num_thr < 1) { resolv_num_thr = 1; } + params->num_resolver_threads = resolv_num_thr; // no limit on the maximum for now - nni_init_set_effective(NNG_INIT_NUM_RESOLVER_THREADS, resolv_num_thr); resolv_thrs = NNI_ALLOC_STRUCTS(resolv_thrs, resolv_num_thr); if (resolv_thrs == NULL) { return (NNG_ENOMEM); diff --git a/src/platform/posix/posix_thread.c b/src/platform/posix/posix_thread.c index b7cd5e1e..343a6ba5 100644 --- a/src/platform/posix/posix_thread.c +++ b/src/platform/posix/posix_thread.c @@ -1,5 +1,5 @@ // -// Copyright 2021 Staysail Systems, Inc. <info@staysail.tech> +// Copyright 2024 Staysail Systems, Inc. <info@staysail.tech> // Copyright 2018 Capitar IT Group BV <info@capitar.com> // // This software is supplied under the terms of the MIT License, a @@ -11,6 +11,7 @@ // POSIX threads. #include "core/nng_impl.h" +#include "nng/nng.h" #ifdef NNG_PLATFORM_POSIX @@ -33,9 +34,8 @@ #include <sys/resource.h> #endif -static pthread_mutex_t nni_plat_init_lock = PTHREAD_MUTEX_INITIALIZER; -static volatile int nni_plat_inited = 0; -static int nni_plat_forked = 0; +static bool nni_plat_inited = 0; +static bool nni_plat_forked = 0; pthread_condattr_t nni_cvattr; pthread_mutexattr_t nni_mxattr; @@ -143,6 +143,9 @@ nni_pthread_cond_timedwait( void nni_plat_mtx_lock(nni_plat_mtx *mtx) { + if (nni_plat_forked) { + nni_panic("nng is not fork-reentrant safe"); + } nni_pthread_mutex_lock(&mtx->mtx); } @@ -339,35 +342,20 @@ nni_atfork_child(void) } int -nni_plat_init(int (*helper)(void)) +nni_plat_init(nng_init_params *params) { int rv; - if (nni_plat_forked) { - nni_panic("nng is not fork-reentrant safe"); - } - if (nni_plat_inited) { - return (0); // fast path - } - - pthread_mutex_lock(&nni_plat_init_lock); - if (nni_plat_inited) { // check again under the lock to be sure - pthread_mutex_unlock(&nni_plat_init_lock); - return (0); - } - if ((pthread_mutexattr_init(&nni_mxattr) != 0) || (pthread_condattr_init(&nni_cvattr) != 0) || (pthread_attr_init(&nni_thrattr) != 0)) { // Technically this is leaking, but it should never // occur, so really not worried about it. - pthread_mutex_unlock(&nni_plat_init_lock); return (NNG_ENOMEM); } #if !defined(NNG_USE_GETTIMEOFDAY) && NNG_USE_CLOCKID != CLOCK_REALTIME if (pthread_condattr_setclock(&nni_cvattr, NNG_USE_CLOCKID) != 0) { - pthread_mutex_unlock(&nni_plat_init_lock); pthread_mutexattr_destroy(&nni_mxattr); pthread_condattr_destroy(&nni_cvattr); pthread_attr_destroy(&nni_thrattr); @@ -381,7 +369,6 @@ nni_plat_init(int (*helper)(void)) (rl.rlim_cur != RLIM_INFINITY) && (rl.rlim_cur >= PTHREAD_STACK_MIN) && (pthread_attr_setstacksize(&nni_thrattr, rl.rlim_cur) != 0)) { - pthread_mutex_unlock(&nni_plat_init_lock); pthread_mutexattr_destroy(&nni_mxattr); pthread_condattr_destroy(&nni_cvattr); pthread_attr_destroy(&nni_thrattr); @@ -393,16 +380,14 @@ nni_plat_init(int (*helper)(void)) (void) pthread_mutexattr_settype( &nni_mxattr, PTHREAD_MUTEX_ERRORCHECK); - if ((rv = nni_posix_pollq_sysinit()) != 0) { - pthread_mutex_unlock(&nni_plat_init_lock); + if ((rv = nni_posix_pollq_sysinit(params)) != 0) { pthread_mutexattr_destroy(&nni_mxattr); pthread_condattr_destroy(&nni_cvattr); pthread_attr_destroy(&nni_thrattr); return (rv); } - if ((rv = nni_posix_resolv_sysinit()) != 0) { - pthread_mutex_unlock(&nni_plat_init_lock); + if ((rv = nni_posix_resolv_sysinit(params)) != 0) { nni_posix_pollq_sysfini(); pthread_mutexattr_destroy(&nni_mxattr); pthread_condattr_destroy(&nni_cvattr); @@ -411,7 +396,6 @@ nni_plat_init(int (*helper)(void)) } if (pthread_atfork(NULL, NULL, nni_atfork_child) != 0) { - pthread_mutex_unlock(&nni_plat_init_lock); nni_posix_resolv_sysfini(); nni_posix_pollq_sysfini(); pthread_mutexattr_destroy(&nni_mxattr); @@ -419,10 +403,7 @@ nni_plat_init(int (*helper)(void)) pthread_attr_destroy(&nni_thrattr); return (NNG_ENOMEM); } - if ((rv = helper()) == 0) { - nni_plat_inited = 1; - } - pthread_mutex_unlock(&nni_plat_init_lock); + nni_plat_inited = 1; return (rv); } @@ -430,7 +411,6 @@ nni_plat_init(int (*helper)(void)) void nni_plat_fini(void) { - pthread_mutex_lock(&nni_plat_init_lock); if (nni_plat_inited) { nni_posix_resolv_sysfini(); nni_posix_pollq_sysfini(); @@ -438,7 +418,6 @@ nni_plat_fini(void) pthread_condattr_destroy(&nni_cvattr); nni_plat_inited = 0; } - pthread_mutex_unlock(&nni_plat_init_lock); } int diff --git a/src/platform/resolver_test.c b/src/platform/resolver_test.c index ed88d09d..64554291 100644 --- a/src/platform/resolver_test.c +++ b/src/platform/resolver_test.c @@ -22,7 +22,6 @@ has_v6(void) nni_plat_udp *u; int rv; - nni_init(); // ensure that platform poller is up sa.s_in6.sa_family = NNG_AF_INET6; sa.s_in6.sa_port = 0; memcpy(sa.s_in6.sa_addr, v6loop, 16); diff --git a/src/platform/windows/win_impl.h b/src/platform/windows/win_impl.h index 1354709b..d94e1547 100644 --- a/src/platform/windows/win_impl.h +++ b/src/platform/windows/win_impl.h @@ -107,7 +107,7 @@ extern int nni_win_error(int); extern int nni_win_tcp_conn_init(nni_tcp_conn **, SOCKET); -extern int nni_win_io_sysinit(void); +extern int nni_win_io_sysinit(nng_init_params *params); extern void nni_win_io_sysfini(void); extern int nni_win_ipc_sysinit(void); @@ -119,7 +119,7 @@ extern void nni_win_tcp_sysfini(void); extern int nni_win_udp_sysinit(void); extern void nni_win_udp_sysfini(void); -extern int nni_win_resolv_sysinit(void); +extern int nni_win_resolv_sysinit(nng_init_params *); extern void nni_win_resolv_sysfini(void); extern void nni_win_io_init(nni_win_io *, nni_win_io_cb, void *); diff --git a/src/platform/windows/win_io.c b/src/platform/windows/win_io.c index 47dd7408..815f9bf3 100644 --- a/src/platform/windows/win_io.c +++ b/src/platform/windows/win_io.c @@ -71,25 +71,16 @@ nni_win_io_init(nni_win_io *io, nni_win_io_cb cb, void *ptr) } int -nni_win_io_sysinit(void) +nni_win_io_sysinit(nng_init_params *params) { - HANDLE h; - int i; - int rv; - int num_thr; - int max_thr; - -#ifndef NNG_MAX_POLLER_THREADS -#define NNG_MAX_POLLER_THREADS 8 -#endif -#ifndef NNG_NUM_POLLER_THREADS -#define NNG_NUM_POLLER_THREADS (nni_plat_ncpu()) -#endif - max_thr = (int) nni_init_get_param( - NNG_INIT_MAX_POLLER_THREADS, NNG_MAX_POLLER_THREADS); - - num_thr = (int) nni_init_get_param( - NNG_INIT_NUM_POLLER_THREADS, NNG_NUM_POLLER_THREADS); + HANDLE h; + int i; + int rv; + int16_t num_thr; + int16_t max_thr; + + max_thr = params->max_poller_threads; + num_thr = params->num_poller_threads; if ((max_thr > 0) && (num_thr > max_thr)) { num_thr = max_thr; @@ -97,7 +88,7 @@ nni_win_io_sysinit(void) if (num_thr < 1) { num_thr = 1; } - nni_init_set_effective(NNG_INIT_NUM_POLLER_THREADS, num_thr); + params->num_poller_threads = num_thr; if ((win_io_thrs = NNI_ALLOC_STRUCTS(win_io_thrs, num_thr)) == NULL) { return (NNG_ENOMEM); } diff --git a/src/platform/windows/win_resolv.c b/src/platform/windows/win_resolv.c index 4f353e08..1b1ae7b9 100644 --- a/src/platform/windows/win_resolv.c +++ b/src/platform/windows/win_resolv.c @@ -27,7 +27,7 @@ static nni_cv resolv_cv = NNI_CV_INITIALIZER(&resolv_mtx); static bool resolv_fini = false; static nni_list resolv_aios; static nni_thr *resolv_thrs; -static int resolv_num_thr; +static int16_t resolv_num_thr; typedef struct resolv_item resolv_item; struct resolv_item { @@ -432,28 +432,24 @@ nni_parse_ip_port(const char *addr, nni_sockaddr *sa) } int -nni_win_resolv_sysinit(void) +nni_win_resolv_sysinit(nng_init_params *params) { nni_aio_list_init(&resolv_aios); resolv_fini = false; -#ifndef NNG_RESOLV_CONCURRENCY -#define NNG_RESOLV_CONCURRENCY 4 -#endif - - resolv_num_thr = (int) nni_init_get_param( - NNG_INIT_NUM_RESOLVER_THREADS, NNG_RESOLV_CONCURRENCY); + resolv_num_thr = params->num_resolver_threads; if (resolv_num_thr < 1) { resolv_num_thr = 1; } + params->num_resolver_threads = resolv_num_thr; + // no limit on the maximum for now - nni_init_set_effective(NNG_INIT_NUM_RESOLVER_THREADS, resolv_num_thr); resolv_thrs = NNI_ALLOC_STRUCTS(resolv_thrs, resolv_num_thr); if (resolv_thrs == NULL) { return (NNG_ENOMEM); } - for (int i = 0; i < resolv_num_thr; i++) { + for (int16_t i = 0; i < resolv_num_thr; i++) { int rv = nni_thr_init(&resolv_thrs[i], resolv_worker, NULL); if (rv != 0) { nni_win_resolv_sysfini(); diff --git a/src/platform/windows/win_thread.c b/src/platform/windows/win_thread.c index 67ff60b2..a6416cd8 100644 --- a/src/platform/windows/win_thread.c +++ b/src/platform/windows/win_thread.c @@ -416,8 +416,6 @@ nni_plat_thr_set_name(nni_plat_thr *thr, const char *name) } } -static LONG plat_inited = 0; - int nni_plat_ncpu(void) { @@ -428,36 +426,26 @@ nni_plat_ncpu(void) } int -nni_plat_init(int (*helper)(void)) +nni_plat_init(nng_init_params *params) { int rv = 0; static SRWLOCK lock = SRWLOCK_INIT; - if (plat_inited) { - return (0); // fast path - } - AcquireSRWLockExclusive(&lock); - if (!plat_inited) { - // Let's look up the function to set thread descriptions. - hKernel32 = LoadLibrary(TEXT("kernel32.dll")); - if (hKernel32 != NULL) { - set_thread_desc = - (pfnSetThreadDescription) GetProcAddress( - hKernel32, "SetThreadDescription"); - } - - if (((rv = nni_win_io_sysinit()) != 0) || - ((rv = nni_win_ipc_sysinit()) != 0) || - ((rv = nni_win_tcp_sysinit()) != 0) || - ((rv = nni_win_udp_sysinit()) != 0) || - ((rv = nni_win_resolv_sysinit()) != 0)) { - goto out; - } + // Let's look up the function to set thread descriptions. + hKernel32 = LoadLibrary(TEXT("kernel32.dll")); + if (hKernel32 != NULL) { + set_thread_desc = (pfnSetThreadDescription) GetProcAddress( + hKernel32, "SetThreadDescription"); + } - helper(); - plat_inited = 1; + if (((rv = nni_win_io_sysinit(params)) != 0) || + ((rv = nni_win_ipc_sysinit()) != 0) || + ((rv = nni_win_tcp_sysinit()) != 0) || + ((rv = nni_win_udp_sysinit()) != 0) || + ((rv = nni_win_resolv_sysinit(params)) != 0)) { + goto out; } out: @@ -478,7 +466,6 @@ nni_plat_fini(void) if (hKernel32 != NULL) { FreeLibrary(hKernel32); } - plat_inited = 0; } #endif diff --git a/src/sp/protocol/pubsub0/sub.c b/src/sp/protocol/pubsub0/sub.c index a4741114..1b17fcbf 100644 --- a/src/sp/protocol/pubsub0/sub.c +++ b/src/sp/protocol/pubsub0/sub.c @@ -711,8 +711,7 @@ nng_sub0_socket_subscribe(nng_socket id, const void *buf, size_t sz) nni_sock *s; sub0_sock *sock; - if (((rv = nni_init()) != 0) || - ((rv = nni_sock_find(&s, id.id)) != 0)) { + if ((rv = nni_sock_find(&s, id.id)) != 0) { return (rv); } // validate the socket type @@ -733,8 +732,7 @@ nng_sub0_socket_unsubscribe(nng_socket id, const void *buf, size_t sz) nni_sock *s; sub0_sock *sock; - if (((rv = nni_init()) != 0) || - ((rv = nni_sock_find(&s, id.id)) != 0)) { + if ((rv = nni_sock_find(&s, id.id)) != 0) { return (rv); } // validate the socket type @@ -755,8 +753,7 @@ nng_sub0_ctx_subscribe(nng_ctx id, const void *buf, size_t sz) nni_ctx *c; sub0_ctx *ctx; - if (((rv = nni_init()) != 0) || - ((rv = nni_ctx_find(&c, id.id, false)) != 0)) { + if ((rv = nni_ctx_find(&c, id.id, false)) != 0) { return (rv); } // validate the socket type @@ -777,8 +774,7 @@ nng_sub0_ctx_unsubscribe(nng_ctx id, const void *buf, size_t sz) nni_ctx *c; sub0_ctx *ctx; - if (((rv = nni_init()) != 0) || - ((rv = nni_ctx_find(&c, id.id, false)) != 0)) { + if ((rv = nni_ctx_find(&c, id.id, false)) != 0) { return (rv); } // validate the socket type diff --git a/src/sp/transport/tls/tls.c b/src/sp/transport/tls/tls.c index 387ce023..8ea92888 100644 --- a/src/sp/transport/tls/tls.c +++ b/src/sp/transport/tls/tls.c @@ -1320,12 +1320,6 @@ static nni_sp_tran tls6_tran = { }; #endif -int -nng_tls_register(void) -{ - return (nni_init()); -} - void nni_sp_tls_register(void) { diff --git a/src/supplemental/http/http_public.c b/src/supplemental/http/http_public.c index e493a994..633f83c7 100644 --- a/src/supplemental/http/http_public.c +++ b/src/supplemental/http/http_public.c @@ -20,7 +20,6 @@ int nng_http_req_alloc(nng_http_req **reqp, const nng_url *url) { #ifdef NNG_SUPP_HTTP - nni_init(); return (nni_http_req_alloc(reqp, url)); #else NNI_ARG_UNUSED(reqp); @@ -53,7 +52,6 @@ int nng_http_res_alloc(nng_http_res **resp) { #ifdef NNG_SUPP_HTTP - nni_init(); return (nni_http_res_alloc(resp)); #else NNI_ARG_UNUSED(resp); @@ -64,7 +62,6 @@ nng_http_res_alloc(nng_http_res **resp) int nng_http_res_alloc_error(nng_http_res **resp, uint16_t code) { - nni_init(); #ifdef NNG_SUPP_HTTP return (nni_http_res_alloc_error(resp, code)); #else @@ -510,7 +507,6 @@ nng_http_handler_alloc( nng_http_handler **hp, const char *uri, void (*cb)(nng_aio *)) { #ifdef NNG_SUPP_HTTP - nni_init(); return (nni_http_handler_init(hp, uri, cb)); #else NNI_ARG_UNUSED(hp); @@ -677,7 +673,6 @@ int nng_http_server_hold(nng_http_server **srvp, const nng_url *url) { #ifdef NNG_SUPP_HTTP - nni_init(); return (nni_http_server_init(srvp, url)); #else NNI_ARG_UNUSED(srvp); @@ -836,7 +831,6 @@ int nng_http_client_alloc(nng_http_client **clip, const nng_url *url) { #ifdef NNG_SUPP_HTTP - nni_init(); return (nni_http_client_init(clip, url)); #else NNI_ARG_UNUSED(clip); diff --git a/src/supplemental/tls/tls_common.c b/src/supplemental/tls/tls_common.c index b197af91..aa34b533 100644 --- a/src/supplemental/tls/tls_common.c +++ b/src/supplemental/tls/tls_common.c @@ -258,9 +258,6 @@ nni_tls_dialer_alloc(nng_stream_dialer **dp, const nng_url *url) my_url.u_scheme += 4; } - if ((rv = nni_init()) != 0) { - return (rv); - } if ((d = NNI_ALLOC_STRUCT(d)) == NULL) { return (NNG_ENOMEM); } @@ -420,9 +417,6 @@ nni_tls_listener_alloc(nng_stream_listener **lp, const nng_url *url) my_url.u_scheme += 4; } - if ((rv = nni_init()) != 0) { - return (rv); - } if ((l = NNI_ALLOC_STRUCT(l)) == NULL) { return (NNG_ENOMEM); } @@ -1193,10 +1187,6 @@ nng_tls_config_alloc(nng_tls_config **cfg_p, nng_tls_mode mode) size_t size; int rv; - if ((rv = nni_init()) != 0) { - return (rv); - } - eng = nni_atomic_get_ptr(&tls_engine); if (eng == NULL) { @@ -1252,8 +1242,6 @@ nng_tls_engine_name(void) { const nng_tls_engine *eng; - nni_init(); - eng = nni_atomic_get_ptr(&tls_engine); return (eng == NULL ? "none" : eng->name); @@ -1264,8 +1252,6 @@ nng_tls_engine_description(void) { const nng_tls_engine *eng; - nni_init(); - eng = nni_atomic_get_ptr(&tls_engine); return (eng == NULL ? "" : eng->description); @@ -1276,8 +1262,6 @@ nng_tls_engine_fips_mode(void) { const nng_tls_engine *eng; - nni_init(); - eng = nni_atomic_get_ptr(&tls_engine); return (eng == NULL ? false : eng->fips_mode); diff --git a/src/testing/nuts.h b/src/testing/nuts.h index 7a60aa32..1c1f4595 100644 --- a/src/testing/nuts.h +++ b/src/testing/nuts.h @@ -23,13 +23,18 @@ extern void nuts_logger( nng_log_level, nng_log_facility, const char *, const char *); // Call nng_fini during test finalization -- this avoids leak warnings. -extern void nng_fini(void); +#ifndef TEST_FINI #define TEST_FINI nng_fini() +#endif + +#ifndef TEST_INIT #define TEST_INIT \ do { \ + nng_init(NULL); \ nng_log_set_logger(nuts_logger); \ nng_log_set_level(NNG_LOG_NONE); \ } while (0) +#endif #include "acutest.h" #include <stdbool.h> diff --git a/src/tools/nngcat/nngcat.c b/src/tools/nngcat/nngcat.c index 87f7c0a6..57bee3e5 100644 --- a/src/tools/nngcat/nngcat.c +++ b/src/tools/nngcat/nngcat.c @@ -723,6 +723,9 @@ main(int ac, char **av) addrend = &addrs; topicend = &topics; + nng_init(NULL); + atexit(nng_fini); + while ((rv = nng_opts_parse(ac, av, opts, &val, &arg, &idx)) == 0) { switch (val) { case OPT_HELP: diff --git a/src/tools/perf/perf.c b/src/tools/perf/perf.c index 2801ce51..c7057c73 100644 --- a/src/tools/perf/perf.c +++ b/src/tools/perf/perf.c @@ -189,6 +189,9 @@ main(int argc, char **argv) open_client = nng_pair0_open; #endif + nng_init(NULL); + atexit(nng_fini); + // Allow -m <remote_lat> or whatever to override argv[0]. if ((argc >= 3) && (strcmp(argv[1], "-m") == 0)) { prog = argv[2]; diff --git a/src/tools/perf/pubdrop.c b/src/tools/perf/pubdrop.c index af715a88..020bae0f 100644 --- a/src/tools/perf/pubdrop.c +++ b/src/tools/perf/pubdrop.c @@ -75,6 +75,9 @@ main(int argc, char **argv) argc--; argv++; + nng_init(NULL); + atexit(nng_fini); + // We calculate a delay factor to roughly delay 1 usec. We don't // need this to be perfect, just reproducible on the same host. unsigned long cnt = 1000000; |