The branch, master has been updated
via c427b29 messaging3: Remove unused messaging_tdb_event
via 9801072 messaging3: remove msg_channel
via 2779e78 smbd: Use messaging_read_send in smbXsrv_session.c
via 0176097 dbwrap_watch: Use messaging_read_send/recv
via 5f3ccfc messaging3: Add messaging_read_send/recv
from eee450f ctdb-daemon: Simplify listing event scripts using scandir
http://gitweb.samba.org/?p=samba.git;a=shortlog;h=master
- Log -----------------------------------------------------------------
commit c427b29be915483eb23791b80b8df265715798d1
Author: Volker Lendecke <[email protected]>
Date: Mon Jan 20 10:56:16 2014 +0100
messaging3: Remove unused messaging_tdb_event
Signed-off-by: Volker Lendecke <[email protected]>
Reviewed-by: Jeremy Allison <[email protected]>
Autobuild-User(master): Volker Lendecke <[email protected]>
Autobuild-Date(master): Tue Jan 21 10:07:21 CET 2014 on sn-devel-104
commit 9801072ce2e991140bb0a799dbdd2badc1aba156
Author: Volker Lendecke <[email protected]>
Date: Sat Jan 18 15:56:49 2014 +0100
messaging3: remove msg_channel
Signed-off-by: Volker Lendecke <[email protected]>
Reviewed-by: Jeremy Allison <[email protected]>
commit 2779e780786180cd393d012690682414098b5437
Author: Volker Lendecke <[email protected]>
Date: Mon Dec 30 21:41:27 2013 +0100
smbd: Use messaging_read_send in smbXsrv_session.c
Signed-off-by: Volker Lendecke <[email protected]>
Reviewed-by: Jeremy Allison <[email protected]>
commit 017609767fff34118cb1f77c40af2037fba6fab2
Author: Volker Lendecke <[email protected]>
Date: Mon Dec 30 21:35:03 2013 +0100
dbwrap_watch: Use messaging_read_send/recv
Signed-off-by: Volker Lendecke <[email protected]>
Reviewed-by: Jeremy Allison <[email protected]>
commit 5f3ccfc16b7c64d2de109ba26d92d22ae1010882
Author: Volker Lendecke <[email protected]>
Date: Mon Dec 30 11:26:52 2013 +0100
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.
Signed-off-by: Volker Lendecke <[email protected]>
Reviewed-by: Jeremy Allison <[email protected]>
-----------------------------------------------------------------------
Summary of changes:
source3/include/messages.h | 13 +-
source3/lib/dbwrap/dbwrap_watch.c | 21 +--
source3/lib/messages.c | 126 ++++++++++++++
source3/lib/messages_local.c | 10 -
source3/lib/msg_channel.c | 337 -------------------------------------
source3/lib/msg_channel.h | 45 -----
source3/selftest/tests.py | 1 -
source3/smbd/smbXsrv_session.c | 20 +--
source3/torture/proto.h | 2 -
source3/torture/test_msg.c | 217 ------------------------
source3/torture/torture.c | 2 -
source3/wscript_build | 2 -
12 files changed, 148 insertions(+), 648 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
Changeset truncated at 500 lines:
diff --git a/source3/include/messages.h b/source3/include/messages.h
index cefa279..5666911 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;
};
@@ -98,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);
@@ -140,6 +140,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..b586b66 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"
@@ -231,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;
};
@@ -249,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);
@@ -273,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));
@@ -287,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;
}
@@ -366,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));
@@ -382,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;
}
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;
}
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.
****************************************************************************/
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;
- }
-
--
Samba Shared Repository
