From d7f7c896c0ede24249ef63b1e45b1878bf4bd473 Mon Sep 17 00:00:00 2001 From: Garrett D'Amore Date: Tue, 31 Jul 2018 12:33:58 -0700 Subject: fixes #599 nng_dial sync should not return until added to socket fixes #208 pipe start should occur before connect / accept fixes #616 Race condition closing between header & body This refactors the transports to handle their own connection handshaking before passing the pipe to the socket. This changes and simplifies the setup. This also fixes a rather challenging race condition described by #616. --- src/core/aio.c | 10 +- src/core/dialer.c | 12 +- src/core/listener.c | 19 +- src/core/pipe.c | 67 +--- src/core/pipe.h | 5 - src/core/socket.c | 52 ++- src/core/socket.h | 2 - src/core/sockimpl.h | 5 +- src/core/transport.h | 12 +- src/platform/posix/posix_ipcconn.c | 8 + src/platform/posix/posix_tcpconn.c | 8 + src/transport/ipc/ipc.c | 567 +++++++++++++--------------- src/transport/tcp/tcp.c | 753 ++++++++++++++++--------------------- src/transport/tls/tls.c | 530 +++++++++++++------------- src/transport/zerotier/zerotier.c | 158 +++----- 15 files changed, 986 insertions(+), 1222 deletions(-) (limited to 'src') diff --git a/src/core/aio.c b/src/core/aio.c index 40638bce..294a0b92 100644 --- a/src/core/aio.c +++ b/src/core/aio.c @@ -217,11 +217,17 @@ void nni_aio_stop(nni_aio *aio) { if (aio != NULL) { + nni_aio_cancelfn cancelfn; + nni_mtx_lock(&nni_aio_lk); - aio->a_stop = true; + cancelfn = aio->a_prov_cancel; + aio->a_prov_cancel = NULL; + aio->a_stop = true; nni_mtx_unlock(&nni_aio_lk); - nni_aio_abort(aio, NNG_ECANCELED); + if (cancelfn != NULL) { + cancelfn(aio, NNG_ECANCELED); + } nni_aio_wait(aio); } diff --git a/src/core/dialer.c b/src/core/dialer.c index 34e90891..4b2f105d 100644 --- a/src/core/dialer.c +++ b/src/core/dialer.c @@ -232,24 +232,18 @@ dialer_timer_cb(void *arg) static void dialer_connect_cb(void *arg) { - nni_dialer *d = arg; - nni_pipe * p; + nni_dialer *d = arg; nni_aio * aio = d->d_con_aio; int rv; bool synch; - if ((rv = nni_aio_result(aio)) == 0) { - void *data = nni_aio_get_output(aio, 0); - NNI_ASSERT(data != NULL); - rv = nni_pipe_create(&p, d->d_sock, d->d_tran, data); - } nni_mtx_lock(&d->d_mtx); synch = d->d_synch; nni_mtx_unlock(&d->d_mtx); - switch (rv) { + switch ((rv = nni_aio_result(aio))) { case 0: - nni_dialer_add_pipe(d, p); + nni_dialer_add_pipe(d, nni_aio_get_output(aio, 0)); break; case NNG_ECLOSED: // No further action. case NNG_ECANCELED: // No further action. diff --git a/src/core/listener.c b/src/core/listener.c index debfa5f1..17062d6e 100644 --- a/src/core/listener.c +++ b/src/core/listener.c @@ -181,6 +181,8 @@ nni_listener_close(nni_listener *l) } l->l_closed = true; nni_mtx_unlock(&listeners_lk); + nni_aio_close(l->l_acc_aio); + nni_aio_close(l->l_tmo_aio); // Remove us from the table so we cannot be found. // This is done fairly early in the teardown process. @@ -228,28 +230,25 @@ listener_timer_cb(void *arg) static void listener_accept_cb(void *arg) { - nni_listener *l = arg; - nni_pipe * p; + nni_listener *l = arg; nni_aio * aio = l->l_acc_aio; - int rv; - if ((rv = nni_aio_result(aio)) == 0) { - void *data = nni_aio_get_output(aio, 0); - NNI_ASSERT(data != NULL); - rv = nni_pipe_create(&p, l->l_sock, l->l_tran, data); - } - switch (rv) { + switch (nni_aio_result(aio)) { case 0: - nni_listener_add_pipe(l, p); + nni_listener_add_pipe(l, nni_aio_get_output(aio, 0)); listener_accept_start(l); break; case NNG_ECONNABORTED: // remote condition, no cooldown case NNG_ECONNRESET: // remote condition, no cooldown + case NNG_EPEERAUTH: // peer validation failure listener_accept_start(l); break; case NNG_ECLOSED: // no further action case NNG_ECANCELED: // no further action break; + case NNG_ENOMEM: + case NNG_ENOFILES: + case NNG_ENOSPC: default: // We don't really know why we failed, but we backoff // here. This is because errors here are probably due diff --git a/src/core/pipe.c b/src/core/pipe.c index 4f50ac7c..374c45c8 100644 --- a/src/core/pipe.c +++ b/src/core/pipe.c @@ -75,9 +75,6 @@ pipe_destroy(nni_pipe *p) } nni_mtx_unlock(&nni_pipe_lk); - // Wait for neg callbacks to finish. (Already closed). - nni_aio_stop(p->p_start_aio); - if (p->p_proto_data != NULL) { p->p_proto_ops.pipe_stop(p->p_proto_data); } @@ -93,7 +90,6 @@ pipe_destroy(nni_pipe *p) if (p->p_tran_data != NULL) { p->p_tran_ops.p_fini(p->p_tran_data); } - nni_aio_fini(p->p_start_aio); nni_cv_fini(&p->p_cv); nni_mtx_fini(&p->p_mtx); NNI_FREE_STRUCT(p); @@ -154,9 +150,6 @@ nni_pipe_send(nni_pipe *p, nni_aio *aio) void nni_pipe_close(nni_pipe *p) { - // abort any pending negotiation/start process. - nni_aio_close(p->p_start_aio); - nni_mtx_lock(&p->p_mtx); if (p->p_closed) { // We already did a close. @@ -178,49 +171,12 @@ nni_pipe_close(nni_pipe *p) nni_reap(&p->p_reap, (nni_cb) pipe_destroy, p); } -bool -nni_pipe_closed(nni_pipe *p) -{ - bool rv; - nni_mtx_lock(&p->p_mtx); - rv = p->p_closed; - nni_mtx_unlock(&p->p_mtx); - return (rv); -} - uint16_t nni_pipe_peer(nni_pipe *p) { return (p->p_tran_ops.p_peer(p->p_tran_data)); } -static void -nni_pipe_start_cb(void *arg) -{ - nni_pipe *p = arg; - nni_sock *s = p->p_sock; - nni_aio * aio = p->p_start_aio; - - if (nni_aio_result(aio) != 0) { - nni_pipe_close(p); - return; - } - - nni_pipe_run_cb(p, NNG_PIPE_EV_ADD_PRE); - if (nni_pipe_closed(p)) { - nni_pipe_close(p); - return; - } - - if ((p->p_proto_ops.pipe_start(p->p_proto_data) != 0) || - nni_sock_closing(s)) { - nni_pipe_close(p); - return; - } - - nni_pipe_run_cb(p, NNG_PIPE_EV_ADD_POST); -} - int nni_pipe_create(nni_pipe **pp, nni_sock *sock, nni_tran *tran, void *tdata) { @@ -228,6 +184,7 @@ nni_pipe_create(nni_pipe **pp, nni_sock *sock, nni_tran *tran, void *tdata) 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... @@ -236,7 +193,6 @@ nni_pipe_create(nni_pipe **pp, nni_sock *sock, nni_tran *tran, void *tdata) } // Make a private copy of the transport ops. - p->p_start_aio = NULL; p->p_tran_ops = *tran->tran_pipe; p->p_tran_data = tdata; p->p_proto_ops = *pops; @@ -253,14 +209,11 @@ nni_pipe_create(nni_pipe **pp, nni_sock *sock, nni_tran *tran, void *tdata) nni_mtx_init(&p->p_mtx); nni_cv_init(&p->p_cv, &nni_pipe_lk); - if ((rv = nni_aio_init(&p->p_start_aio, nni_pipe_start_cb, p)) == 0) { - uint64_t id; - nni_mtx_lock(&nni_pipe_lk); - if ((rv = nni_idhash_alloc(nni_pipes, &id, p)) == 0) { - p->p_id = (uint32_t) id; - } - nni_mtx_unlock(&nni_pipe_lk); + nni_mtx_lock(&nni_pipe_lk); + if ((rv = nni_idhash_alloc(nni_pipes, &id, p)) == 0) { + p->p_id = (uint32_t) id; } + nni_mtx_unlock(&nni_pipe_lk); if ((rv != 0) || ((rv = pops->pipe_init(&p->p_proto_data, p, sdata)) != 0)) { @@ -296,16 +249,6 @@ nni_pipe_getopt( return (NNG_ENOTSUP); } -void -nni_pipe_start(nni_pipe *p) -{ - if (p->p_tran_ops.p_start == NULL) { - nni_aio_finish(p->p_start_aio, 0, 0); - } else { - p->p_tran_ops.p_start(p->p_tran_data, p->p_start_aio); - } -} - void * nni_pipe_get_proto_data(nni_pipe *p) { diff --git a/src/core/pipe.h b/src/core/pipe.h index 1d73ce51..1e2f2b5d 100644 --- a/src/core/pipe.h +++ b/src/core/pipe.h @@ -59,11 +59,6 @@ extern uint32_t nni_pipe_listener_id(nni_pipe *); // nni_pipe_dialer_id returns the dialer id for the pipe (or 0 if none). extern uint32_t nni_pipe_dialer_id(nni_pipe *); -// nni_pipe_closed returns true if nni_pipe_close was called. -// (This is used by the socket to determine if user closed the pipe -// during callback.) -extern bool nni_pipe_closed(nni_pipe *); - // nni_pipe_rele releases the hold on the pipe placed by nni_pipe_find. extern void nni_pipe_rele(nni_pipe *); diff --git a/src/core/socket.c b/src/core/socket.c index f4e59af5..0fa776f1 100644 --- a/src/core/socket.c +++ b/src/core/socket.c @@ -399,16 +399,6 @@ nni_sock_rele(nni_sock *s) nni_mtx_unlock(&sock_lk); } -bool -nni_sock_closing(nni_sock *s) -{ - bool rv; - nni_mtx_lock(&s->s_mx); - rv = s->s_closing; - nni_mtx_unlock(&s->s_mx); - return (rv); -} - static void sock_destroy(nni_sock *s) { @@ -1382,16 +1372,36 @@ nni_dialer_timer_start(nni_dialer *d) nni_mtx_unlock(&s->s_mx); } +static void +pipe_start(nni_pipe *p) +{ + nni_pipe_run_cb(p, NNG_PIPE_EV_ADD_PRE); + + // As the callback above that would close the pipe runs on + // this thread, we can skip the lock. + if (p->p_closed) { + return; + } + + if (p->p_proto_ops.pipe_start(p->p_proto_data) != 0) { + nni_pipe_close(p); + return; + } + + nni_pipe_run_cb(p, NNG_PIPE_EV_ADD_POST); +} + void -nni_dialer_add_pipe(nni_dialer *d, nni_pipe *p) +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) { + if (s->s_closed || d->d_closing || + (nni_pipe_create(&p, d->d_sock, d->d_tran, tpipe) != 0)) { nni_mtx_unlock(&s->s_mx); - nni_pipe_close(p); return; } @@ -1403,7 +1413,7 @@ nni_dialer_add_pipe(nni_dialer *d, nni_pipe *p) nni_mtx_unlock(&s->s_mx); // Start the initial negotiation I/O... - nni_pipe_start(p); + pipe_start(p); } static void @@ -1473,14 +1483,15 @@ nni_dialer_reap(nni_dialer *d) } void -nni_listener_add_pipe(nni_listener *l, nni_pipe *p) +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) { + if (s->s_closed || l->l_closing || + (nni_pipe_create(&p, l->l_sock, l->l_tran, tpipe) != 0)) { nni_mtx_unlock(&s->s_mx); - nni_pipe_close(p); return; } p->p_listener = l; @@ -1489,7 +1500,7 @@ nni_listener_add_pipe(nni_listener *l, nni_pipe *p) nni_mtx_unlock(&s->s_mx); // Start the initial negotiation I/O... - nni_pipe_start(p); + pipe_start(p); } static void @@ -1600,7 +1611,10 @@ nni_pipe_remove(nni_pipe *p) 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) { + dialer_timer_start_locked( + d); // Kick the timer to redial. + } } if (s->s_closing) { nni_cv_wake(&s->s_cv); diff --git a/src/core/socket.h b/src/core/socket.h index 4b9c4642..cebe87ca 100644 --- a/src/core/socket.h +++ b/src/core/socket.h @@ -62,8 +62,6 @@ extern uint32_t nni_sock_flags(nni_sock *); // should be executed. extern void nni_sock_set_pipe_cb(nni_sock *sock, int, nng_pipe_cb, void *); -extern bool nni_sock_closing(nni_sock *sock); - // nni_ctx_open is used to open/create a new context structure. // Contexts are not supported by most protocols, but for those that do, // this can offer some improvements for massive concurrency/scalability. diff --git a/src/core/sockimpl.h b/src/core/sockimpl.h index 569e5cae..29e83f7a 100644 --- a/src/core/sockimpl.h +++ b/src/core/sockimpl.h @@ -78,7 +78,6 @@ struct nni_pipe { nni_mtx p_mtx; nni_cv p_cv; nni_reap_item p_reap; - nni_aio * p_start_aio; }; extern int nni_sock_add_dialer(nni_sock *, nni_dialer *); @@ -87,14 +86,14 @@ extern void nni_sock_remove_dialer(nni_sock *, nni_dialer *); extern int nni_sock_add_listener(nni_sock *, nni_listener *); extern void nni_sock_remove_listener(nni_sock *, nni_listener *); -extern void nni_dialer_add_pipe(nni_dialer *, nni_pipe *); +extern void nni_dialer_add_pipe(nni_dialer *, void *); extern void nni_dialer_shutdown(nni_dialer *); extern void nni_dialer_reap(nni_dialer *); extern void nni_dialer_destroy(nni_dialer *); extern void nni_dialer_timer_start(nni_dialer *); extern void nni_dialer_close_rele(nni_dialer *); -extern void nni_listener_add_pipe(nni_listener *, nni_pipe *); +extern void nni_listener_add_pipe(nni_listener *, void *); extern void nni_listener_shutdown(nni_listener *); extern void nni_listener_reap(nni_listener *); extern void nni_listener_destroy(nni_listener *); diff --git a/src/core/transport.h b/src/core/transport.h index 257d232d..458bfda4 100644 --- a/src/core/transport.h +++ b/src/core/transport.h @@ -30,7 +30,8 @@ enum nni_ep_mode { #define NNI_TRANSPORT_V1 0x54520001 #define NNI_TRANSPORT_V2 0x54520002 #define NNI_TRANSPORT_V3 0x54520003 -#define NNI_TRANSPORT_VERSION NNI_TRANSPORT_V3 +#define NNI_TRANSPORT_V4 0x54520004 +#define NNI_TRANSPORT_VERSION NNI_TRANSPORT_V4 // Option handlers. struct nni_tran_option { @@ -131,19 +132,12 @@ struct nni_tran_pipe_ops { // make further calls on the same pipe. void (*p_fini)(void *); - // p_start starts the pipe running. This gives the transport a - // chance to hook into any transport specific negotiation - // phase. The pipe will not have its p_send or p_recv calls - // started, and will not be access by the "socket" until the - // pipe has indicated its readiness by finishing the aio. - void (*p_start)(void *, nni_aio *); - // p_stop stops the pipe, waiting for any callbacks that are // outstanding to complete. This is done before tearing down // resources with p_fini. void (*p_stop)(void *); - // p_aio_send queues the message for transmit. If this fails, + // p_send queues the message for transmit. If this fails, // then the caller may try again with the same message (or free // it). If the call succeeds, then the transport has taken // ownership of the message, and the caller may not use it diff --git a/src/platform/posix/posix_ipcconn.c b/src/platform/posix/posix_ipcconn.c index 2b46fb12..539a00ba 100644 --- a/src/platform/posix/posix_ipcconn.c +++ b/src/platform/posix/posix_ipcconn.c @@ -47,6 +47,10 @@ ipc_conn_dowrite(nni_ipc_conn *c) int fd; if (c->closed || ((fd = nni_posix_pfd_fd(c->pfd)) < 0)) { + while ((aio = nni_list_first(&c->writeq)) != NULL) { + nni_aio_list_remove(aio); + nni_aio_finish_error(aio, NNG_ECLOSED); + } return; } @@ -129,6 +133,10 @@ ipc_conn_doread(nni_ipc_conn *c) int fd; if (c->closed || ((fd = nni_posix_pfd_fd(c->pfd)) < 0)) { + while ((aio = nni_list_first(&c->readq)) != NULL) { + nni_aio_list_remove(aio); + nni_aio_finish_error(aio, NNG_ECLOSED); + } return; } diff --git a/src/platform/posix/posix_tcpconn.c b/src/platform/posix/posix_tcpconn.c index c0352c55..cf85e8e8 100644 --- a/src/platform/posix/posix_tcpconn.c +++ b/src/platform/posix/posix_tcpconn.c @@ -43,6 +43,10 @@ tcp_conn_dowrite(nni_tcp_conn *c) int fd; if (c->closed || ((fd = nni_posix_pfd_fd(c->pfd)) < 0)) { + while ((aio = nni_list_first(&c->writeq)) != NULL) { + nni_aio_list_remove(aio); + nni_aio_finish_error(aio, NNG_ECLOSED); + } return; } @@ -125,6 +129,10 @@ tcp_conn_doread(nni_tcp_conn *c) int fd; if (c->closed || ((fd = nni_posix_pfd_fd(c->pfd)) < 0)) { + while ((aio = nni_list_first(&c->readq)) != NULL) { + nni_aio_list_remove(aio); + nni_aio_finish_error(aio, NNG_ECLOSED); + } return; } diff --git a/src/transport/ipc/ipc.c b/src/transport/ipc/ipc.c index ee72d6d8..3d798d4c 100644 --- a/src/transport/ipc/ipc.c +++ b/src/transport/ipc/ipc.c @@ -23,14 +23,20 @@ typedef struct ipctran_pipe ipctran_pipe; typedef struct ipctran_dialer ipctran_dialer; typedef struct ipctran_listener ipctran_listener; +typedef struct ipctran_ep ipctran_ep; // ipc_pipe is one end of an IPC connection. struct ipctran_pipe { - nni_ipc_conn *conn; - uint16_t peer; - uint16_t proto; - size_t rcvmax; - nni_sockaddr sa; + nni_ipc_conn * conn; + uint16_t peer; + uint16_t proto; + size_t rcvmax; + nni_atomic_flag reaped; + nni_sockaddr sa; + nni_list_node node; + ipctran_ep * ep; + nni_reap_item reap; + bool closed; uint8_t txhead[1 + sizeof(uint64_t)]; uint8_t rxhead[1 + sizeof(uint64_t)]; @@ -41,32 +47,34 @@ struct ipctran_pipe { nni_list recvq; nni_list sendq; - nni_aio *user_negaio; + nni_aio *useraio; nni_aio *txaio; nni_aio *rxaio; - nni_aio *negaio; + nni_aio *negoaio; + nni_aio *connaio; nni_msg *rxmsg; nni_mtx mtx; }; +struct ipctran_ep { + nni_mtx mtx; + nni_list pipes; + bool fini; + size_t rcvmax; + uint16_t proto; + nni_cb dtor; + nni_sockaddr sa; + nni_reap_item reap; +}; + struct ipctran_dialer { - nni_sockaddr sa; + ipctran_ep ep; nni_ipc_dialer *dialer; - uint16_t proto; - size_t rcvmax; - nni_aio * aio; - nni_aio * user_aio; - nni_mtx mtx; }; struct ipctran_listener { - nni_sockaddr sa; + ipctran_ep ep; nni_ipc_listener *listener; - uint16_t proto; - size_t rcvmax; - nni_aio * aio; - nni_aio * user_aio; - nni_mtx mtx; }; static void ipctran_pipe_send_start(ipctran_pipe *); @@ -74,8 +82,8 @@ static void ipctran_pipe_recv_start(ipctran_pipe *); static void ipctran_pipe_send_cb(void *); static void ipctran_pipe_recv_cb(void *); static void ipctran_pipe_nego_cb(void *); -static void ipctran_dialer_cb(void *); -static void ipctran_listener_cb(void *); +static void ipctran_pipe_conn_cb(void *); +static void ipctran_pipe_reap(ipctran_pipe *); static int ipctran_init(void) @@ -95,7 +103,8 @@ ipctran_pipe_close(void *arg) nni_aio_close(p->rxaio); nni_aio_close(p->txaio); - nni_aio_close(p->negaio); + nni_aio_close(p->negoaio); + nni_aio_close(p->connaio); nni_ipc_conn_close(p->conn); } @@ -105,19 +114,39 @@ ipctran_pipe_stop(void *arg) { ipctran_pipe *p = arg; + nni_mtx_lock(&p->mtx); + p->closed = true; + nni_mtx_unlock(&p->mtx); + nni_aio_stop(p->rxaio); nni_aio_stop(p->txaio); - nni_aio_stop(p->negaio); + nni_aio_stop(p->negoaio); + nni_aio_stop(p->connaio); } static void ipctran_pipe_fini(void *arg) { ipctran_pipe *p = arg; + ipctran_ep * ep; + + if (p == NULL) { + return; + } + ipctran_pipe_stop(p); + if ((ep = p->ep) != NULL) { + nni_mtx_lock(&ep->mtx); + nni_list_remove(&ep->pipes, p); + if (ep->fini && nni_list_empty(&ep->pipes)) { + nni_reap(&ep->reap, ep->dtor, ep); + } + nni_mtx_unlock(&ep->mtx); + } nni_aio_fini(p->rxaio); nni_aio_fini(p->txaio); - nni_aio_fini(p->negaio); + nni_aio_fini(p->negoaio); + nni_aio_fini(p->connaio); if (p->conn != NULL) { nni_ipc_conn_fini(p->conn); } @@ -128,8 +157,16 @@ ipctran_pipe_fini(void *arg) NNI_FREE_STRUCT(p); } +static void +ipctran_pipe_reap(ipctran_pipe *p) +{ + if (!nni_atomic_flag_test_and_set(&p->reaped)) { + nni_reap(&p->reap, ipctran_pipe_fini, p); + } +} + static int -ipctran_pipe_init(ipctran_pipe **pipep, void *conn) +ipctran_pipe_init(ipctran_pipe **pipep, ipctran_ep *ep) { ipctran_pipe *p; int rv; @@ -138,51 +175,75 @@ ipctran_pipe_init(ipctran_pipe **pipep, void *conn) return (NNG_ENOMEM); } nni_mtx_init(&p->mtx); + NNI_LIST_NODE_INIT(&p->node); if (((rv = nni_aio_init(&p->txaio, ipctran_pipe_send_cb, p)) != 0) || ((rv = nni_aio_init(&p->rxaio, ipctran_pipe_recv_cb, p)) != 0) || - ((rv = nni_aio_init(&p->negaio, ipctran_pipe_nego_cb, p)) != 0)) { - ipctran_pipe_fini(p); + ((rv = nni_aio_init(&p->negoaio, ipctran_pipe_nego_cb, p)) != 0) || + ((rv = nni_aio_init(&p->connaio, ipctran_pipe_conn_cb, p)) != 0)) { + ipctran_pipe_reap(p); return (rv); } nni_aio_list_init(&p->sendq); nni_aio_list_init(&p->recvq); - p->conn = conn; -#if 0 - p->proto = ep->proto; - p->rcvmax = ep->rcvmax; - p->sa.s_ipc.sa_family = NNG_AF_IPC; - p->sa = ep->sa; -#endif + nni_list_append(&ep->pipes, p); + p->ep = ep; + p->proto = ep->proto; + p->rcvmax = ep->rcvmax; + p->sa = ep->sa; + *pipep = p; return (0); } static void -ipctran_pipe_nego_cancel(nni_aio *aio, int rv) +ipctran_pipe_conn_cb(void *arg) { - ipctran_pipe *p = nni_aio_get_prov_data(aio); + ipctran_pipe *p = arg; + nni_aio * aio = p->connaio; + nni_iov iov; + int rv; - nni_mtx_lock(&p->mtx); - if (p->user_negaio != aio) { - nni_mtx_unlock(&p->mtx); + nni_mtx_lock(&p->ep->mtx); + if ((rv = nni_aio_result(aio)) != 0) { + nni_aio *uaio; + if ((uaio = p->useraio) != NULL) { + p->useraio = NULL; + nni_aio_finish_error(uaio, rv); + } + nni_mtx_unlock(&p->ep->mtx); + ipctran_pipe_reap(p); return; } - p->user_negaio = NULL; - nni_mtx_unlock(&p->mtx); - nni_aio_abort(p->negaio, rv); - nni_aio_finish_error(aio, rv); + p->conn = nni_aio_get_output(aio, 0); + p->txhead[0] = 0; + p->txhead[1] = 'S'; + p->txhead[2] = 'P'; + p->txhead[3] = 0; + NNI_PUT16(&p->txhead[4], p->proto); + NNI_PUT16(&p->txhead[6], 0); + + p->gotrxhead = 0; + p->gottxhead = 0; + p->wantrxhead = 8; + p->wanttxhead = 8; + iov.iov_len = 8; + iov.iov_buf = &p->txhead[0]; + nni_aio_set_iov(p->negoaio, 1, &iov); + nni_ipc_conn_send(p->conn, p->negoaio); + nni_mtx_unlock(&p->ep->mtx); } static void ipctran_pipe_nego_cb(void *arg) { ipctran_pipe *p = arg; - nni_aio * aio = p->negaio; + nni_aio * aio = p->negoaio; + nni_aio * uaio; int rv; - nni_mtx_lock(&p->mtx); + nni_mtx_lock(&p->ep->mtx); if ((rv = nni_aio_result(aio)) != 0) { goto done; } @@ -201,7 +262,7 @@ ipctran_pipe_nego_cb(void *arg) nni_aio_set_iov(aio, 1, &iov); // send it down... nni_ipc_conn_send(p->conn, aio); - nni_mtx_unlock(&p->mtx); + nni_mtx_unlock(&p->ep->mtx); return; } if (p->gotrxhead < p->wantrxhead) { @@ -210,7 +271,7 @@ ipctran_pipe_nego_cb(void *arg) iov.iov_buf = &p->rxhead[p->gotrxhead]; nni_aio_set_iov(aio, 1, &iov); nni_ipc_conn_recv(p->conn, aio); - nni_mtx_unlock(&p->mtx); + nni_mtx_unlock(&p->ep->mtx); return; } // We have both sent and received the headers. Lets check the @@ -225,11 +286,19 @@ ipctran_pipe_nego_cb(void *arg) NNI_GET16(&p->rxhead[4], p->peer); done: - if ((aio = p->user_negaio) != NULL) { - p->user_negaio = NULL; - nni_aio_finish(aio, rv, 0); + + if ((uaio = p->useraio) != NULL) { + p->useraio = NULL; + if (rv == 0) { + nni_aio_set_output(uaio, 0, p); + nni_aio_finish(uaio, 0, 0); + nni_mtx_unlock(&p->ep->mtx); + return; + } + nni_aio_finish_error(uaio, rv); } - nni_mtx_unlock(&p->mtx); + nni_mtx_unlock(&p->ep->mtx); + ipctran_pipe_reap(p); } static void @@ -243,17 +312,20 @@ ipctran_pipe_send_cb(void *arg) nni_aio * txaio = p->txaio; nni_mtx_lock(&p->mtx); - aio = nni_list_first(&p->sendq); - if ((rv = nni_aio_result(txaio)) != 0) { + rv = p->closed ? NNG_ECLOSED : nni_aio_result(txaio); + + if (rv != 0) { // Intentionally we do not queue up another transfer. // There's an excellent chance that the pipe is no longer // usable, with a partial transfer. // The protocol should see this error, and close the // pipe itself, we hope. - nni_aio_list_remove(aio); + while ((aio = nni_list_first(&p->sendq)) != NULL) { + nni_aio_list_remove(aio); + nni_aio_finish_error(aio, rv); + } nni_mtx_unlock(&p->mtx); - nni_aio_finish_error(aio, rv); return; } @@ -265,6 +337,7 @@ ipctran_pipe_send_cb(void *arg) return; } + aio = nni_list_first(&p->sendq); nni_aio_list_remove(aio); ipctran_pipe_send_start(p); @@ -288,7 +361,6 @@ ipctran_pipe_recv_cb(void *arg) nni_aio * rxaio = p->rxaio; nni_mtx_lock(&p->mtx); - aio = nni_list_first(&p->recvq); if ((rv = nni_aio_result(rxaio)) != 0) { // Error on receive. This has to cause an error back @@ -296,6 +368,10 @@ ipctran_pipe_recv_cb(void *arg) // toss it. goto recv_error; } + if (p->closed) { + rv = NNG_ECLOSED; + goto recv_error; + } n = nni_aio_count(rxaio); nni_aio_iov_advance(rxaio, n); @@ -354,6 +430,7 @@ ipctran_pipe_recv_cb(void *arg) // Otherwise we got a message read completely. Let the user know the // good news. + aio = nni_list_first(&p->recvq); nni_aio_list_remove(aio); msg = p->rxmsg; p->rxmsg = NULL; @@ -367,7 +444,10 @@ ipctran_pipe_recv_cb(void *arg) return; recv_error: - nni_aio_list_remove(aio); + while ((aio = nni_list_first(&p->recvq)) != NULL) { + nni_aio_list_remove(aio); + nni_aio_finish_error(aio, rv); + } msg = p->rxmsg; p->rxmsg = NULL; // Intentionally, we do not queue up another receive. @@ -375,7 +455,6 @@ recv_error: nni_mtx_unlock(&p->mtx); nni_msg_free(msg); - nni_aio_finish_error(aio, rv); } static void @@ -452,6 +531,11 @@ ipctran_pipe_send(void *arg, nni_aio *aio) return; } nni_mtx_lock(&p->mtx); + if (p->closed) { + nni_mtx_unlock(&p->mtx); + nni_aio_finish_error(aio, NNG_ECLOSED); + return; + } if ((rv = nni_aio_schedule(aio, ipctran_pipe_send_cancel, p)) != 0) { nni_mtx_unlock(&p->mtx); nni_aio_finish_error(aio, rv); @@ -514,6 +598,11 @@ ipctran_pipe_recv(void *arg, nni_aio *aio) } nni_mtx_lock(&p->mtx); + if (p->closed) { + nni_mtx_unlock(&p->mtx); + nni_aio_finish_error(aio, NNG_ECLOSED); + return; + } if ((rv = nni_aio_schedule(aio, ipctran_pipe_recv_cancel, p)) != 0) { nni_mtx_unlock(&p->mtx); nni_aio_finish_error(aio, rv); @@ -527,43 +616,6 @@ ipctran_pipe_recv(void *arg, nni_aio *aio) nni_mtx_unlock(&p->mtx); } -static void -ipctran_pipe_start(void *arg, nni_aio *aio) -{ - ipctran_pipe *p = arg; - nni_aio * negaio; - nni_iov iov; - int rv; - - if (nni_aio_begin(aio) != 0) { - return; - } - nni_mtx_lock(&p->mtx); - if ((rv = nni_aio_schedule(aio, ipctran_pipe_nego_cancel, p)) != 0) { - nni_mtx_unlock(&p->mtx); - nni_aio_finish_error(aio, rv); - return; - } - p->txhead[0] = 0; - p->txhead[1] = 'S'; - p->txhead[2] = 'P'; - p->txhead[3] = 0; - NNI_PUT16(&p->txhead[4], p->proto); - NNI_PUT16(&p->txhead[6], 0); - - p->user_negaio = aio; - p->gotrxhead = 0; - p->gottxhead = 0; - p->wantrxhead = 8; - p->wanttxhead = 8; - negaio = p->negaio; - iov.iov_len = 8; - iov.iov_buf = &p->txhead[0]; - nni_aio_set_iov(negaio, 1, &iov); - nni_ipc_conn_send(p->conn, negaio); - nni_mtx_unlock(&p->mtx); -} - static uint16_t ipctran_pipe_peer(void *arg) { @@ -632,21 +684,42 @@ ipctran_dialer_fini(void *arg) { ipctran_dialer *d = arg; - nni_aio_stop(d->aio); + nni_mtx_lock(&d->ep.mtx); if (d->dialer != NULL) { nni_ipc_dialer_fini(d->dialer); + d->dialer = NULL; + } + d->ep.fini = true; + if (!nni_list_empty(&d->ep.pipes)) { + nni_mtx_unlock(&d->ep.mtx); + return; } - nni_aio_fini(d->aio); - nni_mtx_fini(&d->mtx); + nni_mtx_unlock(&d->ep.mtx); + + nni_mtx_fini(&d->ep.mtx); NNI_FREE_STRUCT(d); } +static void +ipctran_ep_close(void *arg) +{ + ipctran_ep * ep = arg; + ipctran_pipe *p; + + nni_mtx_lock(&ep->mtx); + NNI_LIST_FOREACH (&ep->pipes, p) { + nni_aio_close(p->negoaio); + nni_aio_close(p->connaio); + } + nni_mtx_unlock(&ep->mtx); +} + static void ipctran_dialer_close(void *arg) { ipctran_dialer *d = arg; - nni_aio_close(d->aio); + ipctran_ep_close(&d->ep); nni_ipc_dialer_close(d->dialer); } @@ -660,90 +733,46 @@ ipctran_dialer_init(void **dp, nni_url *url, nni_sock *sock) if ((d = NNI_ALLOC_STRUCT(d)) == NULL) { return (NNG_ENOMEM); } - nni_mtx_init(&d->mtx); + nni_mtx_init(&d->ep.mtx); + NNI_LIST_INIT(&d->ep.pipes, ipctran_pipe, node); - sz = sizeof(d->sa.s_ipc.sa_path); - d->sa.s_ipc.sa_family = NNG_AF_IPC; + sz = sizeof(d->ep.sa.s_ipc.sa_path); + d->ep.sa.s_ipc.sa_family = NNG_AF_IPC; + d->ep.proto = nni_sock_proto_id(sock); + d->ep.dtor = ipctran_dialer_fini; - if (nni_strlcpy(d->sa.s_ipc.sa_path, url->u_path, sz) >= sz) { + if (nni_strlcpy(d->ep.sa.s_ipc.sa_path, url->u_path, sz) >= sz) { ipctran_dialer_fini(d); return (NNG_EADDRINVAL); } - if (((rv = nni_ipc_dialer_init(&d->dialer)) != 0) || - ((rv = nni_aio_init(&d->aio, ipctran_dialer_cb, d)) != 0)) { + if ((rv = nni_ipc_dialer_init(&d->dialer)) != 0) { ipctran_dialer_fini(d); return (rv); } - d->proto = nni_sock_proto_id(sock); - *dp = d; return (0); } static void -ipctran_dialer_cb(void *arg) +ipctran_pipe_conn_cancel(nni_aio *aio, int rv) { - ipctran_dialer *d = arg; - ipctran_pipe * p; - nni_ipc_conn * conn; - nni_aio * aio; - int rv; - - nni_mtx_lock(&d->mtx); - aio = d->user_aio; - rv = nni_aio_result(d->aio); - - if (aio == NULL) { - nni_mtx_unlock(&d->mtx); - if (rv == 0) { - conn = nni_aio_get_output(d->aio, 0); - nni_ipc_conn_fini(conn); - } - return; - } - - if (rv != 0) { - d->user_aio = NULL; - nni_mtx_unlock(&d->mtx); - nni_aio_finish_error(aio, rv); - return; - } + ipctran_pipe *p = nni_aio_get_prov_data(aio); - d->user_aio = NULL; - conn = nni_aio_get_output(d->aio, 0); - NNI_ASSERT(conn != NULL); - if ((rv = ipctran_pipe_init(&p, conn)) != 0) { - nni_mtx_unlock(&d->mtx); - nni_ipc_conn_fini(conn); - nni_aio_finish_error(aio, rv); + nni_mtx_lock(&p->ep->mtx); + if (p->useraio != aio) { + nni_mtx_unlock(&p->ep->mtx); return; } - p->proto = d->proto; - p->rcvmax = d->rcvmax; - p->sa = d->sa; - nni_mtx_unlock(&d->mtx); - - nni_aio_set_output(aio, 0, p); - nni_aio_finish(aio, 0, 0); -} - -static void -ipctran_dialer_cancel(nni_aio *aio, int rv) -{ - ipctran_dialer *d = nni_aio_get_prov_data(aio); - - nni_mtx_lock(&d->mtx); - if (d->user_aio != aio) { - nni_mtx_unlock(&d->mtx); - return; - } - d->user_aio = NULL; - nni_mtx_unlock(&d->mtx); + // Close the underlying AIOs. This will abort the operation. + // The pipe is removed from pending list at completion callback. + p->useraio = NULL; + nni_aio_close(p->negoaio); + nni_aio_close(p->connaio); + nni_mtx_unlock(&p->ep->mtx); - nni_aio_abort(d->aio, rv); nni_aio_finish_error(aio, rv); } @@ -751,47 +780,48 @@ static void ipctran_dialer_connect(void *arg, nni_aio *aio) { ipctran_dialer *d = arg; + ipctran_pipe * p = NULL; int rv; if (nni_aio_begin(aio) != 0) { return; } - nni_mtx_lock(&d->mtx); - NNI_ASSERT(d->user_aio == NULL); - if ((rv = nni_aio_schedule(aio, ipctran_dialer_cancel, d)) != 0) { - nni_mtx_unlock(&d->mtx); + nni_mtx_lock(&d->ep.mtx); + if (((rv = ipctran_pipe_init(&p, &d->ep)) != 0) || + ((rv = nni_aio_schedule(aio, ipctran_pipe_conn_cancel, p)) != 0)) { + nni_mtx_unlock(&d->ep.mtx); nni_aio_finish_error(aio, rv); + ipctran_pipe_reap(p); return; } - d->user_aio = aio; + p->useraio = aio; - nni_ipc_dialer_dial(d->dialer, &d->sa, d->aio); - nni_mtx_unlock(&d->mtx); + nni_ipc_dialer_dial(d->dialer, &p->sa, p->connaio); + nni_mtx_unlock(&d->ep.mtx); } static int -ipctran_dialer_get_recvmaxsz(void *arg, void *v, size_t *szp, nni_opt_type t) +ipctran_ep_get_recvmaxsz(void *arg, void *v, size_t *szp, nni_opt_type t) { - ipctran_dialer *d = arg; - int rv; - nni_mtx_lock(&d->mtx); - rv = nni_copyout_size(d->rcvmax, v, szp, t); - nni_mtx_unlock(&d->mtx); + ipctran_ep *ep = arg; + int rv; + nni_mtx_lock(&ep->mtx); + rv = nni_copyout_size(ep->rcvmax, v, szp, t); + nni_mtx_unlock(&ep->mtx); return (rv); } static int -ipctran_dialer_set_recvmaxsz( - void *arg, const void *v, size_t sz, nni_opt_type t) +ipctran_ep_set_recvmaxsz(void *arg, const void *v, size_t sz, nni_opt_type t) { - ipctran_dialer *d = arg; - size_t val; - int rv; + ipctran_ep *ep = arg; + size_t val; + int rv; if ((rv = nni_copyin_size(&val, v, sz, 0, NNI_MAXSZ, t)) == 0) { - nni_mtx_lock(&d->mtx); - d->rcvmax = val; - nni_mtx_unlock(&d->mtx); + nni_mtx_lock(&ep->mtx); + ep->rcvmax = val; + nni_mtx_unlock(&ep->mtx); } return (rv); } @@ -801,12 +831,19 @@ ipctran_listener_fini(void *arg) { ipctran_listener *l = arg; - nni_aio_stop(l->aio); + nni_mtx_lock(&l->ep.mtx); if (l->listener != NULL) { nni_ipc_listener_fini(l->listener); + l->listener = NULL; + } + l->ep.fini = true; + if (!nni_list_empty(&l->ep.pipes)) { + nni_mtx_unlock(&l->ep.mtx); + return; } - nni_aio_fini(l->aio); - nni_mtx_fini(&l->mtx); + nni_mtx_unlock(&l->ep.mtx); + + nni_mtx_fini(&l->ep.mtx); NNI_FREE_STRUCT(l); } @@ -820,24 +857,24 @@ ipctran_listener_init(void **lp, nni_url *url, nni_sock *sock) if ((l = NNI_ALLOC_STRUCT(l)) == NULL) { return (NNG_ENOMEM); } - nni_mtx_init(&l->mtx); + nni_mtx_init(&l->ep.mtx); + NNI_LIST_INIT(&l->ep.pipes, ipctran_pipe, node); - sz = sizeof(l->sa.s_ipc.sa_path); - l->sa.s_ipc.sa_family = NNG_AF_IPC; + sz = sizeof(l->ep.sa.s_ipc.sa_path); + l->ep.sa.s_ipc.sa_family = NNG_AF_IPC; + l->ep.proto = nni_sock_proto_id(sock); + l->ep.dtor = ipctran_listener_fini; - if (nni_strlcpy(l->sa.s_ipc.sa_path, url->u_path, sz) >= sz) { + if (nni_strlcpy(l->ep.sa.s_ipc.sa_path, url->u_path, sz) >= sz) { ipctran_listener_fini(l); return (NNG_EADDRINVAL); } - if (((rv = nni_ipc_listener_init(&l->listener)) != 0) || - ((rv = nni_aio_init(&l->aio, ipctran_listener_cb, l)) != 0)) { + if ((rv = nni_ipc_listener_init(&l->listener)) != 0) { ipctran_listener_fini(l); return (rv); } - l->proto = nni_sock_proto_id(sock); - *lp = l; return (0); } @@ -847,7 +884,7 @@ ipctran_listener_close(void *arg) { ipctran_listener *l = arg; - nni_aio_close(l->aio); + ipctran_ep_close(&l->ep); nni_ipc_listener_close(l->listener); } @@ -857,128 +894,35 @@ ipctran_listener_bind(void *arg) ipctran_listener *l = arg; int rv; - nni_mtx_lock(&l->mtx); - rv = nni_ipc_listener_listen(l->listener, &l->sa); - nni_mtx_unlock(&l->mtx); + nni_mtx_lock(&l->ep.mtx); + rv = nni_ipc_listener_listen(l->listener, &l->ep.sa); + nni_mtx_unlock(&l->ep.mtx); return (rv); } -static void -ipctran_listener_cb(void *arg) -{ - ipctran_listener *l = arg; - nni_aio * aio; - int rv; - ipctran_pipe * p = NULL; - nni_ipc_conn * conn; - - nni_mtx_lock(&l->mtx); - rv = nni_aio_result(l->aio); - aio = l->user_aio; - l->user_aio = NULL; - - if (aio == NULL) { - nni_mtx_unlock(&l->mtx); - if (rv == 0) { - conn = nni_aio_get_output(l->aio, 0); - nni_ipc_conn_fini(conn); - } - return; - } - - if (rv != 0) { - nni_mtx_unlock(&l->mtx); - nni_aio_finish_error(aio, rv); - return; - } - - conn = nni_aio_get_output(l->aio, 0); - NNI_ASSERT(conn != NULL); - - // Attempt to allocate the parent pipe. If this fails we'll - // drop the connection (ENOMEM probably). - if ((rv = ipctran_pipe_init(&p, conn)) != 0) { - nni_mtx_unlock(&l->mtx); - nni_ipc_conn_fini(conn); - nni_aio_finish_error(aio, rv); - return; - } - p->proto = l->proto; - p->rcvmax = l->rcvmax; - p->sa = l->sa; - nni_mtx_unlock(&l->mtx); - - nni_aio_set_output(aio, 0, p); - nni_aio_finish(aio, 0, 0); -} - -static void -ipctran_listener_cancel(nni_aio *aio, int rv) -{ - ipctran_listener *l = nni_aio_get_prov_data(aio); - - NNI_ASSERT(rv != 0); - nni_mtx_lock(&l->mtx); - if (l->user_aio != aio) { - nni_mtx_unlock(&l->mtx); - return; - } - l->user_aio = NULL; - nni_mtx_unlock(&l->mtx); - - nni_aio_abort(l->aio, rv); - nni_aio_finish_error(aio, rv); -} - static void ipctran_listener_accept(void *arg, nni_aio *aio) { ipctran_listener *l = arg; + ipctran_pipe * p; int rv; if (nni_aio_begin(aio) != 0) { return; } - nni_mtx_lock(&l->mtx); - NNI_ASSERT(l->user_aio == NULL); - if ((rv = nni_aio_schedule(aio, ipctran_listener_cancel, l)) != 0) { - nni_mtx_unlock(&l->mtx); + nni_mtx_lock(&l->ep.mtx); + if (((rv = ipctran_pipe_init(&p, &l->ep)) != 0) || + ((rv = nni_aio_schedule(aio, ipctran_pipe_conn_cancel, p)) != 0)) { + nni_mtx_unlock(&l->ep.mtx); nni_aio_finish_error(aio, rv); + ipctran_pipe_reap(p); return; } - l->user_aio = aio; - - nni_ipc_listener_accept(l->listener, l->aio); - nni_mtx_unlock(&l->mtx); -} - -static int -ipctran_listener_set_recvmaxsz( - void *arg, const void *data, size_t sz, nni_opt_type t) -{ - ipctran_listener *l = arg; - size_t val; - int rv; - - if ((rv = nni_copyin_size(&val, data, sz, 0, NNI_MAXSZ, t)) == 0) { - nni_mtx_lock(&l->mtx); - l->rcvmax = val; - nni_mtx_unlock(&l->mtx); - } - return (rv); -} + p->useraio = aio; -static int -ipctran_listener_get_recvmaxsz( - void *arg, void *data, size_t *szp, nni_opt_type t) -{ - ipctran_listener *l = arg; - int rv; - nni_mtx_lock(&l->mtx); - rv = nni_copyout_size(l->rcvmax, data, szp, t); - nni_mtx_unlock(&l->mtx); - return (rv); + nni_ipc_listener_accept(l->listener, p->connaio); + nni_mtx_unlock(&l->ep.mtx); } static int @@ -987,9 +931,9 @@ ipctran_listener_get_locaddr(void *arg, void *buf, size_t *szp, nni_opt_type t) ipctran_listener *l = arg; int rv; - nni_mtx_lock(&l->mtx); - rv = nni_copyout_sockaddr(&l->sa, buf, szp, t); - nni_mtx_unlock(&l->mtx); + nni_mtx_lock(&l->ep.mtx); + rv = nni_copyout_sockaddr(&l->ep.sa, buf, szp, t); + nni_mtx_unlock(&l->ep.mtx); return (rv); } @@ -1010,9 +954,9 @@ ipctran_listener_set_perms( // Probably we could further limit this -- most systems don't have // meaningful chmod beyond the lower 9 bits. if ((rv = nni_copyin_int(&val, data, sz, 0, 0x7FFFFFFF, t)) == 0) { - nni_mtx_lock(&l->mtx); + nni_mtx_lock(&l->ep.mtx); rv = nni_ipc_listener_set_permissions(l->listener, val); - nni_mtx_unlock(&l->mtx); + nni_mtx_unlock(&l->ep.mtx); } return (rv); } @@ -1032,10 +976,10 @@ ipctran_listener_set_sec_desc( int rv; if ((rv = nni_copyin_ptr(&ptr, data, sz, t)) == 0) { - nni_mtx_lock(&l->mtx); + nni_mtx_lock(&l->ep.mtx); rv = nni_ipc_listener_set_security_descriptor(l->listener, ptr); - nni_mtx_unlock(&l->mtx); + nni_mtx_unlock(&l->ep.mtx); } return (rv); } @@ -1085,7 +1029,6 @@ static nni_tran_option ipctran_pipe_options[] = { static nni_tran_pipe_ops ipctran_pipe_ops = { .p_fini = ipctran_pipe_fini, - .p_start = ipctran_pipe_start, .p_stop = ipctran_pipe_stop, .p_send = ipctran_pipe_send, .p_recv = ipctran_pipe_recv, @@ -1098,8 +1041,8 @@ static nni_tran_option ipctran_dialer_options[] = { { .o_name = NNG_OPT_RECVMAXSZ, .o_type = NNI_TYPE_SIZE, - .o_get = ipctran_dialer_get_recvmaxsz, - .o_set = ipctran_dialer_set_recvmaxsz, + .o_get = ipctran_ep_get_recvmaxsz, + .o_set = ipctran_ep_set_recvmaxsz, .o_chk = ipctran_check_recvmaxsz, }, // terminate list @@ -1112,8 +1055,8 @@ static nni_tran_option ipctran_listener_options[] = { { .o_name = NNG_OPT_RECVMAXSZ, .o_type = NNI_TYPE_SIZE, - .o_get = ipctran_listener_get_recvmaxsz, - .o_set = ipctran_listener_set_recvmaxsz, + .o_get = ipctran_ep_get_recvmaxsz, + .o_set = ipctran_ep_set_recvmaxsz, .o_chk = ipctran_check_recvmaxsz, }, { diff --git a/src/transport/tcp/tcp.c b/src/transport/tcp/tcp.c index 0d47529e..6cf69972 100644 --- a/src/transport/tcp/tcp.c +++ b/src/transport/tcp/tcp.c @@ -20,19 +20,26 @@ typedef struct tcptran_pipe tcptran_pipe; typedef struct tcptran_dialer tcptran_dialer; typedef struct tcptran_listener tcptran_listener; +typedef struct tcptran_ep tcptran_ep; // tcp_pipe is one end of a TCP connection. struct tcptran_pipe { - nni_tcp_conn *conn; - uint16_t peer; - uint16_t proto; - size_t rcvmax; - bool nodelay; - bool keepalive; + nni_tcp_conn * conn; + uint16_t peer; + uint16_t proto; + size_t rcvmax; + nni_sockaddr sa; + bool nodelay; + bool keepalive; + bool closed; + nni_atomic_flag reaped; + nni_list_node node; + nni_reap_item reap; + tcptran_ep * ep; nni_list recvq; nni_list sendq; - nni_aio *user_negaio; + nni_aio *useraio; uint8_t txlen[sizeof(uint64_t)]; uint8_t rxlen[sizeof(uint64_t)]; @@ -42,47 +49,48 @@ struct tcptran_pipe { size_t wantrxhead; nni_aio *txaio; nni_aio *rxaio; - nni_aio *negaio; + nni_aio *negoaio; + nni_aio *connaio; + nni_aio *rslvaio; nni_msg *rxmsg; nni_mtx mtx; }; +struct tcptran_ep { + nni_mtx mtx; + nni_list pipes; + nni_url * url; + size_t rcvmax; + bool nodelay; + bool keepalive; + bool fini; + uint16_t proto; + nni_cb dtor; + nni_reap_item reap; +}; + struct tcptran_dialer { + tcptran_ep ep; nni_tcp_dialer *dialer; - uint16_t proto; - uint16_t af; - size_t rcvmax; - bool nodelay; - bool keepalive; - bool resolving; nng_sockaddr sa; - nni_aio * aio; - nni_aio * user_aio; - nni_url * url; - nni_mtx mtx; + uint16_t af; }; struct tcptran_listener { + tcptran_ep ep; nni_tcp_listener *listener; - uint16_t proto; - size_t rcvmax; - bool nodelay; - bool keepalive; - nni_aio * aio; - nni_aio * user_aio; - nni_url * url; nng_sockaddr sa; nng_sockaddr bsa; // bound addr - nni_mtx mtx; }; static void tcptran_pipe_send_start(tcptran_pipe *); static void tcptran_pipe_recv_start(tcptran_pipe *); static void tcptran_pipe_send_cb(void *); static void tcptran_pipe_recv_cb(void *); +static void tcptran_pipe_rslv_cb(void *); static void tcptran_pipe_nego_cb(void *); -static void tcptran_dialer_cb(void *arg); -static void tcptran_listener_cb(void *arg); +static void tcptran_pipe_conn_cb(void *); +static void tcptran_pipe_reap(tcptran_pipe *); static int tcptran_init(void) @@ -100,9 +108,15 @@ tcptran_pipe_close(void *arg) { tcptran_pipe *p = arg; + nni_mtx_lock(&p->mtx); + p->closed = true; + nni_mtx_unlock(&p->mtx); + nni_aio_close(p->rxaio); nni_aio_close(p->txaio); - nni_aio_close(p->negaio); + nni_aio_close(p->negoaio); + nni_aio_close(p->connaio); + nni_aio_close(p->rslvaio); nni_tcp_conn_close(p->conn); } @@ -114,17 +128,34 @@ tcptran_pipe_stop(void *arg) nni_aio_stop(p->rxaio); nni_aio_stop(p->txaio); - nni_aio_stop(p->negaio); + nni_aio_stop(p->negoaio); + nni_aio_stop(p->connaio); + nni_aio_stop(p->rslvaio); } static void tcptran_pipe_fini(void *arg) { tcptran_pipe *p = arg; + tcptran_ep * ep; + if (p == NULL) { + return; + } + tcptran_pipe_stop(p); + if ((ep = p->ep) != NULL) { + nni_mtx_lock(&ep->mtx); + nni_list_remove(&ep->pipes, p); + if (ep->fini && nni_list_empty(&ep->pipes)) { + nni_reap(&ep->reap, ep->dtor, ep); + } + nni_mtx_unlock(&ep->mtx); + } nni_aio_fini(p->rxaio); nni_aio_fini(p->txaio); - nni_aio_fini(p->negaio); + nni_aio_fini(p->negoaio); + nni_aio_fini(p->connaio); + nni_aio_fini(p->rslvaio); if (p->conn != NULL) { nni_tcp_conn_fini(p->conn); } @@ -133,8 +164,16 @@ tcptran_pipe_fini(void *arg) NNI_FREE_STRUCT(p); } +static void +tcptran_pipe_reap(tcptran_pipe *p) +{ + if (!nni_atomic_flag_test_and_set(&p->reaped)) { + nni_reap(&p->reap, tcptran_pipe_fini, p); + } +} + static int -tcptran_pipe_init(tcptran_pipe **pipep, void *conn) +tcptran_pipe_init(tcptran_pipe **pipep, tcptran_ep *ep) { tcptran_pipe *p; int rv; @@ -143,45 +182,108 @@ tcptran_pipe_init(tcptran_pipe **pipep, void *conn) return (NNG_ENOMEM); } nni_mtx_init(&p->mtx); + NNI_LIST_NODE_INIT(&p->node); + nni_aio_list_init(&p->recvq); + nni_aio_list_init(&p->sendq); + nni_atomic_flag_reset(&p->reaped); + if (((rv = nni_aio_init(&p->txaio, tcptran_pipe_send_cb, p)) != 0) || ((rv = nni_aio_init(&p->rxaio, tcptran_pipe_recv_cb, p)) != 0) || - ((rv = nni_aio_init(&p->negaio, tcptran_pipe_nego_cb, p)) != 0)) { - tcptran_pipe_fini(p); + ((rv = nni_aio_init(&p->rslvaio, tcptran_pipe_rslv_cb, p)) != 0) || + ((rv = nni_aio_init(&p->connaio, tcptran_pipe_conn_cb, p)) != 0) || + ((rv = nni_aio_init(&p->negoaio, tcptran_pipe_nego_cb, p)) != 0)) { + tcptran_pipe_reap(p); return (rv); } - nni_aio_list_init(&p->recvq); - nni_aio_list_init(&p->sendq); + nni_list_append(&ep->pipes, p); + p->ep = ep; + p->proto = ep->proto; + p->rcvmax = ep->rcvmax; + p->keepalive = ep->keepalive; + p->nodelay = ep->nodelay; - p->conn = conn; - *pipep = p; + *pipep = p; return (0); } static void -tcptran_pipe_nego_cancel(nni_aio *aio, int rv) +tcptran_pipe_rslv_cb(void *arg) { - tcptran_pipe *p = nni_aio_get_prov_data(aio); + tcptran_pipe * p = arg; + tcptran_dialer *d; + nni_aio * aio = p->rslvaio; + int rv; - nni_mtx_lock(&p->mtx); - if (p->user_negaio != aio) { - nni_mtx_unlock(&p->mtx); + nni_mtx_lock(&p->ep->mtx); + if ((rv = nni_aio_result(aio)) != 0) { + nni_aio *uaio; + if ((uaio = p->useraio) != NULL) { + p->useraio = NULL; + nni_aio_finish_error(uaio, rv); + } + nni_mtx_unlock(&p->ep->mtx); + tcptran_pipe_reap(p); return; } - p->user_negaio = NULL; - nni_mtx_unlock(&p->mtx); + d = (void *) p->ep; + if (d->dialer != NULL) { + nni_tcp_dialer_dial(d->dialer, &p->sa, p->connaio); + } + nni_mtx_unlock(&p->ep->mtx); +} - nni_aio_abort(p->negaio, rv); - nni_aio_finish_error(aio, rv); +static void +tcptran_pipe_conn_cb(void *arg) +{ + tcptran_pipe *p = arg; + nni_aio * aio = p->connaio; + nni_iov iov; + int rv; + + nni_mtx_lock(&p->ep->mtx); + if ((rv = nni_aio_result(aio)) != 0) { + nni_aio *uaio; + if ((uaio = p->useraio) != NULL) { + p->useraio = NULL; + nni_aio_finish_error(uaio, rv); + } + nni_mtx_unlock(&p->ep->mtx); + tcptran_pipe_reap(p); + return; + } + + p->conn = nni_aio_get_output(aio, 0); + + (void) nni_tcp_conn_set_nodelay(p->conn, p->nodelay); + (void) nni_tcp_conn_set_keepalive(p->conn, p->keepalive); + + p->txlen[0] = 0; + p->txlen[1] = 'S'; + p->txlen[2] = 'P'; + p->txlen[3] = 0; + NNI_PUT16(&p->txlen[4], p->proto); + NNI_PUT16(&p->txlen[6], 0); + + p->gotrxhead = 0; + p->gottxhead = 0; + p->wantrxhead = 8; + p->wanttxhead = 8; + iov.iov_len = 8; + iov.iov_buf = &p->txlen[0]; + nni_aio_set_iov(p->negoaio, 1, &iov); + nni_tcp_conn_send(p->conn, p->negoaio); + nni_mtx_unlock(&p->ep->mtx); } static void tcptran_pipe_nego_cb(void *arg) { tcptran_pipe *p = arg; - nni_aio * aio = p->negaio; + nni_aio * aio = p->negoaio; + nni_aio * uaio; int rv; - nni_mtx_lock(&p->mtx); + nni_mtx_lock(&p->ep->mtx); if ((rv = nni_aio_result(aio)) != 0) { goto done; } @@ -200,7 +302,7 @@ tcptran_pipe_nego_cb(void *arg) // send it down... nni_aio_set_iov(aio, 1, &iov); nni_tcp_conn_send(p->conn, aio); - nni_mtx_unlock(&p->mtx); + nni_mtx_unlock(&p->ep->mtx); return; } if (p->gotrxhead < p->wantrxhead) { @@ -209,7 +311,7 @@ tcptran_pipe_nego_cb(void *arg) iov.iov_buf = &p->rxlen[p->gotrxhead]; nni_aio_set_iov(aio, 1, &iov); nni_tcp_conn_recv(p->conn, aio); - nni_mtx_unlock(&p->mtx); + nni_mtx_unlock(&p->ep->mtx); return; } // We have both sent and received the headers. Lets check the @@ -224,11 +326,18 @@ tcptran_pipe_nego_cb(void *arg) NNI_GET16(&p->rxlen[4], p->peer); done: - if ((aio = p->user_negaio) != NULL) { - p->user_negaio = NULL; - nni_aio_finish(aio, rv, 0); + if ((uaio = p->useraio) != NULL) { + p->useraio = NULL; + if (rv == 0) { + nni_aio_set_output(uaio, 0, p); + nni_aio_finish(uaio, 0, 0); + nni_mtx_unlock(&p->ep->mtx); + return; + } + nni_aio_finish_error(uaio, rv); } - nni_mtx_unlock(&p->mtx); + nni_mtx_unlock(&p->ep->mtx); + tcptran_pipe_reap(p); } static void @@ -242,17 +351,19 @@ tcptran_pipe_send_cb(void *arg) nni_aio * txaio = p->txaio; nni_mtx_lock(&p->mtx); - aio = nni_list_first(&p->sendq); + rv = p->closed ? NNG_ECLOSED : nni_aio_result(txaio); - if ((rv = nni_aio_result(txaio)) != 0) { + if (rv != 0) { // Intentionally we do not queue up another transfer. // There's an excellent chance that the pipe is no longer // usable, with a partial transfer. // The protocol should see this error, and close the // pipe itself, we hope. - nni_aio_list_remove(aio); + while ((aio = nni_list_first(&p->sendq)) != NULL) { + nni_aio_list_remove(aio); + nni_aio_finish_error(aio, rv); + } nni_mtx_unlock(&p->mtx); - nni_aio_finish_error(aio, rv); return; } @@ -264,6 +375,7 @@ tcptran_pipe_send_cb(void *arg) return; } + aio = nni_list_first(&p->sendq); nni_aio_list_remove(aio); tcptran_pipe_send_start(p); @@ -287,11 +399,14 @@ tcptran_pipe_recv_cb(void *arg) nni_aio * rxaio = p->rxaio; nni_mtx_lock(&p->mtx); - aio = nni_list_first(&p->recvq); if ((rv = nni_aio_result(rxaio)) != 0) { goto recv_error; } + if (p->closed) { + rv = NNG_ECLOSED; + goto recv_error; + } n = nni_aio_count(rxaio); nni_aio_iov_advance(rxaio, n); @@ -335,6 +450,7 @@ tcptran_pipe_recv_cb(void *arg) } // We read a message completely. Let the user know the good news. + aio = nni_list_first(&p->recvq); nni_aio_list_remove(aio); msg = p->rxmsg; p->rxmsg = NULL; @@ -348,7 +464,10 @@ tcptran_pipe_recv_cb(void *arg) return; recv_error: - nni_aio_list_remove(aio); + while ((aio = nni_list_first(&p->recvq)) != NULL) { + nni_aio_list_remove(aio); + nni_aio_finish_error(aio, rv); + } msg = p->rxmsg; p->rxmsg = NULL; // Intentionally, we do not queue up another receive. @@ -356,7 +475,6 @@ recv_error: nni_mtx_unlock(&p->mtx); nni_msg_free(msg); - nni_aio_finish_error(aio, rv); } static void @@ -432,6 +550,11 @@ tcptran_pipe_send(void *arg, nni_aio *aio) return; } nni_mtx_lock(&p->mtx); + if (p->closed) { + nni_mtx_unlock(&p->mtx); + nni_aio_finish_error(aio, NNG_ECLOSED); + return; + } if ((rv = nni_aio_schedule(aio, tcptran_pipe_send_cancel, p)) != 0) { nni_mtx_unlock(&p->mtx); nni_aio_finish_error(aio, rv); @@ -474,7 +597,7 @@ tcptran_pipe_recv_start(tcptran_pipe *p) nni_iov iov; NNI_ASSERT(p->rxmsg == NULL); - // Schedule a read of the IPC header. + // Schedule a read of the header. rxaio = p->rxaio; iov.iov_buf = p->rxlen; iov.iov_len = sizeof(p->rxlen); @@ -493,6 +616,11 @@ tcptran_pipe_recv(void *arg, nni_aio *aio) return; } nni_mtx_lock(&p->mtx); + if (p->closed) { + nni_mtx_unlock(&p->mtx); + nni_aio_finish_error(aio, NNG_ECLOSED); + return; + } if ((rv = nni_aio_schedule(aio, tcptran_pipe_recv_cancel, p)) != 0) { nni_mtx_unlock(&p->mtx); nni_aio_finish_error(aio, rv); @@ -557,54 +685,36 @@ tcptran_pipe_get_nodelay(void *arg, void *v, size_t *szp, nni_opt_type t) } static void -tcptran_pipe_start(void *arg, nni_aio *aio) +tcptran_dialer_fini(void *arg) { - tcptran_pipe *p = arg; - nni_aio * negaio; - nni_iov iov; - int rv; + tcptran_dialer *d = arg; - if (nni_aio_begin(aio) != 0) { - return; + nni_mtx_lock(&d->ep.mtx); + if (d->dialer != NULL) { + nni_tcp_dialer_fini(d->dialer); + d->dialer = NULL; } - nni_mtx_lock(&p->mtx); - if ((rv = nni_aio_schedule(aio, tcptran_pipe_nego_cancel, p)) != 0) { - nni_mtx_unlock(&p->mtx); - nni_aio_finish_error(aio, rv); + d->ep.fini = true; + if (!nni_list_empty(&d->ep.pipes)) { + nni_mtx_unlock(&d->ep.mtx); return; } - p->txlen[0] = 0; - p->txlen[1] = 'S'; - p->txlen[2] = 'P'; - p->txlen[3] = 0; - NNI_PUT16(&p->txlen[4], p->proto); - NNI_PUT16(&p->txlen[6], 0); - - p->user_negaio = aio; - p->gotrxhead = 0; - p->gottxhead = 0; - p->wantrxhead = 8; - p->wanttxhead = 8; - negaio = p->negaio; - iov.iov_len = 8; - iov.iov_buf = &p->txlen[0]; - nni_aio_set_iov(negaio, 1, &iov); - nni_tcp_conn_send(p->conn, negaio); - nni_mtx_unlock(&p->mtx); + nni_mtx_unlock(&d->ep.mtx); + nni_mtx_fini(&d->ep.mtx); + NNI_FREE_STRUCT(d); } static void -tcptran_dialer_fini(void *arg) +tcptran_ep_close(tcptran_ep *ep) { - tcptran_dialer *d = arg; - - nni_aio_stop(d->aio); - if (d->dialer != NULL) { - nni_tcp_dialer_fini(d->dialer); + tcptran_pipe *p; + nni_mtx_lock(&ep->mtx); + NNI_LIST_FOREACH (&ep->pipes, p) { + nni_aio_close(p->negoaio); + nni_aio_close(p->connaio); + nni_aio_close(p->rslvaio); } - nni_aio_fini(d->aio); - nni_mtx_fini(&d->mtx); - NNI_FREE_STRUCT(d); + nni_mtx_unlock(&ep->mtx); } static void @@ -612,7 +722,7 @@ tcptran_dialer_close(void *arg) { tcptran_dialer *d = arg; - nni_aio_close(d->aio); + tcptran_ep_close(&d->ep); nni_tcp_dialer_close(d->dialer); } @@ -646,127 +756,72 @@ tcptran_dialer_init(void **dp, nni_url *url, nni_sock *sock) if ((d = NNI_ALLOC_STRUCT(d)) == NULL) { return (NNG_ENOMEM); } - nni_mtx_init(&d->mtx); + nni_mtx_init(&d->ep.mtx); + NNI_LIST_INIT(&d->ep.pipes, tcptran_pipe, node); + d->ep.dtor = tcptran_dialer_fini; + d->ep.nodelay = true; + d->ep.keepalive = false; + d->ep.proto = nni_sock_proto_id(sock); + d->ep.url = url; + d->af = af; - if (((rv = nni_tcp_dialer_init(&d->dialer)) != 0) || - ((rv = nni_aio_init(&d->aio, tcptran_dialer_cb, d)) != 0)) { + if ((rv = nni_tcp_dialer_init(&d->dialer)) != 0) { tcptran_dialer_fini(d); return (rv); } - d->url = url; - d->proto = nni_sock_proto_id(sock); - d->nodelay = true; - d->keepalive = false; - d->af = af; - *dp = d; return (0); } static void -tcptran_dialer_cb(void *arg) +tcptran_pipe_conn_cancel(nni_aio *aio, int rv) { - tcptran_dialer *d = arg; - tcptran_pipe * p; - nni_tcp_conn * conn; - nni_aio * aio; - int rv; - - nni_mtx_lock(&d->mtx); - aio = d->user_aio; - rv = nni_aio_result(d->aio); - - if (aio == NULL) { - nni_mtx_unlock(&d->mtx); - if ((rv == 0) && !d->resolving) { - conn = nni_aio_get_output(d->aio, 0); - nni_tcp_conn_fini(conn); - } - return; - } - - if (rv != 0) { - d->user_aio = NULL; - nni_mtx_unlock(&d->mtx); - nni_aio_finish_error(aio, rv); - return; - } - - if (d->resolving) { - // Name resolution complete. Now go to next step. - d->resolving = false; - nni_tcp_dialer_dial(d->dialer, &d->sa, d->aio); - nni_mtx_unlock(&d->mtx); - return; - } + tcptran_pipe *p = nni_aio_get_prov_data(aio); - d->user_aio = NULL; - conn = nni_aio_get_output(d->aio, 0); - NNI_ASSERT(conn != NULL); - if ((rv = tcptran_pipe_init(&p, conn)) != 0) { - nni_mtx_unlock(&d->mtx); - nni_tcp_conn_fini(conn); - nni_aio_finish_error(aio, rv); + nni_mtx_lock(&p->ep->mtx); + if (p->useraio != aio) { + nni_mtx_unlock(&p->ep->mtx); return; } - p->proto = d->proto; - p->rcvmax = d->rcvmax; - p->nodelay = d->nodelay; - p->keepalive = d->keepalive; - nni_mtx_unlock(&d->mtx); + // Close the underlying AIOs. This will abort the operation. + // The pipe is removed from pending list at completion callback. + p->useraio = NULL; + nni_aio_close(p->negoaio); + nni_aio_close(p->connaio); + nni_aio_close(p->rslvaio); + nni_mtx_unlock(&p->ep->mtx); - (void) nni_tcp_conn_set_nodelay(conn, p->nodelay); - (void) nni_tcp_conn_set_keepalive(conn, p->keepalive); - - nni_aio_set_output(aio, 0, p); - nni_aio_finish(aio, 0, 0); -} - -static void -tcptran_dialer_cancel(nni_aio *aio, int rv) -{ - tcptran_dialer *d = nni_aio_get_prov_data(aio); - - nni_mtx_lock(&d->mtx); - if (d->user_aio != aio) { - nni_mtx_unlock(&d->mtx); - return; - } - d->user_aio = NULL; - nni_mtx_unlock(&d->mtx); - - nni_aio_abort(d->aio, rv); nni_aio_finish_error(aio, rv); } static void tcptran_dialer_connect(void *arg, nni_aio *aio) { - tcptran_dialer *d = arg; + tcptran_dialer *d = arg; + tcptran_ep * ep = &d->ep; + tcptran_pipe * p = NULL; int rv; if (nni_aio_begin(aio) != 0) { return; } - nni_mtx_lock(&d->mtx); - NNI_ASSERT(d->user_aio == NULL); - - if ((rv = nni_aio_schedule(aio, tcptran_dialer_cancel, d)) != 0) { - nni_mtx_unlock(&d->mtx); + nni_mtx_lock(&ep->mtx); + if (((rv = tcptran_pipe_init(&p, ep)) != 0) || + ((rv = nni_aio_schedule(aio, tcptran_pipe_conn_cancel, p)) != 0)) { + nni_mtx_unlock(&ep->mtx); nni_aio_finish_error(aio, rv); + tcptran_pipe_reap(p); return; } - d->user_aio = aio; - - d->resolving = true; + p->useraio = aio; - // Start the name resolution. Callback will see resolving, and then - // switch to doing actual connect. - nni_aio_set_input(d->aio, 0, &d->sa); - nni_tcp_resolv(d->url->u_hostname, d->url->u_port, d->af, 0, d->aio); - nni_mtx_unlock(&d->mtx); + // Start the name resolution as first step. + nni_aio_set_input(p->rslvaio, 0, &p->sa); + nni_tcp_resolv( + ep->url->u_hostname, ep->url->u_port, d->af, 0, p->rslvaio); + nni_mtx_unlock(&ep->mtx); } static int @@ -774,79 +829,81 @@ tcptran_dialer_get_url(void *arg, void *v, size_t *szp, nni_opt_type t) { tcptran_dialer *d = arg; - return (nni_copyout_str(d->url->u_rawurl, v, szp, t)); + return (nni_copyout_str(d->ep.url->u_rawurl, v, szp, t)); } static int -tcptran_dialer_get_recvmaxsz(void *arg, void *v, size_t *szp, nni_opt_type t) +tcptran_ep_get_recvmaxsz(void *arg, void *v, size_t *szp, nni_opt_type t) { - tcptran_dialer *d = arg; - int rv; + tcptran_ep *ep = arg; + int rv; - nni_mtx_lock(&d->mtx); - rv = nni_copyout_size(d->rcvmax, v, szp, t); - nni_mtx_unlock(&d->mtx); + nni_mtx_lock(&ep->mtx); + rv = nni_copyout_size(ep->rcvmax, v, szp, t); + nni_mtx_unlock(&ep->mtx); return (rv); } static int -tcptran_dialer_set_recvmaxsz( - void *arg, const void *v, size_t sz, nni_opt_type t) +tcptran_ep_set_recvmaxsz(void *arg, const void *v, size_t sz, nni_opt_type t) { - tcptran_dialer *d = arg; - size_t val; - int rv; + tcptran_ep *ep = arg; + size_t val; + int rv; if ((rv = nni_copyin_size(&val, v, sz, 0, NNI_MAXSZ, t)) == 0) { - nni_mtx_lock(&d->mtx); - d->rcvmax = val; - nni_mtx_unlock(&d->mtx); + nni_mtx_lock(&ep->mtx); + ep->rcvmax = val; + nni_mtx_unlock(&ep->mtx); } return (rv); } static int -tcptran_dialer_get_nodelay(void *arg, void *v, size_t *szp, nni_opt_type t) +tcptran_ep_get_nodelay(void *arg, void *v, size_t *szp, nni_opt_type t) { - tcptran_dialer *d = arg; - int rv; - nni_mtx_lock(&d->mtx); - rv = nni_copyout_bool(d->nodelay, v, szp, t); - nni_mtx_unlock(&d->mtx); + tcptran_ep *ep = arg; + int rv; + nni_mtx_lock(&ep->mtx); + rv = nni_copyout_bool(ep->nodelay, v, szp, t); + nni_mtx_unlock(&ep->mtx); return (rv); } static int -tcptran_dialer_set_nodelay(void *arg, const void *v, size_t sz, nni_opt_type t) +tcptran_ep_set_nodelay(void *arg, const void *v, size_t sz, nni_opt_type t) { - tcptran_dialer *d = arg; - bool val; - int rv; + tcptran_ep *ep = arg; + bool val; + int rv; if ((rv = nni_copyin_bool(&val, v, sz, t)) == 0) { - nni_mtx_lock(&d->mtx); - d->nodelay = val; - nni_mtx_unlock(&d->mtx); + nni_mtx_lock(&ep->mtx); + ep->nodelay = val; + nni_mtx_unlock(&ep->mtx); } return (rv); } static int -tcptran_dialer_get_keepalive(void *arg, void *v, size_t *szp, nni_opt_type t) +tcptran_ep_get_keepalive(void *arg, void *v, size_t *szp, nni_opt_type t) { - tcptran_dialer *d = arg; - return (nni_copyout_bool(d->keepalive, v, szp, t)); + tcptran_ep *ep = arg; + int rv; + nni_mtx_lock(&ep->mtx); + rv = nni_copyout_bool(ep->keepalive, v, szp, t); + nni_mtx_unlock(&ep->mtx); + return (rv); } static int -tcptran_dialer_set_keepalive( - void *arg, const void *v, size_t sz, nni_opt_type t) +tcptran_ep_set_keepalive(void *arg, const void *v, size_t sz, nni_opt_type t) { - tcptran_dialer *d = arg; - bool val; - int rv; + tcptran_ep *ep = arg; + bool val; + int rv; if ((rv = nni_copyin_bool(&val, v, sz, t)) == 0) { - nni_mtx_lock(&d->mtx); - d->keepalive = val; - nni_mtx_unlock(&d->mtx); + nni_mtx_lock(&ep->mtx); + ep->keepalive = val; + nni_mtx_unlock(&ep->mtx); } return (rv); } @@ -856,12 +913,18 @@ tcptran_listener_fini(void *arg) { tcptran_listener *l = arg; - nni_aio_stop(l->aio); + nni_mtx_lock(&l->ep.mtx); if (l->listener != NULL) { nni_tcp_listener_fini(l->listener); + l->listener = NULL; } - nni_aio_fini(l->aio); - nni_mtx_fini(&l->mtx); + l->ep.fini = true; + if (!nni_list_empty(&l->ep.pipes)) { + nni_mtx_unlock(&l->ep.mtx); + return; + } + nni_mtx_unlock(&l->ep.mtx); + nni_mtx_fini(&l->ep.mtx); NNI_FREE_STRUCT(l); } @@ -896,8 +959,13 @@ tcptran_listener_init(void **lp, nni_url *url, nni_sock *sock) if ((l = NNI_ALLOC_STRUCT(l)) == NULL) { return (NNG_ENOMEM); } - nni_mtx_init(&l->mtx); - l->url = url; + nni_mtx_init(&l->ep.mtx); + NNI_LIST_INIT(&l->ep.pipes, tcptran_pipe, node); + l->ep.dtor = tcptran_listener_fini; + l->ep.url = url; + l->ep.proto = nni_sock_proto_id(sock); + l->ep.nodelay = true; + l->ep.keepalive = false; if (strlen(url->u_hostname) == 0) { host = NULL; @@ -923,21 +991,12 @@ tcptran_listener_init(void **lp, nni_url *url, nni_sock *sock) rv = nni_aio_result(aio); nni_aio_fini(aio); - if (rv != 0) { + if ((rv != 0) || ((rv = nni_tcp_listener_init(&l->listener)) != 0)) { tcptran_listener_fini(l); return (rv); } - if (((rv = nni_tcp_listener_init(&l->listener)) != 0) || - ((rv = nni_aio_init(&l->aio, tcptran_listener_cb, l)) != 0)) { - tcptran_listener_fini(l); - return (rv); - } - - l->proto = nni_sock_proto_id(sock); - l->nodelay = true; - l->keepalive = false; - l->bsa = l->sa; + l->bsa = l->sa; *lp = l; return (0); @@ -948,7 +1007,7 @@ tcptran_listener_close(void *arg) { tcptran_listener *l = arg; - nni_aio_close(l->aio); + tcptran_ep_close(&l->ep); nni_tcp_listener_close(l->listener); } @@ -958,171 +1017,36 @@ tcptran_listener_bind(void *arg) tcptran_listener *l = arg; int rv; - nni_mtx_lock(&l->mtx); + nni_mtx_lock(&l->ep.mtx); l->bsa = l->sa; rv = nni_tcp_listener_listen(l->listener, &l->bsa); - nni_mtx_unlock(&l->mtx); + nni_mtx_unlock(&l->ep.mtx); return (rv); } -static void -tcptran_listener_cb(void *arg) -{ - tcptran_listener *l = arg; - nni_aio * aio; - int rv; - tcptran_pipe * p = NULL; - nni_tcp_conn * conn; - - nni_mtx_lock(&l->mtx); - rv = nni_aio_result(l->aio); - aio = l->user_aio; - l->user_aio = NULL; - - if (aio == NULL) { - nni_mtx_unlock(&l->mtx); - if (rv == 0) { - conn = nni_aio_get_output(l->aio, 0); - nni_tcp_conn_fini(conn); - } - return; - } - - if (rv != 0) { - nni_mtx_unlock(&l->mtx); - nni_aio_finish_error(aio, rv); - return; - } - - conn = nni_aio_get_output(l->aio, 0); - - NNI_ASSERT(conn != NULL); - if ((rv = tcptran_pipe_init(&p, conn)) != 0) { - nni_mtx_unlock(&l->mtx); - nni_tcp_conn_fini(conn); - nni_aio_finish_error(aio, rv); - return; - } - - p->proto = l->proto; - p->rcvmax = l->rcvmax; - p->nodelay = l->nodelay; - p->keepalive = l->keepalive; - nni_mtx_unlock(&l->mtx); - - (void) nni_tcp_conn_set_nodelay(conn, p->nodelay); - (void) nni_tcp_conn_set_keepalive(conn, p->keepalive); - - nni_aio_set_output(aio, 0, p); - nni_aio_finish(aio, 0, 0); -} - -static void -tcptran_listener_cancel(nni_aio *aio, int rv) -{ - tcptran_listener *l = nni_aio_get_prov_data(aio); - - nni_mtx_lock(&l->mtx); - if (l->user_aio != aio) { - nni_mtx_unlock(&l->mtx); - return; - } - l->user_aio = NULL; - nni_mtx_unlock(&l->mtx); - - nni_aio_abort(l->aio, rv); - nni_aio_finish_error(aio, rv); -} - static void tcptran_listener_accept(void *arg, nni_aio *aio) { tcptran_listener *l = arg; + tcptran_pipe * p = NULL; int rv; if (nni_aio_begin(aio) != 0) { return; } - nni_mtx_lock(&l->mtx); - NNI_ASSERT(l->user_aio == NULL); - - if ((rv = nni_aio_schedule(aio, tcptran_listener_cancel, l)) != 0) { - nni_mtx_unlock(&l->mtx); + nni_mtx_lock(&l->ep.mtx); + if (((rv = tcptran_pipe_init(&p, &l->ep)) != 0) || + ((rv = nni_aio_schedule(aio, tcptran_pipe_conn_cancel, p)) != 0)) { + nni_mtx_unlock(&l->ep.mtx); nni_aio_finish_error(aio, rv); + tcptran_pipe_reap(p); return; } - l->user_aio = aio; - - nni_tcp_listener_accept(l->listener, l->aio); - nni_mtx_unlock(&l->mtx); -} - -static int -tcptran_listener_set_nodelay( - void *arg, const void *v, size_t sz, nni_opt_type t) -{ - tcptran_listener *l = arg; - bool val; - int rv; - if ((rv = nni_copyin_bool(&val, v, sz, t)) == 0) { - nni_mtx_lock(&l->mtx); - l->nodelay = val; - nni_mtx_unlock(&l->mtx); - } - return (rv); -} - -static int -tcptran_listener_get_nodelay(void *arg, void *v, size_t *szp, nni_opt_type t) -{ - tcptran_listener *l = arg; - int rv; - nni_mtx_lock(&l->mtx); - rv = nni_copyout_bool(l->nodelay, v, szp, t); - nni_mtx_unlock(&l->mtx); - return (rv); -} - -static int -tcptran_listener_set_recvmaxsz( - void *arg, const void *v, size_t sz, nni_opt_type t) -{ - tcptran_listener *l = arg; - size_t val; - int rv; - if ((rv = nni_copyin_size(&val, v, sz, 0, NNI_MAXSZ, t)) == 0) { - nni_mtx_lock(&l->mtx); - l->rcvmax = val; - nni_mtx_unlock(&l->mtx); - } - return (rv); -} + p->useraio = aio; -static int -tcptran_listener_set_keepalive( - void *arg, const void *v, size_t sz, nni_opt_type t) -{ - tcptran_listener *l = arg; - bool val; - int rv; - if ((rv = nni_copyin_bool(&val, v, sz, t)) == 0) { - nni_mtx_lock(&l->mtx); - l->keepalive = val; - nni_mtx_unlock(&l->mtx); - } - return (rv); -} - -static int -tcptran_listener_get_keepalive(void *arg, void *v, size_t *szp, nni_opt_type t) -{ - tcptran_listener *l = arg; - int rv; - nni_mtx_lock(&l->mtx); - rv = nni_copyout_bool(l->keepalive, v, szp, t); - nni_mtx_unlock(&l->mtx); - return (rv); + nni_tcp_listener_accept(l->listener, p->connaio); + nni_mtx_unlock(&l->ep.mtx); } static int @@ -1138,27 +1062,15 @@ tcptran_listener_get_url(void *arg, void *v, size_t *szp, nni_opt_type t) return (nni_copyout_str(ustr, v, szp, t)); } -static int -tcptran_listener_get_recvmaxsz(void *arg, void *v, size_t *szp, nni_opt_type t) -{ - tcptran_listener *l = arg; - int rv; - - nni_mtx_lock(&l->mtx); - rv = nni_copyout_size(l->rcvmax, v, szp, t); - nni_mtx_unlock(&l->mtx); - return (rv); -} - static int tcptran_listener_get_locaddr(void *arg, void *buf, size_t *szp, nni_opt_type t) { tcptran_listener *l = arg; int rv; - nni_mtx_lock(&l->mtx); + nni_mtx_lock(&l->ep.mtx); rv = nni_copyout_sockaddr(&l->bsa, buf, szp, t); - nni_mtx_unlock(&l->mtx); + nni_mtx_unlock(&l->ep.mtx); return (rv); } @@ -1203,7 +1115,6 @@ static nni_tran_option tcptran_pipe_options[] = { static nni_tran_pipe_ops tcptran_pipe_ops = { .p_fini = tcptran_pipe_fini, - .p_start = tcptran_pipe_start, .p_stop = tcptran_pipe_stop, .p_send = tcptran_pipe_send, .p_recv = tcptran_pipe_recv, @@ -1216,8 +1127,8 @@ static nni_tran_option tcptran_dialer_options[] = { { .o_name = NNG_OPT_RECVMAXSZ, .o_type = NNI_TYPE_SIZE, - .o_get = tcptran_dialer_get_recvmaxsz, - .o_set = tcptran_dialer_set_recvmaxsz, + .o_get = tcptran_ep_get_recvmaxsz, + .o_set = tcptran_ep_set_recvmaxsz, .o_chk = tcptran_check_recvmaxsz, }, { @@ -1228,15 +1139,15 @@ static nni_tran_option tcptran_dialer_options[] = { { .o_name = NNG_OPT_TCP_NODELAY, .o_type = NNI_TYPE_BOOL, - .o_get = tcptran_dialer_get_nodelay, - .o_set = tcptran_dialer_set_nodelay, + .o_get = tcptran_ep_get_nodelay, + .o_set = tcptran_ep_set_nodelay, .o_chk = tcptran_check_bool, }, { .o_name = NNG_OPT_TCP_KEEPALIVE, .o_type = NNI_TYPE_BOOL, - .o_get = tcptran_dialer_get_keepalive, - .o_set = tcptran_dialer_set_keepalive, + .o_get = tcptran_ep_get_keepalive, + .o_set = tcptran_ep_set_keepalive, .o_chk = tcptran_check_bool, }, // terminate list @@ -1249,8 +1160,8 @@ static nni_tran_option tcptran_listener_options[] = { { .o_name = NNG_OPT_RECVMAXSZ, .o_type = NNI_TYPE_SIZE, - .o_get = tcptran_listener_get_recvmaxsz, - .o_set = tcptran_listener_set_recvmaxsz, + .o_get = tcptran_ep_get_recvmaxsz, + .o_set = tcptran_ep_set_recvmaxsz, .o_chk = tcptran_check_recvmaxsz, }, { @@ -1266,15 +1177,15 @@ static nni_tran_option tcptran_listener_options[] = { { .o_name = NNG_OPT_TCP_NODELAY, .o_type = NNI_TYPE_BOOL, - .o_get = tcptran_listener_get_nodelay, - .o_set = tcptran_listener_set_nodelay, + .o_get = tcptran_ep_get_nodelay, + .o_set = tcptran_ep_set_nodelay, .o_chk = tcptran_check_bool, }, { .o_name = NNG_OPT_TCP_KEEPALIVE, .o_type = NNI_TYPE_BOOL, - .o_get = tcptran_listener_get_keepalive, - .o_set = tcptran_listener_set_keepalive, + .o_get = tcptran_ep_get_keepalive, + .o_set = tcptran_ep_set_keepalive, .o_chk = tcptran_check_bool, }, // terminate list diff --git a/src/transport/tls/tls.c b/src/transport/tls/tls.c index e6701f5f..88e201b9 100644 --- a/src/transport/tls/tls.c +++ b/src/transport/tls/tls.c @@ -29,16 +29,23 @@ typedef struct tlstran_pipe tlstran_pipe; // tlstran_pipe is one end of a TLS connection. struct tlstran_pipe { - nni_tls *tls; - uint16_t peer; - uint16_t proto; - size_t rcvmax; - bool nodelay; - bool keepalive; + nni_tls * tls; + uint16_t peer; + uint16_t proto; + size_t rcvmax; + bool nodelay; + bool keepalive; + nni_atomic_flag reaped; + nni_list_node node; // When pending, we are on ep list. + tlstran_ep * ep; + nni_tcp_dialer *dialer; // Client side. + nni_sockaddr sa; + nni_reap_item reap; + bool closed; nni_list sendq; nni_list recvq; - nni_aio *user_negaio; + nni_aio *useraio; uint8_t txlen[sizeof(uint64_t)]; uint8_t rxlen[sizeof(uint64_t)]; @@ -48,7 +55,9 @@ struct tlstran_pipe { size_t wantrxhead; nni_aio *txaio; nni_aio *rxaio; - nni_aio *negaio; + nni_aio *negoaio; + nni_aio *connaio; + nni_aio *rslvaio; nni_msg *rxmsg; nni_mtx mtx; }; @@ -59,27 +68,25 @@ struct tlstran_ep { size_t rcvmax; bool nodelay; bool keepalive; + bool fini; + nni_reap_item reap; + nni_cb dtor; int authmode; nng_tls_config *cfg; nni_url * url; nni_mtx mtx; + nni_list pipes; }; struct tlstran_dialer { tlstran_ep ep; // must be first nni_tcp_dialer *dialer; uint16_t af; - nni_aio * aio; - nni_aio * user_aio; - bool resolving; - nng_sockaddr sa; }; struct tlstran_listener { tlstran_ep ep; // must be first nni_tcp_listener *listener; - nni_aio * aio; - nni_aio * user_aio; nng_sockaddr sa; nng_sockaddr bsa; // bound addr }; @@ -88,9 +95,10 @@ static void tlstran_pipe_send_start(tlstran_pipe *); static void tlstran_pipe_recv_start(tlstran_pipe *); static void tlstran_pipe_send_cb(void *); static void tlstran_pipe_recv_cb(void *); +static void tlstran_pipe_rslv_cb(void *); +static void tlstran_pipe_conn_cb(void *); static void tlstran_pipe_nego_cb(void *); -static void tlstran_dialer_cb(void *); -static void tlstran_listener_cb(void *); +static void tlstran_pipe_reap(tlstran_pipe *); static int tlstran_init(void) @@ -108,9 +116,15 @@ tlstran_pipe_close(void *arg) { tlstran_pipe *p = arg; + nni_mtx_lock(&p->mtx); + p->closed = true; + nni_mtx_unlock(&p->mtx); + nni_aio_close(p->rxaio); nni_aio_close(p->txaio); - nni_aio_close(p->negaio); + nni_aio_close(p->negoaio); + nni_aio_close(p->connaio); + nni_aio_close(p->rslvaio); nni_tls_close(p->tls); } @@ -122,26 +136,53 @@ tlstran_pipe_stop(void *arg) nni_aio_stop(p->rxaio); nni_aio_stop(p->txaio); - nni_aio_stop(p->negaio); + nni_aio_stop(p->negoaio); + nni_aio_stop(p->connaio); + nni_aio_stop(p->rslvaio); } static void tlstran_pipe_fini(void *arg) { tlstran_pipe *p = arg; + tlstran_ep * ep; + + if (p == NULL) { + return; + } + tlstran_pipe_stop(p); + if ((ep = p->ep) != NULL) { + nni_mtx_lock(&ep->mtx); + nni_list_remove(&ep->pipes, p); + if (ep->fini && nni_list_empty(&ep->pipes)) { + nni_reap(&ep->reap, ep->dtor, ep); + } + nni_mtx_unlock(&ep->mtx); + } nni_aio_fini(p->rxaio); nni_aio_fini(p->txaio); - nni_aio_fini(p->negaio); + nni_aio_fini(p->negoaio); + nni_aio_fini(p->connaio); + nni_aio_fini(p->rslvaio); if (p->tls != NULL) { nni_tls_fini(p->tls); } nni_msg_free(p->rxmsg); + nni_mtx_fini(&p->mtx); NNI_FREE_STRUCT(p); } +static void +tlstran_pipe_reap(tlstran_pipe *p) +{ + if (!nni_atomic_flag_test_and_set(&p->reaped)) { + nni_reap(&p->reap, tlstran_pipe_fini, p); + } +} + static int -tlstran_pipe_init(tlstran_pipe **pipep, nni_tls *tls) +tlstran_pipe_init(tlstran_pipe **pipep, tlstran_ep *ep) { tlstran_pipe *p; int rv; @@ -150,46 +191,110 @@ tlstran_pipe_init(tlstran_pipe **pipep, nni_tls *tls) return (NNG_ENOMEM); } nni_mtx_init(&p->mtx); + nni_aio_list_init(&p->recvq); + nni_aio_list_init(&p->sendq); + nni_atomic_flag_reset(&p->reaped); + nni_list_append(&ep->pipes, p); + + p->proto = ep->proto; + p->rcvmax = ep->rcvmax; + p->nodelay = ep->nodelay; + p->keepalive = ep->keepalive; + p->ep = ep; if (((rv = nni_aio_init(&p->txaio, tlstran_pipe_send_cb, p)) != 0) || ((rv = nni_aio_init(&p->rxaio, tlstran_pipe_recv_cb, p)) != 0) || - ((rv = nni_aio_init(&p->negaio, tlstran_pipe_nego_cb, p)) != 0)) { - tlstran_pipe_fini(p); + ((rv = nni_aio_init(&p->rslvaio, tlstran_pipe_rslv_cb, p)) != 0) || + ((rv = nni_aio_init(&p->connaio, tlstran_pipe_conn_cb, p)) != 0) || + ((rv = nni_aio_init(&p->negoaio, tlstran_pipe_nego_cb, p)) != 0)) { + tlstran_pipe_reap(p); return (rv); } - nni_aio_list_init(&p->recvq); - nni_aio_list_init(&p->sendq); - p->tls = tls; *pipep = p; return (0); } static void -tlstran_pipe_cancel_nego(nni_aio *aio, int rv) +tlstran_pipe_rslv_cb(void *arg) { - tlstran_pipe *p = nni_aio_get_prov_data(aio); + tlstran_pipe * p = arg; + tlstran_dialer *d = (void *) p->ep; + nni_aio * aio = p->rslvaio; + int rv; - nni_mtx_lock(&p->mtx); - if (p->user_negaio != aio) { - nni_mtx_unlock(&p->mtx); + nni_mtx_lock(&p->ep->mtx); + if ((rv = nni_aio_result(aio)) != 0) { + nni_aio *uaio; + if ((uaio = p->useraio) != NULL) { + p->useraio = NULL; + nni_aio_finish_error(uaio, rv); + } + nni_mtx_unlock(&p->ep->mtx); + tlstran_pipe_reap(p); return; } - p->user_negaio = NULL; - nni_mtx_unlock(&p->mtx); + if (d->dialer != NULL) { + nni_tcp_dialer_dial(d->dialer, &p->sa, p->connaio); + } + nni_mtx_unlock(&p->ep->mtx); +} - nni_aio_abort(p->negaio, rv); - nni_aio_finish_error(aio, rv); +static void +tlstran_pipe_conn_cb(void *arg) +{ + tlstran_pipe *p = arg; + nni_aio * aio = p->connaio; + nni_iov iov; + int rv; + + nni_mtx_lock(&p->ep->mtx); + + if ((rv = nni_aio_result(aio)) == 0) { + nni_tcp_conn *tcp = nni_aio_get_output(aio, 0); + if ((rv = nni_tls_init(&p->tls, p->ep->cfg, tcp)) != 0) { + nni_tcp_conn_fini(tcp); + } + } + + if (rv != 0) { + nni_aio *uaio; + if ((uaio = p->useraio) != NULL) { + p->useraio = NULL; + nni_aio_finish_error(uaio, rv); + } + nni_mtx_unlock(&p->ep->mtx); + tlstran_pipe_reap(p); + return; + } + + p->txlen[0] = 0; + p->txlen[1] = 'S'; + p->txlen[2] = 'P'; + p->txlen[3] = 0; + NNI_PUT16(&p->txlen[4], p->proto); + NNI_PUT16(&p->txlen[6], 0); + + p->gotrxhead = 0; + p->gottxhead = 0; + p->wantrxhead = 8; + p->wanttxhead = 8; + iov.iov_len = 8; + iov.iov_buf = &p->txlen[0]; + nni_aio_set_iov(p->negoaio, 1, &iov); + nni_tls_send(p->tls, p->negoaio); + nni_mtx_unlock(&p->ep->mtx); } static void tlstran_pipe_nego_cb(void *arg) { tlstran_pipe *p = arg; - nni_aio * aio = p->negaio; + nni_aio * aio = p->negoaio; + nni_aio * uaio; int rv; - nni_mtx_lock(&p->mtx); + nni_mtx_lock(&p->ep->mtx); if ((rv = nni_aio_result(aio)) != 0) { goto done; } @@ -208,7 +313,7 @@ tlstran_pipe_nego_cb(void *arg) nni_aio_set_iov(aio, 1, &iov); // send it down... nni_tls_send(p->tls, aio); - nni_mtx_unlock(&p->mtx); + nni_mtx_unlock(&p->ep->mtx); return; } if (p->gotrxhead < p->wantrxhead) { @@ -217,7 +322,7 @@ tlstran_pipe_nego_cb(void *arg) iov.iov_buf = &p->rxlen[p->gotrxhead]; nni_aio_set_iov(aio, 1, &iov); nni_tls_recv(p->tls, aio); - nni_mtx_unlock(&p->mtx); + nni_mtx_unlock(&p->ep->mtx); return; } // We have both sent and received the headers. Lets check the @@ -232,17 +337,23 @@ tlstran_pipe_nego_cb(void *arg) NNI_GET16(&p->rxlen[4], p->peer); done: - if (rv == 0) { - // These can fail. Note that the TLS stack automatically - // starts out in NODELAY to make the handshake performant. - (void) nni_tls_set_nodelay(p->tls, p->nodelay); - (void) nni_tls_set_keepalive(p->tls, p->keepalive); - } - if ((aio = p->user_negaio) != NULL) { - p->user_negaio = NULL; - nni_aio_finish(aio, rv, 0); + if ((uaio = p->useraio) != NULL) { + p->useraio = NULL; + if (rv == 0) { + // These can fail. Note that the TLS stack + // automatically starts out in NODELAY to make the + // handshake performant. + (void) nni_tls_set_nodelay(p->tls, p->nodelay); + (void) nni_tls_set_keepalive(p->tls, p->keepalive); + nni_aio_set_output(uaio, 0, p); + nni_aio_finish(uaio, 0, 0); + nni_mtx_unlock(&p->ep->mtx); + return; + } + nni_aio_finish_error(aio, rv); } - nni_mtx_unlock(&p->mtx); + nni_mtx_unlock(&p->ep->mtx); + tlstran_pipe_reap(p); } static void @@ -256,17 +367,18 @@ tlstran_pipe_send_cb(void *arg) nni_aio * txaio = p->txaio; nni_mtx_lock(&p->mtx); - aio = nni_list_first(&p->sendq); - - if ((rv = nni_aio_result(txaio)) != 0) { + rv = p->closed ? NNG_ECLOSED : nni_aio_result(txaio); + if (rv != 0) { // Intentionally we do not queue up another transfer. // There's an excellent chance that the pipe is no longer // usable, with a partial transfer. // The protocol should see this error, and close the // pipe itself, we hope. - nni_aio_list_remove(aio); + while ((aio = nni_list_first(&p->sendq)) != NULL) { + nni_aio_list_remove(aio); + nni_aio_finish_error(aio, rv); + } nni_mtx_unlock(&p->mtx); - nni_aio_finish_error(aio, rv); return; } @@ -277,6 +389,7 @@ tlstran_pipe_send_cb(void *arg) nni_mtx_unlock(&p->mtx); return; } + aio = nni_list_first(&p->sendq); nni_aio_list_remove(aio); tlstran_pipe_send_start(p); nni_mtx_unlock(&p->mtx); @@ -299,11 +412,14 @@ tlstran_pipe_recv_cb(void *arg) nni_aio * rxaio = p->rxaio; nni_mtx_lock(&p->mtx); - aio = nni_list_first(&p->recvq); if ((rv = nni_aio_result(p->rxaio)) != 0) { goto recv_error; } + if (p->closed) { + rv = NNG_ECLOSED; + goto recv_error; + } n = nni_aio_count(rxaio); nni_aio_iov_advance(rxaio, n); @@ -348,6 +464,7 @@ tlstran_pipe_recv_cb(void *arg) } // We read a message completely. Let the user know the good news. + aio = nni_list_first(&p->recvq); nni_aio_list_remove(aio); msg = p->rxmsg; p->rxmsg = NULL; @@ -361,14 +478,16 @@ tlstran_pipe_recv_cb(void *arg) return; recv_error: - nni_aio_list_remove(aio); + while ((aio = nni_list_first(&p->recvq)) != NULL) { + nni_aio_list_remove(aio); + nni_aio_finish_error(aio, rv); + } msg = p->rxmsg; p->rxmsg = NULL; // Intentionally, we do not queue up another receive. // The protocol should notice this error and close the pipe. nni_mtx_unlock(&p->mtx); nni_msg_free(msg); - nni_aio_finish_error(aio, rv); } static void @@ -444,6 +563,11 @@ tlstran_pipe_send(void *arg, nni_aio *aio) return; } nni_mtx_lock(&p->mtx); + if (p->closed) { + nni_mtx_unlock(&p->mtx); + nni_aio_finish_error(aio, NNG_ECLOSED); + return; + } if ((rv = nni_aio_schedule(aio, tlstran_pipe_send_cancel, p)) != 0) { nni_mtx_unlock(&p->mtx); nni_aio_finish_error(aio, rv); @@ -486,7 +610,7 @@ tlstran_pipe_recv_start(tlstran_pipe *p) nni_iov iov; NNI_ASSERT(p->rxmsg == NULL); - // Schedule a read of the IPC header. + // Schedule a read of the header. rxaio = p->rxaio; iov.iov_buf = p->rxlen; iov.iov_len = sizeof(p->rxlen); @@ -505,6 +629,11 @@ tlstran_pipe_recv(void *arg, nni_aio *aio) return; } nni_mtx_lock(&p->mtx); + if (p->closed) { + nni_mtx_unlock(&p->mtx); + nni_aio_finish_error(aio, NNG_ECLOSED); + return; + } if ((rv = nni_aio_schedule(aio, tlstran_pipe_recv_cancel, p)) != 0) { nni_mtx_unlock(&p->mtx); nni_aio_finish_error(aio, rv); @@ -569,56 +698,40 @@ tlstran_pipe_get_nodelay(void *arg, void *v, size_t *szp, nni_opt_type t) } static void -tlstran_pipe_start(void *arg, nni_aio *aio) +tlstran_ep_close(tlstran_ep *ep) { - tlstran_pipe *p = arg; - nni_aio * negaio; - nni_iov iov; - int rv; - - if (nni_aio_begin(aio) != 0) { - return; - } - nni_mtx_lock(&p->mtx); - if ((rv = nni_aio_schedule(aio, tlstran_pipe_cancel_nego, p)) != 0) { - nni_mtx_unlock(&p->mtx); - nni_aio_finish_error(aio, rv); - return; + tlstran_pipe *p; + nni_mtx_lock(&ep->mtx); + NNI_LIST_FOREACH (&ep->pipes, p) { + nni_aio_close(p->negoaio); + nni_aio_close(p->connaio); + nni_aio_close(p->rslvaio); } - p->txlen[0] = 0; - p->txlen[1] = 'S'; - p->txlen[2] = 'P'; - p->txlen[3] = 0; - NNI_PUT16(&p->txlen[4], p->proto); - NNI_PUT16(&p->txlen[6], 0); - - p->user_negaio = aio; - p->gotrxhead = 0; - p->gottxhead = 0; - p->wantrxhead = 8; - p->wanttxhead = 8; - negaio = p->negaio; - iov.iov_len = 8; - iov.iov_buf = &p->txlen[0]; - nni_aio_set_iov(negaio, 1, &iov); - nni_tls_send(p->tls, negaio); - nni_mtx_unlock(&p->mtx); + nni_mtx_unlock(&ep->mtx); } static void tlstran_dialer_fini(void *arg) { - tlstran_dialer *d = arg; + tlstran_dialer *d = arg; + tlstran_ep * ep = &d->ep; - nni_aio_stop(d->aio); + nni_mtx_lock(&ep->mtx); + d->ep.fini = true; if (d->dialer != NULL) { nni_tcp_dialer_fini(d->dialer); + d->dialer = NULL; + } + if (!nni_list_empty(&ep->pipes)) { + nni_mtx_unlock(&ep->mtx); + return; } - nni_aio_fini(d->aio); - if (d->ep.cfg != NULL) { - nni_tls_config_fini(d->ep.cfg); + nni_mtx_unlock(&ep->mtx); + if (ep->cfg != NULL) { + nni_tls_config_fini(ep->cfg); } - nni_mtx_fini(&d->ep.mtx); + nni_mtx_fini(&ep->mtx); + NNI_FREE_STRUCT(d); } @@ -627,8 +740,8 @@ tlstran_dialer_close(void *arg) { tlstran_dialer *d = arg; - nni_aio_close(d->aio); nni_tcp_dialer_close(d->dialer); + tlstran_ep_close(&d->ep); } static int @@ -664,110 +777,48 @@ tlstran_dialer_init(void **dp, nni_url *url, nni_sock *sock) } nni_mtx_init(&d->ep.mtx); + NNI_LIST_INIT(&d->ep.pipes, tlstran_pipe, node); d->ep.authmode = NNG_TLS_AUTH_MODE_REQUIRED; d->ep.url = url; d->ep.proto = nni_sock_proto_id(sock); d->ep.nodelay = true; d->ep.keepalive = false; + d->ep.dtor = tlstran_dialer_fini; + d->af = af; if (((rv = nni_tcp_dialer_init(&d->dialer)) != 0) || ((rv = nni_tls_config_init(&d->ep.cfg, NNG_TLS_MODE_CLIENT)) != 0) || ((rv = nng_tls_config_auth_mode(d->ep.cfg, d->ep.authmode)) != 0) || - ((rv = nng_tls_config_server_name(d->ep.cfg, host)) != 0) || - ((rv = nni_aio_init(&d->aio, tlstran_dialer_cb, d)) != 0)) { + ((rv = nng_tls_config_server_name(d->ep.cfg, host)) != 0)) { tlstran_dialer_fini(d); return (rv); } - d->af = af; *dp = d; return (0); } static void -tlstran_dialer_cb(void *arg) +tlstran_pipe_conn_cancel(nni_aio *aio, int rv) { - tlstran_dialer *d = arg; - tlstran_pipe * p; - nni_tcp_conn * conn; - nni_tls * tls; - nni_aio * aio; - int rv; - - nni_mtx_lock(&d->ep.mtx); - aio = d->user_aio; - rv = nni_aio_result(d->aio); - - if (aio == NULL) { - nni_mtx_unlock(&d->ep.mtx); - if ((rv == 0) && !d->resolving) { - conn = nni_aio_get_output(d->aio, 0); - nni_tcp_conn_fini(conn); - } - return; - } - - if (rv != 0) { - d->user_aio = NULL; - nni_mtx_unlock(&d->ep.mtx); - nni_aio_finish_error(aio, rv); - return; - } - - if (d->resolving) { - // Name resolution complete. Now go to next step. - d->resolving = false; - nni_tcp_dialer_dial(d->dialer, &d->sa, d->aio); - nni_mtx_unlock(&d->ep.mtx); - return; - } - d->user_aio = NULL; - conn = nni_aio_get_output(d->aio, 0); - NNI_ASSERT(conn != NULL); - - if ((rv = nni_tls_init(&tls, d->ep.cfg, conn)) != 0) { - nni_mtx_unlock(&d->ep.mtx); - nni_tcp_conn_fini(conn); - nni_aio_finish_error(aio, rv); - return; - } + tlstran_pipe *p = nni_aio_get_prov_data(aio); - if ((rv = tlstran_pipe_init(&p, tls)) != 0) { - nni_mtx_unlock(&d->ep.mtx); - nni_tls_fini(tls); - nni_aio_finish_error(aio, rv); + nni_mtx_lock(&p->ep->mtx); + if (p->useraio != aio) { + nni_mtx_unlock(&p->ep->mtx); return; } - p->proto = d->ep.proto; - p->rcvmax = d->ep.rcvmax; - p->nodelay = d->ep.nodelay; - p->keepalive = d->ep.keepalive; - nni_mtx_unlock(&d->ep.mtx); - - (void) nni_tls_set_nodelay(tls, p->nodelay); - (void) nni_tls_set_keepalive(tls, p->keepalive); - - nni_aio_set_output(aio, 0, p); - nni_aio_finish(aio, 0, 0); -} + // Close the underlying AIOs. This will abort the operation. + // The pipe is removed from pending list at completion callback. + p->useraio = NULL; + nni_aio_close(p->negoaio); + nni_aio_close(p->connaio); + nni_aio_close(p->rslvaio); + nni_mtx_unlock(&p->ep->mtx); -static void -tlstran_dialer_cancel(nni_aio *aio, int rv) -{ - tlstran_dialer *d = nni_aio_get_prov_data(aio); - - nni_mtx_lock(&d->ep.mtx); - if (d->user_aio != aio) { - nni_mtx_unlock(&d->ep.mtx); - return; - } - d->user_aio = NULL; - nni_mtx_unlock(&d->ep.mtx); - - nni_aio_abort(d->aio, rv); nni_aio_finish_error(aio, rv); } @@ -775,45 +826,52 @@ static void tlstran_dialer_connect(void *arg, nni_aio *aio) { tlstran_dialer *d = arg; + tlstran_pipe * p = NULL; int rv; if (nni_aio_begin(aio) != 0) { return; } nni_mtx_lock(&d->ep.mtx); - NNI_ASSERT(d->user_aio == NULL); - - if ((rv = nni_aio_schedule(aio, tlstran_dialer_cancel, d)) != 0) { + if (((rv = tlstran_pipe_init(&p, &d->ep)) != 0) || + ((rv = nni_aio_schedule(aio, tlstran_pipe_conn_cancel, p)) != 0)) { nni_mtx_unlock(&d->ep.mtx); nni_aio_finish_error(aio, rv); + tlstran_pipe_reap(p); return; } - d->user_aio = aio; + p->useraio = aio; + p->dialer = d->dialer; - d->resolving = true; - - // Start the name resolution. Callback will see resolving, and then - // switch to doing actual connect. - nni_aio_set_input(d->aio, 0, &d->sa); + // Start the name resolution. + nni_aio_set_input(p->rslvaio, 0, &p->sa); nni_tcp_resolv( - d->ep.url->u_hostname, d->ep.url->u_port, d->af, 0, d->aio); + d->ep.url->u_hostname, d->ep.url->u_port, d->af, 0, p->rslvaio); nni_mtx_unlock(&d->ep.mtx); } static void tlstran_listener_fini(void *arg) { - tlstran_listener *l = arg; + tlstran_listener *l = arg; + tlstran_ep * ep = &l->ep; - nni_aio_stop(l->aio); + nni_mtx_lock(&ep->mtx); + l->ep.fini = true; if (l->listener != NULL) { nni_tcp_listener_fini(l->listener); + l->listener = NULL; } - nni_aio_fini(l->aio); - if (l->ep.cfg != NULL) { - nni_tls_config_fini(l->ep.cfg); + if (!nni_list_empty(&ep->pipes)) { + nni_mtx_unlock(&ep->mtx); + return; + } + nni_mtx_unlock(&ep->mtx); + if (ep->cfg != NULL) { + nni_tls_config_fini(ep->cfg); } - nni_mtx_fini(&l->ep.mtx); + nni_mtx_fini(&ep->mtx); + NNI_FREE_STRUCT(l); } @@ -822,7 +880,7 @@ tlstran_listener_close(void *arg) { tlstran_listener *l = arg; - nni_aio_close(l->aio); + tlstran_ep_close(&l->ep); nni_tcp_listener_close(l->listener); } @@ -859,11 +917,13 @@ tlstran_listener_init(void **lp, nni_url *url, nni_sock *sock) return (NNG_ENOMEM); } nni_mtx_init(&l->ep.mtx); + NNI_LIST_INIT(&l->ep.pipes, tlstran_pipe, node); l->ep.url = url; l->ep.authmode = NNG_TLS_AUTH_MODE_NONE; l->ep.keepalive = false; l->ep.nodelay = true; l->ep.proto = nni_sock_proto_id(sock); + l->ep.dtor = tlstran_listener_fini; if (strlen(host) == 0) { host = NULL; @@ -896,8 +956,7 @@ tlstran_listener_init(void **lp, nni_url *url, nni_sock *sock) ((rv = nni_tls_config_init(&l->ep.cfg, NNG_TLS_MODE_SERVER)) != 0) || ((rv = nng_tls_config_auth_mode(l->ep.cfg, l->ep.authmode)) != - 0) || - ((rv = nni_aio_init(&l->aio, tlstran_listener_cb, l)) != 0)) { + 0)) { tlstran_listener_fini(l); return (rv); } @@ -921,99 +980,27 @@ tlstran_listener_bind(void *arg) return (rv); } -static void -tlstran_listener_cb(void *arg) -{ - tlstran_listener *l = arg; - nni_aio * aio; - int rv; - tlstran_pipe * p = NULL; - nni_tcp_conn * conn; - nni_tls * tls; - - nni_mtx_lock(&l->ep.mtx); - rv = nni_aio_result(l->aio); - aio = l->user_aio; - l->user_aio = NULL; - - if (aio == NULL) { - nni_mtx_unlock(&l->ep.mtx); - if (rv == 0) { - conn = nni_aio_get_output(l->aio, 0); - nni_tcp_conn_fini(conn); - } - return; - } - if (rv != 0) { - nni_mtx_unlock(&l->ep.mtx); - nni_aio_finish_error(aio, rv); - return; - } - - conn = nni_aio_get_output(l->aio, 0); - if ((rv = nni_tls_init(&tls, l->ep.cfg, conn)) != 0) { - nni_mtx_unlock(&l->ep.mtx); - nni_tcp_conn_fini(conn); - nni_aio_finish_error(aio, rv); - return; - } - if ((rv = tlstran_pipe_init(&p, tls)) != 0) { - nni_mtx_unlock(&l->ep.mtx); - nni_tls_fini(tls); - nni_aio_finish_error(aio, rv); - return; - } - p->proto = l->ep.proto; - p->rcvmax = l->ep.rcvmax; - p->nodelay = l->ep.nodelay; - p->keepalive = l->ep.keepalive; - - (void) nni_tls_set_nodelay(tls, p->nodelay); - (void) nni_tls_set_keepalive(tls, p->keepalive); - - nni_mtx_unlock(&l->ep.mtx); - - nni_aio_set_output(aio, 0, p); - nni_aio_finish(aio, 0, 0); -} - -static void -tlstran_listener_cancel(nni_aio *aio, int rv) -{ - tlstran_listener *l = nni_aio_get_prov_data(aio); - - nni_mtx_lock(&l->ep.mtx); - if (l->user_aio != aio) { - nni_mtx_unlock(&l->ep.mtx); - return; - } - l->user_aio = NULL; - nni_mtx_unlock(&l->ep.mtx); - - nni_aio_abort(l->aio, rv); - nni_aio_finish_error(aio, rv); -} - static void tlstran_listener_accept(void *arg, nni_aio *aio) { tlstran_listener *l = arg; + tlstran_pipe * p = NULL; int rv; if (nni_aio_begin(aio) != 0) { return; } nni_mtx_lock(&l->ep.mtx); - NNI_ASSERT(l->user_aio == NULL); - - if ((rv = nni_aio_schedule(aio, tlstran_listener_cancel, l)) != 0) { + if (((rv = tlstran_pipe_init(&p, &l->ep)) != 0) || + ((rv = nni_aio_schedule(aio, tlstran_pipe_conn_cancel, p)) != 0)) { nni_mtx_unlock(&l->ep.mtx); nni_aio_finish_error(aio, rv); + tlstran_pipe_reap(p); return; } - l->user_aio = aio; + p->useraio = aio; - nni_tcp_listener_accept(l->listener, l->aio); + nni_tcp_listener_accept(l->listener, p->connaio); nni_mtx_unlock(&l->ep.mtx); } @@ -1306,7 +1293,6 @@ static nni_tran_option tlstran_pipe_options[] = { static nni_tran_pipe_ops tlstran_pipe_ops = { .p_fini = tlstran_pipe_fini, - .p_start = tlstran_pipe_start, .p_stop = tlstran_pipe_stop, .p_send = tlstran_pipe_send, .p_recv = tlstran_pipe_recv, diff --git a/src/transport/zerotier/zerotier.c b/src/transport/zerotier/zerotier.c index a2163e3f..d1072931 100644 --- a/src/transport/zerotier/zerotier.c +++ b/src/transport/zerotier/zerotier.c @@ -183,27 +183,28 @@ struct zt_fraglist { }; struct zt_pipe { - nni_list_node zp_link; - const char * zp_addr; - zt_node * zp_ztn; - uint64_t zp_nwid; - uint64_t zp_laddr; - uint64_t zp_raddr; - uint16_t zp_peer; - uint16_t zp_proto; - uint16_t zp_next_msgid; - size_t zp_rcvmax; - size_t zp_mtu; - int zp_closed; - nni_aio * zp_user_rxaio; - nni_time zp_last_recv; - zt_fraglist zp_recvq[zt_recvq]; - int zp_ping_try; - int zp_ping_tries; - int zp_ping_active; - nni_duration zp_ping_time; - nni_aio * zp_ping_aio; - uint8_t * zp_send_buf; + nni_list_node zp_link; + const char * zp_addr; + zt_node * zp_ztn; + uint64_t zp_nwid; + uint64_t zp_laddr; + uint64_t zp_raddr; + uint16_t zp_peer; + uint16_t zp_proto; + uint16_t zp_next_msgid; + size_t zp_rcvmax; + size_t zp_mtu; + nni_aio * zp_user_rxaio; + nni_time zp_last_recv; + zt_fraglist zp_recvq[zt_recvq]; + int zp_ping_try; + int zp_ping_tries; + bool zp_closed; + nni_duration zp_ping_time; + nni_aio * zp_ping_aio; + uint8_t * zp_send_buf; + nni_atomic_flag zp_reaped; + nni_reap_item zp_reap; }; typedef struct zt_creq zt_creq; @@ -280,6 +281,7 @@ static void zt_fraglist_free(zt_fraglist *); static void zt_virtual_recv(ZT_Node *, void *, void *, uint64_t, void **, uint64_t, uint64_t, unsigned int, unsigned int, const void *, unsigned int); +static void zt_pipe_start_ping(zt_pipe *); static int64_t zt_now(void) @@ -805,10 +807,8 @@ zt_pipe_close_err(zt_pipe *p, int err, uint8_t code, const char *msg) p->zp_user_rxaio = NULL; nni_aio_finish_error(aio, err); } - if ((aio = p->zp_ping_aio) != NULL) { - nni_aio_finish_error(aio, NNG_ECLOSED); - } - p->zp_closed = 1; + nni_aio_close(p->zp_ping_aio); + p->zp_closed = true; if (msg != NULL) { zt_pipe_send_err(p, code, msg); } @@ -974,7 +974,7 @@ zt_pipe_recv_disc_req(zt_pipe *p, const uint8_t *data, size_t len) // Don't bother to check the length, going to disconnect anyway. if ((aio = p->zp_user_rxaio) != NULL) { p->zp_user_rxaio = NULL; - p->zp_closed = 1; + p->zp_closed = true; nni_aio_finish_error(aio, NNG_ECLOSED); } } @@ -990,7 +990,7 @@ zt_pipe_recv_error(zt_pipe *p, const uint8_t *data, size_t len) // the day, the details are just not that interesting. if ((aio = p->zp_user_rxaio) != NULL) { p->zp_user_rxaio = NULL; - p->zp_closed = 1; + p->zp_closed = true; nni_aio_finish_error(aio, NNG_ETRANERR); } } @@ -1627,7 +1627,8 @@ zt_pipe_close(void *arg) nni_aio *aio; nni_mtx_lock(&zt_lk); - p->zp_closed = 1; + p->zp_closed = true; + nni_aio_close(p->zp_ping_aio); if ((aio = p->zp_user_rxaio) != NULL) { p->zp_user_rxaio = NULL; nni_aio_finish_error(aio, NNG_ECLOSED); @@ -1659,6 +1660,14 @@ zt_pipe_fini(void *arg) NNI_FREE_STRUCT(p); } +static void +zt_pipe_reap(zt_pipe *p) +{ + if (!nni_atomic_flag_test_and_set(&p->zp_reaped)) { + nni_reap(&p->zp_reap, zt_pipe_fini, p); + } +} + static int zt_pipe_init(zt_pipe **pipep, zt_ep *ep, uint64_t raddr, uint64_t laddr) { @@ -1687,6 +1696,7 @@ zt_pipe_init(zt_pipe **pipep, zt_ep *ep, uint64_t raddr, uint64_t laddr) p->zp_ping_time = ep->ze_ping_time; p->zp_next_msgid = (uint16_t) nni_random(); p->zp_ping_try = 0; + nni_atomic_flag_reset(&p->zp_reaped); if (ep->ze_mode == NNI_EP_MODE_DIAL) { rv = nni_idhash_insert(ztn->zn_lpipes, laddr, p); @@ -1696,7 +1706,8 @@ zt_pipe_init(zt_pipe **pipep, zt_ep *ep, uint64_t raddr, uint64_t laddr) if ((rv != 0) || ((rv = nni_idhash_insert(ztn->zn_peers, p->zp_raddr, p)) != 0) || ((rv = nni_aio_init(&p->zp_ping_aio, zt_pipe_ping_cb, p)) != 0)) { - zt_pipe_fini(p); + zt_pipe_reap(p); + return (rv); } // The largest fragment we can accept on this pipe. The MTU is @@ -1717,7 +1728,7 @@ zt_pipe_init(zt_pipe **pipep, zt_ep *ep, uint64_t raddr, uint64_t laddr) fl->fl_missingsz = (maxfrags + 7) / 8; fl->fl_missing = nni_alloc(fl->fl_missingsz); if (fl->fl_missing == NULL) { - zt_pipe_fini(p); + zt_pipe_reap(p); return (NNG_ENOMEM); } } @@ -2015,97 +2026,51 @@ zt_pipe_get_node(void *arg, void *buf, size_t *szp, nni_opt_type t) return (nni_copyout_u64(p->zp_laddr >> 24, buf, szp, t)); } -static void -zt_pipe_cancel_ping(nni_aio *aio, int rv) -{ - zt_pipe *p = nni_aio_get_prov_data(aio); - - nni_mtx_lock(&zt_lk); - if (p->zp_ping_active) { - p->zp_ping_active = 0; - nni_aio_finish_error(aio, rv); - } - nni_mtx_unlock(&zt_lk); -} - static void zt_pipe_ping_cb(void *arg) { zt_pipe *p = arg; nni_aio *aio = p->zp_ping_aio; + int rv; + if ((rv = nni_aio_result(aio)) != 0) { + // We were canceled. That means we're done. + return; + } nni_mtx_lock(&zt_lk); - - p->zp_ping_active = 0; if (p->zp_closed || aio == NULL || (p->zp_ping_tries == 0) || (p->zp_ping_time == NNG_DURATION_INFINITE) || (p->zp_ping_time == NNG_DURATION_ZERO)) { nni_mtx_unlock(&zt_lk); return; } - if (nni_aio_result(aio) != NNG_ETIMEDOUT) { - nni_mtx_unlock(&zt_lk); - return; - } - if (p->zp_ping_try < p->zp_ping_tries) { - nni_time now = nni_clock(); - nni_aio_set_timeout(aio, p->zp_ping_time); - // We want pings. We only send one if needed, but we - // use the the timer to wake us up even if we aren't - // going to send a ping. (We don't increment the try count - // unless we actually do send one though.) - if (nni_aio_begin(aio) == 0) { - int rv; - rv = nni_aio_schedule(aio, zt_pipe_cancel_ping, p); - if (rv != 0) { - nni_mtx_unlock(&zt_lk); - nni_aio_finish_error(aio, rv); - return; - } - p->zp_ping_active = 1; - if (now > (p->zp_last_recv + p->zp_ping_time)) { - p->zp_ping_try++; - zt_pipe_send_ping(p); - } - } - } else { + if (p->zp_ping_try >= p->zp_ping_tries) { // Ping count exceeded; the other side is AFK. // Close the pipe, but no need to send a reason to the peer. zt_pipe_close_err(p, NNG_ECLOSED, 0, NULL); + nni_mtx_unlock(&zt_lk); + return; + } + + if (nni_clock() > (p->zp_last_recv + p->zp_ping_time)) { + p->zp_ping_try++; + zt_pipe_send_ping(p); } + + nni_sleep_aio(p->zp_ping_time, aio); // Schedule a recheck. nni_mtx_unlock(&zt_lk); } static void -zt_pipe_start(void *arg, nni_aio *aio) +zt_pipe_start_ping(zt_pipe *p) { - zt_pipe *p = arg; - - if (nni_aio_begin(aio) != 0) { - return; - } - nni_mtx_lock(&zt_lk); - p->zp_ping_active = 0; // send a gratuitous ping, and start the ping interval timer. if ((p->zp_ping_tries > 0) && (p->zp_ping_time != NNG_DURATION_ZERO) && - (p->zp_ping_time != NNG_DURATION_INFINITE) && - (p->zp_ping_aio != NULL)) { + (p->zp_ping_time != NNG_DURATION_INFINITE)) { p->zp_ping_try = 0; - nni_aio_set_timeout(aio, p->zp_ping_time); - if (nni_aio_begin(p->zp_ping_aio) == 0) { - int rv; - rv = nni_aio_schedule( - p->zp_ping_aio, zt_pipe_cancel_ping, p); - if (rv != 0) { - nni_aio_finish_error(p->zp_ping_aio, rv); - } else { - p->zp_ping_active = 1; - zt_pipe_send_ping(p); - } - } + zt_pipe_send_ping(p); + nni_sleep_aio(p->zp_ping_time, p->zp_ping_aio); } - nni_aio_finish(aio, 0, 0); - nni_mtx_unlock(&zt_lk); } static void @@ -2425,6 +2390,7 @@ zt_ep_doaccept(zt_ep *ep) } p->zp_peer = creq.cr_proto; zt_pipe_send_conn_ack(p); + zt_pipe_start_ping(p); nni_aio_set_output(aio, 0, p); nni_aio_finish(aio, 0, 0); } @@ -2484,6 +2450,7 @@ zt_ep_conn_req_cb(void *arg) // Already canceled, or already handled? if ((uaio = nni_list_first(&ep->ze_aios)) != NULL) { nni_aio_list_remove(uaio); + zt_pipe_start_ping(p); nni_aio_set_output(uaio, 0, p); nni_aio_finish(uaio, 0, 0); } else { @@ -2991,7 +2958,6 @@ static nni_tran_option zt_pipe_options[] = { static nni_tran_pipe_ops zt_pipe_ops = { .p_fini = zt_pipe_fini, - .p_start = zt_pipe_start, .p_send = zt_pipe_send, .p_recv = zt_pipe_recv, .p_close = zt_pipe_close, -- cgit v1.2.3-70-g09d2