Pipe ID calculations & socket error functions.

The use of platform_next_id was a bit off, because it could give
back pipe IDs that were too large (the high order bit must be
clear), and in very long running applications serving many
connections, the IDs could wrap and lead to duplicates.

Also we have added functions to set the recverr or senderr values,
which can be used by protocols -- either during initialization,
or during filters.  (REQ uses this for example.)

3 files changed, 40 insertions, 2 deletions

diff --git a/src/core/pipe.c b/src/core/pipe.c
index 0a7bbed1..15ae393b 100644
--- a/src/core/pipe.c
+++ b/src/core/pipe.c
@@ -89,7 +89,6 @@ nni_pipe_create(nni_pipe **pp, nni_endpt *ep)
 	p->p_trandata = NULL;
 	p->p_protdata = NULL;
 	p->p_ops = *ep->ep_ops.ep_pipe_ops;
-	p->p_id = nni_plat_nextid();
 	p->p_ep = ep;
 	p->p_sock = ep->ep_sock;
 	if (ep->ep_dialer != NULL) {
diff --git a/src/core/socket.c b/src/core/socket.c
index 8c734591..c9b9175f 100644
--- a/src/core/socket.c
+++ b/src/core/socket.c
@@ -344,12 +344,31 @@ int
 nni_socket_add_pipe(nni_socket *sock, nni_pipe *pipe)
 {
 	int rv;
+	int collide;
 
 	nni_mutex_enter(&sock->s_mx);
 	if (sock->s_closing) {
 		nni_mutex_exit(&sock->s_mx);
 		return (NNG_ECLOSED);
 	}
+
+	do {
+		// We generate a new pipe ID, but we make sure it does not
+		// collide with any we already have.  This can only normally
+		// happen if we wrap -- i.e. we've had 4 billion or so pipes.
+		// XXX: consider making this a hash table!!
+		nni_pipe *check;
+		pipe->p_id = nni_plat_nextid() & 0x7FFFFFFF;
+		collide = 0;
+		NNI_LIST_FOREACH (&sock->s_pipes, check) {
+			if (check->p_id == pipe->p_id) {
+				collide = 1;
+				break;
+			}
+		}
+	} while (collide);
+
+	pipe->p_id = nni_plat_nextid();
 	rv = sock->s_ops.proto_add_pipe(sock->s_data, pipe, &pipe->p_protdata);
 	if (rv != 0) {
 		pipe->p_reap = 1;
@@ -418,6 +437,20 @@ nni_socket_listen(nni_socket *sock, const char *addr, nni_endpt **epp,
 }
 
 
+void
+nni_socket_recverr(nni_socket *sock, int err)
+{
+	sock->s_recverr = err;
+}
+
+
+void
+nni_socket_senderr(nni_socket *sock, int err)
+{
+	sock->s_senderr = err;
+}
+
+
 int
 nni_setopt_duration(nni_duration *ptr, const void *val, size_t size)
 {
@@ -459,6 +492,7 @@ nni_getopt_duration(nni_duration *ptr, void *val, size_t *sizep)
 	return (0);
 }
 
+
 int
 nni_getopt_int(int *ptr, void *val, size_t *sizep)
 {
@@ -473,7 +507,6 @@ nni_getopt_int(int *ptr, void *val, size_t *sizep)
 }
 
 
-
 static int
 nni_setopt_buf(nni_msgqueue *mq, const void *val, size_t sz)
 {
diff --git a/src/core/socket.h b/src/core/socket.h
index 35ddc36c..c80af22e 100644
--- a/src/core/socket.h
+++ b/src/core/socket.h
@@ -61,4 +61,10 @@ extern int nni_getopt_duration(nni_duration *, void *, size_t *);
 extern int nni_setopt_int(int *, const void *, size_t);
 extern int nni_getopt_int(int *, void *, size_t *);
 
+// Set error codes for applications.  These are only ever
+// called from the filter functions in protocols, and thus
+// already have the socket lock held.
+extern void nni_socket_recverr(nni_socket *, int);
+extern void nni_socket_senderr(nni_socket *, int);
+
 #endif  // CORE_SOCKET_H