aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/core/defs.h68
-rw-r--r--src/core/dialer.c4
-rw-r--r--src/core/listener.c7
-rw-r--r--src/transport/inproc/inproc.c24
-rw-r--r--src/transport/ipc/ipc.c49
-rw-r--r--src/transport/tcp/tcp.c47
-rw-r--r--src/transport/tls/tls.c62
7 files changed, 95 insertions, 166 deletions
diff --git a/src/core/defs.h b/src/core/defs.h
index eaba7428..1dffdad0 100644
--- a/src/core/defs.h
+++ b/src/core/defs.h
@@ -85,50 +85,50 @@ typedef struct {
#define NNI_ALLOC_STRUCTS(s, n) nni_zalloc(sizeof(*s) * n)
#define NNI_FREE_STRUCTS(s, n) nni_free(s, sizeof(*s) * n)
-#define NNI_PUT16(ptr, u) \
- do { \
- (ptr)[0] = (uint8_t)(((uint16_t)(u)) >> 8); \
- (ptr)[1] = (uint8_t)((uint16_t)(u)); \
+#define NNI_PUT16(ptr, u) \
+ do { \
+ (ptr)[0] = (uint8_t)(((uint16_t)(u)) >> 8u); \
+ (ptr)[1] = (uint8_t)((uint16_t)(u)); \
} while (0)
-#define NNI_PUT32(ptr, u) \
- do { \
- (ptr)[0] = (uint8_t)(((uint32_t)(u)) >> 24); \
- (ptr)[1] = (uint8_t)(((uint32_t)(u)) >> 16); \
- (ptr)[2] = (uint8_t)(((uint32_t)(u)) >> 8); \
- (ptr)[3] = (uint8_t)((uint32_t)(u)); \
+#define NNI_PUT32(ptr, u) \
+ do { \
+ (ptr)[0] = (uint8_t)(((uint32_t)(u)) >> 24u); \
+ (ptr)[1] = (uint8_t)(((uint32_t)(u)) >> 16u); \
+ (ptr)[2] = (uint8_t)(((uint32_t)(u)) >> 8u); \
+ (ptr)[3] = (uint8_t)((uint32_t)(u)); \
} while (0)
-#define NNI_PUT64(ptr, u) \
- do { \
- (ptr)[0] = (uint8_t)(((uint64_t)(u)) >> 56); \
- (ptr)[1] = (uint8_t)(((uint64_t)(u)) >> 48); \
- (ptr)[2] = (uint8_t)(((uint64_t)(u)) >> 40); \
- (ptr)[3] = (uint8_t)(((uint64_t)(u)) >> 32); \
- (ptr)[4] = (uint8_t)(((uint64_t)(u)) >> 24); \
- (ptr)[5] = (uint8_t)(((uint64_t)(u)) >> 16); \
- (ptr)[6] = (uint8_t)(((uint64_t)(u)) >> 8); \
- (ptr)[7] = (uint8_t)((uint64_t)(u)); \
+#define NNI_PUT64(ptr, u) \
+ do { \
+ (ptr)[0] = (uint8_t)(((uint64_t)(u)) >> 56u); \
+ (ptr)[1] = (uint8_t)(((uint64_t)(u)) >> 48u); \
+ (ptr)[2] = (uint8_t)(((uint64_t)(u)) >> 40u); \
+ (ptr)[3] = (uint8_t)(((uint64_t)(u)) >> 32u); \
+ (ptr)[4] = (uint8_t)(((uint64_t)(u)) >> 24u); \
+ (ptr)[5] = (uint8_t)(((uint64_t)(u)) >> 16u); \
+ (ptr)[6] = (uint8_t)(((uint64_t)(u)) >> 8u); \
+ (ptr)[7] = (uint8_t)((uint64_t)(u)); \
} while (0)
-#define NNI_GET16(ptr, v) \
- v = (((uint16_t)((uint8_t)(ptr)[0])) << 8) + \
+#define NNI_GET16(ptr, v) \
+ v = (((uint16_t)((uint8_t)(ptr)[0])) << 8u) + \
(((uint16_t)(uint8_t)(ptr)[1]))
-#define NNI_GET32(ptr, v) \
- v = (((uint32_t)((uint8_t)(ptr)[0])) << 24) + \
- (((uint32_t)((uint8_t)(ptr)[1])) << 16) + \
- (((uint32_t)((uint8_t)(ptr)[2])) << 8) + \
+#define NNI_GET32(ptr, v) \
+ v = (((uint32_t)((uint8_t)(ptr)[0])) << 24u) + \
+ (((uint32_t)((uint8_t)(ptr)[1])) << 16u) + \
+ (((uint32_t)((uint8_t)(ptr)[2])) << 8u) + \
(((uint32_t)(uint8_t)(ptr)[3]))
-#define NNI_GET64(ptr, v) \
- v = (((uint64_t)((uint8_t)(ptr)[0])) << 56) + \
- (((uint64_t)((uint8_t)(ptr)[1])) << 48) + \
- (((uint64_t)((uint8_t)(ptr)[2])) << 40) + \
- (((uint64_t)((uint8_t)(ptr)[3])) << 32) + \
- (((uint64_t)((uint8_t)(ptr)[4])) << 24) + \
- (((uint64_t)((uint8_t)(ptr)[5])) << 16) + \
- (((uint64_t)((uint8_t)(ptr)[6])) << 8) + \
+#define NNI_GET64(ptr, v) \
+ v = (((uint64_t)((uint8_t)(ptr)[0])) << 56u) + \
+ (((uint64_t)((uint8_t)(ptr)[1])) << 48u) + \
+ (((uint64_t)((uint8_t)(ptr)[2])) << 40u) + \
+ (((uint64_t)((uint8_t)(ptr)[3])) << 32u) + \
+ (((uint64_t)((uint8_t)(ptr)[4])) << 24u) + \
+ (((uint64_t)((uint8_t)(ptr)[5])) << 16u) + \
+ (((uint64_t)((uint8_t)(ptr)[6])) << 8u) + \
(((uint64_t)(uint8_t)(ptr)[7]))
// This increments a pointer a fixed number of byte cells.
diff --git a/src/core/dialer.c b/src/core/dialer.c
index 8a463452..fe9cb92f 100644
--- a/src/core/dialer.c
+++ b/src/core/dialer.c
@@ -1,5 +1,5 @@
//
-// Copyright 2019 Staysail Systems, Inc. <info@staysail.tech>
+// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
// Copyright 2018 Capitar IT Group BV <info@capitar.com>
// Copyright 2018 Devolutions <info@devolutions.net>
//
@@ -345,10 +345,12 @@ dialer_connect_cb(void *arg)
break;
case NNG_ECLOSED: // No further action.
case NNG_ECANCELED: // No further action.
+ nni_dialer_bump_error(d, rv);
break;
case NNG_ECONNREFUSED:
case NNG_ETIMEDOUT:
default:
+ nni_dialer_bump_error(d, rv);
if (user_aio == NULL) {
nni_dialer_timer_start(d);
} else {
diff --git a/src/core/listener.c b/src/core/listener.c
index 87d2c532..1fde6e36 100644
--- a/src/core/listener.c
+++ b/src/core/listener.c
@@ -324,8 +324,9 @@ listener_accept_cb(void *arg)
{
nni_listener *l = arg;
nni_aio * aio = l->l_acc_aio;
+ int rv;
- switch (nni_aio_result(aio)) {
+ switch ((rv = nni_aio_result(aio))) {
case 0:
BUMP_STAT(&l->l_stats.s_accept);
nni_listener_add_pipe(l, nni_aio_get_output(aio, 0));
@@ -335,10 +336,12 @@ listener_accept_cb(void *arg)
case NNG_ECONNRESET: // remote condition, no cool down
case NNG_ETIMEDOUT: // No need to sleep, we timed out already.
case NNG_EPEERAUTH: // peer validation failure
+ nni_listener_bump_error(l, rv);
listener_accept_start(l);
break;
case NNG_ECLOSED: // no further action
case NNG_ECANCELED: // no further action
+ nni_listener_bump_error(l, rv);
break;
default:
// We don't really know why we failed, but we back off
@@ -346,6 +349,7 @@ listener_accept_cb(void *arg)
// to system failures (resource exhaustion) and we hope
// by not thrashing we give the system a chance to
// recover. 100 ms is enough to cool down.
+ nni_listener_bump_error(l, rv);
nni_sleep_aio(100, l->l_tmo_aio);
break;
}
@@ -371,6 +375,7 @@ nni_listener_start(nni_listener *l, int flags)
}
if ((rv = l->l_ops.l_bind(l->l_data)) != 0) {
+ nni_listener_bump_error(l, rv);
nni_atomic_flag_reset(&l->l_started);
return (rv);
}
diff --git a/src/transport/inproc/inproc.c b/src/transport/inproc/inproc.c
index 518871f4..a1a2dd5d 100644
--- a/src/transport/inproc/inproc.c
+++ b/src/transport/inproc/inproc.c
@@ -33,7 +33,6 @@ struct inproc_pipe {
inproc_pair * pair;
inproc_queue *recv_queue;
inproc_queue *send_queue;
- nni_pipe * npipe;
uint16_t peer;
uint16_t proto;
};
@@ -62,8 +61,6 @@ struct inproc_ep {
nni_list aios;
size_t rcvmax;
nni_mtx mtx;
- nni_dialer * ndialer;
- nni_listener *nlistener;
};
// nni_inproc is our global state - this contains the list of active endpoints
@@ -114,9 +111,8 @@ inproc_pipe_alloc(inproc_pipe **pipep, inproc_ep *ep)
static int
inproc_pipe_init(void *arg, nni_pipe *p)
{
- inproc_pipe *pipe = arg;
- pipe->npipe = p;
-
+ NNI_ARG_UNUSED(arg);
+ NNI_ARG_UNUSED(p);
return (0);
}
@@ -298,7 +294,6 @@ inproc_dialer_init(void **epp, nni_url *url, nni_dialer *ndialer)
ep->listener = false;
ep->proto = nni_sock_proto_id(sock);
ep->rcvmax = 0;
- ep->ndialer = ndialer;
NNI_LIST_INIT(&ep->clients, inproc_ep, node);
nni_aio_list_init(&ep->aios);
@@ -322,7 +317,6 @@ inproc_listener_init(void **epp, nni_url *url, nni_listener *nlistener)
ep->listener = true;
ep->proto = nni_sock_proto_id(sock);
ep->rcvmax = 0;
- ep->nlistener = nlistener;
NNI_LIST_INIT(&ep->clients, inproc_ep, node);
nni_aio_list_init(&ep->aios);
@@ -353,11 +347,6 @@ inproc_conn_finish(nni_aio *aio, int rv, inproc_ep *ep, inproc_pipe *pipe)
nni_aio_set_output(aio, 0, pipe);
nni_aio_finish(aio, 0, 0);
} else {
- if (ep->ndialer != NULL) {
- nni_dialer_bump_error(ep->ndialer, rv);
- } else {
- nni_listener_bump_error(ep->nlistener, rv);
- }
NNI_ASSERT(pipe == NULL);
nni_aio_finish_error(aio, rv);
}
@@ -474,11 +463,6 @@ inproc_ep_cancel(nni_aio *aio, void *arg, int rv)
nni_list_node_remove(&ep->node);
nni_aio_finish_error(aio, rv);
}
- if (ep->ndialer != NULL) {
- nni_dialer_bump_error(ep->ndialer, rv);
- } else {
- nni_listener_bump_error(ep->nlistener, rv);
- }
nni_mtx_unlock(&nni_inproc.mx);
}
@@ -503,7 +487,6 @@ inproc_ep_connect(void *arg, nni_aio *aio)
}
if (server == NULL) {
nni_mtx_unlock(&nni_inproc.mx);
- nni_dialer_bump_error(ep->ndialer, NNG_ECONNREFUSED);
nni_aio_finish_error(aio, NNG_ECONNREFUSED);
return;
}
@@ -513,7 +496,6 @@ inproc_ep_connect(void *arg, nni_aio *aio)
// that in the upper API.
if ((rv = nni_aio_schedule(aio, inproc_ep_cancel, ep)) != 0) {
nni_mtx_unlock(&nni_inproc.mx);
- nni_dialer_bump_error(ep->ndialer, rv);
nni_aio_finish_error(aio, rv);
return;
}
@@ -536,7 +518,6 @@ inproc_ep_bind(void *arg)
NNI_LIST_FOREACH (list, srch) {
if (strcmp(srch->addr, ep->addr) == 0) {
nni_mtx_unlock(&nni_inproc.mx);
- nni_listener_bump_error(ep->nlistener, NNG_EADDRINUSE);
return (NNG_EADDRINUSE);
}
}
@@ -561,7 +542,6 @@ inproc_ep_accept(void *arg, nni_aio *aio)
// accept was tried -- there is no API to do such a thing.
if ((rv = nni_aio_schedule(aio, inproc_ep_cancel, ep)) != 0) {
nni_mtx_unlock(&nni_inproc.mx);
- nni_listener_bump_error(ep->nlistener, rv);
nni_aio_finish_error(aio, rv);
return;
}
diff --git a/src/transport/ipc/ipc.c b/src/transport/ipc/ipc.c
index 4d5db82d..c09b56be 100644
--- a/src/transport/ipc/ipc.c
+++ b/src/transport/ipc/ipc.c
@@ -1,5 +1,5 @@
//
-// Copyright 2019 Staysail Systems, Inc. <info@staysail.tech>
+// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
// Copyright 2018 Capitar IT Group BV <info@capitar.com>
// Copyright 2019 Devolutions <info@devolutions.net>
//
@@ -11,7 +11,6 @@
#include <stdio.h>
#include <stdlib.h>
-#include <string.h>
#include "core/nng_impl.h"
@@ -46,7 +45,6 @@ struct ipctran_pipe {
size_t wantrxhead;
nni_list recvq;
nni_list sendq;
- nni_aio * useraio;
nni_aio * txaio;
nni_aio * rxaio;
nni_aio * negoaio;
@@ -72,8 +70,6 @@ struct ipctran_ep {
nni_list waitpipes; // pipes waiting to match to socket
nni_list negopipes; // pipes busy negotiating
nni_reap_item reap;
- nni_dialer * ndialer;
- nni_listener * nlistener;
nni_stat_item st_rcvmaxsz;
};
@@ -179,7 +175,8 @@ ipctran_pipe_alloc(ipctran_pipe **pipep)
nni_mtx_init(&p->mtx);
if (((rv = nni_aio_alloc(&p->txaio, ipctran_pipe_send_cb, p)) != 0) ||
((rv = nni_aio_alloc(&p->rxaio, ipctran_pipe_recv_cb, p)) != 0) ||
- ((rv = nni_aio_alloc(&p->negoaio, ipctran_pipe_nego_cb, p)) != 0)) {
+ ((rv = nni_aio_alloc(&p->negoaio, ipctran_pipe_nego_cb, p)) !=
+ 0)) {
ipctran_pipe_fini(p);
return (rv);
}
@@ -268,18 +265,11 @@ ipctran_pipe_nego_cb(void *arg)
return;
error:
- p->useraio = NULL;
-
- if (ep->ndialer != NULL) {
- nni_dialer_bump_error(ep->ndialer, rv);
- } else {
- nni_listener_bump_error(ep->nlistener, rv);
- }
nng_stream_close(p->conn);
// If we are waiting to negotiate on a client side, then a failure
// here has to be passed to the user app.
- if ((ep->dialer != NULL) && ((uaio = ep->useraio) != NULL)) {
+ if ((uaio = ep->useraio) != NULL) {
ep->useraio = NULL;
nni_aio_finish_error(uaio, rv);
}
@@ -347,7 +337,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
@@ -747,10 +736,17 @@ ipctran_accept_cb(void *arg)
return;
error:
- nni_listener_bump_error(ep->nlistener, rv);
+ // When an error here occurs, let's send a notice up to the consumer.
+ // That way it can be reported properly.
+ if ((aio = ep->useraio) != NULL) {
+ ep->useraio = NULL;
+ nni_aio_finish_error(aio, rv);
+ }
+
switch (rv) {
case NNG_ENOMEM:
+ case NNG_ENOFILES:
nng_sleep_aio(10, ep->timeaio);
break;
@@ -786,6 +782,8 @@ ipctran_dial_cb(void *arg)
ipctran_pipe_fini(p);
nng_stream_free(conn);
rv = NNG_ECLOSED;
+ nni_mtx_unlock(&ep->mtx);
+ goto error;
} else {
ipctran_pipe_start(p, conn, ep);
}
@@ -795,14 +793,12 @@ ipctran_dial_cb(void *arg)
error:
// Error connecting. We need to pass this straight back
// to the user.
- nni_dialer_bump_error(ep->ndialer, rv);
nni_mtx_lock(&ep->mtx);
if ((aio = ep->useraio) != NULL) {
ep->useraio = NULL;
nni_aio_finish_error(aio, rv);
}
nni_mtx_unlock(&ep->mtx);
- return;
}
static int
@@ -838,7 +834,6 @@ ipctran_ep_init_dialer(void **dp, nni_url *url, nni_dialer *ndialer)
if ((rv = ipctran_ep_init(&ep, sock)) != 0) {
return (rv);
}
- ep->ndialer = ndialer;
if (((rv = nni_aio_alloc(&ep->connaio, ipctran_dial_cb, ep)) != 0) ||
((rv = nng_stream_dialer_alloc_url(&ep->dialer, url)) != 0)) {
@@ -861,7 +856,6 @@ ipctran_ep_init_listener(void **dp, nni_url *url, nni_listener *nlistener)
if ((rv = ipctran_ep_init(&ep, sock)) != 0) {
return (rv);
}
- ep->nlistener = nlistener;
if (((rv = nni_aio_alloc(&ep->connaio, ipctran_accept_cb, ep)) != 0) ||
((rv = nni_aio_alloc(&ep->timeaio, ipctran_timer_cb, ep)) != 0) ||
@@ -883,11 +877,6 @@ ipctran_ep_cancel(nni_aio *aio, void *arg, int rv)
if (aio == ep->useraio) {
ep->useraio = NULL;
nni_aio_finish_error(aio, rv);
- if (ep->ndialer) {
- nni_dialer_bump_error(ep->ndialer, rv);
- } else {
- nni_listener_bump_error(ep->nlistener, rv);
- }
}
nni_mtx_unlock(&ep->mtx);
}
@@ -905,19 +894,16 @@ ipctran_ep_connect(void *arg, nni_aio *aio)
if (ep->closed) {
nni_mtx_unlock(&ep->mtx);
nni_aio_finish_error(aio, NNG_ECLOSED);
- nni_dialer_bump_error(ep->ndialer, NNG_ECLOSED);
return;
}
if (ep->useraio != NULL) {
nni_mtx_unlock(&ep->mtx);
nni_aio_finish_error(aio, NNG_EBUSY);
- nni_dialer_bump_error(ep->ndialer, NNG_EBUSY);
return;
}
if ((rv = nni_aio_schedule(aio, ipctran_ep_cancel, ep)) != 0) {
nni_mtx_unlock(&ep->mtx);
- nni_dialer_bump_error(ep->ndialer, NNG_EBUSY);
nni_aio_finish_error(aio, rv);
return;
}
@@ -970,9 +956,7 @@ ipctran_ep_bind(void *arg)
int rv;
nni_mtx_lock(&ep->mtx);
- if ((rv = nng_stream_listener_listen(ep->listener)) != 0) {
- nni_listener_bump_error(ep->nlistener, rv);
- }
+ rv = nng_stream_listener_listen(ep->listener);
nni_mtx_unlock(&ep->mtx);
return (rv);
}
@@ -989,20 +973,17 @@ ipctran_ep_accept(void *arg, nni_aio *aio)
nni_mtx_lock(&ep->mtx);
if (ep->closed) {
nni_aio_finish_error(aio, NNG_ECLOSED);
- nni_listener_bump_error(ep->nlistener, NNG_ECLOSED);
nni_mtx_unlock(&ep->mtx);
return;
}
if (ep->useraio != NULL) {
nni_aio_finish_error(aio, NNG_EBUSY);
- nni_listener_bump_error(ep->nlistener, NNG_EBUSY);
nni_mtx_unlock(&ep->mtx);
return;
}
if ((rv = nni_aio_schedule(aio, ipctran_ep_cancel, ep)) != 0) {
nni_mtx_unlock(&ep->mtx);
nni_aio_finish_error(aio, rv);
- nni_listener_bump_error(ep->nlistener, rv);
return;
}
ep->useraio = aio;
diff --git a/src/transport/tcp/tcp.c b/src/transport/tcp/tcp.c
index 9012a5bc..aee917e9 100644
--- a/src/transport/tcp/tcp.c
+++ b/src/transport/tcp/tcp.c
@@ -1,5 +1,5 @@
//
-// Copyright 2019 Staysail Systems, Inc. <info@staysail.tech>
+// Copyright 2020 Staysail Systems, Inc. <info@staysail.tech>
// Copyright 2018 Capitar IT Group BV <info@capitar.com>
// Copyright 2019 Devolutions <info@devolutions.net>
//
@@ -14,7 +14,6 @@
#include <string.h>
#include "core/nng_impl.h"
-#include "core/tcp.h"
// TCP transport. Platform specific TCP operations must be
// supplied as well.
@@ -42,7 +41,6 @@ struct tcptran_pipe {
size_t wantrxhead;
nni_list recvq;
nni_list sendq;
- nni_aio * useraio;
nni_aio * txaio;
nni_aio * rxaio;
nni_aio * negoaio;
@@ -52,7 +50,6 @@ struct tcptran_pipe {
struct tcptran_ep {
nni_mtx mtx;
- uint16_t af;
uint16_t proto;
size_t rcvmax;
bool fini;
@@ -71,8 +68,6 @@ struct tcptran_ep {
nni_reap_item reap;
nng_stream_dialer * dialer;
nng_stream_listener *listener;
- nni_dialer * ndialer;
- nni_listener * nlistener;
nni_stat_item st_rcvmaxsz;
};
@@ -272,17 +267,9 @@ tcptran_pipe_nego_cb(void *arg)
return;
error:
- if (ep->ndialer != NULL) {
- nni_dialer_bump_error(ep->ndialer, rv);
- } else {
- nni_listener_bump_error(ep->nlistener, rv);
- }
-
nng_stream_close(p->conn);
- // If we are waiting to negotiate on a client side, then a failure
- // here has to be passed to the user app.
- if ((ep->dialer != NULL) && ((uaio = ep->useraio) != NULL)) {
+ if ((uaio = ep->useraio) != NULL) {
ep->useraio = NULL;
nni_aio_finish_error(uaio, rv);
}
@@ -793,10 +780,16 @@ tcptran_accept_cb(void *arg)
return;
error:
- nni_listener_bump_error(ep->nlistener, rv);
+ // When an error here occurs, let's send a notice up to the consumer.
+ // That way it can be reported properly.
+ if ((aio = ep->useraio) != NULL) {
+ ep->useraio = NULL;
+ nni_aio_finish_error(aio, rv);
+ }
switch (rv) {
case NNG_ENOMEM:
+ case NNG_ENOFILES:
nng_sleep_aio(10, ep->timeaio);
break;
@@ -832,6 +825,8 @@ tcptran_dial_cb(void *arg)
tcptran_pipe_fini(p);
nng_stream_free(conn);
rv = NNG_ECLOSED;
+ nni_mtx_unlock(&ep->mtx);
+ goto error;
} else {
tcptran_pipe_start(p, conn, ep);
}
@@ -841,14 +836,12 @@ tcptran_dial_cb(void *arg)
error:
// Error connecting. We need to pass this straight back
// to the user.
- nni_dialer_bump_error(ep->ndialer, rv);
nni_mtx_lock(&ep->mtx);
if ((aio = ep->useraio) != NULL) {
ep->useraio = NULL;
nni_aio_finish_error(aio, rv);
}
nni_mtx_unlock(&ep->mtx);
- return;
}
static int
@@ -895,13 +888,12 @@ tcptran_dialer_init(void **dp, nng_url *url, nni_dialer *ndialer)
}
if ((rv = tcptran_url_parse_source(&myurl, &srcsa, url)) != 0) {
- return (NNG_EADDRINVAL);
+ return (rv);
}
if ((rv = tcptran_ep_init(&ep, url, sock)) != 0) {
return (rv);
}
- ep->ndialer = ndialer;
if ((rv != 0) ||
((rv = nni_aio_alloc(&ep->connaio, tcptran_dial_cb, ep)) != 0) ||
@@ -940,7 +932,6 @@ tcptran_listener_init(void **lp, nng_url *url, nni_listener *nlistener)
if ((rv = tcptran_ep_init(&ep, url, sock)) != 0) {
return (rv);
}
- ep->nlistener = nlistener;
if (((rv = nni_aio_alloc(&ep->connaio, tcptran_accept_cb, ep)) != 0) ||
((rv = nni_aio_alloc(&ep->timeaio, tcptran_timer_cb, ep)) != 0) ||
@@ -962,11 +953,6 @@ tcptran_ep_cancel(nni_aio *aio, void *arg, int rv)
if (ep->useraio == aio) {
ep->useraio = NULL;
nni_aio_finish_error(aio, rv);
- if (ep->ndialer) {
- nni_dialer_bump_error(ep->ndialer, rv);
- } else {
- nni_listener_bump_error(ep->nlistener, rv);
- }
}
nni_mtx_unlock(&ep->mtx);
}
@@ -984,18 +970,15 @@ tcptran_ep_connect(void *arg, nni_aio *aio)
if (ep->closed) {
nni_mtx_unlock(&ep->mtx);
nni_aio_finish_error(aio, NNG_ECLOSED);
- nni_dialer_bump_error(ep->ndialer, NNG_ECLOSED);
return;
}
if (ep->useraio != NULL) {
nni_mtx_unlock(&ep->mtx);
nni_aio_finish_error(aio, NNG_EBUSY);
- nni_dialer_bump_error(ep->ndialer, NNG_EBUSY);
return;
}
if ((rv = nni_aio_schedule(aio, tcptran_ep_cancel, ep)) != 0) {
nni_mtx_unlock(&ep->mtx);
- nni_dialer_bump_error(ep->ndialer, rv);
nni_aio_finish_error(aio, rv);
return;
}
@@ -1070,9 +1053,6 @@ tcptran_ep_bind(void *arg)
nni_mtx_lock(&ep->mtx);
rv = nng_stream_listener_listen(ep->listener);
- if (rv != 0) {
- nni_listener_bump_error(ep->nlistener, rv);
- }
nni_mtx_unlock(&ep->mtx);
return (rv);
@@ -1091,19 +1071,16 @@ tcptran_ep_accept(void *arg, nni_aio *aio)
if (ep->closed) {
nni_mtx_unlock(&ep->mtx);
nni_aio_finish_error(aio, NNG_ECLOSED);
- nni_listener_bump_error(ep->nlistener, NNG_ECLOSED);
return;
}
if (ep->useraio != NULL) {
nni_mtx_unlock(&ep->mtx);
nni_aio_finish_error(aio, NNG_EBUSY);
- nni_listener_bump_error(ep->nlistener, NNG_EBUSY);
return;
}
if ((rv = nni_aio_schedule(aio, tcptran_ep_cancel, ep)) != 0) {
nni_mtx_unlock(&ep->mtx);
nni_aio_finish_error(aio, rv);
- nni_listener_bump_error(ep->nlistener, rv);
return;
}
ep->useraio = aio;
diff --git a/src/transport/tls/tls.c b/src/transport/tls/tls.c
index 46ff8117..c291ecf7 100644
--- a/src/transport/tls/tls.c
+++ b/src/transport/tls/tls.c
@@ -57,7 +57,6 @@ struct tlstran_pipe {
// Stuff that is common to both dialers and listeners.
struct tlstran_ep {
nni_mtx mtx;
- uint16_t af;
uint16_t proto;
size_t rcvmax;
bool started;
@@ -79,8 +78,6 @@ struct tlstran_ep {
const char * host;
nng_sockaddr src;
nng_sockaddr sa;
- nni_dialer * ndialer;
- nni_listener * nlistener;
nni_stat_item st_rcvmaxsz;
};
@@ -169,7 +166,8 @@ tlstran_pipe_alloc(tlstran_pipe **pipep)
if (((rv = nni_aio_alloc(&p->txaio, tlstran_pipe_send_cb, p)) != 0) ||
((rv = nni_aio_alloc(&p->rxaio, tlstran_pipe_recv_cb, p)) != 0) ||
- ((rv = nni_aio_alloc(&p->negoaio, tlstran_pipe_nego_cb, p)) != 0)) {
+ ((rv = nni_aio_alloc(&p->negoaio, tlstran_pipe_nego_cb, p)) !=
+ 0)) {
tlstran_pipe_fini(p);
return (rv);
}
@@ -272,17 +270,9 @@ tlstran_pipe_nego_cb(void *arg)
return;
error:
- if (ep->ndialer != NULL) {
- nni_dialer_bump_error(ep->ndialer, rv);
- } else {
- nni_listener_bump_error(ep->nlistener, rv);
- }
-
nng_stream_close(p->tls);
- // If we are waiting to negotiate on a client side, then a failure
- // here has to be passed to the user app.
- if ((ep->dialer != NULL) && ((uaio = ep->useraio) != NULL)) {
+ if ((uaio = ep->useraio) != NULL) {
ep->useraio = NULL;
nni_aio_finish_error(uaio, rv);
}
@@ -532,17 +522,17 @@ tlstran_pipe_recv_cancel(nni_aio *aio, void *arg, int rv)
static void
tlstran_pipe_recv_start(tlstran_pipe *p)
{
- nni_aio *rxaio;
+ nni_aio *aio;
nni_iov iov;
NNI_ASSERT(p->rxmsg == NULL);
// Schedule a read of the IPC header.
- rxaio = p->rxaio;
+ aio = p->rxaio;
iov.iov_buf = p->rxlen;
iov.iov_len = sizeof(p->rxlen);
- nni_aio_set_iov(rxaio, 1, &iov);
+ nni_aio_set_iov(aio, 1, &iov);
- nng_stream_recv(p->tls, rxaio);
+ nng_stream_recv(p->tls, aio);
}
static void
@@ -763,14 +753,24 @@ tlstran_accept_cb(void *arg)
return;
error:
- nni_listener_bump_error(ep->nlistener, rv);
+ // When an error here occurs, let's send a notice up to the consumer.
+ // That way it can be reported properly.
+ if ((aio = ep->useraio) != NULL) {
+ ep->useraio = NULL;
+ nni_aio_finish_error(aio, rv);
+ }
switch (rv) {
case NNG_ENOMEM:
+ case NNG_ENOFILES:
+ // We need to cool down here, to avoid spinning.
nng_sleep_aio(10, ep->timeaio);
break;
default:
+ // Start another accept. This is done because we want to
+ // ensure that TLS negotiations are disconnected from
+ // the upper layer accept logic.
if (!ep->closed) {
nng_stream_listener_accept(ep->listener, ep->connaio);
}
@@ -802,6 +802,8 @@ tlstran_dial_cb(void *arg)
tlstran_pipe_fini(p);
nng_stream_free(conn);
rv = NNG_ECLOSED;
+ nni_mtx_unlock(&ep->mtx);
+ goto error;
} else {
tlstran_pipe_start(p, conn, ep);
}
@@ -809,9 +811,7 @@ tlstran_dial_cb(void *arg)
return;
error:
- // Error connecting. We need to pass this straight back
- // to the user.
- nni_dialer_bump_error(ep->ndialer, rv);
+ // Error connecting. We need to pass this straight back to the user.
nni_mtx_lock(&ep->mtx);
if ((aio = ep->useraio) != NULL) {
ep->useraio = NULL;
@@ -872,7 +872,6 @@ tlstran_ep_init_dialer(void **dp, nni_url *url, nni_dialer *ndialer)
return (rv);
}
ep->authmode = NNG_TLS_AUTH_MODE_REQUIRED;
- ep->ndialer = ndialer;
if ((rv != 0) ||
((rv = nng_stream_dialer_alloc_url(&ep->dialer, &myurl)) != 0)) {
@@ -924,9 +923,7 @@ tlstran_ep_init_listener(void **lp, nni_url *url, nni_listener *nlistener)
return (rv);
}
- ep->authmode = NNG_TLS_AUTH_MODE_NONE;
- ep->af = af;
- ep->nlistener = nlistener;
+ ep->authmode = NNG_TLS_AUTH_MODE_NONE;
if (strlen(host) == 0) {
host = NULL;
@@ -970,11 +967,6 @@ tlstran_ep_cancel(nni_aio *aio, void *arg, int rv)
if (ep->useraio == aio) {
ep->useraio = NULL;
nni_aio_finish_error(aio, rv);
- if (ep->ndialer) {
- nni_dialer_bump_error(ep->ndialer, rv);
- } else {
- nni_listener_bump_error(ep->nlistener, rv);
- }
}
nni_mtx_unlock(&ep->mtx);
}
@@ -993,18 +985,15 @@ tlstran_ep_connect(void *arg, nni_aio *aio)
if (ep->closed) {
nni_mtx_unlock(&ep->mtx);
nni_aio_finish_error(aio, NNG_ECLOSED);
- nni_dialer_bump_error(ep->ndialer, NNG_ECLOSED);
return;
}
if (ep->useraio != NULL) {
nni_mtx_unlock(&ep->mtx);
nni_aio_finish_error(aio, NNG_EBUSY);
- nni_dialer_bump_error(ep->ndialer, NNG_EBUSY);
return;
}
if ((rv = nni_aio_schedule(aio, tlstran_ep_cancel, ep)) != 0) {
nni_mtx_unlock(&ep->mtx);
- nni_dialer_bump_error(ep->ndialer, rv);
nni_aio_finish_error(aio, rv);
return;
}
@@ -1021,9 +1010,7 @@ tlstran_ep_bind(void *arg)
int rv;
nni_mtx_lock(&ep->mtx);
- if ((rv = nng_stream_listener_listen(ep->listener)) != 0) {
- nni_listener_bump_error(ep->nlistener, rv);
- }
+ rv = nng_stream_listener_listen(ep->listener);
nni_mtx_unlock(&ep->mtx);
return (rv);
@@ -1042,19 +1029,16 @@ tlstran_ep_accept(void *arg, nni_aio *aio)
if (ep->closed) {
nni_mtx_unlock(&ep->mtx);
nni_aio_finish_error(aio, NNG_ECLOSED);
- nni_listener_bump_error(ep->nlistener, NNG_ECLOSED);
return;
}
if (ep->useraio != NULL) {
nni_mtx_unlock(&ep->mtx);
nni_aio_finish_error(aio, NNG_EBUSY);
- nni_listener_bump_error(ep->nlistener, NNG_EBUSY);
return;
}
if ((rv = nni_aio_schedule(aio, tlstran_ep_cancel, ep)) != 0) {
nni_mtx_unlock(&ep->mtx);
nni_aio_finish_error(aio, rv);
- nni_listener_bump_error(ep->nlistener, rv);
return;
}
ep->useraio = aio;
generated by cgit v1.2.3-70-g09d2 (git 2.46.0) at 2026-01-18 13:54:39 +0000