Move to generic socket & pipe workers, and up to 4 each.

This should eliminate all need for protocols to do their own
thread management tasks.

6 files changed, 75 insertions, 71 deletions

diff --git a/src/core/defs.h b/src/core/defs.h
index 98f4e661..9c745660 100644
--- a/src/core/defs.h
+++ b/src/core/defs.h
@@ -37,6 +37,7 @@ typedef struct nni_proto		nni_proto;
 typedef int				nni_signal;     // Wakeup channel.
 typedef uint64_t			nni_time;       // Abs. time (usec).
 typedef int64_t				nni_duration;   // Rel. time (usec).
+typedef void (*nni_worker)(void *);
 
 // Used by transports for scatter gather I/O.
 typedef struct {
@@ -53,6 +54,9 @@ typedef struct {
 #define NNI_ALLOC_STRUCT(s)	nni_alloc(sizeof (*s))
 #define NNI_FREE_STRUCT(s)	nni_free((s), sizeof (*s))
 
+// Maximum number of socket or pipe worker threads.
+#define NNI_MAXWORKERS    4
+
 #define NNI_PUT16(ptr, u)				      \
 	do {						      \
 		(ptr)[0] = (uint8_t) (((uint16_t) (u)) >> 8); \
diff --git a/src/core/pipe.c b/src/core/pipe.c
index 150fcd23..249e887c 100644
--- a/src/core/pipe.c
+++ b/src/core/pipe.c
@@ -88,8 +88,11 @@ nni_pipe_peer(nni_pipe *p)
 void
 nni_pipe_destroy(nni_pipe *p)
 {
-	nni_thr_fini(&p->p_send_thr);
-	nni_thr_fini(&p->p_recv_thr);
+	int i;
+
+	for (i = 0; i < NNI_MAXWORKERS; i++) {
+		nni_thr_fini(&p->p_worker_thr[i]);
+	}
 
 	if (p->p_tran_data != NULL) {
 		p->p_tran_ops.pipe_destroy(p->p_tran_data);
@@ -109,6 +112,7 @@ nni_pipe_create(nni_pipe **pp, nni_ep *ep)
 	const nni_proto_pipe_ops *ops = &sock->s_pipe_ops;
 	void *pdata;
 	int rv;
+	int i;
 
 	if ((p = NNI_ALLOC_STRUCT(p)) == NULL) {
 		return (NNG_ENOMEM);
@@ -128,16 +132,19 @@ nni_pipe_create(nni_pipe **pp, nni_ep *ep)
 		return (rv);
 	}
 	p->p_proto_data = pdata;
-	if ((rv = nni_thr_init(&p->p_recv_thr, ops->pipe_recv, pdata)) != 0) {
-		ops->pipe_fini(&p->p_proto_data);
-		NNI_FREE_STRUCT(p);
-		return (rv);
-	}
-	if ((rv = nni_thr_init(&p->p_send_thr, ops->pipe_send, pdata)) != 0) {
-		nni_thr_fini(&p->p_recv_thr);
-		ops->pipe_fini(&p->p_proto_data);
-		NNI_FREE_STRUCT(p);
-		return (rv);
+
+	for (i = 0; i < NNI_MAXWORKERS; i++) {
+		nni_worker fn = ops->pipe_worker[i];
+		rv = nni_thr_init(&p->p_worker_thr[i], fn, pdata);
+		if (rv != 0) {
+			while (i > 0) {
+				i--;
+				nni_thr_fini(&p->p_worker_thr[i]);
+			}
+			ops->pipe_fini(pdata);
+			NNI_FREE_STRUCT(p);
+			return (rv);
+		}
 	}
 
 	*pp = p;
@@ -160,6 +167,7 @@ int
 nni_pipe_start(nni_pipe *pipe)
 {
 	int rv;
+	int i;
 	int collide;
 	nni_sock *sock = pipe->p_sock;
 
@@ -200,8 +208,9 @@ nni_pipe_start(nni_pipe *pipe)
 
 	nni_list_append(&sock->s_pipes, pipe);
 
-	nni_thr_run(&pipe->p_send_thr);
-	nni_thr_run(&pipe->p_recv_thr);
+	for (i = 0; i < NNI_MAXWORKERS; i++) {
+		nni_thr_run(&pipe->p_worker_thr[i]);
+	}
 	pipe->p_active = 1;
 
 	// XXX: Publish event
diff --git a/src/core/pipe.h b/src/core/pipe.h
index cc14de86..65d2ede5 100644
--- a/src/core/pipe.h
+++ b/src/core/pipe.h
@@ -27,8 +27,7 @@ struct nng_pipe {
 	nni_ep *	p_ep;
 	int		p_reap;
 	int		p_active;
-	nni_thr		p_send_thr;
-	nni_thr		p_recv_thr;
+	nni_thr		p_worker_thr[NNI_MAXWORKERS];
 };
 
 // Pipe operations that protocols use.
diff --git a/src/core/protocol.h b/src/core/protocol.h
index 000e9c64..11c4c304 100644
--- a/src/core/protocol.h
+++ b/src/core/protocol.h
@@ -24,35 +24,31 @@
 struct nni_proto_pipe_ops {
 	// pipe_init creates the protocol-specific per pipe data structure.
 	// The last argument is the per-socket protocol private data.
-	int	(*pipe_init)(void **, nni_pipe *, void *);
+	int		(*pipe_init)(void **, nni_pipe *, void *);
 
 	// pipe_fini releases any pipe data structures.  This is called after
 	// the pipe has been removed from the protocol, and the generic
 	// pipe threads have been stopped.
-	void	(*pipe_fini)(void *);
+	void		(*pipe_fini)(void *);
 
 	// pipe_add is called to register a pipe with the protocol.  The
 	// protocol can reject this, for example if another pipe is already
 	// active on a 1:1 protocol.  The protocol may not block during this,
 	// as the socket lock is held.
-	int	(*pipe_add)(void *);
+	int		(*pipe_add)(void *);
 
 	// pipe_rem is called to unregister a pipe from the protocol.
 	// Threads may still acccess data structures, so the protocol
 	// should not free anything yet.  This is called with the socket
 	// lock held, so the protocol may not call back into the socket, and
 	// must not block.
-	void	(*pipe_rem)(void *);
+	void		(*pipe_rem)(void *);
 
-	// pipe_send is a function run in a thread per pipe, to process
-	// send activity.  This can be NULL.
-	void	(*pipe_send)(void *);
-
-	// pipe_recv is a function run in a thread per pipe, to process
-	// receive activity.  While this can be NULL, it should NOT be, as
-	// otherwise the protocol may not be able to discover the closure of
-	// the underlying transport (such as a remote disconnect).
-	void	(*pipe_recv)(void *);
+	// Worker functions.  If non-NULL, each worker is executed and
+	// given the protocol pipe data as a argument.  All workers are
+	// started, or none are started.  The pipe_fini function is obliged
+	// to ensure that workers have exited.
+	nni_worker	pipe_worker[NNI_MAXWORKERS];
 };
 
 struct nni_proto_sock_ops {
@@ -75,15 +71,6 @@ struct nni_proto_sock_ops {
 	int		(*sock_setopt)(void *, int, const void *, size_t);
 	int		(*sock_getopt)(void *, int, void *, size_t *);
 
-	// sock_send is a send worker.  It can really be anything, but it
-	// is run in a separate thread (if it is non-NULL).
-	void		(*sock_send)(void *);
-
-	// sock_recv is a receive worker.  As with send it can really be
-	// anything, its just a thread that runs for the duration of the
-	// socket.
-	void		(*sock_recv)(void *);
-
 	// Receive filter.  This may be NULL, but if it isn't, then
 	// messages coming into the system are routed here just before being
 	// delivered to the application.  To drop the message, the prtocol
@@ -93,6 +80,12 @@ struct nni_proto_sock_ops {
 	// Send filter.  This may be NULL, but if it isn't, then messages
 	// here are filtered just after they come from the application.
 	nni_msg *	(*sock_sfilter)(void *, nni_msg *);
+
+	// Worker functions.  If non-NULL, each worker is executed and given
+	// the protocol socket data as an argument.  These will all be started
+	// at about the same time, and all will be started, or none will be
+	// started.  They are obliged to exit in response to sock_close.
+	nni_worker	sock_worker[NNI_MAXWORKERS];
 };
 
 struct nni_proto {
diff --git a/src/core/socket.c b/src/core/socket.c
index ad8e5703..c6b0408e 100644
--- a/src/core/socket.c
+++ b/src/core/socket.c
@@ -142,6 +142,7 @@ nni_sock_open(nni_sock **sockp, uint16_t pnum)
 	nni_sock *sock;
 	nni_proto *proto;
 	int rv;
+	int i;
 	nni_proto_sock_ops *sops;
 	nni_proto_pipe_ops *pops;
 
@@ -217,7 +218,7 @@ nni_sock_open(nni_sock **sockp, uint16_t pnum)
 	}
 	if ((rv = nni_msgq_init(&sock->s_urq, 0)) != 0) {
 		nni_msgq_fini(sock->s_uwq);
-		nni_thr_fini(&sock->s_recver);
+		nni_thr_fini(&sock->s_reaper);
 		nni_cv_fini(&sock->s_cv);
 		nni_mtx_fini(&sock->s_mx);
 		NNI_FREE_STRUCT(sock);
@@ -234,36 +235,30 @@ nni_sock_open(nni_sock **sockp, uint16_t pnum)
 		return (rv);
 	}
 
-	// NB: If these functions are NULL, the thread initialization is
-	// largely a NO-OP.  The system won't actually create the threads.
-	rv = nni_thr_init(&sock->s_sender, sops->sock_send, sock->s_data);
-	if (rv != 0) {
-		nni_thr_wait(&sock->s_reaper);
-		sops->sock_fini(&sock->s_data);
-		nni_msgq_fini(sock->s_urq);
-		nni_msgq_fini(sock->s_uwq);
-		nni_thr_fini(&sock->s_reaper);
-		nni_cv_fini(&sock->s_cv);
-		nni_mtx_fini(&sock->s_mx);
-		NNI_FREE_STRUCT(sock);
-	}
-	rv = nni_thr_init(&sock->s_recver, sops->sock_recv, sock->s_data);
-	if (rv != 0) {
-		nni_thr_wait(&sock->s_sender);
-		sops->sock_fini(&sock->s_data);
-		nni_msgq_fini(sock->s_urq);
-		nni_msgq_fini(sock->s_uwq);
-		nni_thr_fini(&sock->s_sender);
-		nni_thr_fini(&sock->s_reaper);
-		nni_cv_fini(&sock->s_cv);
-		nni_mtx_fini(&sock->s_mx);
-		NNI_FREE_STRUCT(sock);
+	// NB: If worker functions are null, then the thread initialization
+	// turns into a NOP, and no actual thread will be started.
+	for (i = 0; i < NNI_MAXWORKERS; i++) {
+		nni_worker fn = sops->sock_worker[i];
+		rv = nni_thr_init(&sock->s_worker_thr[i], fn, sock->s_data);
+		if (rv != 0) {
+			while (i > 0) {
+				i--;
+				nni_thr_fini(&sock->s_worker_thr[i]);
+			}
+			sops->sock_fini(&sock->s_data);
+			nni_msgq_fini(sock->s_urq);
+			nni_msgq_fini(sock->s_uwq);
+			nni_cv_fini(&sock->s_cv);
+			nni_mtx_fini(&sock->s_mx);
+			NNI_FREE_STRUCT(sock);
+		}
 	}
 
+	for (i = 0; i < NNI_MAXWORKERS; i++) {
+		nni_thr_run(&sock->s_worker_thr[i]);
+	}
 
 	nni_thr_run(&sock->s_reaper);
-	nni_thr_run(&sock->s_recver);
-	nni_thr_run(&sock->s_sender);
 	*sockp = sock;
 	return (0);
 }
@@ -279,6 +274,7 @@ nni_sock_shutdown(nni_sock *sock)
 	nni_pipe *pipe;
 	nni_ep *ep;
 	nni_time linger;
+	int i;
 
 	nni_mtx_lock(&sock->s_mx);
 	if (sock->s_closing) {
@@ -350,8 +346,9 @@ nni_sock_shutdown(nni_sock *sock)
 	nni_mtx_unlock(&sock->s_mx);
 
 	// Wait for the threads to exit.
-	nni_thr_wait(&sock->s_sender);
-	nni_thr_wait(&sock->s_recver);
+	for (i = 0; i < NNI_MAXWORKERS; i++) {
+		nni_thr_wait(&sock->s_worker_thr[i]);
+	}
 	nni_thr_wait(&sock->s_reaper);
 
 	// At this point, there are no threads blocked inside of us
@@ -368,6 +365,8 @@ nni_sock_shutdown(nni_sock *sock)
 void
 nni_sock_close(nni_sock *sock)
 {
+	int i;
+
 	// Shutdown everything if not already done.  This operation
 	// is idempotent.
 	nni_sock_shutdown(sock);
@@ -383,9 +382,10 @@ nni_sock_close(nni_sock *sock)
 	sock->s_sock_ops.sock_fini(sock->s_data);
 
 	// And we need to clean up *our* state.
+	for (i = 0; i < NNI_MAXWORKERS; i++) {
+		nni_thr_fini(&sock->s_worker_thr[i]);
+	}
 	nni_thr_fini(&sock->s_reaper);
-	nni_thr_fini(&sock->s_sender);
-	nni_thr_fini(&sock->s_recver);
 	nni_msgq_fini(sock->s_urq);
 	nni_msgq_fini(sock->s_uwq);
 	nni_cv_fini(&sock->s_cv);
diff --git a/src/core/socket.h b/src/core/socket.h
index 197b5d0d..4656c5d7 100644
--- a/src/core/socket.h
+++ b/src/core/socket.h
@@ -40,8 +40,7 @@ struct nng_socket {
 
 	nni_list		s_reaps;        // pipes to reap
 	nni_thr			s_reaper;
-	nni_thr			s_sender;
-	nni_thr			s_recver;
+	nni_thr			s_worker_thr[NNI_MAXWORKERS];
 
 	int			s_ep_pend;      // EP dial/listen in progress
 	int			s_closing;      // Socket is closing