[PATCH] simplify messaging3

Volker Lendecke Volker.Lendecke at SerNet.DE
Mon Jan 20 04:01:27 MST 2014


Hi!

Attached find a patchset that cuts 500 lines from the
source3 based messaging subsystem.

Please review & push.

Thanks

Volker

-- 
SerNet GmbH, Bahnhofsallee 1b, 37081 Göttingen
phone: +49-551-370000-0, fax: +49-551-370000-9
AG Göttingen, HRB 2816, GF: Dr. Johannes Loxen
http://www.sernet.de, mailto:kontakt at sernet.de
-------------- next part --------------
From 33ecdfc6ce5216f702d987de3b69e18a4e181912 Mon Sep 17 00:00:00 2001
From: Volker Lendecke <vl at samba.org>
Date: Mon, 30 Dec 2013 11:26:52 +0100
Subject: [PATCH 1/5] messaging3: Add messaging_read_send/recv

This is made to replace the msg_channel abstraction.

msg_channel was created to not miss any messages. For this, some
complex queueing was installed. This complexity has caused quite a
few problems in the past (see bug 10284 for example).

messaging_read_send/recv is able to achieve the same goal with a
lot less complexity. The messaging_read_send atomically installs
the reader into the messaging_context, we will not miss any messages
while this installed. messaging_send_recv will deinstall that
listener, but in the callback function you can directly call
messaging_read_send again without going through the tevent_loop_once.
As long as this is always made sure, no messages will be lost.
---
 source3/include/messages.h        |  10 +++
 source3/lib/dbwrap/dbwrap_watch.c |   1 -
 source3/lib/messages.c            | 126 ++++++++++++++++++++++++++++++++++++++
 3 files changed, 136 insertions(+), 1 deletion(-)

diff --git a/source3/include/messages.h b/source3/include/messages.h
index cefa279..27b3156 100644
--- a/source3/include/messages.h
+++ b/source3/include/messages.h
@@ -80,6 +80,9 @@ struct messaging_context {
 	struct tevent_context *event_ctx;
 	struct messaging_callback *callbacks;
 
+	struct tevent_req **waiters;
+	unsigned num_waiters;
+
 	struct messaging_backend *local;
 	struct messaging_backend *remote;
 };
@@ -140,6 +143,13 @@ NTSTATUS messaging_send_buf(struct messaging_context *msg_ctx,
 void messaging_dispatch_rec(struct messaging_context *msg_ctx,
 			    struct messaging_rec *rec);
 
+struct tevent_req *messaging_read_send(TALLOC_CTX *mem_ctx,
+				       struct tevent_context *ev,
+				       struct messaging_context *msg,
+				       uint32_t msg_type);
+int messaging_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
+			struct messaging_rec **presult);
+
 #include "librpc/gen_ndr/ndr_messaging.h"
 
 #endif
diff --git a/source3/lib/dbwrap/dbwrap_watch.c b/source3/lib/dbwrap/dbwrap_watch.c
index 7bdcd99..e65dbf4 100644
--- a/source3/lib/dbwrap/dbwrap_watch.c
+++ b/source3/lib/dbwrap/dbwrap_watch.c
@@ -22,7 +22,6 @@
 #include "dbwrap/dbwrap.h"
 #include "dbwrap_watch.h"
 #include "dbwrap_open.h"
-#include "msg_channel.h"
 #include "lib/util/util_tdb.h"
 #include "lib/util/tevent_ntstatus.h"
 
diff --git a/source3/lib/messages.c b/source3/lib/messages.c
index ba473ae..58f45d3 100644
--- a/source3/lib/messages.c
+++ b/source3/lib/messages.c
@@ -49,6 +49,7 @@
 #include "dbwrap/dbwrap.h"
 #include "serverid.h"
 #include "messages.h"
+#include "lib/util/tevent_unix.h"
 
 struct messaging_callback {
 	struct messaging_callback *prev, *next;
@@ -425,6 +426,120 @@ NTSTATUS messaging_send_buf(struct messaging_context *msg_ctx,
 	return messaging_send(msg_ctx, server, msg_type, &blob);
 }
 
+static struct messaging_rec *messaging_rec_dup(TALLOC_CTX *mem_ctx,
+					       struct messaging_rec *rec)
+{
+	struct messaging_rec *result;
+
+	result = talloc_pooled_object(mem_ctx, struct messaging_rec,
+				      1, rec->buf.length);
+	if (result == NULL) {
+		return NULL;
+	}
+	*result = *rec;
+
+	/* Doesn't fail, see talloc_pooled_object */
+
+	result->buf.data = talloc_memdup(result, rec->buf.data,
+					 rec->buf.length);
+	return result;
+}
+
+struct messaging_read_state {
+	struct tevent_context *ev;
+	struct messaging_context *msg_ctx;
+	uint32_t msg_type;
+	struct messaging_rec *rec;
+};
+
+static void messaging_read_cleanup(struct tevent_req *req,
+				   enum tevent_req_state req_state);
+
+struct tevent_req *messaging_read_send(TALLOC_CTX *mem_ctx,
+				       struct tevent_context *ev,
+				       struct messaging_context *msg_ctx,
+				       uint32_t msg_type)
+{
+	struct tevent_req *req;
+	struct messaging_read_state *state;
+	size_t waiters_len;
+
+	req = tevent_req_create(mem_ctx, &state,
+				struct messaging_read_state);
+	if (req == NULL) {
+		return NULL;
+	}
+	state->ev = ev;
+	state->msg_ctx = msg_ctx;
+	state->msg_type = msg_type;
+
+	waiters_len = talloc_array_length(msg_ctx->waiters);
+
+	if (waiters_len == msg_ctx->num_waiters) {
+		struct tevent_req **tmp;
+
+		tmp = talloc_realloc(msg_ctx, msg_ctx->waiters,
+				     struct tevent_req *, waiters_len+1);
+		if (tevent_req_nomem(tmp, req)) {
+			return tevent_req_post(req, ev);
+		}
+		msg_ctx->waiters = tmp;
+	}
+
+	msg_ctx->waiters[msg_ctx->num_waiters] = req;
+	msg_ctx->num_waiters += 1;
+	tevent_req_set_cleanup_fn(req, messaging_read_cleanup);
+
+	return req;
+}
+
+static void messaging_read_cleanup(struct tevent_req *req,
+				   enum tevent_req_state req_state)
+{
+	struct messaging_read_state *state = tevent_req_data(
+		req, struct messaging_read_state);
+	struct messaging_context *msg_ctx = state->msg_ctx;
+	struct tevent_req **waiters = msg_ctx->waiters;
+	unsigned i;
+
+	tevent_req_set_cleanup_fn(req, NULL);
+
+	for (i=0; i<msg_ctx->num_waiters; i++) {
+		if (waiters[i] == req) {
+			waiters[i] = waiters[msg_ctx->num_waiters-1];
+			msg_ctx->num_waiters -= 1;
+			return;
+		}
+	}
+}
+
+static void messaging_read_done(struct tevent_req *req, struct messaging_rec *rec)
+{
+	struct messaging_read_state *state = tevent_req_data(
+		req, struct messaging_read_state);
+
+	state->rec = messaging_rec_dup(state, rec);
+	if (tevent_req_nomem(state->rec, req)) {
+		return;
+	}
+	tevent_req_done(req);
+}
+
+int messaging_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
+			struct messaging_rec **presult)
+{
+	struct messaging_read_state *state = tevent_req_data(
+		req, struct messaging_read_state);
+	int err;
+
+	if (tevent_req_is_unix_error(req, &err)) {
+		tevent_req_received(req);
+		return err;
+	}
+	*presult = talloc_move(mem_ctx, &state->rec);
+	return 0;
+}
+
 /*
   Dispatch one messaging_rec
 */
@@ -432,6 +547,7 @@ void messaging_dispatch_rec(struct messaging_context *msg_ctx,
 			    struct messaging_rec *rec)
 {
 	struct messaging_callback *cb, *next;
+	unsigned i;
 
 	for (cb = msg_ctx->callbacks; cb != NULL; cb = next) {
 		next = cb->next;
@@ -445,6 +561,16 @@ void messaging_dispatch_rec(struct messaging_context *msg_ctx,
 			   the same message type */
 		}
 	}
+
+	for (i=0; i<msg_ctx->num_waiters; i++) {
+		struct tevent_req *req = msg_ctx->waiters[i];
+		struct messaging_read_state *state = tevent_req_data(
+			req, struct messaging_read_state);
+
+		if (state->msg_type == rec->msg_type) {
+			messaging_read_done(req, rec);
+		}
+	}
 	return;
 }
 
-- 
1.8.5.2


From a198a947f384e1b9a657bea6e463f5facf1d6dea Mon Sep 17 00:00:00 2001
From: Volker Lendecke <vl at samba.org>
Date: Mon, 30 Dec 2013 21:35:03 +0100
Subject: [PATCH 2/5] dbwrap_watch: Use messaging_read_send/recv

---
 source3/lib/dbwrap/dbwrap_watch.c | 20 +++++++-------------
 1 file changed, 7 insertions(+), 13 deletions(-)

diff --git a/source3/lib/dbwrap/dbwrap_watch.c b/source3/lib/dbwrap/dbwrap_watch.c
index e65dbf4..b586b66 100644
--- a/source3/lib/dbwrap/dbwrap_watch.c
+++ b/source3/lib/dbwrap/dbwrap_watch.c
@@ -230,7 +230,6 @@ struct dbwrap_record_watch_state {
 	struct db_context *db;
 	struct tevent_req *req;
 	struct messaging_context *msg;
-	struct msg_channel *channel;
 	TDB_DATA key;
 	TDB_DATA w_key;
 };
@@ -248,7 +247,6 @@ struct tevent_req *dbwrap_record_watch_send(TALLOC_CTX *mem_ctx,
 	struct dbwrap_record_watch_state *state;
 	struct db_context *watchers_db;
 	NTSTATUS status;
-	int ret;
 
 	req = tevent_req_create(mem_ctx, &state,
 				struct dbwrap_record_watch_state);
@@ -272,12 +270,12 @@ struct tevent_req *dbwrap_record_watch_send(TALLOC_CTX *mem_ctx,
 		return tevent_req_post(req, ev);
 	}
 
-	ret = msg_channel_init(state, state->msg, MSG_DBWRAP_MODIFIED,
-			       &state->channel);
-	if (ret != 0) {
-		tevent_req_nterror(req, map_nt_error_from_unix(ret));
+	subreq = messaging_read_send(state, ev, state->msg,
+				     MSG_DBWRAP_MODIFIED);
+	if (tevent_req_nomem(subreq, req)) {
 		return tevent_req_post(req, ev);
 	}
+	tevent_req_set_callback(subreq, dbwrap_record_watch_done, req);
 
 	status = dbwrap_record_add_watcher(
 		state->w_key, messaging_server_id(state->msg));
@@ -286,11 +284,6 @@ struct tevent_req *dbwrap_record_watch_send(TALLOC_CTX *mem_ctx,
 	}
 	talloc_set_destructor(state, dbwrap_record_watch_state_destructor);
 
-	subreq = msg_read_send(state, state->ev, state->channel);
-	if (tevent_req_nomem(subreq, req)) {
-		return tevent_req_post(req, ev);
-	}
-	tevent_req_set_callback(subreq, dbwrap_record_watch_done, req);
 	return req;
 }
 
@@ -365,7 +358,7 @@ static void dbwrap_record_watch_done(struct tevent_req *subreq)
 	struct messaging_rec *rec;
 	int ret;
 
-	ret = msg_read_recv(subreq, talloc_tos(), &rec);
+	ret = messaging_read_recv(subreq, talloc_tos(), &rec);
 	TALLOC_FREE(subreq);
 	if (ret != 0) {
 		tevent_req_nterror(req, map_nt_error_from_unix(ret));
@@ -381,7 +374,8 @@ static void dbwrap_record_watch_done(struct tevent_req *subreq)
 	/*
 	 * Not our record, wait for the next one
 	 */
-	subreq = msg_read_send(state, state->ev, state->channel);
+	subreq = messaging_read_send(state, state->ev, state->msg,
+				     MSG_DBWRAP_MODIFIED);
 	if (tevent_req_nomem(subreq, req)) {
 		return;
 	}
-- 
1.8.5.2


From 1a0c529b96ee75772dc77ea6766d548b13d79613 Mon Sep 17 00:00:00 2001
From: Volker Lendecke <vl at samba.org>
Date: Mon, 30 Dec 2013 21:41:27 +0100
Subject: [PATCH 3/5] smbd: Use messaging_read_send in smbXsrv_session.c

---
 source3/smbd/smbXsrv_session.c | 20 +++++---------------
 1 file changed, 5 insertions(+), 15 deletions(-)

diff --git a/source3/smbd/smbXsrv_session.c b/source3/smbd/smbXsrv_session.c
index 017880c..fa3033b 100644
--- a/source3/smbd/smbXsrv_session.c
+++ b/source3/smbd/smbXsrv_session.c
@@ -37,7 +37,6 @@
 #include "librpc/gen_ndr/ndr_smbXsrv.h"
 #include "serverid.h"
 #include "lib/util/tevent_ntstatus.h"
-#include "msg_channel.h"
 
 struct smbXsrv_session_table {
 	struct {
@@ -50,7 +49,6 @@ struct smbXsrv_session_table {
 	struct {
 		struct db_context *db_ctx;
 	} global;
-	struct msg_channel *close_channel;
 };
 
 static struct db_context *smbXsrv_session_global_db_ctx = NULL;
@@ -168,7 +166,6 @@ static NTSTATUS smbXsrv_session_table_init(struct smbXsrv_connection *conn,
 	struct smbXsrv_session_table *table;
 	NTSTATUS status;
 	struct tevent_req *subreq;
-	int ret;
 	uint64_t max_range;
 
 	if (lowest_id > highest_id) {
@@ -207,16 +204,8 @@ static NTSTATUS smbXsrv_session_table_init(struct smbXsrv_connection *conn,
 
 	dbwrap_watch_db(table->global.db_ctx, conn->msg_ctx);
 
-	ret = msg_channel_init(table, conn->msg_ctx,
-			       MSG_SMBXSRV_SESSION_CLOSE,
-			       &table->close_channel);
-	if (ret != 0) {
-		status = map_nt_error_from_unix_common(errno);
-		TALLOC_FREE(table);
-		return status;
-	}
-
-	subreq = msg_read_send(table, conn->ev_ctx, table->close_channel);
+	subreq = messaging_read_send(table, conn->ev_ctx, conn->msg_ctx,
+				     MSG_SMBXSRV_SESSION_CLOSE);
 	if (subreq == NULL) {
 		TALLOC_FREE(table);
 		return NT_STATUS_NO_MEMORY;
@@ -243,7 +232,7 @@ static void smbXsrv_session_close_loop(struct tevent_req *subreq)
 	struct timeval tv = timeval_current();
 	NTTIME now = timeval_to_nttime(&tv);
 
-	ret = msg_read_recv(subreq, talloc_tos(), &rec);
+	ret = messaging_read_recv(subreq, talloc_tos(), &rec);
 	TALLOC_FREE(subreq);
 	if (ret != 0) {
 		goto next;
@@ -348,7 +337,8 @@ static void smbXsrv_session_close_loop(struct tevent_req *subreq)
 next:
 	TALLOC_FREE(rec);
 
-	subreq = msg_read_send(table, conn->ev_ctx, table->close_channel);
+	subreq = messaging_read_send(table, conn->ev_ctx, conn->msg_ctx,
+				     MSG_SMBXSRV_SESSION_CLOSE);
 	if (subreq == NULL) {
 		smbd_server_connection_terminate(conn->sconn,
 						 "msg_read_send() failed");
-- 
1.8.5.2


From cfa081794f1161422216315b379c7af1a716749e Mon Sep 17 00:00:00 2001
From: Volker Lendecke <vl at samba.org>
Date: Sat, 18 Jan 2014 15:56:49 +0100
Subject: [PATCH 4/5] messaging3: remove msg_channel

---
 source3/lib/msg_channel.c  | 337 ---------------------------------------------
 source3/lib/msg_channel.h  |  45 ------
 source3/selftest/tests.py  |   1 -
 source3/torture/proto.h    |   2 -
 source3/torture/test_msg.c | 217 -----------------------------
 source3/torture/torture.c  |   2 -
 source3/wscript_build      |   2 -
 7 files changed, 606 deletions(-)
 delete mode 100644 source3/lib/msg_channel.c
 delete mode 100644 source3/lib/msg_channel.h
 delete mode 100644 source3/torture/test_msg.c

diff --git a/source3/lib/msg_channel.c b/source3/lib/msg_channel.c
deleted file mode 100644
index 6be5e2e..0000000
--- a/source3/lib/msg_channel.c
+++ /dev/null
@@ -1,337 +0,0 @@
-/*
-   Unix SMB/CIFS implementation.
-   Samba3 message channels
-   Copyright (C) Volker Lendecke 2012
-
-   This program is free software; you can redistribute it and/or modify
-   it under the terms of the GNU General Public License as published by
-   the Free Software Foundation; either version 3 of the License, or
-   (at your option) any later version.
-
-   This program is distributed in the hope that it will be useful,
-   but WITHOUT ANY WARRANTY; without even the implied warranty of
-   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-   GNU General Public License for more details.
-
-   You should have received a copy of the GNU General Public License
-   along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "includes.h"
-#include "msg_channel.h"
-#include "ctdb_conn.h"
-#include "lib/util/tevent_unix.h"
-
-struct msg_channel {
-	struct ctdb_msg_channel *ctdb_channel;
-	struct messaging_context *msg;
-	uint32_t msg_type;
-
-	struct tevent_req *pending_req;
-	struct tevent_context *ev;
-
-	struct messaging_rec **msgs;
-};
-
-struct msg_channel_init_state {
-	struct msg_channel *channel;
-};
-
-static void msg_channel_init_got_ctdb(struct tevent_req *subreq);
-static void msg_channel_init_got_msg(struct messaging_context *msg,
-			       void *priv, uint32_t msg_type,
-			       struct server_id server_id, DATA_BLOB *data);
-static int msg_channel_destructor(struct msg_channel *s);
-
-struct tevent_req *msg_channel_init_send(TALLOC_CTX *mem_ctx,
-				    struct tevent_context *ev,
-				    struct messaging_context *msg,
-				    uint32_t msg_type)
-{
-	struct tevent_req *req, *subreq;
-	struct msg_channel_init_state *state;
-	struct server_id pid;
-
-	req = tevent_req_create(mem_ctx, &state,
-				struct msg_channel_init_state);
-	if (req == NULL) {
-		return NULL;
-	}
-
-	state->channel = talloc_zero(state, struct msg_channel);
-	if (tevent_req_nomem(state->channel, req)) {
-		return tevent_req_post(req, ev);
-	}
-	state->channel->msg = msg;
-	state->channel->msg_type = msg_type;
-
-	pid = messaging_server_id(msg);
-	subreq = ctdb_msg_channel_init_send(state, ev, lp_ctdbd_socket(),
-					    pid.pid);
-	if (tevent_req_nomem(subreq, req)) {
-		return tevent_req_post(req, ev);
-	}
-	tevent_req_set_callback(subreq, msg_channel_init_got_ctdb, req);
-	return req;
-}
-
-static void msg_channel_init_got_ctdb(struct tevent_req *subreq)
-{
-	struct tevent_req *req = tevent_req_callback_data(
-		subreq, struct tevent_req);
-	struct msg_channel_init_state *state = tevent_req_data(
-		req, struct msg_channel_init_state);
-	struct msg_channel *s = state->channel;
-	NTSTATUS status;
-	int ret;
-
-	ret = ctdb_msg_channel_init_recv(subreq, s, &s->ctdb_channel);
-	TALLOC_FREE(subreq);
-
-	if (ret == ENOSYS) {
-		s->ctdb_channel = NULL;
-		ret = 0;
-	}
-
-	if (tevent_req_error(req, ret)) {
-		return;
-	}
-	status = messaging_register(s->msg, s, s->msg_type,
-				    msg_channel_init_got_msg);
-	if (!NT_STATUS_IS_OK(status)) {
-		tevent_req_error(req, map_errno_from_nt_status(status));
-		return;
-	}
-	talloc_set_destructor(s, msg_channel_destructor);
-	tevent_req_done(req);
-}
-
-static int msg_channel_destructor(struct msg_channel *s)
-{
-	messaging_deregister(s->msg, s->msg_type, s);
-	return 0;
-}
-
-int msg_channel_init_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
-			  struct msg_channel **pchannel)
-{
-	struct msg_channel_init_state *state = tevent_req_data(
-		req, struct msg_channel_init_state);
-	int err;
-
-	if (tevent_req_is_unix_error(req, &err)) {
-		return err;
-	}
-	*pchannel = talloc_move(mem_ctx, &state->channel);
-	return 0;
-}
-
-int msg_channel_init(TALLOC_CTX *mem_ctx, struct messaging_context *msg,
-		     uint32_t msgtype, struct msg_channel **pchannel)
-{
-	TALLOC_CTX *frame = talloc_stackframe();
-	struct tevent_context *ev;
-	struct tevent_req *req;
-	int err = ENOMEM;
-	bool ok;
-
-	ev = samba_tevent_context_init(frame);
-	if (ev == NULL) {
-		goto fail;
-	}
-	req = msg_channel_init_send(frame, ev, msg, msgtype);
-	if (req == NULL) {
-		goto fail;
-	}
-	ok = tevent_req_poll(req, ev);
-	if (!ok) {
-		err = errno;
-		goto fail;
-	}
-	err = msg_channel_init_recv(req, mem_ctx, pchannel);
-fail:
-	TALLOC_FREE(frame);
-	return err;
-}
-
-struct msg_read_state {
-	struct tevent_context *ev;
-	struct msg_channel *channel;
-	struct messaging_rec *rec;
-};
-
-static void msg_channel_init_got_msg(struct messaging_context *msg,
-				     void *priv, uint32_t msg_type,
-				     struct server_id server_id,
-				     DATA_BLOB *data)
-{
-	struct msg_channel *s = talloc_get_type_abort(
-		priv, struct msg_channel);
-	struct messaging_rec *rec;
-	struct messaging_rec **msgs;
-	size_t num_msgs;
-
-	rec = talloc(s, struct messaging_rec);
-	if (rec == NULL) {
-		goto fail;
-	}
-	rec->msg_version = 1;
-	rec->msg_type = msg_type;
-	rec->dest = server_id;
-	rec->src = messaging_server_id(msg);
-	rec->buf.data = (uint8_t *)talloc_memdup(rec, data->data,
-						 data->length);
-	if (rec->buf.data == NULL) {
-		goto fail;
-	}
-	rec->buf.length = data->length;
-
-	if (s->pending_req != NULL) {
-		struct tevent_req *req = s->pending_req;
-		struct msg_read_state *state = tevent_req_data(
-			req, struct msg_read_state);
-
-		s->pending_req = NULL;
-
-		state->rec = talloc_move(state, &rec);
-		tevent_req_defer_callback(req, s->ev);
-		tevent_req_done(req);
-		return;
-	}
-
-	num_msgs = talloc_array_length(s->msgs);
-	msgs = talloc_realloc(s, s->msgs, struct messaging_rec *, num_msgs+1);
-	if (msgs == NULL) {
-		goto fail;
-	}
-	s->msgs = msgs;
-	s->msgs[num_msgs] = talloc_move(s->msgs, &rec);
-
-	return;
-fail:
-	TALLOC_FREE(rec);
-}
-
-static void msg_read_got_ctdb(struct tevent_req *subreq);
-
-struct tevent_req *msg_read_send(TALLOC_CTX *mem_ctx,
-				 struct tevent_context *ev,
-				 struct msg_channel *channel)
-{
-	struct tevent_req *req;
-	struct msg_read_state *state;
-	void *msg_tdb_event;
-	size_t num_msgs;
-
-	req = tevent_req_create(mem_ctx, &state, struct msg_read_state);
-	if (req == NULL) {
-		return NULL;
-	}
-	state->ev = ev;
-	state->channel = channel;
-
-	if (channel->pending_req != NULL) {
-		tevent_req_error(req, EBUSY);
-		return tevent_req_post(req, ev);
-	}
-
-	num_msgs = talloc_array_length(channel->msgs);
-	if (num_msgs != 0) {
-		state->rec = talloc_move(state, &channel->msgs[0]);
-		memmove(channel->msgs, channel->msgs+1,
-			sizeof(struct messaging_rec *) * (num_msgs-1));
-		channel->msgs = talloc_realloc(
-			channel, channel->msgs, struct messaging_rec *,
-			num_msgs - 1);
-		tevent_req_done(req);
-		return tevent_req_post(req, ev);
-	}
-
-	channel->pending_req = req;
-	channel->ev = ev;
-
-	msg_tdb_event = messaging_tdb_event(state, channel->msg, ev);
-	if (tevent_req_nomem(msg_tdb_event, req)) {
-		return tevent_req_post(req, ev);
-
-	}
-	if (channel->ctdb_channel != NULL) {
-		struct tevent_req *subreq;
-
-		subreq = ctdb_msg_read_send(state, ev,
-					    channel->ctdb_channel);
-		if (tevent_req_nomem(subreq, req)) {
-			return tevent_req_post(req, ev);
-		}
-		tevent_req_set_callback(subreq, msg_read_got_ctdb, req);
-	}
-	return req;
-}
-
-static void msg_read_got_ctdb(struct tevent_req *subreq)
-{
-	struct tevent_req *req = tevent_req_callback_data(
-		subreq, struct tevent_req);
-	struct msg_read_state *state = tevent_req_data(
-		req, struct msg_read_state);
-	DATA_BLOB blob;
-	enum ndr_err_code ndr_err;
-	int ret;
-
-	ret = ctdb_msg_read_recv(subreq, talloc_tos(),
-				 &blob.data, &blob.length);
-	TALLOC_FREE(subreq);
-	if (tevent_req_error(req, ret)) {
-		return;
-	}
-
-	state->rec = talloc(state, struct messaging_rec);
-	if (tevent_req_nomem(state->rec, req)) {
-		return;
-	}
-
-	ndr_err = ndr_pull_struct_blob(
-		&blob, state->rec, state->rec,
-		(ndr_pull_flags_fn_t)ndr_pull_messaging_rec);
-
-	if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
-		DEBUG(1, ("ndr_pull_struct_blob failed: %s\n",
-			  ndr_errstr(ndr_err)));
-		tevent_req_error(req, ndr_map_error2errno(ndr_err));
-		return;
-	}
-	if (DEBUGLEVEL >= 10) {
-		NDR_PRINT_DEBUG(messaging_rec, state->rec);
-	}
-	if (state->rec->msg_type == state->channel->msg_type) {
-		tevent_req_done(req);
-		return;
-	}
-	/*
-	 * Got some unexpected msg type, wait for the next one
-	 */
-
-	TALLOC_FREE(state->rec);
-
-	subreq = ctdb_msg_read_send(state, state->ev,
-				    state->channel->ctdb_channel);
-	if (tevent_req_nomem(subreq, req)) {
-		return;
-	}
-	tevent_req_set_callback(subreq, msg_read_got_ctdb, req);
-}
-
-int msg_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
-		  struct messaging_rec **prec)
-{
-	struct msg_read_state *state = tevent_req_data(
-		req, struct msg_read_state);
-	int err;
-
-	if (tevent_req_is_unix_error(req, &err)) {
-		return err;
-	}
-	*prec = talloc_move(mem_ctx, &state->rec);
-	tevent_req_received(req);
-	return 0;
-}
diff --git a/source3/lib/msg_channel.h b/source3/lib/msg_channel.h
deleted file mode 100644
index 4c7ae42..0000000
--- a/source3/lib/msg_channel.h
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
-   Unix SMB/CIFS implementation.
-   Samba3 message streams
-   Copyright (C) Volker Lendecke 2012
-
-   This program is free software; you can redistribute it and/or modify
-   it under the terms of the GNU General Public License as published by
-   the Free Software Foundation; either version 3 of the License, or
-   (at your option) any later version.
-
-   This program is distributed in the hope that it will be useful,
-   but WITHOUT ANY WARRANTY; without even the implied warranty of
-   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-   GNU General Public License for more details.
-
-   You should have received a copy of the GNU General Public License
-   along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#ifndef _MSG_STREAM_H_
-#define _MSG_STREAM_H_
-
-#include <talloc.h>
-#include <tevent.h>
-#include "messages.h"
-#include "librpc/gen_ndr/messaging.h"
-
-struct msg_channel;
-
-struct tevent_req *msg_channel_init_send(TALLOC_CTX *mem_ctx,
-					 struct tevent_context *ev,
-					 struct messaging_context *msg,
-					 uint32_t msgtype);
-int msg_channel_init_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
-			  struct msg_channel **pchannel);
-int msg_channel_init(TALLOC_CTX *mem_ctx, struct messaging_context *msg,
-		     uint32_t msgtype, struct msg_channel **pchannel);
-
-struct tevent_req *msg_read_send(TALLOC_CTX *mem_ctx,
-				 struct tevent_context *ev,
-				 struct msg_channel *channel);
-int msg_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
-		  struct messaging_rec **prec);
-
-#endif
diff --git a/source3/selftest/tests.py b/source3/selftest/tests.py
index 254e942..4ecd9c6 100755
--- a/source3/selftest/tests.py
+++ b/source3/selftest/tests.py
@@ -64,7 +64,6 @@ tests = ["FDPASS", "LOCK1", "LOCK2", "LOCK3", "LOCK4", "LOCK5", "LOCK6", "LOCK7"
         "CLEANUP1",
         "CLEANUP2",
         "CLEANUP4",
-        "LOCAL-MSG2",
         "BAD-NBT-SESSION"]
 
 for t in tests:
diff --git a/source3/torture/proto.h b/source3/torture/proto.h
index b7eacdf..2b27289 100644
--- a/source3/torture/proto.h
+++ b/source3/torture/proto.h
@@ -106,8 +106,6 @@ bool run_cleanup2(int dummy);
 bool run_cleanup3(int dummy);
 bool run_cleanup4(int dummy);
 bool run_ctdb_conn(int dummy);
-bool run_msg_test(int dummy);
-bool run_msg_test2(int dummy);
 bool run_notify_bench2(int dummy);
 bool run_notify_bench3(int dummy);
 bool run_dbwrap_watch1(int dummy);
diff --git a/source3/torture/test_msg.c b/source3/torture/test_msg.c
deleted file mode 100644
index d57379d..0000000
--- a/source3/torture/test_msg.c
+++ /dev/null
@@ -1,217 +0,0 @@
-/*
-   Unix SMB/CIFS implementation.
-   Test msg_stream API
-   Copyright (C) Volker Lendecke 2012
-
-   This program is free software; you can redistribute it and/or modify
-   it under the terms of the GNU General Public License as published by
-   the Free Software Foundation; either version 3 of the License, or
-   (at your option) any later version.
-
-   This program is distributed in the hope that it will be useful,
-   but WITHOUT ANY WARRANTY; without even the implied warranty of
-   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-   GNU General Public License for more details.
-
-   You should have received a copy of the GNU General Public License
-   along with this program.  If not, see <http://www.gnu.org/licenses/>.
-*/
-
-#include "includes.h"
-#include "torture/proto.h"
-#include "lib/util/tevent_unix.h"
-#include "msg_channel.h"
-
-struct msg_test_state {
-	struct tevent_context *ev;
-	struct messaging_context *msg;
-	struct msg_channel *channel;
-};
-
-static void msg_test_got_channel(struct tevent_req *subreq);
-static void msg_test_got_msg(struct tevent_req *subreq);
-
-static struct tevent_req *msg_test_send(TALLOC_CTX *mem_ctx,
-					struct tevent_context *ev)
-{
-	struct tevent_req *req, *subreq;
-	struct msg_test_state *state;
-
-	req = tevent_req_create(mem_ctx, &state, struct msg_test_state);
-	if (req == NULL) {
-		return NULL;
-	}
-	state->ev = ev;
-
-	state->msg = messaging_init(state, state->ev);
-	if (tevent_req_nomem(state->msg, req)) {
-		return tevent_req_post(req, ev);
-	}
-	subreq = msg_channel_init_send(state, state->ev, state->msg, MSG_PING);
-	if (tevent_req_nomem(subreq, req)) {
-		return tevent_req_post(req, ev);
-	}
-	tevent_req_set_callback(subreq, msg_test_got_channel, req);
-	return req;
-}
-
-static void msg_test_got_channel(struct tevent_req *subreq)
-{
-	struct tevent_req *req = tevent_req_callback_data(
-		subreq, struct tevent_req);
-	struct msg_test_state *state = tevent_req_data(
-		req, struct msg_test_state);
-	int ret;
-
-	ret = msg_channel_init_recv(subreq, state, &state->channel);
-	TALLOC_FREE(subreq);
-	if (tevent_req_error(req, ret)) {
-		return;
-	}
-	subreq = msg_read_send(state, state->ev, state->channel);
-	if (tevent_req_nomem(subreq, req)) {
-		return;
-	}
-	tevent_req_set_callback(subreq, msg_test_got_msg, req);
-}
-
-static void msg_test_got_msg(struct tevent_req *subreq)
-{
-	struct tevent_req *req = tevent_req_callback_data(
-		subreq, struct tevent_req);
-	struct msg_test_state *state = tevent_req_data(
-		req, struct msg_test_state);
-	struct messaging_rec *msg;
-	int ret;
-
-	ret = msg_read_recv(subreq, state, &msg);
-	TALLOC_FREE(subreq);
-	if (tevent_req_error(req, ret)) {
-		return;
-	}
-	tevent_req_done(req);
-}
-
-static int msg_test_recv(struct tevent_req *req)
-{
-	int err;
-
-	if (tevent_req_is_unix_error(req, &err)) {
-		return err;
-	}
-	return 0;
-}
-
-bool run_msg_test(int dummy)
-{
-	struct tevent_context *ev;
-	struct tevent_req *req;
-	int ret;
-
-	ev = samba_tevent_context_init(talloc_tos());
-	if (ev == NULL) {
-		fprintf(stderr, "tevent_context_init failed\n");
-		return false;
-	}
-	req = msg_test_send(ev, ev);
-	if (req == NULL) {
-		fprintf(stderr, "msg_test_send failed\n");
-		return false;
-	}
-	if (!tevent_req_poll(req, ev)) {
-		fprintf(stderr, "tevent_req_poll failed\n");
-		return false;
-	}
-	ret = msg_test_recv(req);
-	TALLOC_FREE(req);
-	printf("msg_test_recv returned %s\n",
-	       ret ? strerror(ret) : "success");
-	TALLOC_FREE(ev);
-	return (ret == 0);
-}
-
-/*
- * Reproducer for bug 10284
- */
-
-static void msg_callback(struct tevent_req *subreq);
-
-struct msg_test2_state {
-	struct tevent_context *ev;
-	struct messaging_context *msg;
-	struct msg_channel *channel;
-	struct messaging_rec *rec;
-	struct tevent_req *req;
-};
-
-bool run_msg_test2(int dummy)
-{
-	struct msg_test2_state s;
-	NTSTATUS status;
-	int i, ret;
-
-	s.ev = samba_tevent_context_init(talloc_tos());
-	if (s.ev == NULL) {
-		fprintf(stderr, "tevent_context_init failed\n");
-		return false;
-	}
-
-	s.msg = messaging_init(s.ev, s.ev);
-	if (s.msg == NULL) {
-		fprintf(stderr, "messaging_init failed\n");
-		return false;
-	}
-
-	ret = msg_channel_init(s.ev, s.msg, MSG_PING, &s.channel);
-	if (ret != 0) {
-		fprintf(stderr, "msg_channel_init returned %s\n",
-			strerror(ret));
-		return false;
-	}
-
-	status = messaging_send(s.msg, messaging_server_id(s.msg), MSG_PING,
-				&data_blob_null);
-	if (!NT_STATUS_IS_OK(status)) {
-		fprintf(stderr, "messaging_send returned %s\n",
-			nt_errstr(status));
-		return false;
-	}
-
-	ret = tevent_loop_once(s.ev);
-	if (ret == -1) {
-		fprintf(stderr, "tevent_loop_once failed: %s\n",
-			strerror(errno));
-		return false;
-	}
-
-	s.req = msg_read_send(s.ev, s.ev, s.channel);
-	if (s.req == NULL) {
-		fprintf(stderr, "msg_read_send failed\n");
-		return false;
-	}
-	tevent_req_set_callback(s.req, msg_callback, &s);
-
-	status = messaging_send(s.msg, messaging_server_id(s.msg), MSG_PING,
-				&data_blob_null);
-	if (!NT_STATUS_IS_OK(status)) {
-		fprintf(stderr, "messaging_send returned %s\n",
-			nt_errstr(status));
-		return false;
-	}
-
-	for (i=0; i<5; i++) {
-		tevent_loop_once(s.ev);
-	}
-
-	return true;
-}
-
-static void msg_callback(struct tevent_req *subreq)
-{
-	struct msg_test2_state *s = _tevent_req_callback_data(subreq);
-	struct messaging_rec *rec;
-	msg_read_recv(subreq, NULL, &rec);
-	TALLOC_FREE(subreq);
-	subreq = msg_read_send(s->ev, s->ev, s->channel);
-	tevent_req_set_callback(subreq, msg_callback, s);
-}
diff --git a/source3/torture/torture.c b/source3/torture/torture.c
index edb12b0..1d915fc 100644
--- a/source3/torture/torture.c
+++ b/source3/torture/torture.c
@@ -9572,8 +9572,6 @@ static struct {
 	{ "LOCAL-GENCACHE", run_local_gencache, 0},
 	{ "LOCAL-TALLOC-DICT", run_local_talloc_dict, 0},
 	{ "LOCAL-CTDB-CONN", run_ctdb_conn, 0},
-	{ "LOCAL-MSG", run_msg_test, 0},
-	{ "LOCAL-MSG2", run_msg_test2, 0},
 	{ "LOCAL-DBWRAP-WATCH1", run_dbwrap_watch1, 0 },
 	{ "LOCAL-BASE64", run_local_base64, 0},
 	{ "LOCAL-RBTREE", run_local_rbtree, 0},
diff --git a/source3/wscript_build b/source3/wscript_build
index c5dc521..54f8c14 100755
--- a/source3/wscript_build
+++ b/source3/wscript_build
@@ -286,7 +286,6 @@ bld.SAMBA3_SUBSYSTEM('samba3core',
                    lib/ctdbd_conn.c
                    lib/ctdb_conn.c
                    lib/util_cluster.c
-                   lib/msg_channel.c
                    lib/id_cache.c
                    lib/talloc_dict.c
                    lib/serverid.c
@@ -1209,7 +1208,6 @@ bld.SAMBA3_BINARY('smbtorture' + bld.env.suffix3,
                  torture/test_smbsock_any_connect.c
                  torture/test_cleanup.c
                  torture/test_ctdbconn.c
-                 torture/test_msg.c
                  torture/test_notify.c
                  lib/tevent_barrier.c
                  torture/test_dbwrap_watch.c
-- 
1.8.5.2


From 0f129a5c28cc63f2dabc4d2d88e07f9c52618e3c Mon Sep 17 00:00:00 2001
From: Volker Lendecke <vl at samba.org>
Date: Mon, 20 Jan 2014 10:56:16 +0100
Subject: [PATCH 5/5] messaging3: Remove unused messaging_tdb_event

---
 source3/include/messages.h   |  3 ---
 source3/lib/messages_local.c | 10 ----------
 2 files changed, 13 deletions(-)

diff --git a/source3/include/messages.h b/source3/include/messages.h
index 27b3156..5666911 100644
--- a/source3/include/messages.h
+++ b/source3/include/messages.h
@@ -101,9 +101,6 @@ NTSTATUS messaging_tdb_init(struct messaging_context *msg_ctx,
 
 bool messaging_tdb_parent_init(TALLOC_CTX *mem_ctx);
 
-void *messaging_tdb_event(TALLOC_CTX *mem_ctx, struct messaging_context *msg,
-			  struct tevent_context *ev);
-
 NTSTATUS messaging_ctdbd_init(struct messaging_context *msg_ctx,
 			      TALLOC_CTX *mem_ctx,
 			      struct messaging_backend **presult);
diff --git a/source3/lib/messages_local.c b/source3/lib/messages_local.c
index c74c0aa..6b9c251 100644
--- a/source3/lib/messages_local.c
+++ b/source3/lib/messages_local.c
@@ -77,16 +77,6 @@ static void messaging_tdb_signal_handler(struct tevent_context *ev_ctx,
 	message_dispatch(ctx->msg_ctx);
 }
 
-void *messaging_tdb_event(TALLOC_CTX *mem_ctx, struct messaging_context *msg,
-			  struct tevent_context *ev)
-{
-	struct messaging_tdb_context *msg_tdb = talloc_get_type_abort(
-		msg->local->private_data, struct messaging_tdb_context);
-
-	return tevent_add_signal(ev, mem_ctx, SIGUSR1, 0,
-				 messaging_tdb_signal_handler, msg_tdb);
-}
-
 /****************************************************************************
  Initialise the messaging functions. 
 ****************************************************************************/
-- 
1.8.5.2



More information about the samba-technical mailing list