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;
 		}