diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/transport/ipc/ipc.c | 3 | ||||
| -rw-r--r-- | src/transport/tcp/tcp.c | 970 |
2 files changed, 429 insertions, 544 deletions
diff --git a/src/transport/ipc/ipc.c b/src/transport/ipc/ipc.c index 11a0c890..7ba5bc52 100644 --- a/src/transport/ipc/ipc.c +++ b/src/transport/ipc/ipc.c @@ -149,6 +149,9 @@ static void ipctran_pipe_reap(ipctran_pipe *p) { if (!nni_atomic_flag_test_and_set(&p->reaped)) { + if (p->conn != NULL) { + nni_ipc_conn_close(p->conn); + } nni_reap(&p->reap, ipctran_pipe_fini, p); } } diff --git a/src/transport/tcp/tcp.c b/src/transport/tcp/tcp.c index 0d47529e..5b200644 100644 --- a/src/transport/tcp/tcp.c +++ b/src/transport/tcp/tcp.c @@ -17,72 +17,66 @@ // TCP transport. Platform specific TCP operations must be // supplied as well. -typedef struct tcptran_pipe tcptran_pipe; -typedef struct tcptran_dialer tcptran_dialer; -typedef struct tcptran_listener tcptran_listener; +typedef struct tcptran_pipe tcptran_pipe; +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_list recvq; - nni_list sendq; - nni_aio *user_negaio; - - uint8_t txlen[sizeof(uint64_t)]; - uint8_t rxlen[sizeof(uint64_t)]; - size_t gottxhead; - size_t gotrxhead; - size_t wanttxhead; - size_t wantrxhead; - nni_aio *txaio; - nni_aio *rxaio; - nni_aio *negaio; - nni_msg *rxmsg; - nni_mtx mtx; -}; - -struct tcptran_dialer { - nni_tcp_dialer *dialer; + nni_tcp_conn * conn; + uint16_t peer; 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; + bool closed; + nni_list_node node; + tcptran_ep * ep; + nni_atomic_flag reaped; + nni_reap_item reap; + nni_sockaddr sa; + uint8_t txlen[sizeof(uint64_t)]; + uint8_t rxlen[sizeof(uint64_t)]; + size_t gottxhead; + size_t gotrxhead; + size_t wanttxhead; + size_t wantrxhead; + nni_list recvq; + nni_list sendq; + nni_aio * useraio; + nni_aio * txaio; + nni_aio * rxaio; + nni_aio * negoaio; + nni_aio * connaio; + nni_aio * rslvaio; + nni_msg * rxmsg; nni_mtx mtx; }; -struct tcptran_listener { - nni_tcp_listener *listener; +struct tcptran_ep { + nni_mtx mtx; + uint16_t af; uint16_t proto; size_t rcvmax; bool nodelay; bool keepalive; - nni_aio * aio; - nni_aio * user_aio; + bool fini; nni_url * url; nng_sockaddr sa; - nng_sockaddr bsa; // bound addr - nni_mtx mtx; + nng_sockaddr bsa; + nni_list pipes; + nni_reap_item reap; + nni_tcp_dialer * dialer; + nni_tcp_listener *listener; }; 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_conn_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_rslv_cb(void *); +static void tcptran_ep_fini(void *); static int tcptran_init(void) @@ -100,9 +94,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 +114,35 @@ 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, tcptran_ep_fini, 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 +151,19 @@ 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)) { + if (p->conn != NULL) { + nni_tcp_conn_close(p->conn); + } + 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; @@ -145,45 +174,128 @@ tcptran_pipe_init(tcptran_pipe **pipep, void *conn) nni_mtx_init(&p->mtx); 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)) { + ((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_fini(p); return (rv); } 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->keepalive = ep->keepalive; + p->nodelay = ep->nodelay; + p->rcvmax = ep->rcvmax; + p->proto = ep->proto; + p->ep = ep; + *pipep = p; - p->conn = conn; - *pipep = p; return (0); } static void -tcptran_pipe_nego_cancel(nni_aio *aio, int rv) +tcptran_pipe_conn_cancel(nni_aio *aio, int rv) { tcptran_pipe *p = nni_aio_get_prov_data(aio); - nni_mtx_lock(&p->mtx); - if (p->user_negaio != aio) { - nni_mtx_unlock(&p->mtx); + nni_mtx_lock(&p->ep->mtx); + if (aio == p->useraio) { + nni_aio_close(p->negoaio); + nni_aio_close(p->connaio); + nni_aio_close(p->rslvaio); + p->useraio = NULL; + nni_aio_finish_error(aio, rv); + tcptran_pipe_reap(p); + } + nni_mtx_unlock(&p->ep->mtx); +} + +static void +tcptran_pipe_rslv_cb(void *arg) +{ + tcptran_pipe * p = arg; + tcptran_ep * ep = p->ep; + nni_aio * aio = p->rslvaio; + nni_tcp_dialer *d; + int rv; + + nni_mtx_lock(&ep->mtx); + d = ep->dialer; + rv = (d == NULL) ? NNG_ECLOSED : nni_aio_result(aio); + if (rv != 0) { + nni_aio *uaio; + if ((uaio = p->useraio) != NULL) { + p->useraio = NULL; + nni_aio_finish_error(uaio, rv); + } + nni_mtx_unlock(&ep->mtx); + tcptran_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); + nni_tcp_dialer_dial(d, &p->sa, p->connaio); + nni_mtx_unlock(&ep->mtx); +} + +static void +tcptran_pipe_conn_cb(void *arg) +{ + tcptran_pipe *p = arg; + tcptran_ep * ep = p->ep; + nni_aio * aio = p->connaio; + nni_iov iov; + int rv; + + nni_mtx_lock(&ep->mtx); + if ((rv = nni_aio_result(aio)) != 0) { + nni_aio *uaio = p->useraio; + p->useraio = NULL; + nni_mtx_unlock(&ep->mtx); + if (uaio != NULL) { + nni_aio_finish_error(uaio, rv); + } + tcptran_pipe_reap(p); + return; + } + p->conn = nni_aio_get_output(aio, 0); + 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(&ep->mtx); } static void tcptran_pipe_nego_cb(void *arg) { tcptran_pipe *p = arg; - nni_aio * aio = p->negaio; + tcptran_ep * ep = p->ep; + nni_aio * aio = p->negoaio; + nni_aio * uaio; int rv; - nni_mtx_lock(&p->mtx); + nni_mtx_lock(&ep->mtx); + if ((uaio = p->useraio) == NULL) { + nni_mtx_unlock(&ep->mtx); + tcptran_pipe_reap(p); + return; + } + if ((rv = nni_aio_result(aio)) != 0) { - goto done; + goto error; } // We start transmitting before we receive. @@ -200,7 +312,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(&ep->mtx); return; } if (p->gotrxhead < p->wantrxhead) { @@ -209,7 +321,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(&ep->mtx); return; } // We have both sent and received the headers. Lets check the @@ -218,17 +330,25 @@ tcptran_pipe_nego_cb(void *arg) (p->rxlen[2] != 'P') || (p->rxlen[3] != 0) || (p->rxlen[6] != 0) || (p->rxlen[7] != 0)) { rv = NNG_EPROTO; - goto done; + goto error; } NNI_GET16(&p->rxlen[4], p->peer); + p->useraio = NULL; + (void) nni_tcp_conn_set_nodelay(p->conn, p->nodelay); + (void) nni_tcp_conn_set_keepalive(p->conn, p->keepalive); -done: - if ((aio = p->user_negaio) != NULL) { - p->user_negaio = NULL; - nni_aio_finish(aio, rv, 0); - } - nni_mtx_unlock(&p->mtx); + nni_mtx_unlock(&ep->mtx); + + nni_aio_set_output(uaio, 0, p); + nni_aio_finish(uaio, 0, 0); + return; + +error: + p->useraio = NULL; + nni_mtx_unlock(&ep->mtx); + nni_aio_finish_error(uaio, rv); + tcptran_pipe_reap(p); } static void @@ -338,9 +458,7 @@ tcptran_pipe_recv_cb(void *arg) nni_aio_list_remove(aio); msg = p->rxmsg; p->rxmsg = NULL; - if (!nni_list_empty(&p->recvq)) { - tcptran_pipe_recv_start(p); - } + tcptran_pipe_recv_start(p); nni_mtx_unlock(&p->mtx); nni_aio_set_msg(aio, msg); @@ -393,6 +511,14 @@ tcptran_pipe_send_start(tcptran_pipe *p) nni_iov iov[3]; uint64_t len; + if (p->closed) { + while ((aio = nni_list_first(&p->sendq)) != NULL) { + nni_list_remove(&p->sendq, aio); + nni_aio_finish_error(aio, NNG_ECLOSED); + } + return; + } + if ((aio = nni_list_first(&p->sendq)) == NULL) { return; } @@ -474,7 +600,19 @@ tcptran_pipe_recv_start(tcptran_pipe *p) nni_iov iov; NNI_ASSERT(p->rxmsg == NULL); - // Schedule a read of the IPC header. + if (p->closed) { + nni_aio *aio; + while ((aio = nni_list_first(&p->sendq)) != NULL) { + nni_list_remove(&p->sendq, aio); + nni_aio_finish_error(aio, NNG_ECLOSED); + } + return; + } + if (nni_list_empty(&p->recvq)) { + return; + } + + // Schedule a read of the header. rxaio = p->rxaio; iov.iov_buf = p->rxlen; iov.iov_len = sizeof(p->rxlen); @@ -557,71 +695,58 @@ 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_ep_fini(void *arg) { - tcptran_pipe *p = arg; - nni_aio * negaio; - nni_iov iov; - int rv; + tcptran_ep *ep = arg; - if (nni_aio_begin(aio) != 0) { + nni_mtx_lock(&ep->mtx); + ep->fini = true; + if (!nni_list_empty(&ep->pipes)) { + nni_mtx_unlock(&ep->mtx); return; } - 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); - return; + if (ep->dialer != NULL) { + nni_tcp_dialer_fini(ep->dialer); } - 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); -} - -static void -tcptran_dialer_fini(void *arg) -{ - tcptran_dialer *d = arg; - - nni_aio_stop(d->aio); - if (d->dialer != NULL) { - nni_tcp_dialer_fini(d->dialer); + if (ep->listener != NULL) { + nni_tcp_listener_fini(ep->listener); } - nni_aio_fini(d->aio); - nni_mtx_fini(&d->mtx); - NNI_FREE_STRUCT(d); + nni_mtx_fini(&ep->mtx); + NNI_FREE_STRUCT(ep); } static void -tcptran_dialer_close(void *arg) +tcptran_ep_close(void *arg) { - tcptran_dialer *d = arg; + tcptran_ep * ep = arg; + tcptran_pipe *p; - nni_aio_close(d->aio); - nni_tcp_dialer_close(d->dialer); + 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_close(p->txaio); + nni_aio_close(p->rxaio); + if (p->conn != NULL) { + nni_tcp_conn_close(p->conn); + } + } + if (ep->dialer != NULL) { + nni_tcp_dialer_close(ep->dialer); + } + if (ep->listener != NULL) { + nni_tcp_listener_close(ep->listener); + } + nni_mtx_unlock(&ep->mtx); } static int -tcptran_dialer_init(void **dp, nni_url *url, nni_sock *sock) +tcptran_ep_init_dialer(void **dp, nni_url *url, nni_sock *sock) { - tcptran_dialer *d; - int rv; - uint16_t af; + tcptran_ep *ep; + int rv; + uint16_t af; if (strcmp(url->u_scheme, "tcp") == 0) { af = NNG_AF_UNSPEC; @@ -632,7 +757,6 @@ tcptran_dialer_init(void **dp, nni_url *url, nni_sock *sock) } else { return (NNG_EADDRINVAL); } - // Check for invalid URL components. if ((strlen(url->u_path) != 0) && (strcmp(url->u_path, "/") != 0)) { return (NNG_EADDRINVAL); @@ -643,236 +767,34 @@ tcptran_dialer_init(void **dp, nni_url *url, nni_sock *sock) return (NNG_EADDRINVAL); } - if ((d = NNI_ALLOC_STRUCT(d)) == NULL) { + if ((ep = NNI_ALLOC_STRUCT(ep)) == NULL) { return (NNG_ENOMEM); } - nni_mtx_init(&d->mtx); + nni_mtx_init(&ep->mtx); + NNI_LIST_INIT(&ep->pipes, tcptran_pipe, node); + + ep->af = af; + ep->proto = nni_sock_proto_id(sock); + ep->nodelay = true; + ep->keepalive = false; + ep->url = url; - if (((rv = nni_tcp_dialer_init(&d->dialer)) != 0) || - ((rv = nni_aio_init(&d->aio, tcptran_dialer_cb, d)) != 0)) { - tcptran_dialer_fini(d); + if ((rv = nni_tcp_dialer_init(&ep->dialer)) != 0) { + tcptran_ep_fini(ep); return (rv); } - d->url = url; - d->proto = nni_sock_proto_id(sock); - d->nodelay = true; - d->keepalive = false; - d->af = af; - - *dp = d; + *dp = ep; return (0); } - -static void -tcptran_dialer_cb(void *arg) -{ - 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; - } - - 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); - return; - } - - p->proto = d->proto; - p->rcvmax = d->rcvmax; - p->nodelay = d->nodelay; - p->keepalive = d->keepalive; - nni_mtx_unlock(&d->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; - 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_aio_finish_error(aio, rv); - return; - } - d->user_aio = aio; - - 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); - nni_tcp_resolv(d->url->u_hostname, d->url->u_port, d->af, 0, d->aio); - nni_mtx_unlock(&d->mtx); -} - -static int -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)); -} - -static int -tcptran_dialer_get_recvmaxsz(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_size(d->rcvmax, v, szp, t); - nni_mtx_unlock(&d->mtx); - return (rv); -} - -static int -tcptran_dialer_set_recvmaxsz( - void *arg, const void *v, size_t sz, nni_opt_type t) -{ - tcptran_dialer *d = 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); - } - return (rv); -} - -static int -tcptran_dialer_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); - return (rv); -} - -static int -tcptran_dialer_set_nodelay(void *arg, const void *v, size_t sz, nni_opt_type t) -{ - tcptran_dialer *d = 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); - } - return (rv); -} - -static int -tcptran_dialer_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)); -} - -static int -tcptran_dialer_set_keepalive( - void *arg, const void *v, size_t sz, nni_opt_type t) -{ - tcptran_dialer *d = 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); - } - return (rv); -} - -static void -tcptran_listener_fini(void *arg) -{ - tcptran_listener *l = arg; - - nni_aio_stop(l->aio); - if (l->listener != NULL) { - nni_tcp_listener_fini(l->listener); - } - nni_aio_fini(l->aio); - nni_mtx_fini(&l->mtx); - NNI_FREE_STRUCT(l); -} - static int -tcptran_listener_init(void **lp, nni_url *url, nni_sock *sock) +tcptran_ep_init_listener(void **lp, nni_url *url, nni_sock *sock) { - tcptran_listener *l; - int rv; - char * host; - nni_aio * aio; - uint16_t af; + tcptran_ep *ep; + int rv; + char * host; + nni_aio * aio; + uint16_t af; if (strcmp(url->u_scheme, "tcp") == 0) { af = NNG_AF_UNSPEC; @@ -893,11 +815,16 @@ tcptran_listener_init(void **lp, nni_url *url, nni_sock *sock) return (NNG_EADDRINVAL); } - if ((l = NNI_ALLOC_STRUCT(l)) == NULL) { + if ((ep = NNI_ALLOC_STRUCT(ep)) == NULL) { return (NNG_ENOMEM); } - nni_mtx_init(&l->mtx); - l->url = url; + nni_mtx_init(&ep->mtx); + NNI_LIST_INIT(&ep->pipes, tcptran_pipe, node); + ep->af = af; + ep->proto = nni_sock_proto_id(sock); + ep->nodelay = true; + ep->keepalive = false; + ep->url = url; if (strlen(url->u_hostname) == 0) { host = NULL; @@ -905,260 +832,216 @@ tcptran_listener_init(void **lp, nni_url *url, nni_sock *sock) host = url->u_hostname; } - if ((rv = nni_aio_init(&aio, NULL, NULL)) != 0) { - tcptran_listener_fini(l); - return (rv); - } - // XXX: We are doing lookup at listener initialization. There is // a valid argument that this should be done at bind time, but that // would require making bind asynchronous. In some ways this would // be worse than the cost of just waiting here. We always recommend // using local IP addresses rather than names when possible. - nni_aio_set_input(aio, 0, &l->sa); - + if ((rv = nni_aio_init(&aio, NULL, NULL)) != 0) { + tcptran_ep_fini(ep); + return (rv); + } + nni_aio_set_input(aio, 0, &ep->sa); nni_tcp_resolv(host, url->u_port, af, 1, aio); nni_aio_wait(aio); rv = nni_aio_result(aio); nni_aio_fini(aio); if (rv != 0) { - tcptran_listener_fini(l); + tcptran_ep_fini(ep); 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); + if ((rv = nni_tcp_listener_init(&ep->listener)) != 0) { + tcptran_ep_fini(ep); return (rv); } - l->proto = nni_sock_proto_id(sock); - l->nodelay = true; - l->keepalive = false; - l->bsa = l->sa; + ep->bsa = ep->sa; - *lp = l; + *lp = ep; return (0); } static void -tcptran_listener_close(void *arg) -{ - tcptran_listener *l = arg; - - nni_aio_close(l->aio); - nni_tcp_listener_close(l->listener); -} - -static int -tcptran_listener_bind(void *arg) +tcptran_ep_connect(void *arg, nni_aio *aio) { - tcptran_listener *l = arg; - int rv; - - nni_mtx_lock(&l->mtx); - l->bsa = l->sa; - rv = nni_tcp_listener_listen(l->listener, &l->bsa); - nni_mtx_unlock(&l->mtx); - - return (rv); -} + tcptran_ep * ep = arg; + tcptran_pipe *p; + int 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); - } + if (nni_aio_begin(aio) != 0) { return; } - - if (rv != 0) { - nni_mtx_unlock(&l->mtx); + nni_mtx_lock(&ep->mtx); + if ((rv = tcptran_pipe_init(&p, ep)) != 0) { + nni_mtx_unlock(&ep->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); + if ((rv = nni_aio_schedule(aio, tcptran_pipe_conn_cancel, p)) != 0) { + nni_list_remove(&ep->pipes, p); + p->ep = NULL; + nni_mtx_unlock(&ep->mtx); nni_aio_finish_error(aio, rv); + tcptran_pipe_reap(p); 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); + p->useraio = aio; + // Start the name resolution before we try connecting. + nni_aio_set_input(p->rslvaio, 0, &p->sa); + nni_tcp_resolv( + ep->url->u_hostname, ep->url->u_port, ep->af, 0, p->rslvaio); + nni_mtx_unlock(&ep->mtx); } -static void -tcptran_listener_cancel(nni_aio *aio, int rv) +static int +tcptran_ep_get_url(void *arg, void *v, size_t *szp, nni_opt_type t) { - tcptran_listener *l = nni_aio_get_prov_data(aio); + tcptran_ep *ep = arg; - nni_mtx_lock(&l->mtx); - if (l->user_aio != aio) { - nni_mtx_unlock(&l->mtx); - return; + if (ep->listener != NULL) { + char ustr[128]; + char ipstr[48]; // max for IPv6 addresses including [] + char portstr[6]; // max for 16-bit port + nni_ntop(&ep->bsa, ipstr, portstr); + snprintf(ustr, sizeof(ustr), "tcp://%s:%s", ipstr, portstr); + return (nni_copyout_str(ustr, v, szp, t)); } - l->user_aio = NULL; - nni_mtx_unlock(&l->mtx); - nni_aio_abort(l->aio, rv); - nni_aio_finish_error(aio, rv); + return (nni_copyout_str(ep->url->u_rawurl, v, szp, t)); } -static void -tcptran_listener_accept(void *arg, nni_aio *aio) +static int +tcptran_ep_get_recvmaxsz(void *arg, void *v, size_t *szp, nni_opt_type t) { - tcptran_listener *l = arg; - int rv; + tcptran_ep *ep = arg; + 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_aio_finish_error(aio, rv); - return; - } - l->user_aio = aio; - - nni_tcp_listener_accept(l->listener, l->aio); - nni_mtx_unlock(&l->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_listener_set_nodelay( - 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_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); + 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(&ep->mtx); + ep->rcvmax = val; + nni_mtx_unlock(&ep->mtx); } return (rv); } static int -tcptran_listener_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_listener *l = arg; - int rv; - nni_mtx_lock(&l->mtx); - rv = nni_copyout_bool(l->nodelay, v, szp, t); - nni_mtx_unlock(&l->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_listener_set_recvmaxsz( - 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_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); + tcptran_ep *ep = arg; + bool val; + int rv; + if ((rv = nni_copyin_bool(&val, v, sz, t)) == 0) { + nni_mtx_lock(&ep->mtx); + ep->nodelay = val; + nni_mtx_unlock(&ep->mtx); } return (rv); } static int -tcptran_listener_set_keepalive( - void *arg, const void *v, size_t sz, nni_opt_type t) +tcptran_ep_get_keepalive(void *arg, void *v, size_t *szp, 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); - } + 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_listener_get_keepalive(void *arg, void *v, size_t *szp, nni_opt_type t) +tcptran_ep_set_keepalive(void *arg, const void *v, size_t sz, 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); + tcptran_ep *ep = arg; + bool val; + int rv; + if ((rv = nni_copyin_bool(&val, v, sz, t)) == 0) { + nni_mtx_lock(&ep->mtx); + ep->keepalive = val; + nni_mtx_unlock(&ep->mtx); + } return (rv); } static int -tcptran_listener_get_url(void *arg, void *v, size_t *szp, nni_opt_type t) +tcptran_ep_bind(void *arg) { - tcptran_listener *l = arg; - char ustr[128]; - char ipstr[48]; // max for IPv6 addresses including [] - char portstr[6]; // max for 16-bit port - - nni_ntop(&l->bsa, ipstr, portstr); - snprintf(ustr, sizeof(ustr), "tcp://%s:%s", ipstr, portstr); - return (nni_copyout_str(ustr, v, szp, t)); + tcptran_ep *ep = arg; + int rv; + + nni_mtx_lock(&ep->mtx); + ep->bsa = ep->sa; + rv = nni_tcp_listener_listen(ep->listener, &ep->bsa); + nni_mtx_unlock(&ep->mtx); + + return (rv); } -static int -tcptran_listener_get_recvmaxsz(void *arg, void *v, size_t *szp, nni_opt_type t) +static void +tcptran_ep_accept(void *arg, nni_aio *aio) { - tcptran_listener *l = arg; - int rv; + tcptran_ep * ep = arg; + tcptran_pipe *p; + int rv; - nni_mtx_lock(&l->mtx); - rv = nni_copyout_size(l->rcvmax, v, szp, t); - nni_mtx_unlock(&l->mtx); - return (rv); + if (nni_aio_begin(aio) != 0) { + return; + } + nni_mtx_lock(&ep->mtx); + if ((rv = tcptran_pipe_init(&p, ep)) != 0) { + nni_mtx_unlock(&ep->mtx); + nni_aio_finish_error(aio, rv); + return; + } + if ((rv = nni_aio_schedule(aio, tcptran_pipe_conn_cancel, p)) != 0) { + nni_list_remove(&ep->pipes, p); + p->ep = NULL; + nni_mtx_unlock(&ep->mtx); + nni_aio_finish_error(aio, rv); + tcptran_pipe_reap(p); + return; + } + p->useraio = aio; + nni_tcp_listener_accept(ep->listener, p->connaio); + nni_mtx_unlock(&ep->mtx); } static int -tcptran_listener_get_locaddr(void *arg, void *buf, size_t *szp, nni_opt_type t) +tcptran_ep_get_locaddr(void *arg, void *buf, size_t *szp, nni_opt_type t) { - tcptran_listener *l = arg; - int rv; + tcptran_ep *ep = arg; + int rv; - nni_mtx_lock(&l->mtx); - rv = nni_copyout_sockaddr(&l->bsa, buf, szp, t); - nni_mtx_unlock(&l->mtx); + nni_mtx_lock(&ep->mtx); + rv = nni_copyout_sockaddr(&ep->bsa, buf, szp, t); + nni_mtx_unlock(&ep->mtx); return (rv); } @@ -1203,7 +1086,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,27 +1098,27 @@ 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, }, { .o_name = NNG_OPT_URL, .o_type = NNI_TYPE_STRING, - .o_get = tcptran_dialer_get_url, + .o_get = tcptran_ep_get_url, }, { .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,32 +1131,32 @@ 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, }, { .o_name = NNG_OPT_LOCADDR, .o_type = NNI_TYPE_SOCKADDR, - .o_get = tcptran_listener_get_locaddr, + .o_get = tcptran_ep_get_locaddr, }, { .o_name = NNG_OPT_URL, .o_type = NNI_TYPE_STRING, - .o_get = tcptran_listener_get_url, + .o_get = tcptran_ep_get_url, }, { .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 @@ -1284,19 +1166,19 @@ static nni_tran_option tcptran_listener_options[] = { }; static nni_tran_dialer_ops tcptran_dialer_ops = { - .d_init = tcptran_dialer_init, - .d_fini = tcptran_dialer_fini, - .d_connect = tcptran_dialer_connect, - .d_close = tcptran_dialer_close, + .d_init = tcptran_ep_init_dialer, + .d_fini = tcptran_ep_fini, + .d_connect = tcptran_ep_connect, + .d_close = tcptran_ep_close, .d_options = tcptran_dialer_options, }; static nni_tran_listener_ops tcptran_listener_ops = { - .l_init = tcptran_listener_init, - .l_fini = tcptran_listener_fini, - .l_bind = tcptran_listener_bind, - .l_accept = tcptran_listener_accept, - .l_close = tcptran_listener_close, + .l_init = tcptran_ep_init_listener, + .l_fini = tcptran_ep_fini, + .l_bind = tcptran_ep_bind, + .l_accept = tcptran_ep_accept, + .l_close = tcptran_ep_close, .l_options = tcptran_listener_options, }; |