fixes #112 Need to move some stuff from socket to message queues

5 files changed, 165 insertions, 171 deletions

diff --git a/src/core/msgqueue.c b/src/core/msgqueue.c
index 6fd95f98..c1f3701c 100644
--- a/src/core/msgqueue.c
+++ b/src/core/msgqueue.c
@@ -27,12 +27,17 @@ struct nni_msgq {
 	int       mq_puterr;
 	int       mq_geterr;
 	int       mq_draining;
+	int       mq_besteffort;
 	nni_msg **mq_msgs;
 
 	nni_list mq_aio_putq;
 	nni_list mq_aio_getq;
 	nni_list mq_aio_notify_get;
 	nni_list mq_aio_notify_put;
+
+	// Filters.
+	nni_msgq_filter mq_filter_fn;
+	void *          mq_filter_arg;
 };
 
 int
@@ -157,6 +162,13 @@ nni_msgq_set_error(nni_msgq *mq, int error)
 	nni_mtx_unlock(&mq->mq_lock);
 }
 
+void
+nni_msgq_set_filter(nni_msgq *mq, nni_msgq_filter filter, void *arg)
+{
+	mq->mq_filter_fn  = filter;
+	mq->mq_filter_arg = arg;
+}
+
 static void
 nni_msgq_run_putq(nni_msgq *mq)
 {
@@ -173,12 +185,19 @@ nni_msgq_run_putq(nni_msgq *mq)
 		// the queue is empty, otherwise it would have just taken
 		// data from the queue.
 		if ((raio = nni_list_first(&mq->mq_aio_getq)) != NULL) {
-			nni_aio_set_msg(waio, NULL);
 
-			nni_aio_list_remove(raio);
+			nni_aio_set_msg(waio, NULL);
 			nni_aio_list_remove(waio);
+
+			if (mq->mq_filter_fn != NULL) {
+				msg = mq->mq_filter_fn(mq->mq_filter_arg, msg);
+			}
+			if (msg != NULL) {
+				nni_aio_list_remove(raio);
+				nni_aio_finish_msg(raio, msg);
+			}
+
 			nni_aio_finish(waio, 0, len);
-			nni_aio_finish_msg(raio, msg);
 			continue;
 		}
 
@@ -195,40 +214,76 @@ nni_msgq_run_putq(nni_msgq *mq)
 			continue;
 		}
 
+		// If we are in best effort mode, just drop the message
+		// as if we delivered.
+		if (mq->mq_besteffort) {
+			nni_list_remove(&mq->mq_aio_putq, waio);
+			nni_aio_set_msg(waio, NULL);
+			nni_msg_free(msg);
+			nni_aio_finish(waio, 0, len);
+			continue;
+		}
+
 		// Unable to make progress, leave the aio where it is.
 		break;
 	}
 }
 
+void
+nni_msgq_set_best_effort(nni_msgq *mq, int on)
+{
+	nni_mtx_lock(&mq->mq_lock);
+	mq->mq_besteffort = on;
+	if (on) {
+		nni_msgq_run_putq(mq);
+	}
+	nni_mtx_unlock(&mq->mq_lock);
+}
+
 static void
 nni_msgq_run_getq(nni_msgq *mq)
 {
 	nni_aio *raio;
 	nni_aio *waio;
-	nni_msg *msg;
 
 	while ((raio = nni_list_first(&mq->mq_aio_getq)) != NULL) {
 		// If anything is waiting in the queue, get it first.
 		if (mq->mq_len != 0) {
-			msg = mq->mq_msgs[mq->mq_get++];
+			nni_msg *msg = mq->mq_msgs[mq->mq_get++];
 			if (mq->mq_get == mq->mq_alloc) {
 				mq->mq_get = 0;
 			}
 			mq->mq_len--;
-			nni_aio_list_remove(raio);
-			nni_aio_finish_msg(raio, msg);
+
+			if (mq->mq_filter_fn != NULL) {
+				msg = mq->mq_filter_fn(mq->mq_filter_arg, msg);
+			}
+			if (msg != NULL) {
+				nni_aio_list_remove(raio);
+				nni_aio_finish_msg(raio, msg);
+			}
 			continue;
 		}
 
 		// Nothing queued (unbuffered?), maybe a writer is waiting.
 		if ((waio = nni_list_first(&mq->mq_aio_putq)) != NULL) {
+			nni_msg *msg;
+			size_t   len;
 			msg = nni_aio_get_msg(waio);
+			len = nni_msg_len(msg);
+
 			nni_aio_set_msg(waio, NULL);
 			nni_aio_list_remove(waio);
-			nni_aio_list_remove(raio);
 
-			nni_aio_finish(waio, 0, nni_msg_len(msg));
-			nni_aio_finish_msg(raio, msg);
+			if (mq->mq_filter_fn != NULL) {
+				msg = mq->mq_filter_fn(mq->mq_filter_arg, msg);
+			}
+			if (msg != NULL) {
+				nni_aio_list_remove(raio);
+				nni_aio_finish_msg(raio, msg);
+			}
+
+			nni_aio_finish(waio, 0, len);
 			continue;
 		}
 
@@ -393,44 +448,6 @@ nni_msgq_tryput(nni_msgq *mq, nni_msg *msg)
 	return (NNG_EAGAIN);
 }
 
-int
-nni_msgq_get_until(nni_msgq *mq, nni_msg **msgp, nni_time expire)
-{
-	nni_aio *aio;
-	int      rv;
-
-	if ((rv = nni_aio_init(&aio, NULL, NULL)) != 0) {
-		return (rv);
-	}
-	nni_aio_set_timeout(aio, expire);
-	nni_msgq_aio_get(mq, aio);
-	nni_aio_wait(aio);
-	if ((rv = nni_aio_result(aio)) == 0) {
-		*msgp = nni_aio_get_msg(aio);
-		nni_aio_set_msg(aio, NULL);
-	}
-	nni_aio_fini(aio);
-	return (rv);
-}
-
-int
-nni_msgq_put_until(nni_msgq *mq, nni_msg *msg, nni_time expire)
-{
-	nni_aio *aio;
-	int      rv;
-
-	if ((rv = nni_aio_init(&aio, NULL, NULL)) != 0) {
-		return (rv);
-	}
-	nni_aio_set_timeout(aio, expire);
-	nni_aio_set_msg(aio, msg);
-	nni_msgq_aio_put(mq, aio);
-	nni_aio_wait(aio);
-	rv = nni_aio_result(aio);
-	nni_aio_fini(aio);
-	return (rv);
-}
-
 void
 nni_msgq_drain(nni_msgq *mq, nni_time expire)
 {
diff --git a/src/core/msgqueue.h b/src/core/msgqueue.h
index fa6b845a..bbac505d 100644
--- a/src/core/msgqueue.h
+++ b/src/core/msgqueue.h
@@ -41,28 +41,11 @@ extern void nni_msgq_aio_get(nni_msgq *, nni_aio *);
 extern void nni_msgq_aio_notify_get(nni_msgq *, nni_aio *);
 extern void nni_msgq_aio_notify_put(nni_msgq *, nni_aio *);
 
-// nni_msgq_put puts the message to the queue.  It blocks until it
-// was able to do so, or the queue is closed, or a timeout is reached.
-// It returns 0 on success, NNG_ECLOSED if the queue was closed, or
-// NNG_ETIMEDOUT if the timeout is reached. If an error is returned,
-// the caller is responsible for freeing the message with nni_msg_free(),
-// otherwise the message is "owned" by the queue, and the caller is not
-// permitted to access it further.
-extern int nni_msgq_put_until(nni_msgq *, nni_msg *, nni_time);
-
 // nni_msgq_tryput performs a non-blocking attempt to put a message on
 // the message queue.  It is the same as calling nng_msgq_put_until with
 // a zero time.
 extern int nni_msgq_tryput(nni_msgq *, nni_msg *);
 
-// nni_msgq_get_until gets a message from the queue.  It blocks until a
-// message is available, the queue is closed, or time out is reached.
-// It returns 0 on success, NNG_ECLOSED if the queue was closed, or
-// NNG_ETIMEDOUT if the timeout is reached.  On successful return,
-// the caller assumes ownership of the message and must call
-// nni_msg_free() when it is finished with it.
-extern int nni_msgq_get_until(nni_msgq *, nni_msg **, nni_time);
-
 // nni_msgq_set_error sets an error condition on the message queue,
 // which causes all current and future readers/writes to return the
 // given error condition (if non-zero).  Threads waiting to put or get
@@ -81,6 +64,25 @@ extern void nni_msgq_set_put_error(nni_msgq *, int);
 // Readers (nni_msgq_put*) are unaffected.
 extern void nni_msgq_set_get_error(nni_msgq *, int);
 
+// nni_msgq_set_best_effort marks the message queue best effort on send.
+// What this does is treat the message queue condition as if it were
+// successful, returning 0, and discarding the message.  If zero is
+// passed then this mode is reset to normal.
+extern void nni_msgq_set_best_effort(nni_msgq *, int);
+
+// nni_msgq_filter is a callback function used to filter messages.
+// The function is called on entry (put) or exit (get).  The void
+// argument is an opaque pointer supplied with the function at registration
+// time.  The primary use for these functions is to support the protocol
+// socket needs.
+typedef nni_msg *(*nni_msgq_filter)(void *, nni_msg *);
+
+// nni_msgq_set_filter sets the filter on the queue.  Messages
+// are filtered through this just before they are returned via the get
+// functions.  If the filter returns NULL, then the message is silently
+// discarded instead, and any get waiters remain waiting.
+extern void nni_msgq_set_filter(nni_msgq *, nni_msgq_filter, void *);
+
 // nni_msgq_close closes the queue.  After this all operates on the
 // message queue will return NNG_ECLOSED.  Messages inside the queue
 // are freed.  Unlike closing a go channel, this operation is idempotent.
diff --git a/src/core/protocol.h b/src/core/protocol.h
index 0c0d93ce..253452d8 100644
--- a/src/core/protocol.h
+++ b/src/core/protocol.h
@@ -74,15 +74,17 @@ struct nni_proto_sock_ops {
 	// it can signal the socket worker threads to exit.
 	void (*sock_close)(void *);
 
-	// Receive filter.  This may be NULL, but if it isn't, then
+	// Send a message.
+	void (*sock_send)(void *, nni_aio *);
+
+	// Receive a message.
+	void (*sock_recv)(void *, nni_aio *);
+
+	// Message 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
+	// delivered to the application.  To drop the message, the protocol
 	// should return NULL, otherwise the message (possibly modified).
-	nni_msg *(*sock_rfilter)(void *, nni_msg *);
-
-	// 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 *);
+	nni_msg *(*sock_filter)(void *, nni_msg *);
 
 	// Options. Must not be NULL. Final entry should have NULL name.
 	nni_proto_sock_option *sock_options;
diff --git a/src/core/socket.c b/src/core/socket.c
index 8895f7a7..bc1f446d 100644
--- a/src/core/socket.c
+++ b/src/core/socket.c
@@ -66,12 +66,9 @@ struct nni_socket {
 	nni_list s_eps;   // active endpoints
 	nni_list s_pipes; // active pipes
 
-	int s_ep_pend;    // EP dial/listen in progress
-	int s_closing;    // Socket is closing
-	int s_closed;     // Socket closed, protected by global lock
-	int s_besteffort; // Best effort mode delivery
-	int s_senderr;    // Protocol state machine use
-	int s_recverr;    // Protocol state machine use
+	int s_ep_pend; // EP dial/listen in progress
+	int s_closing; // Socket is closing
+	int s_closed;  // Socket closed, protected by global lock
 
 	nni_event s_recv_ev; // Event for readability
 	nni_event s_send_ev; // Event for sendability
@@ -374,15 +371,19 @@ nni_sock_pipe_remove(nni_sock *sock, nni_pipe *pipe)
 }
 
 void
-nni_sock_lock(nni_sock *sock)
+nni_sock_send_pending(nni_sock *sock)
 {
-	nni_mtx_lock(&sock->s_mx);
+	if (sock->s_send_fd.sn_init) {
+		nni_plat_pipe_clear(sock->s_send_fd.sn_rfd);
+	}
 }
 
 void
-nni_sock_unlock(nni_sock *sock)
+nni_sock_recv_pending(nni_sock *sock)
 {
-	nni_mtx_unlock(&sock->s_mx);
+	if (sock->s_recv_fd.sn_init) {
+		nni_plat_pipe_clear(sock->s_recv_fd.sn_rfd);
+	}
 }
 
 static void
@@ -551,6 +552,11 @@ nni_sock_create(nni_sock **sp, const nni_proto *proto)
 		return (rv);
 	}
 
+	if (s->s_sock_ops.sock_filter != NULL) {
+		nni_msgq_set_filter(
+		    s->s_urq, s->s_sock_ops.sock_filter, s->s_data);
+	}
+
 	*sp = s;
 	return (rv);
 }
@@ -779,13 +785,23 @@ nni_sock_closeall(void)
 	}
 }
 
+void
+nni_sock_send(nni_sock *sock, nni_aio *aio)
+{
+	sock->s_sock_ops.sock_send(sock->s_data, aio);
+}
+
 int
 nni_sock_sendmsg(nni_sock *sock, nni_msg *msg, int flags)
 {
 	int      rv;
-	int      besteffort;
 	nni_time expire;
 	nni_time timeo = sock->s_sndtimeo;
+	nni_aio *aio;
+
+	if ((rv = nni_aio_init(&aio, NULL, NULL)) != 0) {
+		return (rv);
+	}
 
 	if ((flags == NNG_FLAG_NONBLOCK) || (timeo == 0)) {
 		expire = NNI_TIME_ZERO;
@@ -795,47 +811,24 @@ nni_sock_sendmsg(nni_sock *sock, nni_msg *msg, int flags)
 		expire = nni_clock();
 		expire += timeo;
 	}
+	nni_aio_set_timeout(aio, expire);
+	nni_aio_set_msg(aio, msg);
 
-	// Senderr is typically set by protocols when the state machine
-	// indicates that it is no longer valid to send a message.  E.g.
-	// a REP socket with no REQ pending.
-	nni_mtx_lock(&sock->s_mx);
-	if (sock->s_closing) {
-		nni_mtx_unlock(&sock->s_mx);
-		return (NNG_ECLOSED);
-	}
-	if ((rv = sock->s_senderr) != 0) {
-		nni_mtx_unlock(&sock->s_mx);
-		return (rv);
-	}
-	besteffort = sock->s_besteffort;
-
-	if (sock->s_sock_ops.sock_sfilter != NULL) {
-		msg = sock->s_sock_ops.sock_sfilter(sock->s_data, msg);
-	}
-	nni_mtx_unlock(&sock->s_mx);
+	nni_sock_send(sock, aio);
+	nni_aio_wait(aio);
 
-	if (msg == NULL) {
-		return (0);
-	}
+	rv = nni_aio_result(aio);
+	nni_aio_fini(aio);
 
-	if (besteffort) {
-		// BestEffort mode -- if we cannot handle the message due to
-		// backpressure, we just throw it away, and don't complain.
-		expire = NNI_TIME_ZERO;
-	}
-	if (sock->s_send_fd.sn_init) {
-		nni_plat_pipe_clear(sock->s_send_fd.sn_rfd);
-	}
-	rv = nni_msgq_put_until(sock->s_uwq, msg, expire);
-	if (besteffort && (rv == NNG_ETIMEDOUT)) {
-		// Pretend this worked... it didn't, but pretend.
-		nni_msg_free(msg);
-		return (0);
-	}
 	return (rv);
 }
 
+void
+nni_sock_recv(nni_sock *sock, nni_aio *aio)
+{
+	sock->s_sock_ops.sock_recv(sock->s_data, aio);
+}
+
 int
 nni_sock_recvmsg(nni_sock *sock, nni_msg **msgp, int flags)
 {
@@ -843,6 +836,11 @@ nni_sock_recvmsg(nni_sock *sock, nni_msg **msgp, int flags)
 	nni_msg *msg;
 	nni_time expire;
 	nni_time timeo = sock->s_rcvtimeo;
+	nni_aio *aio;
+
+	if ((rv = nni_aio_init(&aio, NULL, NULL)) != 0) {
+		return (rv);
+	}
 
 	if ((flags == NNG_FLAG_NONBLOCK) || (timeo == 0)) {
 		expire = NNI_TIME_ZERO;
@@ -853,39 +851,23 @@ nni_sock_recvmsg(nni_sock *sock, nni_msg **msgp, int flags)
 		expire += timeo;
 	}
 
-	nni_mtx_lock(&sock->s_mx);
-	if (sock->s_closing) {
-		nni_mtx_unlock(&sock->s_mx);
-		return (NNG_ECLOSED);
-	}
-	if ((rv = sock->s_recverr) != 0) {
-		nni_mtx_unlock(&sock->s_mx);
-		return (rv);
-	}
-	nni_mtx_unlock(&sock->s_mx);
-
-	if (sock->s_recv_fd.sn_init) {
-		nni_plat_pipe_clear(sock->s_recv_fd.sn_rfd);
-	}
-
 	for (;;) {
-		rv = nni_msgq_get_until(sock->s_urq, &msg, expire);
+		nni_aio_set_timeout(aio, expire);
+		nni_sock_recv(sock, aio);
+		nni_aio_wait(aio);
+
+		rv = nni_aio_result(aio);
 		if (rv != 0) {
-			return (rv);
-		}
-		if (sock->s_sock_ops.sock_rfilter != NULL) {
-			nni_mtx_lock(&sock->s_mx);
-			msg = sock->s_sock_ops.sock_rfilter(sock->s_data, msg);
-			nni_mtx_unlock(&sock->s_mx);
-		}
-		if (msg != NULL) {
 			break;
 		}
-		// Protocol dropped the message; try again.
-	}
+		msg = nni_aio_get_msg(aio);
+		nni_aio_set_msg(aio, NULL);
 
-	*msgp = msg;
-	return (0);
+		*msgp = msg;
+		break;
+	}
+	nni_aio_fini(aio);
+	return (rv);
 }
 
 // nni_sock_protocol returns the socket's 16-bit protocol number.
@@ -949,18 +931,6 @@ nni_sock_ep_remove(nni_sock *sock, nni_ep *ep)
 	nni_mtx_unlock(&sock->s_mx);
 }
 
-void
-nni_sock_recverr(nni_sock *sock, int err)
-{
-	sock->s_recverr = err;
-}
-
-void
-nni_sock_senderr(nni_sock *sock, int err)
-{
-	sock->s_senderr = err;
-}
-
 int
 nni_sock_setopt(nni_sock *s, const char *name, const void *val, size_t size)
 {
diff --git a/src/core/socket.h b/src/core/socket.h
index 850c4641..edfe6447 100644
--- a/src/core/socket.h
+++ b/src/core/socket.h
@@ -22,15 +22,14 @@ extern void     nni_sock_closeall(void);
 extern int      nni_sock_shutdown(nni_sock *);
 extern uint16_t nni_sock_proto(nni_sock *);
 extern uint16_t nni_sock_peer(nni_sock *);
-extern int nni_sock_setopt(nni_sock *, const char *, const void *, size_t);
-extern int nni_sock_getopt(nni_sock *, const char *, void *, size_t *);
-extern int nni_sock_recvmsg(nni_sock *, nni_msg **, int);
-extern int nni_sock_sendmsg(nni_sock *, nni_msg *, int);
+extern int  nni_sock_setopt(nni_sock *, const char *, const void *, size_t);
+extern int  nni_sock_getopt(nni_sock *, const char *, void *, size_t *);
+extern int  nni_sock_recvmsg(nni_sock *, nni_msg **, int);
+extern int  nni_sock_sendmsg(nni_sock *, nni_msg *, int);
+extern void nni_sock_send(nni_sock *, nni_aio *);
+extern void nni_sock_recv(nni_sock *, nni_aio *);
 extern uint32_t nni_sock_id(nni_sock *);
 
-extern void nni_sock_lock(nni_sock *);
-extern void nni_sock_unlock(nni_sock *);
-
 extern nni_notify *nni_sock_notify(nni_sock *, int, nng_notify_func, void *);
 extern void        nni_sock_unnotify(nni_sock *, nni_notify *);
 
@@ -46,16 +45,20 @@ extern int nni_sock_pipe_start(nni_sock *, nni_pipe *p);
 extern int  nni_sock_ep_add(nni_sock *, nni_ep *);
 extern void nni_sock_ep_remove(nni_sock *, nni_ep *);
 
-// 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_sock_recverr(nni_sock *, int);
-extern void nni_sock_senderr(nni_sock *, int);
-
 // These are socket methods that protocol operations can expect to call.
 // Note that each of these should be called without any locks held, since
 // the socket can reenter the protocol.
 
+// nni_sock_send_pending is called by the protocol when it enqueues
+// a send operation.  The main purpose of this is to clear the raised
+// signal raised on the event descriptor.
+extern void nni_sock_send_pending(nni_sock *);
+
+// nni_sock_recv_pending is called by the protocl when it enqueues
+// a receive operation.  The main purpose of this is to clear the raised
+// signal raised on the event descriptor.
+extern void nni_sock_recv_pending(nni_sock *);
+
 // nni_socket_sendq obtains the upper writeq.  The protocol should
 // recieve messages from this, and place them on the appropriate pipe.
 extern nni_msgq *nni_sock_sendq(nni_sock *);