aboutsummaryrefslogtreecommitdiff
path: root/src/transport
diff options
context:
space:
mode:
authorGarrett D'Amore <garrett@damore.org>2018-08-08 06:40:41 +0300
committerGarrett D'Amore <garrett@damore.org>2018-08-13 09:55:32 +0500
commit5b61fa0e04348ee7661dbb19bc9c50ca22af74dc (patch)
tree456b452db00d26ec245343c8378af4c6e8a4c146 /src/transport
parente7e0306b9c4ff39597c681cb5cba724edc5cca0d (diff)
downloadnng-5b61fa0e04348ee7661dbb19bc9c50ca22af74dc.tar.gz
nng-5b61fa0e04348ee7661dbb19bc9c50ca22af74dc.tar.bz2
nng-5b61fa0e04348ee7661dbb19bc9c50ca22af74dc.zip
fixes #632 TCP start elimination
While here, perform a more aggressive close of the pipe on reaping (IPC).
Diffstat (limited to 'src/transport')
-rw-r--r--src/transport/ipc/ipc.c3
-rw-r--r--src/transport/tcp/tcp.c970
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,
};