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)

8 files changed, 747 insertions, 423 deletions

diff --git a/source4/librpc/rpc/dcerpc.c b/source4/librpc/rpc/dcerpc.c
index b605b4d110..1a8fe7373a 100644
--- a/source4/librpc/rpc/dcerpc.c
+++ b/source4/librpc/rpc/dcerpc.c
@@ -28,18 +28,12 @@ struct dcerpc_pipe *dcerpc_pipe_init(void)
 {
 	struct dcerpc_pipe *p;
 
-	TALLOC_CTX *mem_ctx = talloc_init("dcerpc_tree");
-	if (mem_ctx == NULL)
-		return NULL;
-
-	p = talloc(mem_ctx, sizeof(*p));
+	p = talloc_p(NULL, struct dcerpc_pipe);
 	if (!p) {
-		talloc_destroy(mem_ctx);
 		return NULL;
 	}
 
 	p->reference_count = 0;
-	p->mem_ctx = mem_ctx;
 	p->call_id = 1;
 	p->security_state.auth_info = NULL;
 	p->security_state.generic_state = NULL;
@@ -48,10 +42,23 @@ struct dcerpc_pipe *dcerpc_pipe_init(void)
 	p->srv_max_xmit_frag = 0;
 	p->srv_max_recv_frag = 0;
 	p->last_fault_code = 0;
+	p->pending = NULL;
 
 	return p;
 }
 
+/* 
+   choose the next call id to use
+*/
+static uint32_t next_call_id(struct dcerpc_pipe *p)
+{
+	p->call_id++;
+	if (p->call_id == 0) {
+		p->call_id++;
+	}
+	return p->call_id;
+}
+
 /* close down a dcerpc over SMB pipe */
 void dcerpc_pipe_close(struct dcerpc_pipe *p)
 {
@@ -62,7 +69,7 @@ void dcerpc_pipe_close(struct dcerpc_pipe *p)
 			gensec_end(&p->security_state.generic_state);
 		}
 		p->transport.shutdown_pipe(p);
-		talloc_destroy(p->mem_ctx);
+		talloc_free(p);
 	}
 }
 
@@ -320,6 +327,67 @@ static void init_dcerpc_hdr(struct dcerpc_pipe *p, struct dcerpc_packet *pkt)
 	pkt->drep[3] = 0;
 }
 
+/*
+  hold the state of pending full requests
+*/
+struct full_request_state {
+	DATA_BLOB *reply_blob;
+	NTSTATUS status;
+};
+
+/*
+  receive a reply to a full request
+ */
+static void full_request_recv(struct dcerpc_pipe *p, DATA_BLOB *blob, 
+			      NTSTATUS status)
+{
+	struct full_request_state *state = p->full_request_private;
+
+	if (!NT_STATUS_IS_OK(status)) {
+		state->status = status;
+		return;
+	}
+	state->reply_blob[0] = data_blob_talloc(state, blob->data, blob->length);
+	state->reply_blob = NULL;
+}
+
+/*
+  perform a synchronous request - used for the bind code
+  this cannot be mixed with normal async requests
+*/
+static NTSTATUS full_request(struct dcerpc_pipe *p, 
+			     TALLOC_CTX *mem_ctx,
+			     DATA_BLOB *request_blob,
+			     DATA_BLOB *reply_blob)
+{
+	struct full_request_state *state = talloc_p(mem_ctx, struct full_request_state);
+	NTSTATUS status;
+
+	if (state == NULL) {
+		return NT_STATUS_NO_MEMORY;
+	}
+
+	state->reply_blob = reply_blob;
+	state->status = NT_STATUS_OK;
+
+	p->transport.recv_data = full_request_recv;
+	p->full_request_private = state;
+
+	status = p->transport.send_request(p, request_blob);
+	if (!NT_STATUS_IS_OK(status)) {
+		return status;
+	}
+
+	p->transport.send_read(p);
+
+	while (NT_STATUS_IS_OK(state->status) && state->reply_blob) {
+		struct event_context *ctx = p->transport.event_context(p);
+		event_loop_once(ctx);
+	}
+
+	return state->status;
+}
+
 
 /* 
    perform a bind using the given syntax 
@@ -367,7 +435,7 @@ NTSTATUS dcerpc_bind(struct dcerpc_pipe *p,
 	}
 
 	/* send it on its way */
-	status = p->transport.full_request(p, mem_ctx, &blob, &blob);
+	status = full_request(p, mem_ctx, &blob, &blob);
 	if (!NT_STATUS_IS_OK(status)) {
 		return status;
 	}
@@ -441,7 +509,7 @@ NTSTATUS dcerpc_alter(struct dcerpc_pipe *p,
 	}
 
 	/* send it on its way */
-	status = p->transport.full_request(p, mem_ctx, &blob, &blob);
+	status = full_request(p, mem_ctx, &blob, &blob);
 	if (!NT_STATUS_IS_OK(status)) {
 		return status;
 	}
@@ -483,7 +551,7 @@ NTSTATUS dcerpc_auth3(struct dcerpc_pipe *p,
 
 	pkt.ptype = DCERPC_PKT_AUTH3;
 	pkt.pfc_flags = DCERPC_PFC_FLAG_FIRST | DCERPC_PFC_FLAG_LAST;
-	pkt.call_id = p->call_id++;
+	pkt.call_id = next_call_id(p);
 	pkt.auth_length = 0;
 	pkt.u.auth._pad = 0;
 	pkt.u.auth.auth_info = data_blob(NULL, 0);
@@ -495,7 +563,7 @@ NTSTATUS dcerpc_auth3(struct dcerpc_pipe *p,
 	}
 
 	/* send it on its way */
-	status = p->transport.initial_request(p, mem_ctx, &blob);
+	status = p->transport.send_request(p, &blob);
 	if (!NT_STATUS_IS_OK(status)) {
 		return status;
 	}
@@ -530,33 +598,141 @@ NTSTATUS dcerpc_bind_byuuid(struct dcerpc_pipe *p,
 }
 
 /*
-  perform a full request/response pair on a dcerpc pipe
+  process a fragment received from the transport layer during a
+  request
 */
-NTSTATUS dcerpc_request(struct dcerpc_pipe *p, 
-			uint16_t opnum,
-			TALLOC_CTX *mem_ctx,
-			DATA_BLOB *stub_data_in,
-			DATA_BLOB *stub_data_out)
+static void dcerpc_request_recv_data(struct dcerpc_pipe *p, 
+				     DATA_BLOB *data,
+				     NTSTATUS status)
 {
+	struct dcerpc_packet pkt;
+	struct rpc_request *req;
+	uint_t length;
 	
+	if (!NT_STATUS_IS_OK(status)) {
+		/* all pending requests get the error */
+		while (p->pending) {
+			req = p->pending;
+			req->state = RPC_REQUEST_DONE;
+			req->status = status;
+			DLIST_REMOVE(p->pending, req);
+		}
+		return;
+	}
+
+	pkt.call_id = 0;
+
+	status = dcerpc_pull_request_sign(p, data, (TALLOC_CTX *)data->data, &pkt);
+
+	/* find the matching request. Notice we match before we check
+	   the status.  this is ok as a pending call_id can never be
+	   zero */
+	for (req=p->pending;req;req=req->next) {
+		if (pkt.call_id == req->call_id) break;
+	}
+
+	if (req == NULL) {
+		DEBUG(2,("dcerpc_request: unmatched call_id in response packet\n"));
+		return;
+	}
+
+	if (!NT_STATUS_IS_OK(status)) {
+		req->status = status;
+		req->state = RPC_REQUEST_DONE;
+		DLIST_REMOVE(p->pending, req);
+		return;
+	}
+
+	if (pkt.ptype == DCERPC_PKT_FAULT) {
+		DEBUG(5,("rpc fault 0x%x\n", pkt.u.fault.status));
+		req->fault_code = pkt.u.fault.status;
+		req->status = NT_STATUS_NET_WRITE_FAULT;
+		req->state = RPC_REQUEST_DONE;
+		DLIST_REMOVE(p->pending, req);
+		return;
+	}
+
+	if (pkt.ptype != DCERPC_PKT_RESPONSE) {
+		DEBUG(2,("Unexpected packet type %d in dcerpc response\n",
+			 (int)pkt.ptype)); 
+		req->fault_code = DCERPC_FAULT_OTHER;
+		req->status = NT_STATUS_NET_WRITE_FAULT;
+		req->state = RPC_REQUEST_DONE;
+		DLIST_REMOVE(p->pending, req);
+		return;
+	}
+
+	length = pkt.u.response.stub_and_verifier.length;
+
+	if (length > 0) {
+		req->payload.data = talloc_realloc(req->payload.data, 
+						   req->payload.length + length);
+		if (!req->payload.data) {
+			req->status = NT_STATUS_NO_MEMORY;
+			req->state = RPC_REQUEST_DONE;
+			DLIST_REMOVE(p->pending, req);
+			return;
+		}
+		memcpy(req->payload.data+req->payload.length, 
+		       pkt.u.response.stub_and_verifier.data, length);
+		req->payload.length += length;
+	}
+
+	if (!(pkt.pfc_flags & DCERPC_PFC_FLAG_LAST)) {
+		p->transport.send_read(p);
+		return;
+	}
+
+	/* we've got the full payload */
+	req->state = RPC_REQUEST_DONE;
+	DLIST_REMOVE(p->pending, req);
+
+	if (!(pkt.drep[0] & DCERPC_DREP_LE)) {
+		req->flags |= DCERPC_PULL_BIGENDIAN;
+	} else {
+		req->flags &= ~DCERPC_PULL_BIGENDIAN;
+	}
+}
+
+
+/*
+  perform a full request/response pair on a dcerpc pipe
+*/
+struct rpc_request *dcerpc_request_send(struct dcerpc_pipe *p, 
+					uint16_t opnum,
+					TALLOC_CTX *mem_ctx,
+					DATA_BLOB *stub_data)
+{
+	struct rpc_request *req;
 	struct dcerpc_packet pkt;
-	NTSTATUS status;
-	DATA_BLOB blob, payload;
+	DATA_BLOB blob;
 	uint32_t remaining, chunk_size;
 
-	/* allow the application to tell when a fault has happened */
-	p->last_fault_code = 0;
+	p->transport.recv_data = dcerpc_request_recv_data;
+
+	req = talloc_p(mem_ctx, struct rpc_request);
+	if (req == NULL) {
+		return NULL;
+	}
+
+	req->p = p;
+	req->call_id = next_call_id(p);
+	req->status = NT_STATUS_OK;
+	req->state = RPC_REQUEST_PENDING;
+	req->payload = data_blob(NULL, 0);
+	req->flags = 0;
+	req->fault_code = 0;
 
 	init_dcerpc_hdr(p, &pkt);
 
-	remaining = stub_data_in->length;
+	remaining = stub_data->length;
 
 	/* we can write a full max_recv_frag size, minus the dcerpc
 	   request header size */
 	chunk_size = p->srv_max_recv_frag - (DCERPC_MAX_SIGN_SIZE+DCERPC_REQUEST_LENGTH);
 
 	pkt.ptype = DCERPC_PKT_REQUEST;
-	pkt.call_id = p->call_id++;
+	pkt.call_id = req->call_id;
 	pkt.auth_length = 0;
 	pkt.u.request.alloc_hint = remaining;
 	pkt.u.request.context_id = 0;
@@ -565,24 +741,26 @@ NTSTATUS dcerpc_request(struct dcerpc_pipe *p,
 	/* we send a series of pdus without waiting for a reply until
 	   the last pdu */
 	while (remaining > chunk_size) {
-		if (remaining == stub_data_in->length) {
+		if (remaining == stub_data->length) {
 			pkt.pfc_flags = DCERPC_PFC_FLAG_FIRST;
 		} else {
 			pkt.pfc_flags = 0;
 		}
 
-		pkt.u.request.stub_and_verifier.data = stub_data_in->data + 
-			(stub_data_in->length - remaining);
+		pkt.u.request.stub_and_verifier.data = stub_data->data + 
+			(stub_data->length - remaining);
 		pkt.u.request.stub_and_verifier.length = chunk_size;
 
-		status = dcerpc_push_request_sign(p, &blob, mem_ctx, &pkt);
-		if (!NT_STATUS_IS_OK(status)) {
-			return status;
+		req->status = dcerpc_push_request_sign(p, &blob, mem_ctx, &pkt);
+		if (!NT_STATUS_IS_OK(req->status)) {
+			req->state = RPC_REQUEST_DONE;
+			return req;
 		}
 		
-		status = p->transport.initial_request(p, mem_ctx, &blob);
-		if (!NT_STATUS_IS_OK(status)) {
-			return status;
+		req->status = p->transport.send_request(p, &blob);
+		if (!NT_STATUS_IS_OK(req->status)) {
+			req->state = RPC_REQUEST_DONE;
+			return req;
 		}		
 
 		remaining -= chunk_size;
@@ -590,102 +768,88 @@ NTSTATUS dcerpc_request(struct dcerpc_pipe *p,
 
 	/* now we send a pdu with LAST_FRAG sent and get the first
 	   part of the reply */
-	if (remaining == stub_data_in->length) {
+	if (remaining == stub_data->length) {
 		pkt.pfc_flags = DCERPC_PFC_FLAG_FIRST | DCERPC_PFC_FLAG_LAST;
 	} else {
 		pkt.pfc_flags = DCERPC_PFC_FLAG_LAST;
 	}
-	pkt.u.request.stub_and_verifier.data = stub_data_in->data + 
-		(stub_data_in->length - remaining);
+	pkt.u.request.stub_and_verifier.data = stub_data->data + 
+		(stub_data->length - remaining);
 	pkt.u.request.stub_and_verifier.length = remaining;
 
-	status = dcerpc_push_request_sign(p, &blob, mem_ctx, &pkt);
-	if (!NT_STATUS_IS_OK(status)) {
-		return status;
+	req->status = dcerpc_push_request_sign(p, &blob, mem_ctx, &pkt);
+	if (!NT_STATUS_IS_OK(req->status)) {
+		req->state = RPC_REQUEST_DONE;
+		return req;
 	}
 
-	/* send the pdu and get the initial response pdu */
-	status = p->transport.full_request(p, mem_ctx, &blob, &blob);
-	if (!NT_STATUS_IS_OK(status)) {
-		return status;
-	}
+	/* send the pdu */
+	req->status = p->transport.send_request(p, &blob);
 
-	status = dcerpc_pull_request_sign(p, &blob, mem_ctx, &pkt);
-	if (!NT_STATUS_IS_OK(status)) {
-		return status;
+	if (!NT_STATUS_IS_OK(req->status)) {
+		req->state = RPC_REQUEST_DONE;
 	}
 
-	if (pkt.ptype == DCERPC_PKT_FAULT) {
-		DEBUG(5,("rpc fault 0x%x\n", pkt.u.fault.status));
-		p->last_fault_code = pkt.u.fault.status;
-		return NT_STATUS_NET_WRITE_FAULT;
-	}
+	DLIST_ADD(p->pending, req);
 
-	if (pkt.ptype != DCERPC_PKT_RESPONSE) {
-		return NT_STATUS_UNSUCCESSFUL;
-	}
+	p->transport.send_read(p);
 
-	if (!(pkt.pfc_flags & DCERPC_PFC_FLAG_FIRST)) {
-		/* something is badly wrong! */
-		return NT_STATUS_UNSUCCESSFUL;
-	}
-
-	payload = pkt.u.response.stub_and_verifier;
-
-	/* continue receiving fragments */
-	while (!(pkt.pfc_flags & DCERPC_PFC_FLAG_LAST)) {
-		uint32_t length;
-
-		status = p->transport.secondary_request(p, mem_ctx, &blob);
-		if (!NT_STATUS_IS_OK(status)) {
-			return status;
-		}
-
-		status = dcerpc_pull_request_sign(p, &blob, mem_ctx, &pkt);
-		if (!NT_STATUS_IS_OK(status)) {
-			return status;
-		}
-
-		if (pkt.pfc_flags & DCERPC_PFC_FLAG_FIRST) {
-			/* start of another packet!? */
-			return NT_STATUS_UNSUCCESSFUL;
-		}
-
-		if (pkt.ptype == DCERPC_PKT_FAULT) {
-			p->last_fault_code = pkt.u.fault.status;
-			return NT_STATUS_NET_WRITE_FAULT;
-		}
+	return req;
+}
 
-		if (pkt.ptype != DCERPC_PKT_RESPONSE) {
-			return NT_STATUS_UNSUCCESSFUL;
-		}
+/*
+  return the event context for a dcerpc pipe
+  used by callers who wish to operate asynchronously
+*/
+struct event_context *dcerpc_event_context(struct dcerpc_pipe *p)
+{
+	return p->transport.event_context(p);
+}
 
-		length = pkt.u.response.stub_and_verifier.length;
 
-		payload.data = talloc_realloc(payload.data, 
-					      payload.length + length);
-		if (!payload.data) {
-			return NT_STATUS_NO_MEMORY;
-		}
 
-		memcpy(payload.data + payload.length,
-		       pkt.u.response.stub_and_verifier.data,
-		       length);
+/*
+  perform a full request/response pair on a dcerpc pipe
+*/
+NTSTATUS dcerpc_request_recv(struct rpc_request *req,
+			     TALLOC_CTX *mem_ctx,
+			     DATA_BLOB *stub_data)
+{
+	NTSTATUS status;
 
-		payload.length += length;
+	while (req->state == RPC_REQUEST_PENDING) {
+		struct event_context *ctx = dcerpc_event_context(req->p);
+		event_loop_once(ctx);
 	}
-
-	if (stub_data_out) {
-		*stub_data_out = payload;
+	*stub_data = req->payload;
+	status = req->status;
+	if (stub_data->data) {
+		stub_data->data = talloc_steal(mem_ctx, stub_data->data);
+	}
+	if (NT_STATUS_EQUAL(status, NT_STATUS_NET_WRITE_FAULT)) {
+		req->p->last_fault_code = req->fault_code;
 	}
+	talloc_free(req);
+	return status;
+}
 
-	if (!(pkt.drep[0] & DCERPC_DREP_LE)) {
-		p->flags |= DCERPC_PULL_BIGENDIAN;
-	} else {
-		p->flags &= ~DCERPC_PULL_BIGENDIAN;
+/*
+  perform a full request/response pair on a dcerpc pipe
+*/
+NTSTATUS dcerpc_request(struct dcerpc_pipe *p, 
+			uint16_t opnum,
+			TALLOC_CTX *mem_ctx,
+			DATA_BLOB *stub_data_in,
+			DATA_BLOB *stub_data_out)
+{
+	struct rpc_request *req;
+
+	req = dcerpc_request_send(p, opnum, mem_ctx, stub_data_in);
+	if (req == NULL) {
+		return NT_STATUS_NO_MEMORY;
 	}
 
-	return status;
+	return dcerpc_request_recv(req, mem_ctx, stub_data_out);
 }
 
 
@@ -830,30 +994,29 @@ static NTSTATUS dcerpc_ndr_validate_out(TALLOC_CTX *mem_ctx,
 	return NT_STATUS_OK;
 }
 
+
 /*
-  a useful helper function for synchronous rpc requests 
+  send a rpc request with a given set of ndr helper functions
 
-  this can be used when you have ndr push/pull functions in the
-  standard format
+  call dcerpc_ndr_request_recv() to receive the answer
 */
-NTSTATUS dcerpc_ndr_request(struct dcerpc_pipe *p,
-			    uint32_t opnum,
-			    TALLOC_CTX *mem_ctx,
-			    NTSTATUS (*ndr_push)(struct ndr_push *, int, void *),
-			    NTSTATUS (*ndr_pull)(struct ndr_pull *, int, void *),
-			    void *struct_ptr,
-			    size_t struct_size)
+struct rpc_request *dcerpc_ndr_request_send(struct dcerpc_pipe *p,
+					    uint32_t opnum,
+					    TALLOC_CTX *mem_ctx,
+					    NTSTATUS (*ndr_push)(struct ndr_push *, int, void *),
+					    NTSTATUS (*ndr_pull)(struct ndr_pull *, int, void *),
+					    void *struct_ptr,
+					    size_t struct_size)
 {
 	struct ndr_push *push;
-	struct ndr_pull *pull;
 	NTSTATUS status;
-	DATA_BLOB request, response;
+	DATA_BLOB request;
+	struct rpc_request *req;
 
 	/* setup for a ndr_push_* call */
 	push = ndr_push_init();
 	if (!push) {
-		talloc_destroy(mem_ctx);
-		return NT_STATUS_NO_MEMORY;
+		return NULL;
 	}
 
 	if (p->flags & DCERPC_PUSH_BIGENDIAN) {
@@ -863,7 +1026,10 @@ NTSTATUS dcerpc_ndr_request(struct dcerpc_pipe *p,
 	/* push the structure into a blob */
 	status = ndr_push(push, NDR_IN, struct_ptr);
 	if (!NT_STATUS_IS_OK(status)) {
-		goto failed;
+		DEBUG(2,("Unable to ndr_push structure in dcerpc_ndr_request_send - %s\n",
+			 nt_errstr(status)));
+		ndr_push_free(push);
+		return NULL;
 	}
 
 	/* retrieve the blob */
@@ -873,7 +1039,10 @@ NTSTATUS dcerpc_ndr_request(struct dcerpc_pipe *p,
 		status = dcerpc_ndr_validate_in(mem_ctx, request, struct_size, 
 						ndr_push, ndr_pull);
 		if (!NT_STATUS_IS_OK(status)) {
-			goto failed;
+			DEBUG(2,("Validation failed in dcerpc_ndr_request_send - %s\n",
+				 nt_errstr(status)));
+			ndr_push_free(push);
+			return NULL;
 		}
 	}
 
@@ -881,18 +1050,45 @@ NTSTATUS dcerpc_ndr_request(struct dcerpc_pipe *p,
 	dump_data(10, request.data, request.length);
 
 	/* make the actual dcerpc request */
-	status = dcerpc_request(p, opnum, mem_ctx, &request, &response);
+	req = dcerpc_request_send(p, opnum, mem_ctx, &request);
+
+	if (req != NULL) {
+		req->ndr.ndr_push = ndr_push;
+		req->ndr.ndr_pull = ndr_pull;
+		req->ndr.struct_ptr = struct_ptr;
+		req->ndr.struct_size = struct_size;
+		req->ndr.mem_ctx = mem_ctx;
+	}
+
+	ndr_push_free(push);
+	
+	return req;
+}
+
+/*
+  receive the answer from a dcerpc_ndr_request_send()
+*/
+NTSTATUS dcerpc_ndr_request_recv(struct rpc_request *req)
+{
+	struct dcerpc_pipe *p = req->p;
+	NTSTATUS status;
+	DATA_BLOB response;
+	struct ndr_pull *pull;
+	struct rpc_request_ndr ndr = req->ndr;
+	uint_t flags = req->flags;
+
+	status = dcerpc_request_recv(req, ndr.mem_ctx, &response);
 	if (!NT_STATUS_IS_OK(status)) {
-		goto failed;
+		return status;
 	}
 
 	/* prepare for ndr_pull_* */
-	pull = ndr_pull_init_blob(&response, mem_ctx);
+	pull = ndr_pull_init_blob(&response, ndr.mem_ctx);
 	if (!pull) {
-		goto failed;
+		return NT_STATUS_NO_MEMORY;
 	}
 
-	if (p->flags & DCERPC_PULL_BIGENDIAN) {
+	if (flags & DCERPC_PULL_BIGENDIAN) {
 		pull->flags |= LIBNDR_FLAG_BIGENDIAN;
 	}
 
@@ -900,19 +1096,16 @@ NTSTATUS dcerpc_ndr_request(struct dcerpc_pipe *p,
 	dump_data(10, pull->data, pull->data_size);
 
 	/* pull the structure from the blob */
-	status = ndr_pull(pull, NDR_OUT, struct_ptr);
+	status = ndr.ndr_pull(pull, NDR_OUT, ndr.struct_ptr);
 	if (!NT_STATUS_IS_OK(status)) {
-		goto failed;
+		return status;
 	}
 
-	/* possibly check the packet signature */
-	
-
 	if (p->flags & DCERPC_DEBUG_VALIDATE_OUT) {
-		status = dcerpc_ndr_validate_out(mem_ctx, struct_ptr, struct_size, 
-						 ndr_push, ndr_pull);
+		status = dcerpc_ndr_validate_out(ndr.mem_ctx, ndr.struct_ptr, ndr.struct_size, 
+						 ndr.ndr_push, ndr.ndr_pull);
 		if (!NT_STATUS_IS_OK(status)) {
-			goto failed;
+			return status;
 		}
 	}
 
@@ -926,9 +1119,32 @@ NTSTATUS dcerpc_ndr_request(struct dcerpc_pipe *p,
 		   those versions then we need to ignore this error */
 	}
 
-failed:
-	ndr_push_free(push);
-	return status;
+	return NT_STATUS_OK;
+}
+
+
+/*
+  a useful helper function for synchronous rpc requests 
+
+  this can be used when you have ndr push/pull functions in the
+  standard format
+*/
+NTSTATUS dcerpc_ndr_request(struct dcerpc_pipe *p,
+			    uint32_t opnum,
+			    TALLOC_CTX *mem_ctx,
+			    NTSTATUS (*ndr_push)(struct ndr_push *, int, void *),
+			    NTSTATUS (*ndr_pull)(struct ndr_pull *, int, void *),
+			    void *struct_ptr,
+			    size_t struct_size)
+{
+	struct rpc_request *req;
+
+	req = dcerpc_ndr_request_send(p, opnum, mem_ctx, ndr_push, ndr_pull, struct_ptr, struct_size);
+	if (req == NULL) {
+		return NT_STATUS_NO_MEMORY;
+	}
+
+	return dcerpc_ndr_request_recv(req);
 }
 
 
diff --git a/source4/librpc/rpc/dcerpc.h b/source4/librpc/rpc/dcerpc.h
index 2c36241020..2dce5df92a 100644
--- a/source4/librpc/rpc/dcerpc.h
+++ b/source4/librpc/rpc/dcerpc.h
@@ -32,7 +32,6 @@ struct dcerpc_security {
 };
 
 struct dcerpc_pipe {
-	TALLOC_CTX *mem_ctx;
 	int reference_count;
 	uint32_t call_id;
 	uint32_t srv_max_xmit_frag;
@@ -47,16 +46,33 @@ struct dcerpc_pipe {
 	struct dcerpc_transport {
 		enum dcerpc_transport_t transport;
 		void *private;
-		NTSTATUS (*full_request)(struct dcerpc_pipe *, 
-					 TALLOC_CTX *, DATA_BLOB *, DATA_BLOB *);
-		NTSTATUS (*secondary_request)(struct dcerpc_pipe *, TALLOC_CTX *, DATA_BLOB *);
-		NTSTATUS (*initial_request)(struct dcerpc_pipe *, TALLOC_CTX *, DATA_BLOB *);
+
 		NTSTATUS (*shutdown_pipe)(struct dcerpc_pipe *);
+
 		const char *(*peer_name)(struct dcerpc_pipe *);
+
+		/* send a request to the server */
+		NTSTATUS (*send_request)(struct dcerpc_pipe *, DATA_BLOB *);
+
+		/* send a read request to the server */
+		NTSTATUS (*send_read)(struct dcerpc_pipe *);
+
+		/* get an event context for the connection */
+		struct event_context *(*event_context)(struct dcerpc_pipe *);
+
+		/* a callback to the dcerpc code when a full fragment
+		   has been received */
+		void (*recv_data)(struct dcerpc_pipe *, DATA_BLOB *, NTSTATUS status);
 	} transport;
 
 	/* the last fault code from a DCERPC fault */
 	uint32_t last_fault_code;
+
+	/* pending requests */
+	struct rpc_request *pending;
+
+	/* private pointer for pending full requests */
+	void *full_request_private;
 };
 
 /* dcerpc pipe flags */
@@ -119,3 +135,32 @@ struct dcerpc_binding {
 	const char **options;
 	uint32_t flags;
 };
+
+
+enum rpc_request_state {
+	RPC_REQUEST_PENDING,
+	RPC_REQUEST_DONE
+};
+
+/*
+  handle for an async dcerpc request
+*/
+struct rpc_request {
+	struct rpc_request *next, *prev;
+	struct dcerpc_pipe *p;
+	NTSTATUS status;
+	uint32_t call_id;
+	enum rpc_request_state state;
+	DATA_BLOB payload;
+	uint_t flags;
+	uint32_t fault_code;
+
+	/* use by the ndr level async recv call */
+	struct rpc_request_ndr {
+		NTSTATUS (*ndr_push)(struct ndr_push *, int, void *);
+		NTSTATUS (*ndr_pull)(struct ndr_pull *, int, void *);
+		void *struct_ptr;
+		size_t struct_size;
+		TALLOC_CTX *mem_ctx;
+	} ndr;
+};
diff --git a/source4/librpc/rpc/dcerpc_auth.c b/source4/librpc/rpc/dcerpc_auth.c
index 6ae6a80596..9587fb9390 100644
--- a/source4/librpc/rpc/dcerpc_auth.c
+++ b/source4/librpc/rpc/dcerpc_auth.c
@@ -70,7 +70,7 @@ NTSTATUS dcerpc_bind_auth3(struct dcerpc_pipe *p, uint8_t auth_type, uint8_t aut
 		}
 	}
 
-	p->security_state.auth_info = talloc(p->mem_ctx, sizeof(*p->security_state.auth_info));
+	p->security_state.auth_info = talloc(p, sizeof(*p->security_state.auth_info));
 	if (!p->security_state.auth_info) {
 		status = NT_STATUS_NO_MEMORY;
 		goto done;
@@ -146,7 +146,7 @@ NTSTATUS dcerpc_bind_alter(struct dcerpc_pipe *p, uint8_t auth_type, uint8_t aut
 		}
 	}
 
-	p->security_state.auth_info = talloc(p->mem_ctx, sizeof(*p->security_state.auth_info));
+	p->security_state.auth_info = talloc(p, sizeof(*p->security_state.auth_info));
 	if (!p->security_state.auth_info) {
 		status = NT_STATUS_NO_MEMORY;
 		goto done;
diff --git a/source4/librpc/rpc/dcerpc_error.c b/source4/librpc/rpc/dcerpc_error.c
index d1456cad72..c9434060a1 100644
--- a/source4/librpc/rpc/dcerpc_error.c
+++ b/source4/librpc/rpc/dcerpc_error.c
@@ -40,10 +40,8 @@ static const struct dcerpc_fault_table dcerpc_faults[] =
 	{ NULL,					0}	
 };
 
-const char *dcerpc_errstr(uint32_t fault_code)
+const char *dcerpc_errstr(TALLOC_CTX *mem_ctx, uint32_t fault_code)
 {
-	/* TODO: remove static pstring! */
-	static pstring msg;
 	int idx = 0;
 
 	while (dcerpc_faults[idx].errstr != NULL) {
@@ -53,7 +51,5 @@ const char *dcerpc_errstr(uint32_t fault_code)
 		idx++;
 	}
 
-	slprintf(msg, sizeof(msg), "DCERPC fault 0x%08x", fault_code);
-
-	return msg;
+	return talloc_asprintf(mem_ctx, "DCERPC fault 0x%08x", fault_code);
 }
diff --git a/source4/librpc/rpc/dcerpc_schannel.c b/source4/librpc/rpc/dcerpc_schannel.c
index e7b1d08988..efe609f9a3 100644
--- a/source4/librpc/rpc/dcerpc_schannel.c
+++ b/source4/librpc/rpc/dcerpc_schannel.c
@@ -369,14 +369,14 @@ static NTSTATUS dcerpc_schannel_key(struct dcerpc_pipe *p,
 	/*
 	  step 2 - request a netlogon challenge
 	*/
-	r.in.server_name = talloc_asprintf(p->mem_ctx, "\\\\%s", dcerpc_server_name(p));
+	r.in.server_name = talloc_asprintf(p, "\\\\%s", dcerpc_server_name(p));
 	r.in.computer_name = workstation;
 	r.in.credentials = &credentials1;
 	r.out.credentials = &credentials2;
 
 	generate_random_buffer(credentials1.data, sizeof(credentials1.data));
 
-	status = dcerpc_netr_ServerReqChallenge(p2, p->mem_ctx, &r);
+	status = dcerpc_netr_ServerReqChallenge(p2, p, &r);
 	if (!NT_STATUS_IS_OK(status)) {
 		return status;
 	}
@@ -389,7 +389,7 @@ static NTSTATUS dcerpc_schannel_key(struct dcerpc_pipe *p,
 			  negotiate_flags);
 
 	a.in.server_name = r.in.server_name;
-	a.in.account_name = talloc_asprintf(p->mem_ctx, "%s$", workstation);
+	a.in.account_name = talloc_asprintf(p, "%s$", workstation);
 	a.in.secure_channel_type = chan_type;
 	a.in.computer_name = workstation;
 	a.in.negotiate_flags = &negotiate_flags;
@@ -397,7 +397,7 @@ static NTSTATUS dcerpc_schannel_key(struct dcerpc_pipe *p,
 	a.in.credentials = &credentials3;
 	a.out.credentials = &credentials3;
 
-	status = dcerpc_netr_ServerAuthenticate2(p2, p->mem_ctx, &a);
+	status = dcerpc_netr_ServerAuthenticate2(p2, p, &a);
 	if (!NT_STATUS_IS_OK(status)) {
 		return status;
 	}
diff --git a/source4/librpc/rpc/dcerpc_smb.c b/source4/librpc/rpc/dcerpc_smb.c
index d3a0a42d10..fffc945403 100644
--- a/source4/librpc/rpc/dcerpc_smb.c
+++ b/source4/librpc/rpc/dcerpc_smb.c
@@ -29,221 +29,156 @@ struct smb_private {
 	struct smbcli_tree *tree;
 };
 
-static struct smbcli_request *dcerpc_raw_send(struct dcerpc_pipe *p, DATA_BLOB *blob)
-{
-	struct smb_private *smb = p->transport.private;
-	struct smb_trans2 trans;
-	uint16_t setup[2];
-	struct smbcli_request *req;
-	TALLOC_CTX *mem_ctx;
-
-	mem_ctx = talloc_init("dcerpc_raw_send");
-	if (!mem_ctx) return NULL;
-
-	trans.in.data = *blob;
-	trans.in.params = data_blob(NULL, 0);
-	
-	setup[0] = TRANSACT_DCERPCCMD;
-	setup[1] = smb->fnum;
-
-	trans.in.max_param = 0;
-	trans.in.max_data = 0x8000;
-	trans.in.max_setup = 0;
-	trans.in.setup_count = 2;
-	trans.in.flags = 0;
-	trans.in.timeout = 0;
-	trans.in.setup = setup;
-	trans.in.trans_name = "\\PIPE\\";
 
-	req = smb_raw_trans_send(smb->tree, &trans);
-
-	talloc_destroy(mem_ctx);
-
-	return req;
+/*
+  tell the dcerpc layer that the transport is dead
+*/
+static void pipe_dead(struct dcerpc_pipe *p, NTSTATUS status)
+{
+	p->transport.recv_data(p, NULL, status);
 }
 
 
-static NTSTATUS dcerpc_raw_recv(struct dcerpc_pipe *p, 
-				struct smbcli_request *req,
-				TALLOC_CTX *mem_ctx,
-				DATA_BLOB *blob)
+/* 
+   this holds the state of an in-flight call
+*/
+struct smb_read_state {
+	struct dcerpc_pipe *p;
+	struct smbcli_request *req;
+	size_t received;
+	DATA_BLOB data;
+	union smb_read *io;
+};
+
+/*
+  called when a read request has completed
+*/
+static void smb_read_callback(struct smbcli_request *req)
 {
-	struct smb_private *smb = p->transport.private;
-	struct smb_trans2 trans;
-	NTSTATUS status;
+	struct smb_private *smb;
+	struct smb_read_state *state;
+	union smb_read *io;
 	uint16_t frag_length;
-	DATA_BLOB payload;
+	NTSTATUS status;
 
-	status = smb_raw_trans_recv(req, mem_ctx, &trans);
+	state = req->async.private;
+	smb = state->p->transport.private;
+	io = state->io;
 
-	/* STATUS_BUFFER_OVERFLOW means that there is more data
-	   available via SMBreadX */
-	if (!NT_STATUS_IS_OK(status) && 
-	    !NT_STATUS_EQUAL(status, STATUS_BUFFER_OVERFLOW)) {
-		return status;
+	if (!NT_STATUS_IS_OK(req->status)) {
+		pipe_dead(state->p, req->status);
+		talloc_free(state);
+		return;
 	}
 
-	payload = trans.out.data;
-
-	if (trans.out.data.length < 16 || 
-	    !NT_STATUS_EQUAL(status, STATUS_BUFFER_OVERFLOW)) {
-		goto done;
+	status = smb_raw_read_recv(state->req, io);
+	if (!NT_STATUS_IS_OK(status)) {
+		pipe_dead(state->p, status);
+		talloc_free(state);
+		return;
 	}
 
-	/* we might have recieved a partial fragment, in which case we
-	   need to pull the rest of it */
-	frag_length = dcerpc_get_frag_length(&payload);
-	if (frag_length <= payload.length) {
-		goto done;
-	}
+	state->received += io->readx.out.nread;
 
-	/* make sure the payload can hold the whole fragment */
-	payload.data = talloc_realloc(payload.data, frag_length);
-	if (!payload.data) {
-		return NT_STATUS_NO_MEMORY;
+	if (state->received < 16) {
+		DEBUG(0,("dcerpc_smb: short packet (length %d) in read callback!\n",
+			 state->received));
+		pipe_dead(state->p, NT_STATUS_INFO_LENGTH_MISMATCH);
+		talloc_free(state);
+		return;
 	}
 
-	/* the rest of the data is available via SMBreadX */
-	while (frag_length > payload.length) {
-		uint32_t n;
-		union smb_read io;
-
-		n = frag_length - payload.length;
-		if (n > 0xFF00) {
-			n = 0xFF00;
-		}
-
-		io.generic.level = RAW_READ_READX;
-		io.readx.in.fnum = smb->fnum;
-		io.readx.in.mincnt = n;
-		io.readx.in.maxcnt = n;
-		io.readx.in.offset = 0;
-		io.readx.in.remaining = 0;
-		io.readx.out.data = payload.data + payload.length;
-		status = smb_raw_read(smb->tree, &io);
-		if (!NT_STATUS_IS_OK(status) &&
-		    !NT_STATUS_EQUAL(status, STATUS_BUFFER_OVERFLOW)) {
-			break;
-		}
-		
-		n = io.readx.out.nread;
-		if (n == 0) {
-			status = NT_STATUS_UNSUCCESSFUL;
-			break;
-		}
-		
-		payload.length += n;
-
-		/* if the SMBreadX returns NT_STATUS_OK then there
-		   isn't any more data to be read */
-		if (NT_STATUS_IS_OK(status)) {
-			break;
-		}
+	frag_length = dcerpc_get_frag_length(&state->data);
+	if (frag_length <= state->received) {
+		state->data.length = state->received;
+		state->p->transport.recv_data(state->p, &state->data, NT_STATUS_OK);
+		talloc_free(state);
+		return;
 	}
 
-done:
-	if (blob) {
-		*blob = payload;
-	}
+	/* initiate another read request, as we only got part of a fragment */
+	state->data.data = talloc_realloc(state->data.data, frag_length);
 
-	return status;
-}
+	io->readx.in.mincnt = frag_length - state->received;
+	io->readx.in.maxcnt = io->readx.in.mincnt;
+	io->readx.out.data = state->data.data + state->received;
 
-static NTSTATUS smb_full_request(struct dcerpc_pipe *p, 
-				 TALLOC_CTX *mem_ctx,
-				 DATA_BLOB *request_blob,
-				 DATA_BLOB *reply_blob)
-{
-	struct smbcli_request *req;
-	req = dcerpc_raw_send(p, request_blob);
-	return dcerpc_raw_recv(p, req, mem_ctx, reply_blob);
+	req = smb_raw_read_send(smb->tree, io);
+	if (req == NULL) {
+		pipe_dead(state->p, NT_STATUS_NO_MEMORY);
+		talloc_free(state);
+		return;
+	}
+
+	req->async.fn = smb_read_callback;
+	req->async.private = state;
 }
-	      
 
-/* 
-   retrieve a secondary pdu from a pipe 
+/*
+  trigger a read request from the server
 */
-static NTSTATUS smb_secondary_request(struct dcerpc_pipe *p, 
-			       TALLOC_CTX *mem_ctx,
-			       DATA_BLOB *blob)
+static NTSTATUS send_read_request(struct dcerpc_pipe *p)
 {
 	struct smb_private *smb = p->transport.private;
-	union smb_read io;
-	uint32_t n = 0x2000;
-	uint32_t frag_length;
-	NTSTATUS status;
+	union smb_read *io;
+	struct smb_read_state *state;
+	struct smbcli_request *req;
 
-	*blob = data_blob_talloc(mem_ctx, NULL, n);
-	if (!blob->data) {
+	state = talloc_p(smb, struct smb_read_state);
+	if (state == NULL) {
 		return NT_STATUS_NO_MEMORY;
 	}
 
-	io.generic.level = RAW_READ_READX;
-	io.readx.in.fnum = smb->fnum;
-	io.readx.in.mincnt = n;
-	io.readx.in.maxcnt = n;
-	io.readx.in.offset = 0;
-	io.readx.in.remaining = 0;
-	io.readx.out.data = blob->data;
-
-	status = smb_raw_read(smb->tree, &io);
-	if (!NT_STATUS_IS_OK(status) &&
-	    !NT_STATUS_EQUAL(status, STATUS_BUFFER_OVERFLOW)) {
-		return status;
+	state->p = p;
+	state->received = 0;
+	state->data = data_blob_talloc(state, NULL, 0x2000);
+	state->io = talloc_p(state, union smb_read);
+
+	io = state->io;
+	io->generic.level = RAW_READ_READX;
+	io->readx.in.fnum = smb->fnum;
+	io->readx.in.mincnt = state->data.length;
+	io->readx.in.maxcnt = state->data.length;
+	io->readx.in.offset = 0;
+	io->readx.in.remaining = 0;
+	io->readx.out.data = state->data.data;
+	req = smb_raw_read_send(smb->tree, io);
+	if (req == NULL) {
+		return NT_STATUS_NO_MEMORY;
 	}
 
-	blob->length = io.readx.out.nread;
+	req->async.fn = smb_read_callback;
+	req->async.private = state;
 
-	if (blob->length < 16) {
-		return status;
-	}
+	state->req = req;
 
-	frag_length = dcerpc_get_frag_length(blob);
-	if (frag_length <= blob->length) {
-		return status;
-	}
+	return NT_STATUS_OK;
+}
 
-	blob->data = talloc_realloc(blob->data, frag_length);
-	if (!blob->data) {
-		return NT_STATUS_NO_MEMORY;
-	}
 
-	while (frag_length > blob->length &&
-	       NT_STATUS_EQUAL(status, STATUS_BUFFER_OVERFLOW)) {
-
-		n = frag_length - blob->length;
-		if (n > 0xFF00) {
-			n = 0xFF00;
-		}
-
-		io.readx.in.mincnt = n;
-		io.readx.in.maxcnt = n;
-		io.readx.out.data = blob->data + blob->length;
-		status = smb_raw_read(smb->tree, &io);
-
-		if (!NT_STATUS_IS_OK(status) &&
-		    !NT_STATUS_EQUAL(status, STATUS_BUFFER_OVERFLOW)) {
-			return status;
-		}
-		
-		n = io.readx.out.nread;
-		blob->length += n;
+/*
+  called when a write request has completed
+*/
+static void smb_write_callback(struct smbcli_request *req)
+{
+	struct dcerpc_pipe *p = req->async.private;
+
+	if (!NT_STATUS_IS_OK(req->status)) {
+		DEBUG(0,("dcerpc_smb: write callback error\n"));
+		pipe_dead(p, req->status);
 	}
 
-	return status;
+	smbcli_request_destroy(req);
 }
 
-
 /* 
-   send an initial pdu in a multi-pdu sequence
+   send a packet to the server
 */
-static NTSTATUS smb_initial_request(struct dcerpc_pipe *p, 
-				    TALLOC_CTX *mem_ctx,
-				    DATA_BLOB *blob)
+static NTSTATUS smb_send_request(struct dcerpc_pipe *p, DATA_BLOB *blob)
 {
 	struct smb_private *smb = p->transport.private;
 	union smb_write io;
-	NTSTATUS status;
+	struct smbcli_request *req;
 
 	io.generic.level = RAW_WRITE_WRITEX;
 	io.writex.in.fnum = smb->fnum;
@@ -253,17 +188,26 @@ static NTSTATUS smb_initial_request(struct dcerpc_pipe *p,
 	io.writex.in.count = blob->length;
 	io.writex.in.data = blob->data;
 
-	status = smb_raw_write(smb->tree, &io);
-	if (NT_STATUS_IS_OK(status)) {
-		return status;
+	req = smb_raw_write_send(smb->tree, &io);
+	if (req == NULL) {
+		return NT_STATUS_NO_MEMORY;
 	}
 
-	/* make sure it accepted it all */
-	if (io.writex.out.nwritten != blob->length) {
-		return NT_STATUS_UNSUCCESSFUL;
-	}
+	req->async.fn = smb_write_callback;
+	req->async.private = p;
+
+	return NT_STATUS_OK;
+}
+
+/* 
+   return the event context for the pipe, so the caller can wait
+   for events asynchronously
+*/
+static struct event_context *smb_event_context(struct dcerpc_pipe *p)
+{
+	struct smb_private *smb = p->transport.private;
 
-	return status;
+	return smb->tree->session->transport->event.ctx;
 }
 
 
@@ -356,13 +300,15 @@ NTSTATUS dcerpc_pipe_open_smb(struct dcerpc_pipe **p,
 	*/
 	(*p)->transport.transport = NCACN_NP;
 	(*p)->transport.private = NULL;
-	(*p)->transport.full_request = smb_full_request;
-	(*p)->transport.secondary_request = smb_secondary_request;
-	(*p)->transport.initial_request = smb_initial_request;
 	(*p)->transport.shutdown_pipe = smb_shutdown_pipe;
 	(*p)->transport.peer_name = smb_peer_name;
+
+	(*p)->transport.send_request = smb_send_request;
+	(*p)->transport.send_read = send_read_request;
+	(*p)->transport.event_context = smb_event_context;
+	(*p)->transport.recv_data = NULL;
 	
-	smb = talloc((*p)->mem_ctx, sizeof(*smb));
+	smb = talloc((*p), sizeof(*smb));
 	if (!smb) {
 		dcerpc_pipe_close(*p);
 		return NT_STATUS_NO_MEMORY;
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;
 
diff --git a/source4/librpc/rpc/dcerpc_util.c b/source4/librpc/rpc/dcerpc_util.c
index 21325659f7..5a3b875c01 100644
--- a/source4/librpc/rpc/dcerpc_util.c
+++ b/source4/librpc/rpc/dcerpc_util.c
@@ -76,7 +76,7 @@ NTSTATUS dcerpc_epm_map_tcp_port(const char *server,
 
 	/* we can use the pipes memory context here as we will have a short
 	   lived connection */
-	status = dcerpc_bind_byuuid(p, p->mem_ctx, 
+	status = dcerpc_bind_byuuid(p, p, 
 				    DCERPC_EPMAPPER_UUID,
 				    DCERPC_EPMAPPER_VERSION);
 	if (!NT_STATUS_IS_OK(status)) {
@@ -88,7 +88,7 @@ NTSTATUS dcerpc_epm_map_tcp_port(const char *server,
 	ZERO_STRUCT(guid);
 
 	twr.towers.num_floors = 5;
-	twr.towers.floors = talloc(p->mem_ctx, sizeof(twr.towers.floors[0]) * 5);
+	twr.towers.floors = talloc(p, sizeof(twr.towers.floors[0]) * 5);
 
 	/* what I'd like for christmas ... */
 
@@ -96,28 +96,28 @@ NTSTATUS dcerpc_epm_map_tcp_port(const char *server,
 	twr.towers.floors[0].lhs.protocol = EPM_PROTOCOL_UUID;
 	GUID_from_string(uuid, &twr.towers.floors[0].lhs.info.uuid.uuid);
 	twr.towers.floors[0].lhs.info.uuid.version = version;
-	twr.towers.floors[0].rhs.rhs_data = data_blob_talloc_zero(p->mem_ctx, 2);
+	twr.towers.floors[0].rhs.rhs_data = data_blob_talloc_zero(p, 2);
 
 	/* encoded with NDR ... */
 	twr.towers.floors[1].lhs.protocol = EPM_PROTOCOL_UUID;
 	GUID_from_string(NDR_GUID, &twr.towers.floors[1].lhs.info.uuid.uuid);
 	twr.towers.floors[1].lhs.info.uuid.version = NDR_GUID_VERSION;
-	twr.towers.floors[1].rhs.rhs_data = data_blob_talloc_zero(p->mem_ctx, 2);
+	twr.towers.floors[1].rhs.rhs_data = data_blob_talloc_zero(p, 2);
 
 	/* on an RPC connection ... */
 	twr.towers.floors[2].lhs.protocol = EPM_PROTOCOL_NCACN_RPC_C;
 	twr.towers.floors[2].lhs.info.lhs_data = data_blob(NULL, 0);
-	twr.towers.floors[2].rhs.rhs_data = data_blob_talloc_zero(p->mem_ctx, 2);
+	twr.towers.floors[2].rhs.rhs_data = data_blob_talloc_zero(p, 2);
 
 	/* on a TCP port ... */
 	twr.towers.floors[3].lhs.protocol = EPM_PROTOCOL_NCACN_TCP;
 	twr.towers.floors[3].lhs.info.lhs_data = data_blob(NULL, 0);
-	twr.towers.floors[3].rhs.rhs_data = data_blob_talloc_zero(p->mem_ctx, 2);
+	twr.towers.floors[3].rhs.rhs_data = data_blob_talloc_zero(p, 2);
 
 	/* on an IP link ... */
 	twr.towers.floors[4].lhs.protocol = EPM_PROTOCOL_NCACN_IP;
 	twr.towers.floors[4].lhs.info.lhs_data = data_blob(NULL, 0);
-	twr.towers.floors[4].rhs.rhs_data = data_blob_talloc_zero(p->mem_ctx, 4);
+	twr.towers.floors[4].rhs.rhs_data = data_blob_talloc_zero(p, 4);
 
 	/* with some nice pretty paper around it of course */
 	r.in.object = &guid;
@@ -126,7 +126,7 @@ NTSTATUS dcerpc_epm_map_tcp_port(const char *server,
 	r.in.max_towers = 1;
 	r.out.entry_handle = &handle;
 
-	status = dcerpc_epm_Map(p, p->mem_ctx, &r);
+	status = dcerpc_epm_Map(p, p, &r);
 	if (!NT_STATUS_IS_OK(status)) {
 		dcerpc_pipe_close(p);
 		return status;
@@ -490,7 +490,7 @@ static NTSTATUS dcerpc_pipe_connect_ncacn_np(struct dcerpc_pipe **p,
 	(*p)->flags = binding->flags;
 
 	/* remember the binding string for possible secondary connections */
-	(*p)->binding_string = dcerpc_binding_string((*p)->mem_ctx, binding);
+	(*p)->binding_string = dcerpc_binding_string((*p), binding);
 
 	if (username && username[0] && (binding->flags & DCERPC_SCHANNEL_ANY)) {
 		status = dcerpc_bind_auth_schannel(*p, pipe_uuid, pipe_version, 
@@ -556,7 +556,7 @@ static NTSTATUS dcerpc_pipe_connect_ncacn_ip_tcp(struct dcerpc_pipe **p,
 	(*p)->flags = binding->flags;
 
 	/* remember the binding string for possible secondary connections */
-	(*p)->binding_string = dcerpc_binding_string((*p)->mem_ctx, binding);
+	(*p)->binding_string = dcerpc_binding_string((*p), binding);
 
 	if (username && username[0] && (binding->flags & DCERPC_SCHANNEL_ANY)) {
 		status = dcerpc_bind_auth_schannel(*p, pipe_uuid, pipe_version, 
@@ -665,7 +665,7 @@ NTSTATUS dcerpc_secondary_connection(struct dcerpc_pipe *p, struct dcerpc_pipe *
 		break;
 
 	case NCACN_IP_TCP:
-		status = dcerpc_parse_binding(p->mem_ctx, p->binding_string, &b);
+		status = dcerpc_parse_binding(p, p->binding_string, &b);
 		if (!NT_STATUS_IS_OK(status)) {
 			return status;
 		}