// // Copyright 2021 Staysail Systems, Inc. // Copyright 2018 Capitar IT Group BV // // This software is supplied under the terms of the MIT License, a // copy of which should be located in the distribution where this // file was obtained (LICENSE.txt). A copy of the license may also be // found online at https://opensource.org/licenses/MIT. // #include "core/nng_impl.h" #include "sockimpl.h" #include #include // Socket implementation. static nni_list sock_list; static nni_id_map sock_ids; static nni_mtx sock_lk; static nni_id_map ctx_ids; static bool inited; struct nni_ctx { nni_list_node c_node; nni_sock * c_sock; nni_proto_ctx_ops c_ops; void * c_data; size_t c_size; bool c_closed; unsigned c_ref; // protected by global lock uint32_t c_id; nng_duration c_sndtimeo; nng_duration c_rcvtimeo; }; typedef struct nni_sockopt { nni_list_node node; char * name; nni_type typ; size_t sz; void * data; } nni_sockopt; typedef struct nni_sock_pipe_cb { nng_pipe_cb cb_fn; void * cb_arg; } nni_sock_pipe_cb; struct nni_socket { nni_list_node s_node; nni_mtx s_mx; nni_cv s_cv; nni_cv s_close_cv; uint32_t s_id; uint32_t s_flags; unsigned s_ref; // protected by global lock void * s_data; // Protocol private size_t s_size; nni_msgq *s_uwq; // Upper write queue nni_msgq *s_urq; // Upper read queue nni_proto_id s_self_id; nni_proto_id s_peer_id; nni_proto_pipe_ops s_pipe_ops; nni_proto_sock_ops s_sock_ops; nni_proto_ctx_ops s_ctx_ops; // options nni_duration s_sndtimeo; // send timeout nni_duration s_rcvtimeo; // receive timeout nni_duration s_reconn; // reconnect time nni_duration s_reconnmax; // max reconnect time size_t s_rcvmaxsz; // max receive size nni_list s_options; // opts not handled by sock/proto char s_name[64]; // socket name (legacy compat) nni_list s_listeners; // active listeners nni_list s_dialers; // active dialers nni_list s_pipes; // active pipes nni_list s_ctxs; // active contexts (protected by global sock_lk) bool s_closing; // Socket is closing bool s_closed; // Socket closed, protected by global lock bool s_ctxwait; // Waiting for contexts to close. nni_mtx s_pipe_cbs_mtx; nni_sock_pipe_cb s_pipe_cbs[NNG_PIPE_EV_NUM]; #ifdef NNG_ENABLE_STATS nni_stat_item st_root; // socket scope nni_stat_item st_id; // socket id nni_stat_item st_name; // socket name nni_stat_item st_protocol; // socket protocol nni_stat_item st_dialers; // number of dialers nni_stat_item st_listeners; // number of listeners nni_stat_item st_pipes; // number of pipes nni_stat_item st_rx_bytes; // number of bytes received nni_stat_item st_tx_bytes; // number of bytes received nni_stat_item st_rx_msgs; // number of msgs received nni_stat_item st_tx_msgs; // number of msgs sent nni_stat_item st_rejects; // pipes rejected #endif }; static void nni_ctx_destroy(nni_ctx *); static void dialer_shutdown_locked(nni_dialer *); static void listener_shutdown_locked(nni_listener *); #define SOCK(s) ((nni_sock *) (s)) static int sock_get_fd(void *s, unsigned flag, int *fdp) { int rv; nni_pollable *p; if ((flag & nni_sock_flags(SOCK(s))) == 0) { return (NNG_ENOTSUP); } if (flag == NNI_PROTO_FLAG_SND) { rv = nni_msgq_get_sendable(SOCK(s)->s_uwq, &p); } else { rv = nni_msgq_get_recvable(SOCK(s)->s_urq, &p); } if (rv == 0) { rv = nni_pollable_getfd(p, fdp); } return (rv); } static int sock_get_sendfd(void *s, void *buf, size_t *szp, nni_type t) { int fd; int rv; if ((rv = sock_get_fd(SOCK(s), NNI_PROTO_FLAG_SND, &fd)) != 0) { return (rv); } return (nni_copyout_int(fd, buf, szp, t)); } static int sock_get_recvfd(void *s, void *buf, size_t *szp, nni_type t) { int fd; int rv; if ((rv = sock_get_fd(SOCK(s), NNI_PROTO_FLAG_RCV, &fd)) != 0) { return (rv); } return (nni_copyout_int(fd, buf, szp, t)); } static int sock_get_raw(void *s, void *buf, size_t *szp, nni_type t) { bool raw = ((nni_sock_flags(SOCK(s)) & NNI_PROTO_FLAG_RAW) != 0); return (nni_copyout_bool(raw, buf, szp, t)); } static int sock_set_recvtimeo(void *s, const void *buf, size_t sz, nni_type t) { return (nni_copyin_ms(&SOCK(s)->s_rcvtimeo, buf, sz, t)); } static int sock_get_recvtimeo(void *s, void *buf, size_t *szp, nni_type t) { return (nni_copyout_ms(SOCK(s)->s_rcvtimeo, buf, szp, t)); } static int sock_set_sendtimeo(void *s, const void *buf, size_t sz, nni_type t) { return (nni_copyin_ms(&SOCK(s)->s_sndtimeo, buf, sz, t)); } static int sock_get_sendtimeo(void *s, void *buf, size_t *szp, nni_type t) { return (nni_copyout_ms(SOCK(s)->s_sndtimeo, buf, szp, t)); } static int sock_set_recvbuf(void *s, const void *buf, size_t sz, nni_type t) { int len; int rv; if ((rv = nni_copyin_int(&len, buf, sz, 0, 8192, t)) != 0) { return (rv); } return (nni_msgq_resize(SOCK(s)->s_urq, len)); } static int sock_get_recvbuf(void *s, void *buf, size_t *szp, nni_type t) { int len = nni_msgq_cap(SOCK(s)->s_urq); return (nni_copyout_int(len, buf, szp, t)); } static int sock_set_sendbuf(void *s, const void *buf, size_t sz, nni_type t) { int len; int rv; if ((rv = nni_copyin_int(&len, buf, sz, 0, 8192, t)) != 0) { return (rv); } return (nni_msgq_resize(SOCK(s)->s_uwq, len)); } static int sock_get_sendbuf(void *s, void *buf, size_t *szp, nni_type t) { int len = nni_msgq_cap(SOCK(s)->s_uwq); return (nni_copyout_int(len, buf, szp, t)); } static int sock_get_sockname(void *s, void *buf, size_t *szp, nni_type t) { return (nni_copyout_str(SOCK(s)->s_name, buf, szp, t)); } static int sock_set_sockname(void *s, const void *buf, size_t sz, nni_type t) { return (nni_copyin_str( SOCK(s)->s_name, buf, sizeof(SOCK(s)->s_name), sz, t)); } static int sock_get_proto(void *s, void *buf, size_t *szp, nni_type t) { return (nni_copyout_int(nni_sock_proto_id(SOCK(s)), buf, szp, t)); } static int sock_get_peer(void *s, void *buf, size_t *szp, nni_type t) { return (nni_copyout_int(nni_sock_peer_id(SOCK(s)), buf, szp, t)); } static int sock_get_protoname(void *s, void *buf, size_t *szp, nni_type t) { return (nni_copyout_str(nni_sock_proto_name(SOCK(s)), buf, szp, t)); } static int sock_get_peername(void *s, void *buf, size_t *szp, nni_type t) { return (nni_copyout_str(nni_sock_peer_name(SOCK(s)), buf, szp, t)); } static const nni_option sock_options[] = { { .o_name = NNG_OPT_RECVTIMEO, .o_get = sock_get_recvtimeo, .o_set = sock_set_recvtimeo, }, { .o_name = NNG_OPT_SENDTIMEO, .o_get = sock_get_sendtimeo, .o_set = sock_set_sendtimeo, }, { .o_name = NNG_OPT_RECVFD, .o_get = sock_get_recvfd, }, { .o_name = NNG_OPT_SENDFD, .o_get = sock_get_sendfd, }, { .o_name = NNG_OPT_RECVBUF, .o_get = sock_get_recvbuf, .o_set = sock_set_recvbuf, }, { .o_name = NNG_OPT_SENDBUF, .o_get = sock_get_sendbuf, .o_set = sock_set_sendbuf, }, { .o_name = NNG_OPT_SOCKNAME, .o_get = sock_get_sockname, .o_set = sock_set_sockname, }, { .o_name = NNG_OPT_RAW, .o_get = sock_get_raw, }, { .o_name = NNG_OPT_PROTO, .o_get = sock_get_proto, }, { .o_name = NNG_OPT_PEER, .o_get = sock_get_peer, }, { .o_name = NNG_OPT_PROTONAME, .o_get = sock_get_protoname, }, { .o_name = NNG_OPT_PEERNAME, .o_get = sock_get_peername, }, // terminate list { .o_name = NULL, }, }; static void nni_free_opt(nni_sockopt *opt) { nni_strfree(opt->name); nni_free(opt->data, opt->sz); NNI_FREE_STRUCT(opt); } uint32_t nni_sock_id(nni_sock *s) { return (s->s_id); } // nni_sock_sendq and nni_sock_recvq are called by the protocol to obtain // the upper read and write queues. nni_msgq * nni_sock_sendq(nni_sock *s) { return (s->s_uwq); } nni_msgq * nni_sock_recvq(nni_sock *s) { return (s->s_urq); } int nni_sock_find(nni_sock **sockp, uint32_t id) { int rv; 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) { rv = NNG_ECLOSED; } else { s->s_ref++; *sockp = s; } } else { rv = NNG_ECLOSED; } nni_mtx_unlock(&sock_lk); return (rv); } void nni_sock_rele(nni_sock *s) { nni_mtx_lock(&sock_lk); s->s_ref--; if (s->s_closed && (s->s_ref < 2)) { nni_cv_wake(&s->s_close_cv); } nni_mtx_unlock(&sock_lk); } #ifdef NNG_ENABLE_STATS static void sock_stat_init(nni_sock *s, nni_stat_item *item, const nni_stat_info *info) { nni_stat_init(item, info); nni_stat_add(&s->st_root, item); } static void sock_stats_init(nni_sock *s) { static const nni_stat_info root_info = { .si_name = "socket", .si_desc = "socket statistics", .si_type = NNG_STAT_SCOPE, }; static const nni_stat_info id_info = { .si_name = "id", .si_desc = "socket identifier", .si_type = NNG_STAT_ID, }; static const nni_stat_info name_info = { .si_name = "name", .si_desc = "socket name", .si_type = NNG_STAT_STRING, .si_alloc = true, }; static const nni_stat_info protocol_info = { .si_name = "protocol", .si_desc = "socket protocol", .si_type = NNG_STAT_STRING, }; static const nni_stat_info dialers_info = { .si_name = "dialers", .si_desc = "open dialers", .si_type = NNG_STAT_LEVEL, .si_atomic = true, }; static const nni_stat_info listeners_info = { .si_name = "listeners", .si_desc = "open listeners", .si_type = NNG_STAT_LEVEL, .si_atomic = true, }; static const nni_stat_info pipes_info = { .si_name = "pipes", .si_desc = "open pipes", .si_type = NNG_STAT_LEVEL, .si_atomic = true, }; static const nni_stat_info reject_info = { .si_name = "reject", .si_desc = "rejected pipes", .si_type = NNG_STAT_COUNTER, .si_atomic = true, }; static const nni_stat_info tx_msgs_info = { .si_name = "tx_msgs", .si_desc = "sent messages", .si_type = NNG_STAT_COUNTER, .si_unit = NNG_UNIT_MESSAGES, .si_atomic = true, }; static const nni_stat_info rx_msgs_info = { .si_name = "rx_msgs", .si_desc = "received messages", .si_type = NNG_STAT_COUNTER, .si_unit = NNG_UNIT_MESSAGES, .si_atomic = true, }; static const nni_stat_info tx_bytes_info = { .si_name = "tx_bytes", .si_desc = "sent bytes", .si_type = NNG_STAT_COUNTER, .si_unit = NNG_UNIT_BYTES, .si_atomic = true, }; static const nni_stat_info rx_bytes_info = { .si_name = "rx_bytes", .si_desc = "received messages", .si_type = NNG_STAT_COUNTER, .si_unit = NNG_UNIT_BYTES, .si_atomic = true, }; // To make collection cheap and atomic for the socket, // we just use a single lock for the entire chain. nni_stat_init(&s->st_root, &root_info); sock_stat_init(s, &s->st_id, &id_info); sock_stat_init(s, &s->st_name, &name_info); sock_stat_init(s, &s->st_protocol, &protocol_info); sock_stat_init(s, &s->st_dialers, &dialers_info); sock_stat_init(s, &s->st_listeners, &listeners_info); sock_stat_init(s, &s->st_pipes, &pipes_info); sock_stat_init(s, &s->st_rejects, &reject_info); sock_stat_init(s, &s->st_tx_msgs, &tx_msgs_info); sock_stat_init(s, &s->st_rx_msgs, &rx_msgs_info); sock_stat_init(s, &s->st_tx_bytes, &tx_bytes_info); sock_stat_init(s, &s->st_rx_bytes, &rx_bytes_info); nni_stat_set_id(&s->st_id, (int) s->s_id); nni_stat_set_string(&s->st_name, s->s_name); nni_stat_set_string(&s->st_protocol, nni_sock_proto_name(s)); } #endif static void sock_destroy(nni_sock *s) { nni_sockopt *sopt; #ifdef NNG_ENABLE_STATS nni_stat_unregister(&s->st_root); #endif // The protocol needs to clean up its state. if (s->s_data != NULL) { s->s_sock_ops.sock_fini(s->s_data); } nni_mtx_lock(&s->s_mx); while ((sopt = nni_list_first(&s->s_options)) != NULL) { nni_list_remove(&s->s_options, sopt); nni_free_opt(sopt); } nni_mtx_unlock(&s->s_mx); nni_msgq_fini(s->s_urq); nni_msgq_fini(s->s_uwq); nni_cv_fini(&s->s_close_cv); nni_cv_fini(&s->s_cv); nni_mtx_fini(&s->s_mx); nni_mtx_fini(&s->s_pipe_cbs_mtx); nni_free(s, s->s_size); } static int nni_sock_create(nni_sock **sp, const nni_proto *proto) { int rv; nni_sock *s; bool on; size_t sz; sz = NNI_ALIGN_UP(sizeof(*s)) + proto->proto_sock_ops->sock_size; if ((s = nni_zalloc(sz)) == NULL) { return (NNG_ENOMEM); } s->s_data = s + 1; s->s_sndtimeo = -1; s->s_rcvtimeo = -1; s->s_reconn = NNI_SECOND; s->s_reconnmax = 0; s->s_rcvmaxsz = 0; // unlimited by default s->s_id = 0; s->s_ref = 0; s->s_self_id = proto->proto_self; s->s_peer_id = proto->proto_peer; s->s_flags = proto->proto_flags; s->s_sock_ops = *proto->proto_sock_ops; s->s_pipe_ops = *proto->proto_pipe_ops; s->s_closed = false; s->s_closing = false; if (proto->proto_ctx_ops != NULL) { s->s_ctx_ops = *proto->proto_ctx_ops; } NNI_ASSERT(s->s_sock_ops.sock_open != NULL); NNI_ASSERT(s->s_sock_ops.sock_close != NULL); NNI_LIST_NODE_INIT(&s->s_node); NNI_LIST_INIT(&s->s_options, nni_sockopt, node); NNI_LIST_INIT(&s->s_ctxs, nni_ctx, c_node); NNI_LIST_INIT(&s->s_pipes, nni_pipe, p_sock_node); NNI_LIST_INIT(&s->s_listeners, nni_listener, l_node); NNI_LIST_INIT(&s->s_dialers, nni_dialer, d_node); nni_mtx_init(&s->s_mx); nni_mtx_init(&s->s_pipe_cbs_mtx); nni_cv_init(&s->s_cv, &s->s_mx); nni_cv_init(&s->s_close_cv, &sock_lk); #ifdef NNG_ENABLE_STATS sock_stats_init(s); #endif if (((rv = nni_msgq_init(&s->s_uwq, 0)) != 0) || ((rv = nni_msgq_init(&s->s_urq, 1)) != 0) || ((rv = s->s_sock_ops.sock_init(s->s_data, s)) != 0) || ((rv = nni_sock_setopt(s, NNG_OPT_SENDTIMEO, &s->s_sndtimeo, sizeof(nni_duration), NNI_TYPE_DURATION)) != 0) || ((rv = nni_sock_setopt(s, NNG_OPT_RECVTIMEO, &s->s_rcvtimeo, sizeof(nni_duration), NNI_TYPE_DURATION)) != 0) || ((rv = nni_sock_setopt(s, NNG_OPT_RECONNMINT, &s->s_reconn, sizeof(nni_duration), NNI_TYPE_DURATION)) != 0) || ((rv = nni_sock_setopt(s, NNG_OPT_RECONNMAXT, &s->s_reconnmax, sizeof(nni_duration), NNI_TYPE_DURATION)) != 0) || ((rv = nni_sock_setopt(s, NNG_OPT_RECVMAXSZ, &s->s_rcvmaxsz, sizeof(size_t), NNI_TYPE_SIZE)) != 0)) { sock_destroy(s); return (rv); } // These we *attempt* to call so that we are likely to have initial // values loaded. They should not fail, but if they do we don't // worry about it. on = true; (void) nni_sock_setopt( s, NNG_OPT_TCP_NODELAY, &on, sizeof(on), NNI_TYPE_BOOL); on = false; (void) nni_sock_setopt( s, NNG_OPT_TCP_KEEPALIVE, &on, sizeof(on), NNI_TYPE_BOOL); *sp = s; return (rv); } int nni_sock_sys_init(void) { NNI_LIST_INIT(&sock_list, nni_sock, s_node); nni_mtx_init(&sock_lk); nni_id_map_init(&sock_ids, 1, 0x7fffffff, false); nni_id_map_init(&ctx_ids, 1, 0x7fffffff, false); inited = true; return (0); } void nni_sock_sys_fini(void) { nni_id_map_fini(&sock_ids); nni_id_map_fini(&ctx_ids); nni_mtx_fini(&sock_lk); inited = false; } int nni_sock_open(nni_sock **sockp, const nni_proto *proto) { nni_sock *s = NULL; int rv; if (proto->proto_version != NNI_PROTOCOL_VERSION) { // unsupported protocol version return (NNG_ENOTSUP); } if (((rv = nni_init()) != 0) || ((rv = nni_sock_create(&s, proto)) != 0)) { return (rv); } nni_mtx_lock(&sock_lk); if (nni_id_alloc(&sock_ids, &s->s_id, s) != 0) { sock_destroy(s); } else { nni_list_append(&sock_list, s); s->s_sock_ops.sock_open(s->s_data); *sockp = s; } nni_mtx_unlock(&sock_lk); // Set the socket name. (void) snprintf(s->s_name, sizeof(s->s_name), "%u", s->s_id); #ifdef NNG_ENABLE_STATS // Set up basic stat values. nni_stat_set_id(&s->st_id, (int) s->s_id); // Add our stats chain. nni_stat_register(&s->st_root); #endif return (0); } // nni_sock_shutdown shuts down the socket; after this point no // further access to the socket will function, and any threads blocked // in entry points will be woken (and the functions they are blocked // in will return NNG_ECLOSED.) int nni_sock_shutdown(nni_sock *sock) { nni_pipe * pipe; nni_dialer * d; nni_listener *l; nni_ctx * ctx; nni_ctx * nctx; nni_mtx_lock(&sock->s_mx); if (sock->s_closing) { nni_mtx_unlock(&sock->s_mx); return (NNG_ECLOSED); } // Mark us closing, so no more EPs or changes can occur. sock->s_closing = true; // Close the EPs. This prevents new connections from forming // but but allows existing ones to drain. NNI_LIST_FOREACH (&sock->s_listeners, l) { listener_shutdown_locked(l); } NNI_LIST_FOREACH (&sock->s_dialers, d) { dialer_shutdown_locked(d); } nni_mtx_unlock(&sock->s_mx); // We now mark any owned contexts as closing. // XXX: Add context draining support here! nni_mtx_lock(&sock_lk); nctx = nni_list_first(&sock->s_ctxs); while ((ctx = nctx) != NULL) { nctx = nni_list_next(&sock->s_ctxs, ctx); ctx->c_closed = true; if (ctx->c_ref == 0) { // No open operations. So close it. nni_id_remove(&ctx_ids, ctx->c_id); nni_list_remove(&sock->s_ctxs, ctx); nni_ctx_destroy(ctx); } // If still has a reference count, then wait for last // reference to close before nuking it. } nni_mtx_unlock(&sock_lk); // Generally, unless the protocol is blocked trying to perform // writes (e.g. a slow reader on the other side), it should be // trying to shut things down. We wait to give it // a chance to do so gracefully. nni_mtx_lock(&sock_lk); while (!nni_list_empty(&sock->s_ctxs)) { sock->s_ctxwait = true; nni_cv_wait(&sock->s_close_cv); } nni_mtx_unlock(&sock_lk); nni_mtx_lock(&sock->s_mx); // At this point, we've done everything we politely can to // give the protocol a chance to flush its write side. Now // its time to be a little more insistent. // Close the upper queues immediately. This can happen // safely while we hold the lock. nni_msgq_close(sock->s_urq); nni_msgq_close(sock->s_uwq); // Go through the dialers and listeners, attempting to close them. // We might already have a close in progress, in which case // we skip past it; it will be removed from another thread. NNI_LIST_FOREACH (&sock->s_listeners, l) { if (nni_listener_hold(l) == 0) { nni_listener_close_rele(l); } } NNI_LIST_FOREACH (&sock->s_dialers, d) { if (nni_dialer_hold(d) == 0) { nni_dialer_close_rele(d); } } // For each pipe, arrange for it to teardown hard. We would // expect there not to be any here. However, it is possible for // a pipe to have been added by an endpoint due to racing conditions // in the shutdown. Therefore it is important that we shutdown pipes // *last*. NNI_LIST_FOREACH (&sock->s_pipes, pipe) { nni_pipe_close(pipe); } // We have to wait for *both* endpoints and pipes to be // removed. while ((!nni_list_empty(&sock->s_pipes)) || (!nni_list_empty(&sock->s_listeners)) || (!nni_list_empty(&sock->s_dialers))) { nni_cv_wait(&sock->s_cv); } sock->s_sock_ops.sock_close(sock->s_data); nni_cv_wake(&sock->s_cv); NNI_ASSERT(nni_list_first(&sock->s_pipes) == NULL); nni_mtx_unlock(&sock->s_mx); // At this point, there are no threads blocked inside of us // that are referencing socket state. User code should call // nng_close to release the last resources. return (0); } // nni_sock_close shuts down the socket, then releases any resources // associated with it. It is a programmer error to reference the // socket after this function is called, as the pointer may reference // invalid memory or other objects. void nni_sock_close(nni_sock *s) { // Shutdown everything if not already done. This operation // is idempotent. nni_sock_shutdown(s); nni_mtx_lock(&sock_lk); if (s->s_closed) { // Some other thread called close. All we need to do // is drop our reference count. nni_mtx_unlock(&sock_lk); nni_sock_rele(s); return; } s->s_closed = true; nni_id_remove(&sock_ids, s->s_id); // We might have been removed from the list already, e.g. by // nni_sock_closeall. This is idempotent. nni_list_node_remove(&s->s_node); // Wait for all other references to drop. Note that we // have a reference already (from our caller). s->s_ctxwait = true; while ((s->s_ref > 1) || (!nni_list_empty(&s->s_ctxs))) { nni_cv_wait(&s->s_close_cv); } nni_mtx_unlock(&sock_lk); // Because we already shut everything down before, we should not // have any child objects. nni_mtx_lock(&s->s_mx); NNI_ASSERT(nni_list_empty(&s->s_dialers)); NNI_ASSERT(nni_list_empty(&s->s_listeners)); NNI_ASSERT(nni_list_empty(&s->s_pipes)); nni_mtx_unlock(&s->s_mx); sock_destroy(s); } void nni_sock_closeall(void) { nni_sock *s; if (!inited) { return; } for (;;) { nni_mtx_lock(&sock_lk); if ((s = nni_list_first(&sock_list)) == NULL) { nni_mtx_unlock(&sock_lk); return; } // Bump the reference count. The close call below // will drop it. s->s_ref++; nni_list_node_remove(&s->s_node); nni_mtx_unlock(&sock_lk); nni_sock_close(s); } } void nni_sock_send(nni_sock *sock, nni_aio *aio) { nni_aio_normalize_timeout(aio, sock->s_sndtimeo); sock->s_sock_ops.sock_send(sock->s_data, aio); } void nni_sock_recv(nni_sock *sock, nni_aio *aio) { nni_aio_normalize_timeout(aio, sock->s_rcvtimeo); sock->s_sock_ops.sock_recv(sock->s_data, aio); } // nni_sock_proto_id returns the socket's 16-bit protocol number. uint16_t nni_sock_proto_id(nni_sock *sock) { return (sock->s_self_id.p_id); } // nni_sock_peer_id returns the socket peer's 16-bit protocol number. uint16_t nni_sock_peer_id(nni_sock *sock) { return (sock->s_peer_id.p_id); } const char * nni_sock_proto_name(nni_sock *sock) { return (sock->s_self_id.p_name); } const char * nni_sock_peer_name(nni_sock *sock) { return (sock->s_peer_id.p_name); } struct nni_proto_pipe_ops * nni_sock_proto_pipe_ops(nni_sock *sock) { return (&sock->s_pipe_ops); } void * nni_sock_proto_data(nni_sock *sock) { return (sock->s_data); } int nni_sock_add_listener(nni_sock *s, nni_listener *l) { nni_sockopt *sopt; nni_mtx_lock(&s->s_mx); if (s->s_closing) { nni_mtx_unlock(&s->s_mx); return (NNG_ECLOSED); } NNI_LIST_FOREACH (&s->s_options, sopt) { int rv; rv = nni_listener_setopt( l, sopt->name, sopt->data, sopt->sz, sopt->typ); if ((rv != 0) && (rv != NNG_ENOTSUP)) { nni_mtx_unlock(&s->s_mx); return (rv); } } nni_list_append(&s->s_listeners, l); #ifdef NNG_ENABLE_STATS nni_stat_inc(&s->st_listeners, 1); #endif nni_mtx_unlock(&s->s_mx); return (0); } int nni_sock_add_dialer(nni_sock *s, nni_dialer *d) { nni_sockopt *sopt; nni_mtx_lock(&s->s_mx); if (s->s_closing) { nni_mtx_unlock(&s->s_mx); return (NNG_ECLOSED); } NNI_LIST_FOREACH (&s->s_options, sopt) { int rv; rv = nni_dialer_setopt( d, sopt->name, sopt->data, sopt->sz, sopt->typ); if ((rv != 0) && (rv != NNG_ENOTSUP)) { nni_mtx_unlock(&s->s_mx); return (rv); } } nni_list_append(&s->s_dialers, d); #ifdef NNG_ENABLE_STATS nni_stat_inc(&s->st_dialers, 1); #endif nni_mtx_unlock(&s->s_mx); return (0); } int nni_sock_setopt( nni_sock *s, const char *name, const void *v, size_t sz, nni_type t) { int rv; nni_sockopt *optv; nni_sockopt *oldv = NULL; nni_mtx_lock(&s->s_mx); if (s->s_closing) { nni_mtx_unlock(&s->s_mx); return (NNG_ECLOSED); } // Protocol options. The protocol can override options that // the socket framework would otherwise supply, like buffer // sizes. rv = nni_setopt(s->s_sock_ops.sock_options, name, s->s_data, v, sz, t); if (rv != NNG_ENOTSUP) { nni_mtx_unlock(&s->s_mx); return (rv); } // Some options do not go down to transports. Handle them directly. rv = nni_setopt(sock_options, name, s, v, sz, t); if (rv != NNG_ENOTSUP) { nni_mtx_unlock(&s->s_mx); return (rv); } nni_mtx_unlock(&s->s_mx); // Validation of generic and transport options. // NOTE: Setting transport options via socket is deprecated. // These options should be set on the endpoint to which they apply. if ((strcmp(name, NNG_OPT_RECONNMINT) == 0) || (strcmp(name, NNG_OPT_RECONNMAXT) == 0)) { if ((rv = nni_copyin_ms(NULL, v, sz, t)) != 0) { return (rv); } } else if (strcmp(name, NNG_OPT_RECVMAXSZ) == 0) { if ((rv = nni_copyin_size(NULL, v, sz, 0, NNI_MAXSZ, t)) != 0) { return (rv); } #if !defined(NNG_ELIDE_DEPRECATED) // TCP options, set via socket is deprecated. } else if ((strcmp(name, NNG_OPT_TCP_KEEPALIVE) == 0) || (strcmp(name, NNG_OPT_TCP_NODELAY)) == 0) { if ((rv = nni_copyin_bool(NULL, v, sz, t)) != 0) { return (rv); } #endif #if defined(NNG_SUPP_TLS) && !defined(NNG_ELIDE_DEPRECATED) // TLS options may not be supported if TLS is not // compiled in. Supporting all these is deprecated. } else if (strcmp(name, NNG_OPT_TLS_CONFIG) == 0) { if ((rv = nni_copyin_ptr(NULL, v, sz, t)) != 0) { return (rv); } } else if ((strcmp(name, NNG_OPT_TLS_SERVER_NAME) == 0) || (strcmp(name, NNG_OPT_TLS_CA_FILE) == 0) || (strcmp(name, NNG_OPT_TLS_CERT_KEY_FILE) == 0)) { if ((t != NNI_TYPE_OPAQUE) && (t != NNI_TYPE_STRING)) { return (NNG_EBADTYPE); } if (nni_strnlen(v, sz) >= sz) { return (NNG_EINVAL); } } else if ((strcmp(name, NNG_OPT_TLS_AUTH_MODE) == 0)) { // 0, 1, or 2 (none, optional, required) if ((rv = nni_copyin_int(NULL, v, sz, 0, 2, t)) != 0) { return (rv); } #endif #if defined(NNG_PLATFORM_POSIX) && !defined(NNG_SUPPRESS_DEPRECATED) } else if (strcmp(name, NNG_OPT_IPC_PERMISSIONS) == 0) { // UNIX mode bits are 0777, but allow set id and sticky bits if ((rv = nni_copyin_int(NULL, v, sz, 0, 07777, t)) != 0) { return (rv); } #endif #if defined(NNG_PLATFORM_WINDOWS) && !defined(NNG_SUPPRESS_DEPRECATED) } else if (strcmp(name, NNG_OPT_IPC_SECURITY_DESCRIPTOR) == 0) { if ((rv = nni_copyin_ptr(NULL, v, sz, t)) == 0) { return (rv); } #endif } // Prepare a copy of the socket option. if ((optv = NNI_ALLOC_STRUCT(optv)) == NULL) { return (NNG_ENOMEM); } if ((optv->data = nni_alloc(sz)) == NULL) { NNI_FREE_STRUCT(optv); return (NNG_ENOMEM); } if ((optv->name = nni_strdup(name)) == NULL) { nni_free(optv->data, sz); NNI_FREE_STRUCT(optv); return (NNG_ENOMEM); } memcpy(optv->data, v, sz); optv->sz = sz; optv->typ = t; NNI_LIST_NODE_INIT(&optv->node); nni_mtx_lock(&s->s_mx); NNI_LIST_FOREACH (&s->s_options, oldv) { if (strcmp(oldv->name, name) == 0) { if ((oldv->sz != sz) || (memcmp(oldv->data, v, sz) != 0)) { break; } // The values are the same. This is a no-op. nni_mtx_unlock(&s->s_mx); nni_free_opt(optv); return (0); } } #ifndef NNG_ELIDE_DEPRCATED nni_dialer * d; nni_listener *l; // Apply the options. Failure to set any option on any // transport (other than ENOTSUP) stops the operation // altogether. Its important that transport wide checks // properly pre-validate. NNI_LIST_FOREACH (&s->s_listeners, l) { int x; x = nni_listener_setopt(l, optv->name, optv->data, sz, t); if (x != NNG_ENOTSUP) { if ((rv = x) != 0) { nni_mtx_unlock(&s->s_mx); nni_free_opt(optv); return (rv); } } } NNI_LIST_FOREACH (&s->s_dialers, d) { int x; x = nni_dialer_setopt(d, optv->name, optv->data, sz, t); if (x != NNG_ENOTSUP) { if ((rv = x) != 0) { nni_mtx_unlock(&s->s_mx); nni_free_opt(optv); return (rv); } } } #endif if (rv == 0) { // Remove and toss the old value; we are using a new one. if (oldv != NULL) { nni_list_remove(&s->s_options, oldv); nni_free_opt(oldv); } // Insert our new value. This permits it to be // compared against later, and for new endpoints to // automatically receive these values, nni_list_append(&s->s_options, optv); } else { nni_free_opt(optv); } nni_mtx_unlock(&s->s_mx); return (rv); } int nni_sock_getopt( nni_sock *s, const char *name, void *val, size_t *szp, nni_type t) { int rv; nni_sockopt *sopt; nni_mtx_lock(&s->s_mx); if (s->s_closing) { nni_mtx_unlock(&s->s_mx); return (NNG_ECLOSED); } // Protocol specific options. The protocol can override // options like the send buffer or notification descriptors // this way. rv = nni_getopt( s->s_sock_ops.sock_options, name, s->s_data, val, szp, t); if (rv != NNG_ENOTSUP) { nni_mtx_unlock(&s->s_mx); return (rv); } // Socket generic options. rv = nni_getopt(sock_options, name, s, val, szp, t); if (rv != NNG_ENOTSUP) { nni_mtx_unlock(&s->s_mx); return (rv); } NNI_LIST_FOREACH (&s->s_options, sopt) { if (strcmp(sopt->name, name) == 0) { size_t sz = sopt->sz; if ((sopt->typ != NNI_TYPE_OPAQUE) && (t != sopt->typ)) { if (t != NNI_TYPE_OPAQUE) { nni_mtx_unlock(&s->s_mx); return (NNG_EBADTYPE); } if (*szp != sopt->sz) { nni_mtx_unlock(&s->s_mx); return (NNG_EINVAL); } } if (szp != NULL) { if (sopt->sz > *szp) { sz = *szp; } *szp = sopt->sz; } memcpy(val, sopt->data, sz); rv = 0; break; } } nni_mtx_unlock(&s->s_mx); return (rv); } uint32_t nni_sock_flags(nni_sock *sock) { return (sock->s_flags); } void nni_sock_set_pipe_cb(nni_sock *s, int ev, nng_pipe_cb cb, void *arg) { if ((ev >= 0) && (ev < NNG_PIPE_EV_NUM)) { nni_mtx_lock(&s->s_pipe_cbs_mtx); s->s_pipe_cbs[ev].cb_fn = cb; s->s_pipe_cbs[ev].cb_arg = arg; nni_mtx_unlock(&s->s_pipe_cbs_mtx); } } int nni_ctx_find(nni_ctx **cp, uint32_t id, bool closing) { int rv; 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 // closed, or the context is closed. (If the socket // is closed, and we are only getting the reference so // we can close it, then we still allow. In the case // the only valid operation will be to close the // socket.) if (ctx->c_closed || ((!closing) && ctx->c_sock->s_closed)) { rv = NNG_ECLOSED; } else { ctx->c_ref++; *cp = ctx; } } else { rv = NNG_ECLOSED; } nni_mtx_unlock(&sock_lk); return (rv); } static void nni_ctx_destroy(nni_ctx *ctx) { if (ctx->c_data != NULL) { ctx->c_ops.ctx_fini(ctx->c_data); } // Let the socket go, our hold on it is done. nni_free(ctx, ctx->c_size); } void nni_ctx_rele(nni_ctx *ctx) { nni_sock *sock = ctx->c_sock; nni_mtx_lock(&sock_lk); ctx->c_ref--; if ((ctx->c_ref > 0) || (!ctx->c_closed)) { // Either still have an active reference, or not // actually closing yet. nni_mtx_unlock(&sock_lk); return; } // Remove us from the hash, so we can't be found any more. // This allows our ID to be reused later, although the system // tries to avoid ID reuse. nni_id_remove(&ctx_ids, ctx->c_id); nni_list_remove(&sock->s_ctxs, ctx); if (sock->s_closed || sock->s_ctxwait) { nni_cv_wake(&sock->s_close_cv); } nni_mtx_unlock(&sock_lk); nni_ctx_destroy(ctx); } int nni_ctx_open(nni_ctx **ctxp, nni_sock *sock) { nni_ctx *ctx; int rv; size_t sz; if (sock->s_ctx_ops.ctx_init == NULL) { return (NNG_ENOTSUP); } sz = NNI_ALIGN_UP(sizeof(*ctx)) + sock->s_ctx_ops.ctx_size; if ((ctx = nni_zalloc(sz)) == NULL) { return (NNG_ENOMEM); } ctx->c_size = sz; ctx->c_data = ctx + 1; ctx->c_closed = false; ctx->c_ref = 1; // Caller implicitly gets a reference. ctx->c_sock = sock; ctx->c_ops = sock->s_ctx_ops; ctx->c_rcvtimeo = sock->s_rcvtimeo; ctx->c_sndtimeo = sock->s_sndtimeo; nni_mtx_lock(&sock_lk); if (sock->s_closed) { nni_mtx_unlock(&sock_lk); nni_free(ctx, ctx->c_size); return (NNG_ECLOSED); } if ((rv = nni_id_alloc(&ctx_ids, &ctx->c_id, ctx)) != 0) { nni_mtx_unlock(&sock_lk); nni_free(ctx, ctx->c_size); return (rv); } if ((rv = sock->s_ctx_ops.ctx_init(ctx->c_data, sock->s_data)) != 0) { nni_id_remove(&ctx_ids, ctx->c_id); nni_mtx_unlock(&sock_lk); nni_free(ctx, ctx->c_size); return (rv); } nni_list_append(&sock->s_ctxs, ctx); nni_mtx_unlock(&sock_lk); // Paranoia, fixing a possible race in close. Don't let us // give back a context if the socket is being shutdown (it // might not have reached the "closed" state yet.) nni_mtx_lock(&sock->s_mx); if (sock->s_closing) { nni_mtx_unlock(&sock->s_mx); nni_ctx_rele(ctx); return (NNG_ECLOSED); } nni_mtx_unlock(&sock->s_mx); *ctxp = ctx; return (0); } void nni_ctx_close(nni_ctx *ctx) { nni_mtx_lock(&sock_lk); ctx->c_closed = true; nni_mtx_unlock(&sock_lk); nni_ctx_rele(ctx); } uint32_t nni_ctx_id(nni_ctx *ctx) { return (ctx->c_id); } void nni_ctx_send(nni_ctx *ctx, nni_aio *aio) { nni_aio_normalize_timeout(aio, ctx->c_sndtimeo); ctx->c_ops.ctx_send(ctx->c_data, aio); } void nni_ctx_recv(nni_ctx *ctx, nni_aio *aio) { nni_aio_normalize_timeout(aio, ctx->c_rcvtimeo); ctx->c_ops.ctx_recv(ctx->c_data, aio); } int nni_ctx_getopt(nni_ctx *ctx, const char *opt, void *v, size_t *szp, nni_type t) { nni_sock * sock = ctx->c_sock; nni_option *o; int rv = NNG_ENOTSUP; nni_mtx_lock(&sock->s_mx); if (strcmp(opt, NNG_OPT_RECVTIMEO) == 0) { rv = nni_copyout_ms(ctx->c_rcvtimeo, v, szp, t); } else if (strcmp(opt, NNG_OPT_SENDTIMEO) == 0) { rv = nni_copyout_ms(ctx->c_sndtimeo, v, szp, t); } else if (ctx->c_ops.ctx_options != NULL) { for (o = ctx->c_ops.ctx_options; o->o_name != NULL; o++) { if (strcmp(opt, o->o_name) != 0) { continue; } if (o->o_get == NULL) { rv = NNG_EWRITEONLY; break; } rv = o->o_get(ctx->c_data, v, szp, t); break; } } nni_mtx_unlock(&sock->s_mx); return (rv); } int nni_ctx_setopt( nni_ctx *ctx, const char *opt, const void *v, size_t sz, nni_type t) { nni_sock * sock = ctx->c_sock; nni_option *o; int rv = NNG_ENOTSUP; nni_mtx_lock(&sock->s_mx); if (strcmp(opt, NNG_OPT_RECVTIMEO) == 0) { rv = nni_copyin_ms(&ctx->c_rcvtimeo, v, sz, t); } else if (strcmp(opt, NNG_OPT_SENDTIMEO) == 0) { rv = nni_copyin_ms(&ctx->c_sndtimeo, v, sz, t); } else if (ctx->c_ops.ctx_options != NULL) { for (o = ctx->c_ops.ctx_options; o->o_name != NULL; o++) { if (strcmp(opt, o->o_name) != 0) { continue; } if (o->o_set == NULL) { rv = NNG_EREADONLY; break; } rv = o->o_set(ctx->c_data, v, sz, t); break; } } nni_mtx_unlock(&sock->s_mx); return (rv); } static void dialer_timer_start_locked(nni_dialer *d) { nni_duration back_off; nni_sock * sock = d->d_sock; if (d->d_closing || sock->s_closed) { return; } back_off = d->d_currtime; if (d->d_maxrtime > 0) { d->d_currtime *= 2; if (d->d_currtime > d->d_maxrtime) { d->d_currtime = d->d_maxrtime; } } // To minimize damage from storms, etc., we select a back-off // value randomly, in the range of [0, back_off-1]; this is // pretty similar to 802 style back-off, except that we have a // nearly uniform time period instead of discrete slot times. // This algorithm may lead to slight biases because we don't // have a statistically perfect distribution with the modulo of // the random number, but this really doesn't matter. nni_sleep_aio( back_off ? (int) nni_random() % back_off : 0, &d->d_tmo_aio); } void nni_dialer_timer_start(nni_dialer *d) { nni_sock *s = d->d_sock; nni_mtx_lock(&s->s_mx); dialer_timer_start_locked(d); nni_mtx_unlock(&s->s_mx); } void nni_dialer_add_pipe(nni_dialer *d, void *tpipe) { nni_sock *s = d->d_sock; nni_pipe *p; nni_mtx_lock(&s->s_mx); if (s->s_closed || d->d_closing) { d->d_tran->tran_pipe->p_fini(tpipe); nni_mtx_unlock(&s->s_mx); return; } if (nni_pipe_create_dialer(&p, d, tpipe) != 0) { nni_mtx_unlock(&s->s_mx); return; } nni_list_append(&d->d_pipes, p); nni_list_append(&s->s_pipes, p); d->d_pipe = p; d->d_currtime = d->d_inirtime; nni_mtx_unlock(&s->s_mx); #ifdef NNG_ENABLE_STATS nni_stat_inc(&s->st_pipes, 1); nni_stat_inc(&d->st_pipes, 1); #endif nni_pipe_run_cb(p, NNG_PIPE_EV_ADD_PRE); nni_mtx_lock(&s->s_mx); if (p->p_closed) { nni_mtx_unlock(&s->s_mx); #ifdef NNG_ENABLE_STATS nni_stat_inc(&d->st_reject, 1); nni_stat_inc(&s->st_rejects, 1); #endif nni_pipe_rele(p); return; } if (p->p_proto_ops.pipe_start(p->p_proto_data) != 0) { nni_mtx_unlock(&s->s_mx); #ifdef NNG_ENABLE_STATS nni_stat_inc(&d->st_reject, 1); nni_stat_inc(&s->st_rejects, 1); #endif nni_pipe_close(p); nni_pipe_rele(p); return; } nni_mtx_unlock(&s->s_mx); #ifdef NNG_ENABLE_STATS nni_stat_register(&p->st_root); #endif nni_pipe_run_cb(p, NNG_PIPE_EV_ADD_POST); nni_pipe_rele(p); } static void dialer_shutdown_impl(nni_dialer *d) { nni_pipe *p; // Abort any remaining in-flight operations. nni_aio_close(&d->d_con_aio); nni_aio_close(&d->d_tmo_aio); // Stop the underlying transport. d->d_ops.d_close(d->d_data); NNI_LIST_FOREACH (&d->d_pipes, p) { nni_pipe_close(p); } } static void dialer_shutdown_locked(nni_dialer *d) { if (!d->d_closing) { d->d_closing = true; dialer_shutdown_impl(d); } } void nni_dialer_shutdown(nni_dialer *d) { nni_sock *s = d->d_sock; nni_mtx_lock(&s->s_mx); dialer_shutdown_locked(d); nni_mtx_unlock(&s->s_mx); } static void dialer_reap(void *); static nni_reap_list dialer_reap_list = { .rl_offset = offsetof(nni_dialer, d_reap), .rl_func = dialer_reap, }; static void dialer_reap(void *arg) { nni_dialer *d = arg; nni_sock * s = d->d_sock; nni_aio_stop(&d->d_tmo_aio); nni_aio_stop(&d->d_con_aio); #ifdef NNG_ENABLE_STATS nni_stat_unregister(&d->st_root); #endif nni_mtx_lock(&s->s_mx); if (!nni_list_empty(&d->d_pipes)) { nni_pipe *p; // This should already have been done, but be certain! NNI_LIST_FOREACH (&d->d_pipes, p) { nni_pipe_close(p); } nni_mtx_unlock(&s->s_mx); // Go back to the end of reap list. nni_dialer_reap(d); return; } nni_list_remove(&s->s_dialers, d); if ((s->s_closing) && (nni_list_empty(&s->s_dialers))) { nni_cv_wake(&s->s_cv); } nni_mtx_unlock(&s->s_mx); nni_dialer_destroy(d); } void nni_dialer_reap(nni_dialer *d) { nni_reap(&dialer_reap_list, d); } void nni_listener_add_pipe(nni_listener *l, void *tpipe) { nni_sock *s = l->l_sock; nni_pipe *p; nni_mtx_lock(&s->s_mx); if (s->s_closed || l->l_closing) { l->l_tran->tran_pipe->p_fini(tpipe); nni_mtx_unlock(&s->s_mx); return; } if (nni_pipe_create_listener(&p, l, tpipe) != 0) { nni_mtx_unlock(&s->s_mx); return; } nni_list_append(&l->l_pipes, p); nni_list_append(&s->s_pipes, p); nni_mtx_unlock(&s->s_mx); #ifdef NNG_ENABLE_STATS nni_stat_inc(&l->st_pipes, 1); nni_stat_inc(&s->st_pipes, 1); #endif nni_pipe_run_cb(p, NNG_PIPE_EV_ADD_PRE); nni_mtx_lock(&s->s_mx); if (p->p_closed) { nni_mtx_unlock(&s->s_mx); #ifdef NNG_ENABLE_STATS nni_stat_inc(&l->st_reject, 1); nni_stat_inc(&s->st_rejects, 1); #endif nni_pipe_rele(p); return; } if (p->p_proto_ops.pipe_start(p->p_proto_data) != 0) { nni_mtx_unlock(&s->s_mx); #ifdef NNG_ENABLE_STATS nni_stat_inc(&l->st_reject, 1); nni_stat_inc(&s->st_rejects, 1); #endif nni_pipe_close(p); nni_pipe_rele(p); return; } nni_mtx_unlock(&s->s_mx); #ifdef NNG_ENABLE_STATS nni_stat_register(&p->st_root); #endif nni_pipe_run_cb(p, NNG_PIPE_EV_ADD_POST); nni_pipe_rele(p); } static void listener_shutdown_impl(nni_listener *l) { nni_pipe *p; // Abort any remaining in-flight accepts. nni_aio_close(&l->l_acc_aio); nni_aio_close(&l->l_tmo_aio); // Stop the underlying transport. l->l_ops.l_close(l->l_data); NNI_LIST_FOREACH (&l->l_pipes, p) { nni_pipe_close(p); } } static void listener_shutdown_locked(nni_listener *l) { if (!l->l_closing) { l->l_closing = true; listener_shutdown_impl(l); } } void nni_listener_shutdown(nni_listener *l) { nni_sock *s = l->l_sock; nni_mtx_lock(&s->s_mx); listener_shutdown_locked(l); nni_mtx_unlock(&s->s_mx); } static void listener_reap(void *); static nni_reap_list listener_reap_list = { .rl_offset = offsetof(nni_listener, l_reap), .rl_func = listener_reap, }; static void listener_reap(void *arg) { nni_listener *l = arg; nni_sock * s = l->l_sock; nni_aio_stop(&l->l_tmo_aio); nni_aio_stop(&l->l_acc_aio); #ifdef NNG_ENABLE_STATS nni_stat_unregister(&l->st_root); #endif nni_mtx_lock(&s->s_mx); if (!nni_list_empty(&l->l_pipes)) { nni_pipe *p; // This should already have been done, but be certain! NNI_LIST_FOREACH (&l->l_pipes, p) { nni_pipe_close(p); } nni_mtx_unlock(&s->s_mx); // Go back to the end of reap list. nni_reap(&listener_reap_list, l); return; } nni_list_remove(&s->s_listeners, l); if ((s->s_closing) && (nni_list_empty(&s->s_listeners))) { nni_cv_wake(&s->s_cv); } nni_mtx_unlock(&s->s_mx); nni_listener_destroy(l); } void nni_listener_reap(nni_listener *l) { nni_reap(&listener_reap_list, l); } void nni_pipe_run_cb(nni_pipe *p, nng_pipe_ev ev) { nni_sock * s = p->p_sock; nng_pipe_cb cb; void * arg; nni_mtx_lock(&s->s_pipe_cbs_mtx); if (!p->p_cbs) { if (ev == NNG_PIPE_EV_ADD_PRE) { // First event, after this we want all other events. p->p_cbs = true; } else { nni_mtx_unlock(&s->s_pipe_cbs_mtx); return; } } cb = s->s_pipe_cbs[ev].cb_fn; arg = s->s_pipe_cbs[ev].cb_arg; nni_mtx_unlock(&s->s_pipe_cbs_mtx); if (cb != NULL) { nng_pipe pid; pid.id = p->p_id; cb(pid, ev, arg); } } void nni_pipe_remove(nni_pipe *p) { nni_sock * s = p->p_sock; nni_dialer *d = p->p_dialer; nni_mtx_lock(&s->s_mx); #ifdef NNG_ENABLE_STATS if (nni_list_node_active(&p->p_sock_node)) { nni_stat_dec(&s->st_pipes, 1); } if (p->p_listener != NULL) { nni_stat_dec(&p->p_listener->st_pipes, 1); } if (p->p_dialer != NULL) { nni_stat_dec(&p->p_dialer->st_pipes, 1); } #endif nni_list_node_remove(&p->p_sock_node); nni_list_node_remove(&p->p_ep_node); p->p_listener = NULL; p->p_dialer = NULL; if ((d != NULL) && (d->d_pipe == p)) { d->d_pipe = NULL; dialer_timer_start_locked(d); // Kick the timer to redial. } if (s->s_closing) { nni_cv_wake(&s->s_cv); } nni_mtx_unlock(&s->s_mx); } void nni_sock_add_stat(nni_sock *s, nni_stat_item *stat) { #ifdef NNG_ENABLE_STATS nni_stat_add(&s->st_root, stat); #else NNI_ARG_UNUSED(s); NNI_ARG_UNUSED(stat); #endif } void nni_sock_bump_tx(nni_sock *s, uint64_t sz) { #ifdef NNG_ENABLE_STATS nni_stat_inc(&s->st_tx_msgs, 1); nni_stat_inc(&s->st_tx_bytes, sz); #else NNI_ARG_UNUSED(s); NNI_ARG_UNUSED(sz); #endif } void nni_sock_bump_rx(nni_sock *s, uint64_t sz) { #ifdef NNG_ENABLE_STATS nni_stat_inc(&s->st_rx_msgs, 1); nni_stat_inc(&s->st_rx_bytes, sz); #else NNI_ARG_UNUSED(s); NNI_ARG_UNUSED(sz); #endif }