// // Copyright 2018 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 // This file contains functions relating to pipes. // // Operations on pipes (to the transport) are generally blocking operations, // performed in the context of the protocol. static nni_idhash *nni_pipes; static nni_mtx nni_pipe_lk; int nni_pipe_sys_init(void) { int rv; nni_mtx_init(&nni_pipe_lk); if ((rv = nni_idhash_init(&nni_pipes)) != 0) { return (rv); } // Note that pipes have their own namespace. ID hash will // guarantee the that the first value is reasonable (non-zero), // if we supply an out of range value (0). (Consequently the // value "1" has a bias -- its roughly twice as likely to be // chosen as any other value. This does not mater.) nni_idhash_set_limits( nni_pipes, 1, 0x7fffffff, nni_random() & 0x7fffffff); return (0); } void nni_pipe_sys_fini(void) { nni_reap_drain(); nni_mtx_fini(&nni_pipe_lk); if (nni_pipes != NULL) { nni_idhash_fini(nni_pipes); nni_pipes = NULL; } } static void pipe_destroy(nni_pipe *p) { if (p == NULL) { return; } nni_pipe_run_cb(p, NNG_PIPE_EV_REM_POST); // Make sure any unlocked holders are done with this. // This happens during initialization for example. nni_mtx_lock(&nni_pipe_lk); if (p->p_id != 0) { nni_idhash_remove(nni_pipes, p->p_id); } // This wait guarantees that all callers are done with us. while (p->p_refcnt != 0) { nni_cv_wait(&p->p_cv); } nni_mtx_unlock(&nni_pipe_lk); if (p->p_proto_data != NULL) { p->p_proto_ops.pipe_stop(p->p_proto_data); } if ((p->p_tran_data != NULL) && (p->p_tran_ops.p_stop != NULL)) { p->p_tran_ops.p_stop(p->p_tran_data); } nni_pipe_remove(p); if (p->p_proto_data != NULL) { p->p_proto_ops.pipe_fini(p->p_proto_data); } if (p->p_tran_data != NULL) { p->p_tran_ops.p_fini(p->p_tran_data); } nni_cv_fini(&p->p_cv); nni_mtx_fini(&p->p_mtx); NNI_FREE_STRUCT(p); } int nni_pipe_find(nni_pipe **pp, uint32_t id) { int rv; nni_pipe *p; nni_mtx_lock(&nni_pipe_lk); // We don't care if the pipe is "closed". End users only have // access to the pipe in order to obtain properties (which may // be retried during the post-close notification callback) or to // close the pipe. if ((rv = nni_idhash_find(nni_pipes, id, (void **) &p)) == 0) { p->p_refcnt++; *pp = p; } nni_mtx_unlock(&nni_pipe_lk); return (rv); } void nni_pipe_rele(nni_pipe *p) { nni_mtx_lock(&nni_pipe_lk); p->p_refcnt--; if (p->p_refcnt == 0) { nni_cv_wake(&p->p_cv); } nni_mtx_unlock(&nni_pipe_lk); } // nni_pipe_id returns the 32-bit pipe id, which can be used in backtraces. uint32_t nni_pipe_id(nni_pipe *p) { return (p->p_id); } void nni_pipe_recv(nni_pipe *p, nni_aio *aio) { p->p_tran_ops.p_recv(p->p_tran_data, aio); } void nni_pipe_send(nni_pipe *p, nni_aio *aio) { p->p_tran_ops.p_send(p->p_tran_data, aio); } // nni_pipe_close closes the underlying connection. It is expected that // subsequent attempts to receive or send (including any waiting receive) will // simply return NNG_ECLOSED. void nni_pipe_close(nni_pipe *p) { nni_mtx_lock(&p->p_mtx); if (p->p_closed) { // We already did a close. nni_mtx_unlock(&p->p_mtx); return; } p->p_closed = true; nni_mtx_unlock(&p->p_mtx); if (p->p_proto_data != NULL) { p->p_proto_ops.pipe_close(p->p_proto_data); } // Close the underlying transport. if (p->p_tran_data != NULL) { p->p_tran_ops.p_close(p->p_tran_data); } nni_reap(&p->p_reap, (nni_cb) pipe_destroy, p); } uint16_t nni_pipe_peer(nni_pipe *p) { return (p->p_tran_ops.p_peer(p->p_tran_data)); } int nni_pipe_create(nni_pipe **pp, nni_sock *sock, nni_tran *tran, void *tdata) { nni_pipe * p; int rv; void * sdata = nni_sock_proto_data(sock); nni_proto_pipe_ops *pops = nni_sock_proto_pipe_ops(sock); uint64_t id; if ((p = NNI_ALLOC_STRUCT(p)) == NULL) { // In this case we just toss the pipe... tran->tran_pipe->p_fini(tdata); return (NNG_ENOMEM); } // Make a private copy of the transport ops. p->p_tran_ops = *tran->tran_pipe; p->p_tran_data = tdata; p->p_proto_ops = *pops; p->p_proto_data = NULL; p->p_sock = sock; p->p_closed = false; p->p_cbs = false; p->p_refcnt = 0; nni_atomic_flag_reset(&p->p_stop); NNI_LIST_NODE_INIT(&p->p_sock_node); NNI_LIST_NODE_INIT(&p->p_ep_node); nni_mtx_init(&p->p_mtx); nni_cv_init(&p->p_cv, &nni_pipe_lk); nni_mtx_lock(&nni_pipe_lk); if ((rv = nni_idhash_alloc(nni_pipes, &id, p)) == 0) { p->p_id = id; p->p_refcnt = 1; } nni_mtx_unlock(&nni_pipe_lk); if ((rv != 0) || ((rv = pops->pipe_init(&p->p_proto_data, p, sdata)) != 0)) { nni_pipe_close(p); nni_pipe_rele(p); return (rv); } *pp = p; return (0); } int nni_pipe_getopt( nni_pipe *p, const char *name, void *val, size_t *szp, nni_opt_type t) { nni_tran_option *o; for (o = p->p_tran_ops.p_options; o && o->o_name; o++) { if (strcmp(o->o_name, name) != 0) { continue; } return (o->o_get(p->p_tran_data, val, szp, t)); } // Maybe the endpoint knows? The guarantees on pipes ensure that the // pipe will not outlive its creating endpoint. if (p->p_dialer != NULL) { return (nni_dialer_getopt(p->p_dialer, name, val, szp, t)); } if (p->p_listener != NULL) { return (nni_listener_getopt(p->p_listener, name, val, szp, t)); } return (NNG_ENOTSUP); } void * nni_pipe_get_proto_data(nni_pipe *p) { return (p->p_proto_data); } uint32_t nni_pipe_sock_id(nni_pipe *p) { return (nni_sock_id(p->p_sock)); } uint32_t nni_pipe_listener_id(nni_pipe *p) { return (p->p_listener ? nni_listener_id(p->p_listener) : 0); } uint32_t nni_pipe_dialer_id(nni_pipe *p) { return (p->p_dialer ? nni_dialer_id(p->p_dialer) : 0); }