r2100: rework the dcerpc client side library so that it is async. We now

generate a separate *_send() async function for every RPC call, and
there is a single dcerpc_ndr_request_recv() call that processes the
receive side of any rpc call. The caller can use
dcerpc_event_context() to get a pointer to the event context for the
pipe so that events can be waited for asynchronously.

The only part that remains synchronous is the initial bind
calls. These could also be made async if necessary, although I suspect
most applications won't need them to be.
(This used to be commit f5d004d8eb8c76c03342cace1976b27266cfa1f0)

1 files changed, 194 insertions, 73 deletions

diff --git a/source4/librpc/rpc/dcerpc_tcp.c b/source4/librpc/rpc/dcerpc_tcp.c
index 05c700ea89..7cf7cc98bb 100644
--- a/source4/librpc/rpc/dcerpc_tcp.c
+++ b/source4/librpc/rpc/dcerpc_tcp.c
@@ -22,136 +22,239 @@
 
 #include "includes.h"
 
+#define MIN_HDR_SIZE 16
+
+struct tcp_blob {
+	struct tcp_blob *next, *prev;
+	DATA_BLOB data;
+};
+
 /* transport private information used by TCP pipe transport */
 struct tcp_private {
+	struct event_context *event_ctx;
+	struct fd_event *fde;
 	int fd;
 	char *server_name;
 	uint32_t port;
+
+	struct tcp_blob *pending_send;
+
+	struct {
+		size_t received;
+		DATA_BLOB data;
+		uint_t pending_count;
+	} recv;
 };
 
 
 /*
   mark the socket dead
 */
-static void tcp_sock_dead(struct tcp_private *tcp)
+static void tcp_sock_dead(struct dcerpc_pipe *p, NTSTATUS status)
 {
+	struct tcp_private *tcp = p->transport.private;
+
 	if (tcp && tcp->fd != -1) {
 		close(tcp->fd);
 		tcp->fd = -1;
 	}
+
+	/* wipe any pending sends */
+	while (tcp->pending_send) {
+		struct tcp_blob *blob = tcp->pending_send;
+		DLIST_REMOVE(tcp->pending_send, blob);
+		talloc_free(blob);
+	}
+
+	if (!NT_STATUS_IS_OK(status)) {
+		p->transport.recv_data(p, NULL, status);
+	}
 }
 
-static NTSTATUS tcp_raw_recv(struct dcerpc_pipe *p, 
-			     TALLOC_CTX *mem_ctx,
-			     DATA_BLOB *blob)
+/*
+  process send requests
+*/
+static void tcp_process_send(struct dcerpc_pipe *p)
 {
 	struct tcp_private *tcp = p->transport.private;
-	ssize_t ret;
-	uint32_t frag_length;
-	DATA_BLOB blob1;
 
-	blob1 = data_blob_talloc(mem_ctx, NULL, 16);
-	if (!blob1.data) {
-		return NT_STATUS_NO_MEMORY;
+	while (tcp->pending_send) {
+		struct tcp_blob *blob = tcp->pending_send;
+		ssize_t ret = write(tcp->fd, blob->data.data, blob->data.length);
+		if (ret == -1) {
+			if (errno != EAGAIN && errno != EINTR) {
+				tcp_sock_dead(p, NT_STATUS_NET_WRITE_FAULT);
+			}
+			break;
+		}
+
+		blob->data.data += ret;
+		blob->data.length -= ret;
+
+		if (blob->data.length != 0) {
+			break;
+		}
+
+		DLIST_REMOVE(tcp->pending_send, blob);
+		talloc_free(blob);
 	}
 
-	ret = read_data(tcp->fd, blob1.data, blob1.length);
-	if (ret != blob1.length) {
-		tcp_sock_dead(tcp);
-		return NT_STATUS_UNEXPECTED_NETWORK_ERROR;
+	if (tcp->pending_send == NULL) {
+		tcp->fde->flags &= ~EVENT_FD_WRITE;
 	}
+}
+
+
+/*
+  process recv requests
+*/
+static void tcp_process_recv(struct dcerpc_pipe *p)
+{
+	struct tcp_private *tcp = p->transport.private;
+	ssize_t ret;
 
-	/* this could be a ncacn_http endpoint - this doesn't work
-	   yet, but it goes close */
-	if (strncmp(blob1.data, "ncacn_http/1.0", 14) == 0) {
-		memmove(blob1.data, blob1.data+14, 2);
-		ret = read_data(tcp->fd, blob1.data+2, 14);
-		if (ret != 14) {
-			tcp_sock_dead(tcp);
-			return NT_STATUS_UNEXPECTED_NETWORK_ERROR;
+	/* read in the base header to get the fragment length */
+	if (tcp->recv.received < MIN_HDR_SIZE) {
+		uint32_t frag_length;
+
+		ret = read(tcp->fd, tcp->recv.data.data, 
+			   MIN_HDR_SIZE - tcp->recv.received);
+		if (ret == -1) {
+			if (errno != EAGAIN && errno != EINTR) {
+				tcp_sock_dead(p, NT_STATUS_NET_WRITE_FAULT);
+			}
+			return;
 		}
-	}
+		if (ret == 0) {
+			tcp_sock_dead(p, NT_STATUS_NET_WRITE_FAULT);
+			return;
+		}
+
+		tcp->recv.received += ret;
+
+		if (tcp->recv.received != MIN_HDR_SIZE) {
+			return;
+		}
+		frag_length = dcerpc_get_frag_length(&tcp->recv.data);
 
-	/* we might have recieved a partial fragment, in which case we
-	   need to pull the rest of it */
-	frag_length = dcerpc_get_frag_length(&blob1);
-	if (frag_length == blob1.length) {
-		*blob = blob1;
-		return NT_STATUS_OK;
+		tcp->recv.data.data = talloc_realloc(tcp->recv.data.data,
+						     frag_length);
+		if (tcp->recv.data.data == NULL) {
+			tcp_sock_dead(p, NT_STATUS_NO_MEMORY);
+			return;
+		}
+		tcp->recv.data.length = frag_length;
 	}
 
-	*blob = data_blob_talloc(mem_ctx, NULL, frag_length);
-	if (!blob->data) {
-		return NT_STATUS_NO_MEMORY;
+	/* read in the rest of the packet */
+	ret = read(tcp->fd, tcp->recv.data.data + tcp->recv.received,
+		   tcp->recv.data.length - tcp->recv.received);
+	if (ret == -1) {
+		if (errno != EAGAIN && errno != EINTR) {
+			tcp_sock_dead(p, NT_STATUS_NET_WRITE_FAULT);
+		}
+		return;
+	}
+	if (ret == 0) {
+		tcp_sock_dead(p, NT_STATUS_NET_WRITE_FAULT);
+		return;
 	}
-	memcpy(blob->data, blob1.data, blob1.length);
 
-	ret = read_data(tcp->fd, blob->data + blob1.length, frag_length - blob1.length);
-	if (ret != frag_length - blob1.length) {
-		tcp_sock_dead(tcp);
-		return NT_STATUS_UNEXPECTED_NETWORK_ERROR;
+	tcp->recv.received += ret;
+
+	if (tcp->recv.received != tcp->recv.data.length) {
+		return;
 	}
 
-	return NT_STATUS_OK;
+	/* we have a full packet */
+	p->transport.recv_data(p, &tcp->recv.data, NT_STATUS_OK);
+
+	tcp->recv.received = 0;
+	tcp->recv.pending_count--;
+	if (tcp->recv.pending_count == 0) {
+		tcp->fde->flags &= ~EVENT_FD_READ;
+	}
 }
 
-static NTSTATUS tcp_full_request(struct dcerpc_pipe *p, 
-				 TALLOC_CTX *mem_ctx,
-				 DATA_BLOB *request_blob,
-				 DATA_BLOB *reply_blob)
+/*
+  called when a IO is triggered by the events system
+*/
+static void tcp_io_handler(struct event_context *ev, struct fd_event *fde, 
+			   time_t t, uint16_t flags)
 {
+	struct dcerpc_pipe *p = fde->private;
 	struct tcp_private *tcp = p->transport.private;
-	ssize_t ret;
 
-	ret = write_data(tcp->fd, request_blob->data, request_blob->length);
-	if (ret != request_blob->length) {
-		tcp_sock_dead(tcp);
-		return NT_STATUS_UNEXPECTED_NETWORK_ERROR;
+	if (flags & EVENT_FD_WRITE) {
+		tcp_process_send(p);
 	}
 
-	return tcp_raw_recv(p, mem_ctx, reply_blob);
+	if (tcp->fd == -1) {
+		return;
+	}
+
+	if (flags & EVENT_FD_READ) {
+		tcp_process_recv(p);
+	}
 }
-	      
 
 /* 
-   retrieve a secondary pdu from a pipe 
+   send an initial pdu in a multi-pdu sequence
 */
-static NTSTATUS tcp_secondary_request(struct dcerpc_pipe *p, 
-			       TALLOC_CTX *mem_ctx,
-			       DATA_BLOB *blob)
+static NTSTATUS tcp_send_request(struct dcerpc_pipe *p, 
+				 DATA_BLOB *data)
 {
-	return tcp_raw_recv(p, mem_ctx, blob);
-}
+	struct tcp_private *tcp = p->transport.private;
+	struct tcp_blob *blob;
+
+	blob = talloc_p(tcp, struct tcp_blob);
+	if (blob == NULL) {
+		return NT_STATUS_NO_MEMORY;
+	}
+
+	blob->data = data_blob_talloc(blob, data->data, data->length);
+	if (blob->data.data == NULL) {
+		talloc_free(blob);
+		return NT_STATUS_NO_MEMORY;
+	}
 
+	DLIST_ADD_END(tcp->pending_send, blob, struct tcp_blob *);
+
+	tcp->fde->flags |= EVENT_FD_WRITE;
+
+	return NT_STATUS_OK;
+}
 
 /* 
-   send an initial pdu in a multi-pdu sequence
+   initiate a read request 
 */
-static NTSTATUS tcp_initial_request(struct dcerpc_pipe *p, 
-				    TALLOC_CTX *mem_ctx,
-				    DATA_BLOB *blob)
+static NTSTATUS tcp_send_read(struct dcerpc_pipe *p)
 {
 	struct tcp_private *tcp = p->transport.private;
-	ssize_t ret;
 
-	ret = write_data(tcp->fd, blob->data, blob->length);
-	if (ret != blob->length) {
-		tcp_sock_dead(tcp);
-		return NT_STATUS_UNEXPECTED_NETWORK_ERROR;
+	tcp->recv.pending_count++;
+	if (tcp->recv.pending_count == 1) {
+		tcp->fde->flags |= EVENT_FD_READ;
 	}
-
 	return NT_STATUS_OK;
 }
 
+/* 
+   return the event context so the caller can process asynchronously
+*/
+static struct event_context *tcp_event_context(struct dcerpc_pipe *p)
+{
+	struct tcp_private *tcp = p->transport.private;
+
+	return tcp->event_ctx;
+}
 
 /* 
    shutdown TCP pipe connection
 */
 static NTSTATUS tcp_shutdown_pipe(struct dcerpc_pipe *p)
 {
-	struct tcp_private *tcp = p->transport.private;
-
-	tcp_sock_dead(tcp);
+	tcp_sock_dead(p, NT_STATUS_OK);
 
 	return NT_STATUS_OK;
 }
@@ -176,6 +279,7 @@ NTSTATUS dcerpc_pipe_open_tcp(struct dcerpc_pipe **p,
 	struct tcp_private *tcp;
 	int fd;
 	struct in_addr addr;
+	struct fd_event fde;
 
 	if (port == 0) {
 		port = EPMAPPER_PORT;
@@ -191,6 +295,8 @@ NTSTATUS dcerpc_pipe_open_tcp(struct dcerpc_pipe **p,
 		return NT_STATUS_PORT_CONNECTION_REFUSED;
 	}
 
+	set_blocking(fd, False);
+
         if (!(*p = dcerpc_pipe_init())) {
                 return NT_STATUS_NO_MEMORY;
 	}
@@ -200,20 +306,35 @@ NTSTATUS dcerpc_pipe_open_tcp(struct dcerpc_pipe **p,
 	*/
 	(*p)->transport.transport = NCACN_IP_TCP;
 	(*p)->transport.private = NULL;
-	(*p)->transport.full_request = tcp_full_request;
-	(*p)->transport.secondary_request = tcp_secondary_request;
-	(*p)->transport.initial_request = tcp_initial_request;
+
+	(*p)->transport.send_request = tcp_send_request;
+	(*p)->transport.send_read = tcp_send_read;
+	(*p)->transport.event_context = tcp_event_context;
+	(*p)->transport.recv_data = NULL;
+
 	(*p)->transport.shutdown_pipe = tcp_shutdown_pipe;
 	(*p)->transport.peer_name = tcp_peer_name;
 	
-	tcp = talloc((*p)->mem_ctx, sizeof(*tcp));
+	tcp = talloc((*p), sizeof(*tcp));
 	if (!tcp) {
 		dcerpc_pipe_close(*p);
 		return NT_STATUS_NO_MEMORY;
 	}
 
 	tcp->fd = fd;
-	tcp->server_name = talloc_strdup((*p)->mem_ctx, server);
+	tcp->server_name = talloc_strdup((*p), server);
+	tcp->event_ctx = event_context_init();
+	tcp->pending_send = NULL;
+	tcp->recv.received = 0;
+	tcp->recv.data = data_blob_talloc(tcp, NULL, MIN_HDR_SIZE);
+	tcp->recv.pending_count = 0;
+
+	fde.fd = fd;
+	fde.flags = 0;
+	fde.handler = tcp_io_handler;
+	fde.private = *p;
+
+	tcp->fde = event_add_fd(tcp->event_ctx, &fde);
 
 	(*p)->transport.private = tcp;