aboutsummaryrefslogtreecommitdiff
path: root/src/transport
diff options
context:
space:
mode:
Diffstat (limited to 'src/transport')
-rw-r--r--src/transport/inproc/inproc.c135
-rw-r--r--src/transport/ipc/ipc.c8
-rw-r--r--src/transport/tcp/tcp.c8
-rw-r--r--src/transport/tls/tls.c8
-rw-r--r--src/transport/ws/websocket.c16
-rw-r--r--src/transport/zerotier/zerotier.c12
6 files changed, 87 insertions, 100 deletions
diff --git a/src/transport/inproc/inproc.c b/src/transport/inproc/inproc.c
index 6329a627..f9f438ab 100644
--- a/src/transport/inproc/inproc.c
+++ b/src/transport/inproc/inproc.c
@@ -217,14 +217,11 @@ nni_inproc_ep_fini(void *arg)
}
static void
-nni_inproc_conn_finish(nni_aio *aio, int rv)
+nni_inproc_conn_finish(nni_aio *aio, int rv, nni_inproc_pipe *pipe)
{
nni_inproc_ep *ep = nni_aio_get_prov_data(aio);
- void * pipe;
nni_aio_list_remove(aio);
- pipe = nni_aio_get_pipe(aio);
- nni_aio_set_pipe(aio, NULL);
if ((ep != NULL) && (ep->mode != NNI_EP_MODE_LISTEN) &&
nni_list_empty(&ep->aios)) {
@@ -232,11 +229,10 @@ nni_inproc_conn_finish(nni_aio *aio, int rv)
}
if (rv == 0) {
- nni_aio_finish_pipe(aio, pipe);
+ nni_aio_set_output(aio, 0, pipe);
+ nni_aio_finish(aio, 0, 0);
} else {
- if (pipe != NULL) {
- nni_inproc_pipe_fini(pipe);
- }
+ NNI_ASSERT(pipe == NULL);
nni_aio_finish_error(aio, rv);
}
}
@@ -255,67 +251,81 @@ nni_inproc_ep_close(void *arg)
// Notify any waiting clients that we are closed.
while ((client = nni_list_first(&ep->clients)) != NULL) {
while ((aio = nni_list_first(&client->aios)) != NULL) {
- nni_inproc_conn_finish(aio, NNG_ECONNREFUSED);
+ nni_inproc_conn_finish(aio, NNG_ECONNREFUSED, NULL);
}
nni_list_remove(&ep->clients, client);
}
while ((aio = nni_list_first(&ep->aios)) != NULL) {
- nni_inproc_conn_finish(aio, NNG_ECLOSED);
+ nni_inproc_conn_finish(aio, NNG_ECLOSED, NULL);
}
nni_mtx_unlock(&nni_inproc.mx);
}
static void
-nni_inproc_accept_clients(nni_inproc_ep *server)
+nni_inproc_accept_clients(nni_inproc_ep *srv)
{
- nni_inproc_ep * client, *nclient;
- nni_aio * saio, *caio;
- nni_inproc_pair *pair;
- int rv;
+ nni_inproc_ep *cli, *nclient;
+
+ nclient = nni_list_first(&srv->clients);
+ while ((cli = nclient) != NULL) {
+ nni_aio *caio;
+ nclient = nni_list_next(&srv->clients, nclient);
+ NNI_LIST_FOREACH (&cli->aios, caio) {
- nclient = nni_list_first(&server->clients);
- while ((client = nclient) != NULL) {
- nclient = nni_list_next(&server->clients, nclient);
- NNI_LIST_FOREACH (&client->aios, caio) {
- if ((saio = nni_list_first(&server->aios)) == NULL) {
+ nni_inproc_pipe *cpipe;
+ nni_inproc_pipe *spipe;
+ nni_inproc_pair *pair;
+ nni_aio * saio;
+ int rv;
+
+ if ((saio = nni_list_first(&srv->aios)) == NULL) {
// No outstanding accept() calls.
break;
}
if ((pair = NNI_ALLOC_STRUCT(pair)) == NULL) {
- nni_inproc_conn_finish(caio, NNG_ENOMEM);
- nni_inproc_conn_finish(saio, NNG_ENOMEM);
+ nni_inproc_conn_finish(caio, NNG_ENOMEM, NULL);
+ nni_inproc_conn_finish(saio, NNG_ENOMEM, NULL);
continue;
}
+ nni_mtx_init(&pair->mx);
- if (((rv = nni_msgq_init(&pair->q[0], 4)) != 0) ||
+ spipe = cpipe = NULL;
+ if (((rv = nni_inproc_pipe_init(&cpipe, cli)) != 0) ||
+ ((rv = nni_inproc_pipe_init(&spipe, srv)) != 0) ||
+ ((rv = nni_msgq_init(&pair->q[0], 4)) != 0) ||
((rv = nni_msgq_init(&pair->q[1], 4)) != 0)) {
+
+ if (cpipe != NULL) {
+ nni_inproc_pipe_fini(cpipe);
+ }
+ if (spipe != NULL) {
+ nni_inproc_pipe_fini(spipe);
+ }
+ nni_inproc_conn_finish(caio, rv, NULL);
+ nni_inproc_conn_finish(saio, rv, NULL);
nni_inproc_pair_destroy(pair);
- nni_inproc_conn_finish(caio, rv);
- nni_inproc_conn_finish(saio, rv);
continue;
}
- nni_mtx_init(&pair->mx);
-
- pair->pipes[0] = nni_aio_get_pipe(caio);
- pair->pipes[1] = nni_aio_get_pipe(saio);
- pair->pipes[0]->rq = pair->pipes[1]->wq = pair->q[0];
- pair->pipes[1]->rq = pair->pipes[0]->wq = pair->q[1];
- pair->pipes[0]->pair = pair->pipes[1]->pair = pair;
- pair->pipes[1]->peer = pair->pipes[0]->proto;
- pair->pipes[0]->peer = pair->pipes[1]->proto;
- pair->refcnt = 2;
-
- nni_inproc_conn_finish(caio, 0);
- nni_inproc_conn_finish(saio, 0);
+ spipe->peer = cpipe->proto;
+ cpipe->peer = spipe->proto;
+ pair->pipes[0] = cpipe;
+ pair->pipes[1] = spipe;
+ pair->refcnt = 2;
+ cpipe->pair = spipe->pair = pair;
+ cpipe->rq = spipe->wq = pair->q[0];
+ cpipe->wq = spipe->rq = pair->q[1];
+
+ nni_inproc_conn_finish(caio, 0, cpipe);
+ nni_inproc_conn_finish(saio, 0, spipe);
}
- if (nni_list_first(&client->aios) == NULL) {
- // No more outstanding client connects. Normally
- // there should only be one.
- if (nni_list_active(&server->clients, client)) {
- nni_list_remove(&server->clients, client);
+ if (nni_list_first(&cli->aios) == NULL) {
+ // No more outstanding client connects.
+ // Normally there should only be one.
+ if (nni_list_active(&srv->clients, cli)) {
+ nni_list_remove(&srv->clients, cli);
}
}
}
@@ -324,17 +334,12 @@ nni_inproc_accept_clients(nni_inproc_ep *server)
static void
nni_inproc_ep_cancel(nni_aio *aio, int rv)
{
- nni_inproc_ep * ep = nni_aio_get_prov_data(aio);
- nni_inproc_pipe *pipe;
+ nni_inproc_ep *ep = nni_aio_get_prov_data(aio);
nni_mtx_lock(&nni_inproc.mx);
if (nni_aio_list_active(aio)) {
nni_aio_list_remove(aio);
nni_list_node_remove(&ep->node);
- if ((pipe = nni_aio_get_pipe(aio)) != NULL) {
- nni_aio_set_pipe(aio, NULL);
- nni_inproc_pipe_fini(pipe);
- }
nni_aio_finish_error(aio, rv);
}
nni_mtx_unlock(&nni_inproc.mx);
@@ -343,10 +348,9 @@ nni_inproc_ep_cancel(nni_aio *aio, int rv)
static void
nni_inproc_ep_connect(void *arg, nni_aio *aio)
{
- nni_inproc_ep * ep = arg;
- nni_inproc_ep * server;
- int rv;
- nni_inproc_pipe *pipe;
+ nni_inproc_ep *ep = arg;
+ nni_inproc_ep *server;
+ int rv;
if (ep->mode != NNI_EP_MODE_DIAL) {
nni_aio_finish_error(aio, NNG_EINVAL);
@@ -359,13 +363,6 @@ nni_inproc_ep_connect(void *arg, nni_aio *aio)
return;
}
- if ((rv = nni_inproc_pipe_init(&pipe, ep)) != 0) {
- nni_aio_finish_error(aio, rv);
- nni_mtx_unlock(&nni_inproc.mx);
- return;
- }
- nni_aio_set_pipe(aio, pipe);
-
// Find a server.
NNI_LIST_FOREACH (&nni_inproc.servers, server) {
if (strcmp(server->addr, ep->addr) == 0) {
@@ -373,7 +370,7 @@ nni_inproc_ep_connect(void *arg, nni_aio *aio)
}
}
if (server == NULL) {
- nni_inproc_conn_finish(aio, NNG_ECONNREFUSED);
+ nni_inproc_conn_finish(aio, NNG_ECONNREFUSED, NULL);
nni_mtx_unlock(&nni_inproc.mx);
return;
}
@@ -407,9 +404,8 @@ nni_inproc_ep_bind(void *arg)
static void
nni_inproc_ep_accept(void *arg, nni_aio *aio)
{
- nni_inproc_ep * ep = arg;
- nni_inproc_pipe *pipe;
- int rv;
+ nni_inproc_ep *ep = arg;
+ int rv;
nni_mtx_lock(&nni_inproc.mx);
@@ -419,16 +415,7 @@ nni_inproc_ep_accept(void *arg, nni_aio *aio)
}
// We are already on the master list of servers, thanks to bind.
-
- if ((rv = nni_inproc_pipe_init(&pipe, ep)) != 0) {
- nni_aio_finish_error(aio, rv);
- nni_mtx_unlock(&nni_inproc.mx);
- return;
- }
- nni_aio_set_pipe(aio, pipe);
-
- // Insert us into the pending server aios, and then run the
- // accept list.
+ // Insert us into pending server aios, and then run accept list.
nni_aio_list_append(&ep->aios, aio);
nni_inproc_accept_clients(ep);
nni_mtx_unlock(&nni_inproc.mx);
diff --git a/src/transport/ipc/ipc.c b/src/transport/ipc/ipc.c
index 6475e43b..9ebf3145 100644
--- a/src/transport/ipc/ipc.c
+++ b/src/transport/ipc/ipc.c
@@ -579,20 +579,20 @@ nni_ipc_ep_finish(nni_ipc_ep *ep)
if ((rv = nni_aio_result(ep->aio)) != 0) {
goto done;
}
- NNI_ASSERT(nni_aio_get_pipe(ep->aio) != NULL);
+ NNI_ASSERT(nni_aio_get_output(ep->aio, 0) != NULL);
// Attempt to allocate the parent pipe. If this fails we'll
// drop the connection (ENOMEM probably).
- rv = nni_ipc_pipe_init(&pipe, ep, nni_aio_get_pipe(ep->aio));
+ rv = nni_ipc_pipe_init(&pipe, ep, nni_aio_get_output(ep->aio, 0));
done:
- nni_aio_set_pipe(ep->aio, NULL);
aio = ep->user_aio;
ep->user_aio = NULL;
if ((aio != NULL) && (rv == 0)) {
NNI_ASSERT(pipe != NULL);
- nni_aio_finish_pipe(aio, pipe);
+ nni_aio_set_output(aio, 0, pipe);
+ nni_aio_finish(aio, 0, 0);
return;
}
diff --git a/src/transport/tcp/tcp.c b/src/transport/tcp/tcp.c
index 1504d0ee..475a77ff 100644
--- a/src/transport/tcp/tcp.c
+++ b/src/transport/tcp/tcp.c
@@ -631,19 +631,19 @@ nni_tcp_ep_finish(nni_tcp_ep *ep)
if ((rv = nni_aio_result(ep->aio)) != 0) {
goto done;
}
- NNI_ASSERT(nni_aio_get_pipe(ep->aio) != NULL);
+ NNI_ASSERT(nni_aio_get_output(ep->aio, 0) != NULL);
// Attempt to allocate the parent pipe. If this fails we'll
// drop the connection (ENOMEM probably).
- rv = nni_tcp_pipe_init(&pipe, ep, nni_aio_get_pipe(ep->aio));
+ rv = nni_tcp_pipe_init(&pipe, ep, nni_aio_get_output(ep->aio, 0));
done:
- nni_aio_set_pipe(ep->aio, NULL);
aio = ep->user_aio;
ep->user_aio = NULL;
if ((aio != NULL) && (rv == 0)) {
- nni_aio_finish_pipe(aio, pipe);
+ nni_aio_set_output(aio, 0, pipe);
+ nni_aio_finish(aio, 0, 0);
return;
}
if (pipe != NULL) {
diff --git a/src/transport/tls/tls.c b/src/transport/tls/tls.c
index 09c59582..fc993501 100644
--- a/src/transport/tls/tls.c
+++ b/src/transport/tls/tls.c
@@ -653,19 +653,19 @@ nni_tls_ep_finish(nni_tls_ep *ep)
if ((rv = nni_aio_result(ep->aio)) != 0) {
goto done;
}
- NNI_ASSERT(nni_aio_get_pipe(ep->aio) != NULL);
+ NNI_ASSERT(nni_aio_get_output(ep->aio, 0) != NULL);
// Attempt to allocate the parent pipe. If this fails we'll
// drop the connection (ENOMEM probably).
- rv = nni_tls_pipe_init(&pipe, ep, nni_aio_get_pipe(ep->aio));
+ rv = nni_tls_pipe_init(&pipe, ep, nni_aio_get_output(ep->aio, 0));
done:
- nni_aio_set_pipe(ep->aio, NULL);
aio = ep->user_aio;
ep->user_aio = NULL;
if ((aio != NULL) && (rv == 0)) {
- nni_aio_finish_pipe(aio, pipe);
+ nni_aio_set_output(aio, 0, pipe);
+ nni_aio_finish(aio, 0, 0);
return;
}
if (pipe != NULL) {
diff --git a/src/transport/ws/websocket.c b/src/transport/ws/websocket.c
index 2fe7804c..9740cdde 100644
--- a/src/transport/ws/websocket.c
+++ b/src/transport/ws/websocket.c
@@ -205,7 +205,6 @@ ws_pipe_init(ws_pipe **pipep, ws_ep *ep, void *ws)
{
ws_pipe *p;
int rv;
- nni_aio *aio;
if ((p = NNI_ALLOC_STRUCT(p)) == NULL) {
return (NNG_ENOMEM);
@@ -225,11 +224,6 @@ ws_pipe_init(ws_pipe **pipep, ws_ep *ep, void *ws)
p->lproto = ep->lproto;
p->ws = ws;
- if ((aio = nni_list_first(&ep->aios)) != NULL) {
- nni_aio_list_remove(aio);
- nni_aio_finish_pipe(aio, p);
- }
-
*pipep = p;
return (0);
}
@@ -621,7 +615,7 @@ ws_ep_conn_cb(void *arg)
nni_mtx_lock(&ep->mtx);
if (nni_aio_result(caio) == 0) {
- ws = nni_aio_get_pipe(caio);
+ ws = nni_aio_get_output(caio, 0);
}
if ((uaio = nni_list_first(&ep->aios)) == NULL) {
// The client stopped caring about this!
@@ -639,7 +633,8 @@ ws_ep_conn_cb(void *arg)
nni_ws_fini(ws);
nni_aio_finish_error(uaio, rv);
} else {
- nni_aio_finish_pipe(uaio, p);
+ nni_aio_set_output(uaio, 0, p);
+ nni_aio_finish(uaio, 0, 0);
}
nni_mtx_unlock(&ep->mtx);
}
@@ -672,7 +667,7 @@ ws_ep_acc_cb(void *arg)
nni_aio_finish_error(uaio, rv);
}
} else {
- nni_ws *ws = nni_aio_get_pipe(aaio);
+ nni_ws *ws = nni_aio_get_output(aaio, 0);
if (uaio != NULL) {
ws_pipe *p;
// Make a pipe
@@ -681,7 +676,8 @@ ws_ep_acc_cb(void *arg)
nni_ws_close(ws);
nni_aio_finish_error(uaio, rv);
} else {
- nni_aio_finish_pipe(uaio, p);
+ nni_aio_set_output(uaio, 0, p);
+ nni_aio_finish(uaio, 0, 0);
}
}
}
diff --git a/src/transport/zerotier/zerotier.c b/src/transport/zerotier/zerotier.c
index d6762d25..6dddbb9a 100644
--- a/src/transport/zerotier/zerotier.c
+++ b/src/transport/zerotier/zerotier.c
@@ -690,7 +690,8 @@ zt_ep_recv_conn_ack(zt_ep *ep, uint64_t raddr, const uint8_t *data, size_t len)
nni_idhash_remove(ztn->zn_eps, ep->ze_laddr);
ep->ze_laddr = 0;
- nni_aio_finish_pipe(aio, p);
+ nni_aio_set_output(aio, 0, p);
+ nni_aio_finish(aio, 0, 0);
}
static void
@@ -2351,7 +2352,8 @@ zt_ep_doaccept(zt_ep *ep)
}
p->zp_peer = creq.cr_proto;
zt_pipe_send_conn_ack(p);
- nni_aio_finish_pipe(aio, p);
+ nni_aio_set_output(aio, 0, p);
+ nni_aio_finish(aio, 0, 0);
}
}
@@ -2398,14 +2400,16 @@ zt_ep_conn_req_cb(void *arg)
ep->ze_creq_active = 0;
switch ((rv = nni_aio_result(aio))) {
case 0:
+ p = nni_aio_get_output(aio, 0);
// Already canceled, or already handled?
if ((uaio = nni_list_first(&ep->ze_aios)) != NULL) {
nni_aio_list_remove(uaio);
- nni_aio_finish_pipe(uaio, nni_aio_get_pipe(aio));
+ nni_aio_set_output(uaio, 0, p);
+ nni_aio_finish(uaio, 0, 0);
} else {
// We have a pipe, but nowhere to stick it.
// Just discard it.
- zt_pipe_fini(nni_aio_get_pipe(aio));
+ zt_pipe_fini(p);
}
ep->ze_creq_try = 0;
break;
generated by cgit v1.2.3-70-g09d2 (git 2.46.0) at 2026-01-19 06:49:49 +0000