Gdamore/udp

CMakeLists.txt

@@ -207,6 +207,24 @@ if (NOT WIN32 AND NOT CYGWIN)
     endif ()
 endif ()
 
+# Detect endianness -- this only handles little and big endian.
+# PDP users, sorry, but you're out of luck.'
+if (DEFINED CMAKE_LANG_C_BYTE_ORDER)
+    if (CMAKE_LANG_BYTE_ORDER EQUAL BIG_ENDIAN)
+        add_definitions(-DNNG_BIG_ENDIAN=1)
+    else()
+        add_definitions(-DNNG_LITTLE_ENDIAN=1)
+    endif()
+else()
+    include(TestBigEndian)
+    test_big_endian(NNG_BIG_ENDIAN)
+    if (NNG_BIG_ENDIAN)
+        add_definitions(-DNNG_BIG_EDNDIAN=1)
+    else()
+        add_definitions(-DNNG_LITTLE_ENDIAN=1)
+    endif()
+endif()
+
 if (CMAKE_SYSTEM_NAME MATCHES "Linux")
     add_definitions(-DNNG_PLATFORM_POSIX)
     add_definitions(-DNNG_PLATFORM_LINUX)

src/core/defs.h

@@ -12,6 +12,7 @@
 #define CORE_DEFS_H
 
 #include <stdint.h>
+#include <stdbool.h>
 
 // C compilers may get unhappy when named arguments are not used.  While
 // there are things like __attribute__((unused)) which are arguably
@@ -121,6 +122,54 @@ typedef void (*nni_cb)(void *);
 	    (((uint64_t) ((uint8_t) (ptr)[6])) << 8u) +  \
 	    (((uint64_t) (uint8_t) (ptr)[7]))
 
+// Modern CPUs are all little endian.  Let's stop paying the endian tax.
+
+#define NNI_PUT16LE(ptr, u)                                    \
+	do {                                                   \
+		(ptr)[1] = (uint8_t) (((uint16_t) (u)) >> 8u); \
+		(ptr)[0] = (uint8_t) ((uint16_t) (u));         \
+	} while (0)
+
+#define NNI_PUT32LE(ptr, u)                                     \
+	do {                                                    \
+		(ptr)[3] = (uint8_t) (((uint32_t) (u)) >> 24u); \
+		(ptr)[2] = (uint8_t) (((uint32_t) (u)) >> 16u); \
+		(ptr)[1] = (uint8_t) (((uint32_t) (u)) >> 8u);  \
+		(ptr)[0] = (uint8_t) ((uint32_t) (u));          \
+	} while (0)
+
+#define NNI_PUT64LE(ptr, u)                                     \
+	do {                                                    \
+		(ptr)[7] = (uint8_t) (((uint64_t) (u)) >> 56u); \
+		(ptr)[6] = (uint8_t) (((uint64_t) (u)) >> 48u); \
+		(ptr)[5] = (uint8_t) (((uint64_t) (u)) >> 40u); \
+		(ptr)[4] = (uint8_t) (((uint64_t) (u)) >> 32u); \
+		(ptr)[3] = (uint8_t) (((uint64_t) (u)) >> 24u); \
+		(ptr)[2] = (uint8_t) (((uint64_t) (u)) >> 16u); \
+		(ptr)[1] = (uint8_t) (((uint64_t) (u)) >> 8u);  \
+		(ptr)[0] = (uint8_t) ((uint64_t) (u));          \
+	} while (0)
+
+#define NNI_GET16LE(ptr, v)                             \
+	v = (((uint16_t) ((uint8_t) (ptr)[1])) << 8u) + \
+	    (((uint16_t) (uint8_t) (ptr)[0]))
+
+#define NNI_GET32LE(ptr, v)                              \
+	v = (((uint32_t) ((uint8_t) (ptr)[3])) << 24u) + \
+	    (((uint32_t) ((uint8_t) (ptr)[2])) << 16u) + \
+	    (((uint32_t) ((uint8_t) (ptr)[1])) << 8u) +  \
+	    (((uint32_t) (uint8_t) (ptr)[0]))
+
+#define NNI_GET64LE(ptr, v)                              \
+	v = (((uint64_t) ((uint8_t) (ptr)[7])) << 56u) + \
+	    (((uint64_t) ((uint8_t) (ptr)[6])) << 48u) + \
+	    (((uint64_t) ((uint8_t) (ptr)[5])) << 40u) + \
+	    (((uint64_t) ((uint8_t) (ptr)[4])) << 32u) + \
+	    (((uint64_t) ((uint8_t) (ptr)[3])) << 24u) + \
+	    (((uint64_t) ((uint8_t) (ptr)[2])) << 16u) + \
+	    (((uint64_t) ((uint8_t) (ptr)[1])) << 8u) +  \
+	    (((uint64_t) (uint8_t) (ptr)[0]))
+
 // This increments a pointer a fixed number of byte cells.
 #define NNI_INCPTR(ptr, n) ((ptr) = (void *) ((char *) (ptr) + (n)))
 
@@ -178,4 +227,19 @@ typedef nni_type nni_opt_type;
 	(__GNUC__ * 10000 + __GNUC_MINOR__ * 100 + __GNUC_PATCHLEVEL__)
 #endif
 
+#if !defined(NNG_BIG_ENDIAN) && !defined(NNG_LITTLE_ENDIAN)
+#if defined(__BYTE_ORDER__)
+#if __BYTE_ORDER__ == __ORDER_BIG_ENDIAN__
+#define NNG_BIG_ENDIAN 1
+#elif __BYTE_ORDER__ == __ORDER_LITTLE_ENDIAN__
+#define NNG_LITTLE_ENDIAN 1
+#else // middle-endian? (aka PDP-11)
+#error "Unsupported or unknown endian"
+#endif // __BYTE_ORDER__
+#else  // defined(__BYTE_ORDER__)
+#define NNG_LITTLE_ENDIAN 1
+#error "Unknown endian: specify -DNNG_BIG_ENDIAN=1 or -DNNG_LITTLE_ENDIAN=1"
+#endif // defined(__BYTE_ORDER)
+#endif // defined() endianness
+
 #endif // CORE_DEFS_H

src/core/idhash.c

@@ -389,3 +389,9 @@
 	*cursor = index;
 	return (false);
 }
+
+uint32_t
+nni_id_count(const nni_id_map *m)
+{
+	return (m->id_count);
+}

src/core/idhash.h

@@ -46,15 +46,16 @@ struct nni_id_map {
 #define NNI_ID_FLAG_RANDOM 2   // start at a random value
 #define NNI_ID_FLAG_REGISTER 4 // map is registered for finalization
 
-extern void  nni_id_map_init(nni_id_map *, uint64_t, uint64_t, bool);
-extern void  nni_id_map_fini(nni_id_map *);
-extern void *nni_id_get(nni_id_map *, uint64_t);
-extern int   nni_id_set(nni_id_map *, uint64_t, void *);
-extern int   nni_id_alloc(nni_id_map *, uint64_t *, void *);
-extern int   nni_id_alloc32(nni_id_map *, uint32_t *, void *);
-extern int   nni_id_remove(nni_id_map *, uint64_t);
-extern void  nni_id_map_sys_fini(void);
-extern bool  nni_id_visit(nni_id_map *, uint64_t *, void **, uint32_t *);
+extern void     nni_id_map_init(nni_id_map *, uint64_t, uint64_t, bool);
+extern void     nni_id_map_fini(nni_id_map *);
+extern void    *nni_id_get(nni_id_map *, uint64_t);
+extern int      nni_id_set(nni_id_map *, uint64_t, void *);
+extern int      nni_id_alloc(nni_id_map *, uint64_t *, void *);
+extern int      nni_id_alloc32(nni_id_map *, uint32_t *, void *);
+extern int      nni_id_remove(nni_id_map *, uint64_t);
+extern void     nni_id_map_sys_fini(void);
+extern bool     nni_id_visit(nni_id_map *, uint64_t *, void **, uint32_t *);
+extern uint32_t nni_id_count(const nni_id_map *);
 
 #define NNI_ID_MAP_INITIALIZER(min, max, flags)            \
 	{                                                  \

src/core/message.c

@@ -29,6 +29,7 @@
 	nni_chunk      m_body;
 	uint32_t       m_pipe; // set on receive
 	nni_atomic_int m_refcnt;
+	nng_sockaddr   m_addr; // set on receive, transport use
 };
 
 #if 0
@@ -656,3 +657,15 @@
 {
 	return (m->m_pipe);
 }
+
+const nng_sockaddr *
+nni_msg_address(const nni_msg *msg)
+{
+	return (&msg->m_addr);
+}
+
+void
+nni_msg_set_address(nng_msg *msg, const nng_sockaddr *addr)
+{
+	msg->m_addr = *addr;
+}

src/core/message.h

@@ -20,9 +20,9 @@ extern int      nni_msg_realloc(nni_msg *, size_t);
 extern int      nni_msg_reserve(nni_msg *, size_t);
 extern size_t   nni_msg_capacity(nni_msg *);
 extern int      nni_msg_dup(nni_msg **, const nni_msg *);
-extern void *   nni_msg_header(nni_msg *);
+extern void    *nni_msg_header(nni_msg *);
 extern size_t   nni_msg_header_len(const nni_msg *);
-extern void *   nni_msg_body(nni_msg *);
+extern void    *nni_msg_body(nni_msg *);
 extern size_t   nni_msg_len(const nni_msg *);
 extern int      nni_msg_append(nni_msg *, const void *, size_t);
 extern int      nni_msg_insert(nni_msg *, const void *, size_t);
@@ -55,6 +55,14 @@ extern void     nni_msg_clone(nni_msg *);
 extern nni_msg *nni_msg_unique(nni_msg *);
 extern bool     nni_msg_shared(nni_msg *);
 
+// Socket address access. Principally useful for transports like UDP,
+// which may need to remember or add the socket address later.
+// SP transports will generally not support upper layers setting the
+// address on send, but will take the information from the pipe.
+// It may be set on receive, depending upon the transport.
+extern const nng_sockaddr *nni_msg_address(const nni_msg *);
+extern void nni_msg_set_address(nng_msg *, const nng_sockaddr *);
+
 // nni_msg_pull_up ensures that the message is unique, and that any
 // header present is "pulled up" into the message body.  If the function
 // cannot do this for any reason (out of space in the body), then NULL

src/core/tcp.c

@@ -14,19 +14,20 @@
 
 #include <nng/nng.h>
 
-#include "core/nng_impl.h"
-#include "core/tcp.h"
+#include "core/url.h"
+#include "nng_impl.h"
+#include "tcp.h"
 
 typedef struct {
 	nng_stream_dialer ops;
-	char *            host;
-	char *            port;
+	char             *host;
+	char             *port;
 	int               af; // address family
 	bool              closed;
 	nng_sockaddr      sa;
-	nni_tcp_dialer *  d;      // platform dialer implementation
-	nni_aio *         resaio; // resolver aio
-	nni_aio *         conaio; // platform connection aio
+	nni_tcp_dialer   *d;      // platform dialer implementation
+	nni_aio          *resaio; // resolver aio
+	nni_aio          *conaio; // platform connection aio
 	nni_list          conaios;
 	nni_mtx           mtx;
 } tcp_dialer;
@@ -62,7 +63,7 @@
 tcp_dial_res_cb(void *arg)
 {
 	tcp_dialer *d = arg;
-	nni_aio *   aio;
+	nni_aio    *aio;
 	int         rv;
 
 	nni_mtx_lock(&d->mtx);
@@ -94,7 +95,7 @@
 tcp_dial_con_cb(void *arg)
 {
 	tcp_dialer *d = arg;
-	nng_aio *   aio;
+	nng_aio    *aio;
 	int         rv;
 
 	nni_mtx_lock(&d->mtx);
@@ -124,7 +125,7 @@
 tcp_dialer_close(void *arg)
 {
 	tcp_dialer *d = arg;
-	nni_aio *   aio;
+	nni_aio    *aio;
 	nni_mtx_lock(&d->mtx);
 	d->closed = true;
 	while ((aio = nni_list_first(&d->conaios)) != NULL) {
@@ -186,8 +187,7 @@
 }
 
 static int
-tcp_dialer_get(
-    void *arg, const char *name, void *buf, size_t *szp, nni_type t)
+tcp_dialer_get(void *arg, const char *name, void *buf, size_t *szp, nni_type t)
 {
 	tcp_dialer *d = arg;
 	return (nni_tcp_dialer_get(d->d, name, buf, szp, t));
@@ -238,10 +238,6 @@
 	int         rv;
 	const char *p;
 
-	if ((rv = nni_init()) != 0) {
-		return (rv);
-	}
-
 	if ((rv = tcp_dialer_alloc(&d)) != 0) {
 		return (rv);
 	}
@@ -276,7 +272,7 @@
 
 typedef struct {
 	nng_stream_listener ops;
-	nni_tcp_listener *  l;
+	nni_tcp_listener   *l;
 	nng_sockaddr        sa;
 } tcp_listener;
 
@@ -317,7 +313,7 @@
 	nng_sockaddr  sa;
 	size_t        sz;
 	int           port;
-	uint8_t *     paddr;
+	uint8_t      *paddr;
 
 	sz = sizeof(sa);
 	rv = nni_tcp_listener_get(
@@ -373,6 +369,9 @@
 	tcp_listener *l;
 	int           rv;
 
+	if ((rv = nni_init()) != 0) {
+		return (rv);
+	}
 	if ((l = NNI_ALLOC_STRUCT(l)) == NULL) {
 		return (NNG_ENOMEM);
 	}
@@ -396,41 +395,15 @@
 int
 nni_tcp_listener_alloc(nng_stream_listener **lp, const nng_url *url)
 {
-	nni_aio *    aio;
-	int          af;
 	int          rv;
 	nng_sockaddr sa;
-	const char * h;
 
 	if ((rv = nni_init()) != 0) {
 		return (rv);
 	}
-	if (strchr(url->u_scheme, '4') != NULL) {
-		af = NNG_AF_INET;
-	} else if (strchr(url->u_scheme, '6') != NULL) {
-		af = NNG_AF_INET6;
-	} else {
-		af = NNG_AF_UNSPEC;
-	}
-
-	if ((rv = nng_aio_alloc(&aio, NULL, NULL)) != 0) {
-		return (rv);
-	}
-
-	h = url->u_hostname;
-
-	// Wildcard special case, which means bind to INADDR_ANY.
-	if ((h != NULL) && ((strcmp(h, "*") == 0) || (strcmp(h, "") == 0))) {
-		h = NULL;
-	}
-	nni_resolv_ip(h, url->u_port, af, true, &sa, aio);
-	nni_aio_wait(aio);
-
-	if ((rv = nni_aio_result(aio)) != 0) {
-		nni_aio_free(aio);
+	if ((rv = nni_url_to_address(&sa, url)) != 0) {
 		return (rv);
 	}
-	nni_aio_free(aio);
 
 	return (tcp_listener_alloc_addr(lp, &sa));
 }

src/core/url.c

@@ -598,3 +598,37 @@
 }
 
 #undef URL_COPYSTR
+
+// nni_url_to_address resolves a URL into a sockaddr, assuming the URL is for
+// an IP address.
+int
+nni_url_to_address(nng_sockaddr *sa, const nng_url *url)
+{
+	int         af;
+	nni_aio     aio;
+	const char *h;
+	int         rv;
+
+	// This assumes the scheme is one that uses TCP/IP addresses.
+
+	if (strchr(url->u_scheme, '4') != NULL) {
+		af = NNG_AF_INET;
+	} else if (strchr(url->u_scheme, '6') != NULL) {
+		af = NNG_AF_INET6;
+	} else {
+		af = NNG_AF_UNSPEC;
+	}
+
+	nni_aio_init(&aio, NULL, NULL);
+
+	h = url->u_hostname;
+	if ((h != NULL) && ((strcmp(h, "*") == 0) || (strcmp(h, "") == 0))) {
+		h = NULL;
+	}
+
+	nni_resolv_ip(h, url->u_port, af, true, sa, &aio);
+	nni_aio_wait(&aio);
+	rv = nni_aio_result(&aio);
+	nni_aio_fini(&aio);
+	return (rv);
+}

src/core/url.h

@@ -20,5 +20,6 @@ extern const char *nni_url_default_port(const char *);
 extern int         nni_url_asprintf(char **, const nni_url *);
 extern int         nni_url_asprintf_port(char **, const nni_url *, int);
 extern size_t      nni_url_decode(uint8_t *, const char *, size_t);
+extern int         nni_url_to_address(nng_sockaddr *, const nni_url *);
 
 #endif // CORE_URL_H