messaging3: Push down the self-send callback

In the messaging_read receivers we already defer the callback: We need to
reply on potentially different tevent contexts, thus the defer_callback.

The callback case in messaging_dispatch_rec was direct before this
patch. This changes messaging_dispatch_rec to also defer the callback
in the self-send case.

Now we need only two roundtrips in local-messaging-read1 :-)

Signed-off-by: Volker Lendecke <vl@samba.org>
Reviewed-by: Stefan Metzmacher <metze@samba.org>
Reviewed-by: Jeremy Allison <jra@samba.org>

2 files changed, 95 insertions, 53 deletions

diff --git a/source3/lib/messages.c b/source3/lib/messages.c
index 50c79a128d9..1c62809b0d9 100644
--- a/source3/lib/messages.c
+++ b/source3/lib/messages.c
@@ -341,15 +341,6 @@ void messaging_deregister(struct messaging_context *ctx, uint32_t msg_type,
 	}
 }
 
-struct messaging_selfsend_state {
-	struct messaging_context *msg;
-	struct messaging_rec rec;
-};
-
-static void messaging_trigger_self(struct tevent_context *ev,
-				   struct tevent_immediate *im,
-				   void *private_data);
-
 /*
   Send a message to a particular server
 */
@@ -368,33 +359,13 @@ NTSTATUS messaging_send(struct messaging_context *msg_ctx,
 	}
 
 	if (server_id_equal(&msg_ctx->id, &server)) {
-		struct messaging_selfsend_state *state;
-		struct tevent_immediate *im;
-
-		state = talloc_pooled_object(
-			msg_ctx, struct messaging_selfsend_state,
-			1, data->length);
-		if (state == NULL) {
-			return NT_STATUS_NO_MEMORY;
-		}
-		state->msg = msg_ctx;
-		state->rec.msg_version = MESSAGE_VERSION;
-		state->rec.msg_type = msg_type & MSG_TYPE_MASK;
-		state->rec.dest = server;
-		state->rec.src = msg_ctx->id;
-
-		/* Can't fail, it's a pooled_object */
-		state->rec.buf = data_blob_talloc(
-			state, data->data, data->length);
-
-		im = tevent_create_immediate(state);
-		if (im == NULL) {
-			TALLOC_FREE(state);
-			return NT_STATUS_NO_MEMORY;
-		}
-
-		tevent_schedule_immediate(im, msg_ctx->event_ctx,
-					  messaging_trigger_self, state);
+		struct messaging_rec rec;
+		rec.msg_version = MESSAGE_VERSION;
+		rec.msg_type = msg_type & MSG_TYPE_MASK;
+		rec.dest = server;
+		rec.src = msg_ctx->id;
+		rec.buf = *data;
+		messaging_dispatch_rec(msg_ctx, &rec);
 		return NT_STATUS_OK;
 	}
 
@@ -402,16 +373,6 @@ NTSTATUS messaging_send(struct messaging_context *msg_ctx,
 				       msg_ctx->local);
 }
 
-static void messaging_trigger_self(struct tevent_context *ev,
-				   struct tevent_immediate *im,
-				   void *private_data)
-{
-	struct messaging_selfsend_state *state = talloc_get_type_abort(
-		private_data, struct messaging_selfsend_state);
-	messaging_dispatch_rec(state->msg, &state->rec);
-	TALLOC_FREE(state);
-}
-
 NTSTATUS messaging_send_buf(struct messaging_context *msg_ctx,
 			    struct server_id server, uint32_t msg_type,
 			    const uint8_t *buf, size_t len)
@@ -697,6 +658,68 @@ static bool messaging_append_new_waiters(struct messaging_context *msg_ctx)
 	return true;
 }
 
+struct messaging_defer_callback_state {
+	struct messaging_context *msg_ctx;
+	struct messaging_rec *rec;
+	void (*fn)(struct messaging_context *msg, void *private_data,
+		   uint32_t msg_type, struct server_id server_id,
+		   DATA_BLOB *data);
+	void *private_data;
+};
+
+static void messaging_defer_callback_trigger(struct tevent_context *ev,
+					     struct tevent_immediate *im,
+					     void *private_data);
+
+static void messaging_defer_callback(
+	struct messaging_context *msg_ctx, struct messaging_rec *rec,
+	void (*fn)(struct messaging_context *msg, void *private_data,
+		   uint32_t msg_type, struct server_id server_id,
+		   DATA_BLOB *data),
+	void *private_data)
+{
+	struct messaging_defer_callback_state *state;
+	struct tevent_immediate *im;
+
+	state = talloc(msg_ctx, struct messaging_defer_callback_state);
+	if (state == NULL) {
+		DEBUG(1, ("talloc failed\n"));
+		return;
+	}
+	state->msg_ctx = msg_ctx;
+	state->fn = fn;
+	state->private_data = private_data;
+
+	state->rec = messaging_rec_dup(state, rec);
+	if (state->rec == NULL) {
+		DEBUG(1, ("talloc failed\n"));
+		TALLOC_FREE(state);
+		return;
+	}
+
+	im = tevent_create_immediate(state);
+	if (im == NULL) {
+		DEBUG(1, ("tevent_create_immediate failed\n"));
+		TALLOC_FREE(state);
+		return;
+	}
+	tevent_schedule_immediate(im, msg_ctx->event_ctx,
+				  messaging_defer_callback_trigger, state);
+}
+
+static void messaging_defer_callback_trigger(struct tevent_context *ev,
+					     struct tevent_immediate *im,
+					     void *private_data)
+{
+	struct messaging_defer_callback_state *state = talloc_get_type_abort(
+		private_data, struct messaging_defer_callback_state);
+	struct messaging_rec *rec = state->rec;
+
+	state->fn(state->msg_ctx, state->private_data, rec->msg_type, rec->src,
+		  &rec->buf);
+	TALLOC_FREE(state);
+}
+
 /*
   Dispatch one messaging_rec
 */
@@ -708,15 +731,34 @@ void messaging_dispatch_rec(struct messaging_context *msg_ctx,
 
 	for (cb = msg_ctx->callbacks; cb != NULL; cb = next) {
 		next = cb->next;
-		if (cb->msg_type == rec->msg_type) {
+		if (cb->msg_type != rec->msg_type) {
+			continue;
+		}
+
+		if (server_id_equal(&msg_ctx->id, &rec->dest)) {
+			/*
+			 * This is a self-send. We are called here from
+			 * messaging_send(), and we don't want to directly
+			 * recurse into the callback but go via a
+			 * tevent_loop_once
+			 */
+			messaging_defer_callback(msg_ctx, rec, cb->fn,
+						 cb->private_data);
+		} else {
+			/*
+			 * This comes from a different process. we are called
+			 * from the event loop, so we should call back
+			 * directly.
+			 */
 			cb->fn(msg_ctx, cb->private_data, rec->msg_type,
 			       rec->src, &rec->buf);
-			/* we continue looking for matching messages
-			   after finding one. This matters for
-			   subsystems like the internal notify code
-			   which register more than one handler for
-			   the same message type */
 		}
+		/*
+		 * we continue looking for matching messages after finding
+		 * one. This matters for subsystems like the internal notify
+		 * code which register more than one handler for the same
+		 * message type
+		 */
 	}
 
 	if (!messaging_append_new_waiters(msg_ctx)) {
diff --git a/source3/torture/test_messaging_read.c b/source3/torture/test_messaging_read.c
index 188b0216e71..0bb3128cd88 100644
--- a/source3/torture/test_messaging_read.c
+++ b/source3/torture/test_messaging_read.c
@@ -122,7 +122,7 @@ bool run_messaging_read1(int dummy)
 		goto fail;
 	}
 
-	for (i=0; i<3; i++) {
+	for (i=0; i<2; i++) {
 		if (tevent_loop_once(ev) != 0) {
 			fprintf(stderr, "tevent_loop_once failed\n");
 			goto fail;