aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
authorGarrett D'Amore <garrett@damore.org>2018-07-31 12:33:58 -0700
committerGarrett D'Amore <garrett@damore.org>2018-08-05 18:45:04 +0300
commitd7f7c896c0ede24249ef63b1e45b1878bf4bd473 (patch)
tree32eece7d91a648f24cb174096fb9667cab978f37 /src
parentccc24a8e508131a2226474642a038baaa2cbcc8c (diff)
downloadnng-d7f7c896c0ede24249ef63b1e45b1878bf4bd473.tar.gz
nng-d7f7c896c0ede24249ef63b1e45b1878bf4bd473.tar.bz2
nng-d7f7c896c0ede24249ef63b1e45b1878bf4bd473.zip
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.
Diffstat (limited to 'src')
-rw-r--r--src/core/aio.c10
-rw-r--r--src/core/dialer.c12
-rw-r--r--src/core/listener.c19
-rw-r--r--src/core/pipe.c67
-rw-r--r--src/core/pipe.h5
-rw-r--r--src/core/socket.c52
-rw-r--r--src/core/socket.h2
-rw-r--r--src/core/sockimpl.h5
-rw-r--r--src/core/transport.h12
-rw-r--r--src/platform/posix/posix_ipcconn.c8
-rw-r--r--src/platform/posix/posix_tcpconn.c8
-rw-r--r--src/transport/ipc/ipc.c567
-rw-r--r--src/transport/tcp/tcp.c753
-rw-r--r--src/transport/tls/tls.c530
-rw-r--r--src/transport/zerotier/zerotier.c158
15 files changed, 986 insertions, 1222 deletions
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
@@ -1139,26 +1063,14 @@ tcptran_listener_get_url(void *arg, void *v, size_t *szp, nni_opt_type 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);
}
@@ -922,98 +981,26 @@ tlstran_listener_bind(void *arg)
}
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);
}
}
@@ -2016,96 +2027,50 @@ zt_pipe_get_node(void *arg, void *buf, size_t *szp, nni_opt_type 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,