More pipe details, and clarified locking / call order comments in transport.h.

5 files changed, 83 insertions, 12 deletions

diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index d6bd960b..af29cfac 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -34,6 +34,7 @@ set (NNG_SOURCES
     core/nng_impl.h
     core/panic.c
     core/panic.h
+    core/pipe.c
     core/platform.c
     core/platform.h
     core/protocol.h
diff --git a/src/core/pipe.c b/src/core/pipe.c
index ab43dbd4..408964ab 100644
--- a/src/core/pipe.c
+++ b/src/core/pipe.c
@@ -24,12 +24,16 @@
 
 /*
  * This file contains functions relating to pipes.
+ *
+ * Operations on pipes (to the transport) are generally blocking operations,
+ * performed in the context of the protocol.
  */
 
 struct nng_pipe {
         uint32_t                p_id;
         struct nni_pipe_ops     p_ops;
         void                    *p_tran;
+        nni_list_node_t         p_node;
 };
 
 /* nni_pipe_id returns the 32-bit pipe id, which can be used in backtraces. */
@@ -51,8 +55,19 @@ nni_pipe_recv(nng_pipe_t p, nng_msg_t *msgp)
         return (p->p_ops.p_recv(p->p_tran, msgp));
 }
 
+/*
+ * nni_pipe_close closes the underlying connection.  It is expected that
+ * subsequent attempts receive or send (including any waiting receive) will
+ * simply return NNG_ECLOSED.
+ */
+void
+nni_pipe_close(nng_pipe_t p)
+{
+        return (p->p_ops.p_close(p->p_tran));
+}
+
 uint16_t
-nni_pipe_peer_proto(nng_pipe_t p)
+nni_pipe_peer(nng_pipe_t p)
 {
         return (p->p_ops.p_peer(p->p_tran));
 }
\ No newline at end of file
diff --git a/src/core/protocol.h b/src/core/protocol.h
index b0e579a3..ae97e896 100644
--- a/src/core/protocol.h
+++ b/src/core/protocol.h
@@ -122,6 +122,6 @@ extern void nni_socket_send_err(nng_socket_t, int);
 extern int nni_pipe_recv(nng_pipe_t, nng_msg_t *);
 extern int nni_pipe_send(nng_pipe_t, nng_msg_t);
 extern uint32_t nni_pipe_id(nng_pipe_t);
-extern uint32_t nni_pipe_close(nng_pipe_t);
+extern void nni_pipe_close(nng_pipe_t);
 
 #endif /* CORE_PROTOCOL_H */
diff --git a/src/core/socket.c b/src/core/socket.c
index 6bb1d5d5..dffc8a2d 100644
--- a/src/core/socket.c
+++ b/src/core/socket.c
@@ -27,14 +27,13 @@
  */
 
 struct nng_socket {
-	int		s_proto;
 	nni_mutex_t	s_mx;
 
 	nni_msgqueue_t	s_uwq;	/* Upper write queue. */
 	nni_msgqueue_t	s_urq;	/* Upper read queue. */
 
-	/* uwq */
-	/* urq */
+	struct nni_protocol s_ops;
+
 	/* options */
 	/* pipes */
 	/* endpoints */
@@ -43,6 +42,22 @@ struct nng_socket {
 	int		s_senderr;	/* Protocol state machine use. */
 };
 
+/*
+ * nni_socket_sendq and nni_socket_recvq are called by the protocol
+ * to obtain the upper read and write queues.
+ */
+nni_msgqueue_t
+nng_socket_sendq(nng_socket_t s)
+{
+	return (s->s_uwq);
+}
+
+nni_msgqueue_t
+nni_socket_recvq(nng_socket_t s)
+{
+	return (s->s_urq);
+}
+
 int
 nng_socket_create(nng_socket_t *sockp, int proto)
 {
diff --git a/src/core/transport.h b/src/core/transport.h
index 93eeafc0..c79c12ad 100644
--- a/src/core/transport.h
+++ b/src/core/transport.h
@@ -67,6 +67,12 @@ struct nni_transport {
 	void (*tran_fork)(int prefork);
 };
 
+/*
+ * Endpoint operations are called by the socket in a protocol-independent
+ * fashion.  The socket makes individual calls, which are expected to block
+ * if appropriate (except for destroy). Endpoints are unable to call back
+ * into the socket, to prevent recusive entry and deadlock.
+ */
 struct nni_endpt_ops {
 	/*
 	 * ep_create creates a vanilla endpoint. The value created is
@@ -116,23 +122,57 @@ struct nni_endpt_ops {
 	int (*ep_getopt)(void *, int, void *, size_t *);
 };
 
+/*
+ * Pipe operations are entry points called by the socket. These may be called
+ * with socket locks held, so it is forbidden for the transport to call
+ * back into the socket at this point.  (Which is one reason pointers back
+ * to socket or even enclosing pipe state, are not provided.)
+ */
 struct nni_pipe_ops {
-	/* p_destroy destroys the pipe */
+	/*
+	 * p_destroy destroys the pipe.  This should clean up all local resources,
+	 * including closing files and freeing memory, used by the pipe.  After
+	 * this call returns, the system will not make further calls on the same
+	 * pipe.
+	 */
 	void (*p_destroy)(void *);
 
-	/* p_send sends the message */
+	/*
+	 * p_send sends the message.  If the message cannot be received, then
+	 * the caller may try again with the same message (or free it).  If the
+	 * call succeeds, then the transport has taken ownership of the message,
+	 * and the caller may not use it again.  The transport will have the
+	 * responsibility to free the message (nng_msg_free()) when it is
+	 * finished with it.
+	 */
 	int (*p_send)(void *, nng_msg_t);
 
-	/* p_recv recvs the message */
+	/*
+	 * p_recv recvs the message. This is a blocking operation, and a read
+	 * will be performed even for cases where no data is expected.  This
+	 * allows the socket to detect a closed socket, by the returned error
+	 * NNG_ECLOSED. Note that the closed socket condition can arise as either
+	 * a result of a remote peer closing the connection, or a synchronous
+	 * call to p_close.
+	 */
 	int (*p_recv)(void *, nng_msg_t *);
 
-	/* p_close closes the pipe */
+	/*
+	 * p_close closes the pipe.  Further recv or send operations should
+	 * return back NNG_ECLOSED.
+	 */
 	void (*p_close)(void *);
 
-	/* p_proto returns the peer protocol */
-	uint16_t (*p_proto)(void *);
+	/*
+	 * p_peer returns the peer protocol. This may arrive in whatever
+	 * transport specific manner is appropriate.
+	 */
+	uint16_t (*p_peer)(void *);
 
-	/* p_getopt gets an pipe (transport-specific) property */
+	/*
+	 * p_getopt gets an pipe (transport-specific) property.  These values
+	 * may not be changed once the pipe is created.
+	 */
 	int (*p_getopt)(void *, int, void *, size_t *);
 };