refactor/dtls: Use message oriented send/receive for DTLS.

The protocol here needs to know and respect message boundaries.

Author: gdamore

src/sp/transport/dtls/dtls.c

@@ -215,8 +215,6 @@ static void
 dtls_bio_recv_done(dtls_pipe *p)
 {
 	nng_aio *aio;
-	uint8_t *ptr;
-	size_t   resid;
 	nni_msg *msg;
 
 	while ((!nni_lmq_empty(&p->rx_mq)) &&
@@ -225,22 +223,7 @@ dtls_bio_recv_done(dtls_pipe *p)
 		nni_aio_list_remove(aio);
 		nni_lmq_get(&p->rx_mq, &msg);
 
-		// assumption we only have a body, because we don't bother to
-		// fill in the header for raw UDP.
-
-		resid = nni_msg_len(msg);
-		ptr   = nni_msg_body(msg);
-
-		for (unsigned i = 0; i < aio->a_nio && resid > 0; i++) {
-			size_t num = resid > aio->a_iov[i].iov_len
-			    ? aio->a_iov[i].iov_len
-			    : resid;
-			memcpy(aio->a_iov[i].iov_buf, ptr, num);
-			ptr += num;
-			resid -= num;
-		}
-		nni_aio_finish(aio, NNG_OK, nni_msg_len(msg));
-		nni_msg_free(msg);
+		nni_aio_finish_msg(aio, msg);
 	}
 }
 
@@ -264,11 +247,19 @@ static void
 dtls_bio_send(void *arg, nng_aio *aio)
 {
 	dtls_pipe *p = arg;
+	nni_iov    iov;
+	nni_msg   *msg;
 
 	nni_mtx_lock(&p->lower_mtx);
 	if (!p->closed) {
 		nni_aio_set_input(aio, 0, &p->peer_addr);
+		msg         = nni_aio_get_msg(aio);
+		iov.iov_buf = nni_msg_body(msg);
+		iov.iov_len = nni_msg_len(msg);
+		nng_aio_set_iov(aio, 1, &iov);
 		nng_udp_send(p->ep->udp, aio);
+	} else {
+		nni_aio_finish_error(aio, NNG_ECLOSED);
 	}
 	nni_mtx_unlock(&p->lower_mtx);
 }
@@ -766,7 +757,7 @@ dtls_pipe_alloc(dtls_ep *ep, dtls_pipe **pp, const nng_sockaddr *sa)
 	p->recv_max  = ep->rcvmax;
 	*pp          = p;
 
-	if (((rv = nni_tls_init(&p->tls, ep->tlscfg)) != NNG_OK) ||
+	if (((rv = nni_tls_init(&p->tls, ep->tlscfg, true)) != NNG_OK) ||
 	    ((rv = nni_tls_start(&p->tls, &dtls_bio_ops, p, sa)) != NNG_OK) ||
 	    ((rv = dtls_add_pipe(ep, p)) != NNG_OK)) {
 		nni_pipe_close(p->npipe);
@@ -919,6 +910,7 @@ dtls_add_pipe(dtls_ep *ep, dtls_pipe *p)
 			id = 1;
 		}
 	}
+	p->id = id;
 	return (nni_id_set(&ep->pipes, id, p));
 }
 
@@ -979,11 +971,12 @@ dtls_rx_cb(void *arg)
 	}
 	NNI_ASSERT(p != NULL);
 
-	if (nni_msg_alloc(&msg, nni_aio_count(aio)) != NNG_OK) {
+	size_t len = nni_aio_count(aio);
+	if (nni_msg_alloc(&msg, len) != NNG_OK) {
 		// TODO BUMP A NO RECV ALLOC STAT
 		goto fail;
 	}
-	memcpy(nni_msg_body(msg), ep->rx_buf, nni_aio_count(aio));
+	memcpy(nni_msg_body(msg), ep->rx_buf, len);
 	dtls_start_rx(ep);
 	nni_pipe_hold(p->npipe);
 	nni_mtx_unlock(&ep->mtx);
@@ -1405,13 +1398,18 @@ dtls_resolv_cb(void *arg)
 		ep->self_sa.s_family = ep->peer_sa.s_family;
 	}
 
-	if (ep->udp == NULL) {
-		if ((rv = nng_udp_open(&ep->udp, &ep->self_sa)) != NNG_OK) {
-			nni_aio_list_remove(aio);
-			nni_aio_finish_error(aio, rv);
-			nni_mtx_unlock(&ep->mtx);
-			return;
-		}
+	// Close the socket if it was open, because we need to
+	// start with a fresh port.
+	if (ep->udp != NULL) {
+		nng_udp_close(ep->udp);
+		ep->udp = NULL;
+	}
+
+	if ((rv = nng_udp_open(&ep->udp, &ep->self_sa)) != NNG_OK) {
+		nni_aio_list_remove(aio);
+		nni_aio_finish_error(aio, rv);
+		nni_mtx_unlock(&ep->mtx);
+		return;
 	}
 
 	if ((rv = dtls_pipe_alloc(ep, &p, &ep->peer_sa)) != NNG_OK) {

src/sp/transport/dtls/dtls_tran_test.c

@@ -449,7 +449,7 @@ test_dtls_reqrep_multi(void)
 	nng_tls_config_free(c1);
 }
 
-#define NCLIENT 3
+#define NCLIENT 10
 void
 test_dtls_pub_multi(void)
 {
@@ -493,7 +493,7 @@ test_dtls_pub_multi(void)
 
 	// send a bunch of messages - we're hoping that by serializing we won't
 	// overwhelm the network.
-	for (int i = 0; i < 100; i++) {
+	for (int i = 0; i < 1000; i++) {
 		size_t len = nng_random() % (sizeof(msg) - 1);
 		memset(msg, 'a' + i % 26, sizeof(buf));
 		msg[len] = 0;

src/supplemental/tls/mbedtls/mbedtls.c

@@ -211,8 +211,8 @@ tls_mk_err(int err)
 static int
 net_send(void *tls, const unsigned char *buf, size_t len)
 {
-	size_t sz = len;
-	int    rv;
+	size_t  sz = len;
+	nng_err rv;
 
 	rv = nng_tls_engine_send(tls, buf, &sz);
 	switch (rv) {
@@ -228,8 +228,8 @@ net_send(void *tls, const unsigned char *buf, size_t len)
 static int
 net_recv(void *tls, unsigned char *buf, size_t len)
 {
-	size_t sz = len;
-	int    rv;
+	size_t  sz = len;
+	nng_err rv;
 
 	rv = nng_tls_engine_recv(tls, buf, &sz);
 	switch (rv) {
@@ -993,7 +993,7 @@ tls_engine_init(void)
 #endif
 	// Uncomment the following to have noisy debug from mbedTLS.
 	// This may be useful when trying to debug failures.
-	// mbedtls_debug_set_threshold(9);
+	mbedtls_debug_set_threshold(1);
 
 	mbedtls_ssl_cookie_init(&mbed_ssl_cookie_ctx);
 	rv = mbedtls_ssl_cookie_setup(&mbed_ssl_cookie_ctx, tls_random, NULL);

src/supplemental/tls/openssl/openssl.c

@@ -25,6 +25,7 @@
 #include "../../../core/list.h"
 #include "../../../core/strs.h"
 #include "../tls_engine.h"
+#include "nng/nng.h"
 
 // library code for openssl
 static int ossl_libcode;
@@ -116,8 +117,8 @@ tls_log_err(const char *msgid, const char *context, int errnum)
 static int
 ossl_net_send(BIO *bio, const char *buf, size_t len, size_t *lenp)
 {
-	void *ctx = BIO_get_data(bio);
-	int   rv;
+	void   *ctx = BIO_get_data(bio);
+	nng_err rv;
 
 	switch (rv = nng_tls_engine_send(ctx, (const uint8_t *) buf, &len)) {
 	case NNG_OK:
@@ -135,8 +136,8 @@ ossl_net_send(BIO *bio, const char *buf, size_t len, size_t *lenp)
 static int
 ossl_net_recv(BIO *bio, char *buf, size_t len, size_t *lenp)
 {
-	void *ctx = BIO_get_data(bio);
-	int   rv;
+	void   *ctx = BIO_get_data(bio);
+	nng_err rv;
 
 	switch (rv = nng_tls_engine_recv(ctx, (uint8_t *) buf, &len)) {
 	case NNG_OK:
@@ -348,7 +349,8 @@ ossl_conn_handshake(nng_tls_engine_conn *ec)
 
 	rv = SSL_do_handshake(ec->ssl);
 	if (rv == 1) {
-		nng_log_debug("NNG-TLS-HS", "TLS handshake complete");
+		nng_log_debug("NNG-TLS-HS", "TLS handshake complete %s",
+		    ec->mode == NNG_TLS_MODE_CLIENT ? "client" : "server");
 		return (NNG_OK);
 	}
 	rv = SSL_get_error(ec->ssl, rv);