Copyright (C) Andrew Tridgell 2000
Copyright (C) 2001 by Martin Pool
Copyright (C) 2002 by Jeremy Allison
-
+ Copyright (C) 2007 by Volker Lendecke
+
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 2 of the License, or
+ 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, write to the Free Software
- Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
/**
@defgroup messages Internal messaging framework
@{
@file messages.c
-
+
@brief Module for internal messaging between Samba daemons.
The idea is that if a part of Samba wants to do communication with
*/
#include "includes.h"
+#include "dbwrap/dbwrap.h"
+#include "serverid.h"
+#include "messages.h"
+#include "lib/util/tevent_unix.h"
+#include "lib/background.h"
-/* the locking database handle */
-static TDB_CONTEXT *tdb;
-static int received_signal;
-
-/* change the message version with any incompatible changes in the protocol */
-#define MESSAGE_VERSION 1
-
-struct message_rec {
- int msg_version;
- int msg_type;
- struct process_id dest;
- struct process_id src;
- size_t len;
-};
-
-/* we have a linked list of dispatch handlers */
-static struct dispatch_fns {
- struct dispatch_fns *next, *prev;
- int msg_type;
- void (*fn)(int msg_type, struct process_id pid, void *buf, size_t len,
- void *private_data);
+struct messaging_callback {
+ struct messaging_callback *prev, *next;
+ uint32 msg_type;
+ void (*fn)(struct messaging_context *msg, void *private_data,
+ uint32_t msg_type,
+ struct server_id server_id, DATA_BLOB *data);
void *private_data;
-} *dispatch_fns;
+};
/****************************************************************************
- Free global objects.
+ A useful function for testing the message system.
****************************************************************************/
-void gfree_messages(void)
+static void ping_message(struct messaging_context *msg_ctx,
+ void *private_data,
+ uint32_t msg_type,
+ struct server_id src,
+ DATA_BLOB *data)
{
- struct dispatch_fns *dfn, *next;
+ const char *msg = "none";
+ char *free_me = NULL;
- /* delete the dispatch_fns list */
- dfn = dispatch_fns;
- while( dfn ) {
- next = dfn->next;
- DLIST_REMOVE(dispatch_fns, dfn);
- SAFE_FREE(dfn);
- dfn = next;
+ if (data->data != NULL) {
+ free_me = talloc_strndup(talloc_tos(), (char *)data->data,
+ data->length);
+ msg = free_me;
}
+ DEBUG(1,("INFO: Received PING message from PID %s [%s]\n",
+ procid_str_static(&src), msg));
+ TALLOC_FREE(free_me);
+ messaging_send(msg_ctx, src, MSG_PONG, data);
}
/****************************************************************************
- Notifications come in as signals.
-****************************************************************************/
-
-static void sig_usr1(void)
-{
- received_signal = 1;
- sys_select_signal(SIGUSR1);
-}
-
-/****************************************************************************
- A useful function for testing the message system.
+ Register/replace a dispatch function for a particular message type.
+ JRA changed Dec 13 2006. Only one message handler now permitted per type.
+ *NOTE*: Dispatch functions must be able to cope with incoming
+ messages on an *odd* byte boundary.
****************************************************************************/
-static void ping_message(int msg_type, struct process_id src,
- void *buf, size_t len, void *private_data)
-{
- const char *msg = buf ? (const char *)buf : "none";
-
- DEBUG(1,("INFO: Received PING message from PID %s [%s]\n",
- procid_str_static(&src), msg));
- message_send_pid(src, MSG_PONG, buf, len, True);
-}
+struct msg_all {
+ struct messaging_context *msg_ctx;
+ int msg_type;
+ uint32 msg_flag;
+ const void *buf;
+ size_t len;
+ int n_sent;
+};
/****************************************************************************
- Initialise the messaging functions.
+ Send one of the messages for the broadcast.
****************************************************************************/
-BOOL message_init(void)
+static int traverse_fn(struct db_record *rec, const struct server_id *id,
+ uint32_t msg_flags, void *state)
{
- sec_init();
-
- if (tdb)
- return True;
+ struct msg_all *msg_all = (struct msg_all *)state;
+ NTSTATUS status;
- tdb = tdb_open_log(lock_path("messages.tdb"),
- 0, TDB_CLEAR_IF_FIRST|TDB_DEFAULT,
- O_RDWR|O_CREAT,0600);
+ /* Don't send if the receiver hasn't registered an interest. */
- if (!tdb) {
- DEBUG(0,("ERROR: Failed to initialise messages database\n"));
- return False;
+ if((msg_flags & msg_all->msg_flag) == 0) {
+ return 0;
}
- CatchSignal(SIGUSR1, SIGNAL_CAST sig_usr1);
+ /* If the msg send fails because the pid was not found (i.e. smbd died),
+ * the msg has already been deleted from the messages.tdb.*/
- message_register(MSG_PING, ping_message, NULL);
+ status = messaging_send_buf(msg_all->msg_ctx, *id, msg_all->msg_type,
+ (const uint8_t *)msg_all->buf, msg_all->len);
- /* Register some debugging related messages */
+ if (NT_STATUS_EQUAL(status, NT_STATUS_INVALID_HANDLE)) {
- register_msg_pool_usage();
- register_dmalloc_msgs();
+ /*
+ * If the pid was not found delete the entry from
+ * serverid.tdb
+ */
- return True;
-}
+ DEBUG(2, ("pid %s doesn't exist\n", procid_str_static(id)));
-/*******************************************************************
- Form a static tdb key from a pid.
-******************************************************************/
+ dbwrap_record_delete(rec);
+ }
+ msg_all->n_sent++;
+ return 0;
+}
-static TDB_DATA message_key_pid(struct process_id pid)
+/**
+ * Send a message to all smbd processes.
+ *
+ * It isn't very efficient, but should be OK for the sorts of
+ * applications that use it. When we need efficient broadcast we can add
+ * it.
+ *
+ * @param n_sent Set to the number of messages sent. This should be
+ * equal to the number of processes, but be careful for races.
+ *
+ * @retval True for success.
+ **/
+bool message_send_all(struct messaging_context *msg_ctx,
+ int msg_type,
+ const void *buf, size_t len,
+ int *n_sent)
{
- static char key[20];
- TDB_DATA kbuf;
-
- slprintf(key, sizeof(key)-1, "PID/%s", procid_str_static(&pid));
-
- kbuf.dptr = (char *)key;
- kbuf.dsize = strlen(key)+1;
- return kbuf;
-}
+ struct msg_all msg_all;
-/****************************************************************************
- Notify a process that it has a message. If the process doesn't exist
- then delete its record in the database.
-****************************************************************************/
+ msg_all.msg_type = msg_type;
+ if (msg_type < 0x100) {
+ msg_all.msg_flag = FLAG_MSG_GENERAL;
+ } else if (msg_type > 0x100 && msg_type < 0x200) {
+ msg_all.msg_flag = FLAG_MSG_NMBD;
+ } else if (msg_type > 0x200 && msg_type < 0x300) {
+ msg_all.msg_flag = FLAG_MSG_PRINT_GENERAL;
+ } else if (msg_type > 0x300 && msg_type < 0x400) {
+ msg_all.msg_flag = FLAG_MSG_SMBD;
+ } else if (msg_type > 0x400 && msg_type < 0x600) {
+ msg_all.msg_flag = FLAG_MSG_WINBIND;
+ } else if (msg_type > 4000 && msg_type < 5000) {
+ msg_all.msg_flag = FLAG_MSG_DBWRAP;
+ } else {
+ return false;
+ }
-static BOOL message_notify(struct process_id procid)
-{
- pid_t pid = procid.pid;
- int ret;
- uid_t euid = geteuid();
+ msg_all.buf = buf;
+ msg_all.len = len;
+ msg_all.n_sent = 0;
+ msg_all.msg_ctx = msg_ctx;
- /*
- * Doing kill with a non-positive pid causes messages to be
- * sent to places we don't want.
- */
+ serverid_traverse(traverse_fn, &msg_all);
+ if (n_sent)
+ *n_sent = msg_all.n_sent;
+ return true;
+}
- SMB_ASSERT(pid > 0);
+struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx,
+ struct tevent_context *ev)
+{
+ struct messaging_context *ctx;
+ NTSTATUS status;
- if (euid != 0) {
- become_root_uid_only();
+ if (!(ctx = talloc_zero(mem_ctx, struct messaging_context))) {
+ return NULL;
}
- ret = kill(pid, SIGUSR1);
+ ctx->id = procid_self();
+ ctx->event_ctx = ev;
- if (euid != 0) {
- unbecome_root_uid_only();
+ status = messaging_dgm_init(ctx, ctx, &ctx->local);
+
+ if (!NT_STATUS_IS_OK(status)) {
+ DEBUG(2, ("messaging_dgm_init failed: %s\n",
+ nt_errstr(status)));
+ TALLOC_FREE(ctx);
+ return NULL;
}
- if (ret == -1) {
- if (errno == ESRCH) {
- DEBUG(2,("pid %d doesn't exist - deleting messages record\n", (int)pid));
- tdb_delete(tdb, message_key_pid(procid));
- } else {
- DEBUG(2,("message to process %d failed - %s\n", (int)pid, strerror(errno)));
+ if (lp_clustering()) {
+ status = messaging_ctdbd_init(ctx, ctx, &ctx->remote);
+
+ if (!NT_STATUS_IS_OK(status)) {
+ DEBUG(2, ("messaging_ctdbd_init failed: %s\n",
+ nt_errstr(status)));
+ TALLOC_FREE(ctx);
+ return NULL;
}
- return False;
}
+ ctx->id.vnn = get_my_vnn();
- return True;
-}
+ messaging_register(ctx, NULL, MSG_PING, ping_message);
-/****************************************************************************
- Send a message to a particular pid.
-****************************************************************************/
+ /* Register some debugging related messages */
-static BOOL message_send_pid_internal(struct process_id pid, int msg_type,
- const void *buf, size_t len,
- BOOL duplicates_allowed,
- unsigned int timeout)
-{
- TDB_DATA kbuf;
- TDB_DATA dbuf;
- TDB_DATA old_dbuf;
- struct message_rec rec;
- char *ptr;
- struct message_rec prec;
+ register_msg_pool_usage(ctx);
+ register_dmalloc_msgs(ctx);
+ debug_register_msgs(ctx);
- /* NULL pointer means implicit length zero. */
- if (!buf) {
- SMB_ASSERT(len == 0);
- }
-
- /*
- * Doing kill with a non-positive pid causes messages to be
- * sent to places we don't want.
- */
+ return ctx;
+}
- SMB_ASSERT(procid_to_pid(&pid) > 0);
+struct server_id messaging_server_id(const struct messaging_context *msg_ctx)
+{
+ return msg_ctx->id;
+}
- rec.msg_version = MESSAGE_VERSION;
- rec.msg_type = msg_type;
- rec.dest = pid;
- rec.src = procid_self();
- rec.len = buf ? len : 0;
+/*
+ * re-init after a fork
+ */
+NTSTATUS messaging_reinit(struct messaging_context *msg_ctx)
+{
+ NTSTATUS status;
- kbuf = message_key_pid(pid);
+ TALLOC_FREE(msg_ctx->local);
- dbuf.dptr = (char *)SMB_MALLOC(len + sizeof(rec));
- if (!dbuf.dptr)
- return False;
+ msg_ctx->id = procid_self();
- memcpy(dbuf.dptr, &rec, sizeof(rec));
- if (len > 0 && buf)
- memcpy((void *)((char*)dbuf.dptr+sizeof(rec)), buf, len);
+ status = messaging_dgm_init(msg_ctx, msg_ctx, &msg_ctx->local);
+ if (!NT_STATUS_IS_OK(status)) {
+ DEBUG(0, ("messaging_dgm_init failed: %s\n",
+ nt_errstr(status)));
+ return status;
+ }
- dbuf.dsize = len + sizeof(rec);
+ TALLOC_FREE(msg_ctx->remote);
- if (duplicates_allowed) {
+ if (lp_clustering()) {
+ status = messaging_ctdbd_init(msg_ctx, msg_ctx,
+ &msg_ctx->remote);
- /* If duplicates are allowed we can just append the message and return. */
+ if (!NT_STATUS_IS_OK(status)) {
+ DEBUG(1, ("messaging_ctdbd_init failed: %s\n",
+ nt_errstr(status)));
+ return status;
+ }
+ }
- /* lock the record for the destination */
- if (timeout) {
- if (tdb_chainlock_with_timeout(tdb, kbuf, timeout) == -1) {
- DEBUG(0,("message_send_pid_internal: failed to get chainlock with timeout %ul.\n", timeout));
- return False;
- }
- } else {
- if (tdb_chainlock(tdb, kbuf) == -1) {
- DEBUG(0,("message_send_pid_internal: failed to get chainlock.\n"));
- return False;
- }
- }
- tdb_append(tdb, kbuf, dbuf);
- tdb_chainunlock(tdb, kbuf);
+ return NT_STATUS_OK;
+}
- SAFE_FREE(dbuf.dptr);
- errno = 0; /* paranoia */
- return message_notify(pid);
- }
- /* lock the record for the destination */
- if (timeout) {
- if (tdb_chainlock_with_timeout(tdb, kbuf, timeout) == -1) {
- DEBUG(0,("message_send_pid_internal: failed to get chainlock with timeout %ul.\n", timeout));
- return False;
- }
- } else {
- if (tdb_chainlock(tdb, kbuf) == -1) {
- DEBUG(0,("message_send_pid_internal: failed to get chainlock.\n"));
- return False;
- }
- }
+/*
+ * Register a dispatch function for a particular message type. Allow multiple
+ * registrants
+*/
+NTSTATUS messaging_register(struct messaging_context *msg_ctx,
+ void *private_data,
+ uint32_t msg_type,
+ void (*fn)(struct messaging_context *msg,
+ void *private_data,
+ uint32_t msg_type,
+ struct server_id server_id,
+ DATA_BLOB *data))
+{
+ struct messaging_callback *cb;
- old_dbuf = tdb_fetch(tdb, kbuf);
+ DEBUG(5, ("Registering messaging pointer for type %u - "
+ "private_data=%p\n",
+ (unsigned)msg_type, private_data));
- if (!old_dbuf.dptr) {
- /* its a new record */
+ /*
+ * Only one callback per type
+ */
- tdb_store(tdb, kbuf, dbuf, TDB_REPLACE);
- tdb_chainunlock(tdb, kbuf);
+ for (cb = msg_ctx->callbacks; cb != NULL; cb = cb->next) {
+ /* we allow a second registration of the same message
+ type if it has a different private pointer. This is
+ needed in, for example, the internal notify code,
+ which creates a new notify context for each tree
+ connect, and expects to receive messages to each of
+ them. */
+ if (cb->msg_type == msg_type && private_data == cb->private_data) {
+ DEBUG(5,("Overriding messaging pointer for type %u - private_data=%p\n",
+ (unsigned)msg_type, private_data));
+ cb->fn = fn;
+ cb->private_data = private_data;
+ return NT_STATUS_OK;
+ }
+ }
- SAFE_FREE(dbuf.dptr);
- errno = 0; /* paranoia */
- return message_notify(pid);
+ if (!(cb = talloc(msg_ctx, struct messaging_callback))) {
+ return NT_STATUS_NO_MEMORY;
}
- /* Not a new record. Check for duplicates. */
+ cb->msg_type = msg_type;
+ cb->fn = fn;
+ cb->private_data = private_data;
- for(ptr = (char *)old_dbuf.dptr; ptr < old_dbuf.dptr + old_dbuf.dsize; ) {
- /*
- * First check if the message header matches, then, if it's a non-zero
- * sized message, check if the data matches. If so it's a duplicate and
- * we can discard it. JRA.
- */
+ DLIST_ADD(msg_ctx->callbacks, cb);
+ return NT_STATUS_OK;
+}
- if (!memcmp(ptr, &rec, sizeof(rec))) {
- if (!len || (len && !memcmp( ptr + sizeof(rec), buf, len))) {
- tdb_chainunlock(tdb, kbuf);
- DEBUG(10,("message_send_pid_internal: discarding duplicate message.\n"));
- SAFE_FREE(dbuf.dptr);
- SAFE_FREE(old_dbuf.dptr);
- return True;
- }
+/*
+ De-register the function for a particular message type.
+*/
+void messaging_deregister(struct messaging_context *ctx, uint32_t msg_type,
+ void *private_data)
+{
+ struct messaging_callback *cb, *next;
+
+ for (cb = ctx->callbacks; cb; cb = next) {
+ next = cb->next;
+ if ((cb->msg_type == msg_type)
+ && (cb->private_data == private_data)) {
+ DEBUG(5,("Deregistering messaging pointer for type %u - private_data=%p\n",
+ (unsigned)msg_type, private_data));
+ DLIST_REMOVE(ctx->callbacks, cb);
+ TALLOC_FREE(cb);
}
- memcpy(&prec, ptr, sizeof(prec));
- ptr += sizeof(rec) + prec.len;
}
+}
- /* we're adding to an existing entry */
+static bool messaging_is_self_send(const struct messaging_context *msg_ctx,
+ const struct server_id *dst)
+{
+ return ((msg_ctx->id.vnn == dst->vnn) &&
+ (msg_ctx->id.pid == dst->pid));
+}
- tdb_append(tdb, kbuf, dbuf);
- tdb_chainunlock(tdb, kbuf);
+/*
+ Send a message to a particular server
+*/
+NTSTATUS messaging_send(struct messaging_context *msg_ctx,
+ struct server_id server, uint32_t msg_type,
+ const DATA_BLOB *data)
+{
+ struct iovec iov;
- SAFE_FREE(old_dbuf.dptr);
- SAFE_FREE(dbuf.dptr);
+ iov.iov_base = data->data;
+ iov.iov_len = data->length;
- errno = 0; /* paranoia */
- return message_notify(pid);
+ return messaging_send_iov(msg_ctx, server, msg_type, &iov, 1);
}
-/****************************************************************************
- Send a message to a particular pid - no timeout.
-****************************************************************************/
-
-BOOL message_send_pid(struct process_id pid, int msg_type, const void *buf, size_t len, BOOL duplicates_allowed)
+NTSTATUS messaging_send_buf(struct messaging_context *msg_ctx,
+ struct server_id server, uint32_t msg_type,
+ const uint8_t *buf, size_t len)
{
- return message_send_pid_internal(pid, msg_type, buf, len, duplicates_allowed, 0);
+ DATA_BLOB blob = data_blob_const(buf, len);
+ return messaging_send(msg_ctx, server, msg_type, &blob);
}
-/****************************************************************************
- Send a message to a particular pid, with timeout in seconds.
-****************************************************************************/
-
-BOOL message_send_pid_with_timeout(struct process_id pid, int msg_type, const void *buf, size_t len,
- BOOL duplicates_allowed, unsigned int timeout)
+NTSTATUS messaging_send_iov(struct messaging_context *msg_ctx,
+ struct server_id server, uint32_t msg_type,
+ const struct iovec *iov, int iovlen)
{
- return message_send_pid_internal(pid, msg_type, buf, len, duplicates_allowed, timeout);
-}
+ if (server_id_is_disconnected(&server)) {
+ return NT_STATUS_INVALID_PARAMETER_MIX;
+ }
-/****************************************************************************
- Count the messages pending for a particular pid. Expensive....
-****************************************************************************/
+ if (!procid_is_local(&server)) {
+ return msg_ctx->remote->send_fn(msg_ctx->id, server,
+ msg_type, iov, iovlen,
+ msg_ctx->remote);
+ }
-unsigned int messages_pending_for_pid(struct process_id pid)
-{
- TDB_DATA kbuf;
- TDB_DATA dbuf;
- char *buf;
- unsigned int message_count = 0;
+ if (messaging_is_self_send(msg_ctx, &server)) {
+ struct messaging_rec rec;
+ uint8_t *buf;
+ DATA_BLOB data;
- kbuf = message_key_pid(pid);
+ buf = iov_buf(talloc_tos(), iov, iovlen);
+ if (buf == NULL) {
+ return NT_STATUS_NO_MEMORY;
+ }
- dbuf = tdb_fetch(tdb, kbuf);
- if (dbuf.dptr == NULL || dbuf.dsize == 0) {
- SAFE_FREE(dbuf.dptr);
- return 0;
- }
+ data = data_blob_const(buf, talloc_get_size(buf));
- for (buf = dbuf.dptr; dbuf.dsize > sizeof(struct message_rec);) {
- struct message_rec rec;
- memcpy(&rec, buf, sizeof(rec));
- buf += (sizeof(rec) + rec.len);
- dbuf.dsize -= (sizeof(rec) + rec.len);
- message_count++;
+ 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);
+ TALLOC_FREE(buf);
+ return NT_STATUS_OK;
}
- SAFE_FREE(dbuf.dptr);
- return message_count;
+ return msg_ctx->local->send_fn(msg_ctx->id, server, msg_type,
+ iov, iovlen, msg_ctx->local);
}
-/****************************************************************************
- Retrieve all messages for the current process.
-****************************************************************************/
-
-static BOOL retrieve_all_messages(char **msgs_buf, size_t *total_len)
+static struct messaging_rec *messaging_rec_dup(TALLOC_CTX *mem_ctx,
+ struct messaging_rec *rec)
{
- TDB_DATA kbuf;
- TDB_DATA dbuf;
- TDB_DATA null_dbuf;
+ struct messaging_rec *result;
- ZERO_STRUCT(null_dbuf);
+ result = talloc_pooled_object(mem_ctx, struct messaging_rec,
+ 1, rec->buf.length);
+ if (result == NULL) {
+ return NULL;
+ }
+ *result = *rec;
- *msgs_buf = NULL;
- *total_len = 0;
+ /* Doesn't fail, see talloc_pooled_object */
- kbuf = message_key_pid(pid_to_procid(sys_getpid()));
+ result->buf.data = talloc_memdup(result, rec->buf.data,
+ rec->buf.length);
+ return result;
+}
- if (tdb_chainlock(tdb, kbuf) == -1)
- return False;
+struct messaging_filtered_read_state {
+ struct tevent_context *ev;
+ struct messaging_context *msg_ctx;
+ void *tevent_handle;
- dbuf = tdb_fetch(tdb, kbuf);
- /*
- * Replace with an empty record to keep the allocated
- * space in the tdb.
- */
- tdb_store(tdb, kbuf, null_dbuf, TDB_REPLACE);
- tdb_chainunlock(tdb, kbuf);
+ bool (*filter)(struct messaging_rec *rec, void *private_data);
+ void *private_data;
- if (dbuf.dptr == NULL || dbuf.dsize == 0) {
- SAFE_FREE(dbuf.dptr);
- return False;
- }
+ struct messaging_rec *rec;
+};
- *msgs_buf = dbuf.dptr;
- *total_len = dbuf.dsize;
+static void messaging_filtered_read_cleanup(struct tevent_req *req,
+ enum tevent_req_state req_state);
- return True;
-}
+struct tevent_req *messaging_filtered_read_send(
+ TALLOC_CTX *mem_ctx, struct tevent_context *ev,
+ struct messaging_context *msg_ctx,
+ bool (*filter)(struct messaging_rec *rec, void *private_data),
+ void *private_data)
+{
+ struct tevent_req *req;
+ struct messaging_filtered_read_state *state;
+ size_t new_waiters_len;
-/****************************************************************************
- Parse out the next message for the current process.
-****************************************************************************/
+ req = tevent_req_create(mem_ctx, &state,
+ struct messaging_filtered_read_state);
+ if (req == NULL) {
+ return NULL;
+ }
+ state->ev = ev;
+ state->msg_ctx = msg_ctx;
+ state->filter = filter;
+ state->private_data = private_data;
-static BOOL message_recv(char *msgs_buf, size_t total_len, int *msg_type,
- struct process_id *src, char **buf, size_t *len)
-{
- struct message_rec rec;
- char *ret_buf = *buf;
+ /*
+ * We have to defer the callback here, as we might be called from
+ * within a different tevent_context than state->ev
+ */
+ tevent_req_defer_callback(req, state->ev);
- *buf = NULL;
- *len = 0;
+ state->tevent_handle = messaging_dgm_register_tevent_context(
+ state, msg_ctx, ev);
+ if (tevent_req_nomem(state, req)) {
+ return tevent_req_post(req, ev);
+ }
- if (total_len - (ret_buf - msgs_buf) < sizeof(rec))
- return False;
+ /*
+ * We add ourselves to the "new_waiters" array, not the "waiters"
+ * array. If we are called from within messaging_read_done,
+ * messaging_dispatch_rec will be in an active for-loop on
+ * "waiters". We must be careful not to mess with this array, because
+ * it could mean that a single event is being delivered twice.
+ */
- memcpy(&rec, ret_buf, sizeof(rec));
- ret_buf += sizeof(rec);
+ new_waiters_len = talloc_array_length(msg_ctx->new_waiters);
- if (rec.msg_version != MESSAGE_VERSION) {
- DEBUG(0,("message version %d received (expected %d)\n", rec.msg_version, MESSAGE_VERSION));
- return False;
- }
+ if (new_waiters_len == msg_ctx->num_new_waiters) {
+ struct tevent_req **tmp;
- if (rec.len > 0) {
- if (total_len - (ret_buf - msgs_buf) < rec.len)
- return False;
+ tmp = talloc_realloc(msg_ctx, msg_ctx->new_waiters,
+ struct tevent_req *, new_waiters_len+1);
+ if (tevent_req_nomem(tmp, req)) {
+ return tevent_req_post(req, ev);
+ }
+ msg_ctx->new_waiters = tmp;
}
- *len = rec.len;
- *msg_type = rec.msg_type;
- *src = rec.src;
- *buf = ret_buf;
+ msg_ctx->new_waiters[msg_ctx->num_new_waiters] = req;
+ msg_ctx->num_new_waiters += 1;
+ tevent_req_set_cleanup_fn(req, messaging_filtered_read_cleanup);
- return True;
+ return req;
}
-/****************************************************************************
- Receive and dispatch any messages pending for this process.
- JRA changed Dec 13 2006. Only one message handler now permitted per type.
- *NOTE*: Dispatch functions must be able to cope with incoming
- messages on an *odd* byte boundary.
-****************************************************************************/
-
-void message_dispatch(void)
+static void messaging_filtered_read_cleanup(struct tevent_req *req,
+ enum tevent_req_state req_state)
{
- int msg_type;
- struct process_id src;
- char *buf;
- char *msgs_buf;
- size_t len, total_len;
- int n_handled;
-
- if (!received_signal)
- return;
+ struct messaging_filtered_read_state *state = tevent_req_data(
+ req, struct messaging_filtered_read_state);
+ struct messaging_context *msg_ctx = state->msg_ctx;
+ unsigned i;
- DEBUG(10,("message_dispatch: received_signal = %d\n", received_signal));
+ tevent_req_set_cleanup_fn(req, NULL);
- received_signal = 0;
+ TALLOC_FREE(state->tevent_handle);
- if (!retrieve_all_messages(&msgs_buf, &total_len))
- return;
+ /*
+ * Just set the [new_]waiters entry to NULL, be careful not to mess
+ * with the other "waiters" array contents. We are often called from
+ * within "messaging_dispatch_rec", which loops over
+ * "waiters". Messing with the "waiters" array will mess up that
+ * for-loop.
+ */
- for (buf = msgs_buf; message_recv(msgs_buf, total_len, &msg_type, &src, &buf, &len); buf += len) {
- struct dispatch_fns *dfn;
-
- DEBUG(10,("message_dispatch: received msg_type=%d "
- "src_pid=%u\n", msg_type,
- (unsigned int) procid_to_pid(&src)));
-
- n_handled = 0;
- for (dfn = dispatch_fns; dfn; dfn = dfn->next) {
- if (dfn->msg_type == msg_type) {
- DEBUG(10,("message_dispatch: processing message of type %d.\n", msg_type));
- dfn->fn(msg_type, src,
- len ? (void *)buf : NULL, len,
- dfn->private_data);
- n_handled++;
- break;
- }
- }
- if (!n_handled) {
- DEBUG(5,("message_dispatch: warning: no handler registed for "
- "msg_type %d in pid %u\n",
- msg_type, (unsigned int)sys_getpid()));
+ for (i=0; i<msg_ctx->num_waiters; i++) {
+ if (msg_ctx->waiters[i] == req) {
+ msg_ctx->waiters[i] = NULL;
+ return;
}
}
- SAFE_FREE(msgs_buf);
-}
-
-/****************************************************************************
- Register/replace a dispatch function for a particular message type.
- JRA changed Dec 13 2006. Only one message handler now permitted per type.
- *NOTE*: Dispatch functions must be able to cope with incoming
- messages on an *odd* byte boundary.
-****************************************************************************/
-
-void message_register(int msg_type,
- void (*fn)(int msg_type, struct process_id pid,
- void *buf, size_t len,
- void *private_data),
- void *private_data)
-{
- struct dispatch_fns *dfn;
- for (dfn = dispatch_fns; dfn; dfn = dfn->next) {
- if (dfn->msg_type == msg_type) {
- dfn->fn = fn;
+ for (i=0; i<msg_ctx->num_new_waiters; i++) {
+ if (msg_ctx->new_waiters[i] == req) {
+ msg_ctx->new_waiters[i] = NULL;
return;
}
}
+}
- dfn = SMB_MALLOC_P(struct dispatch_fns);
-
- if (dfn != NULL) {
-
- ZERO_STRUCTPN(dfn);
-
- dfn->msg_type = msg_type;
- dfn->fn = fn;
- dfn->private_data = private_data;
+static void messaging_filtered_read_done(struct tevent_req *req,
+ struct messaging_rec *rec)
+{
+ struct messaging_filtered_read_state *state = tevent_req_data(
+ req, struct messaging_filtered_read_state);
- DLIST_ADD(dispatch_fns, dfn);
- }
- else {
-
- DEBUG(0,("message_register: Not enough memory. malloc failed!\n"));
+ state->rec = messaging_rec_dup(state, rec);
+ if (tevent_req_nomem(state->rec, req)) {
+ return;
}
+ tevent_req_done(req);
}
-/****************************************************************************
- De-register the function for a particular message type.
-****************************************************************************/
-
-void message_deregister(int msg_type)
+int messaging_filtered_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
+ struct messaging_rec **presult)
{
- struct dispatch_fns *dfn, *next;
+ struct messaging_filtered_read_state *state = tevent_req_data(
+ req, struct messaging_filtered_read_state);
+ int err;
- for (dfn = dispatch_fns; dfn; dfn = next) {
- next = dfn->next;
- if (dfn->msg_type == msg_type) {
- DLIST_REMOVE(dispatch_fns, dfn);
- SAFE_FREE(dfn);
- return;
- }
- }
+ if (tevent_req_is_unix_error(req, &err)) {
+ tevent_req_received(req);
+ return err;
+ }
+ *presult = talloc_move(mem_ctx, &state->rec);
+ return 0;
}
-struct msg_all {
- int msg_type;
- uint32 msg_flag;
- const void *buf;
- size_t len;
- BOOL duplicates;
- int n_sent;
+struct messaging_read_state {
+ uint32_t msg_type;
+ struct messaging_rec *rec;
};
-/****************************************************************************
- Send one of the messages for the broadcast.
-****************************************************************************/
+static bool messaging_read_filter(struct messaging_rec *rec,
+ void *private_data);
+static void messaging_read_done(struct tevent_req *subreq);
-static int traverse_fn(TDB_CONTEXT *the_tdb, TDB_DATA kbuf, TDB_DATA dbuf, void *state)
+struct tevent_req *messaging_read_send(TALLOC_CTX *mem_ctx,
+ struct tevent_context *ev,
+ struct messaging_context *msg,
+ uint32_t msg_type)
{
- struct connections_data crec;
- struct msg_all *msg_all = (struct msg_all *)state;
+ struct tevent_req *req, *subreq;
+ struct messaging_read_state *state;
- if (dbuf.dsize != sizeof(crec))
- return 0;
-
- memcpy(&crec, dbuf.dptr, sizeof(crec));
+ req = tevent_req_create(mem_ctx, &state,
+ struct messaging_read_state);
+ if (req == NULL) {
+ return NULL;
+ }
+ state->msg_type = msg_type;
- if (crec.cnum != -1)
- return 0;
+ subreq = messaging_filtered_read_send(state, ev, msg,
+ messaging_read_filter, state);
+ if (tevent_req_nomem(subreq, req)) {
+ return tevent_req_post(req, ev);
+ }
+ tevent_req_set_callback(subreq, messaging_read_done, req);
+ return req;
+}
- /* Don't send if the receiver hasn't registered an interest. */
+static bool messaging_read_filter(struct messaging_rec *rec,
+ void *private_data)
+{
+ struct messaging_read_state *state = talloc_get_type_abort(
+ private_data, struct messaging_read_state);
- if(!(crec.bcast_msg_flags & msg_all->msg_flag))
- return 0;
+ return rec->msg_type == state->msg_type;
+}
- /* If the msg send fails because the pid was not found (i.e. smbd died),
- * the msg has already been deleted from the messages.tdb.*/
+static void messaging_read_done(struct tevent_req *subreq)
+{
+ struct tevent_req *req = tevent_req_callback_data(
+ subreq, struct tevent_req);
+ struct messaging_read_state *state = tevent_req_data(
+ req, struct messaging_read_state);
+ int ret;
- if (!message_send_pid(crec.pid, msg_all->msg_type,
- msg_all->buf, msg_all->len,
- msg_all->duplicates)) {
-
- /* If the pid was not found delete the entry from connections.tdb */
-
- if (errno == ESRCH) {
- DEBUG(2,("pid %s doesn't exist - deleting connections %d [%s]\n",
- procid_str_static(&crec.pid),
- crec.cnum, crec.name));
- tdb_delete(the_tdb, kbuf);
- }
+ ret = messaging_filtered_read_recv(subreq, state, &state->rec);
+ TALLOC_FREE(subreq);
+ if (tevent_req_error(req, ret)) {
+ return;
}
- msg_all->n_sent++;
- return 0;
+ tevent_req_done(req);
}
-/**
- * Send a message to all smbd processes.
- *
- * It isn't very efficient, but should be OK for the sorts of
- * applications that use it. When we need efficient broadcast we can add
- * it.
- *
- * @param n_sent Set to the number of messages sent. This should be
- * equal to the number of processes, but be careful for races.
- *
- * @retval True for success.
- **/
-BOOL message_send_all(TDB_CONTEXT *conn_tdb, int msg_type,
- const void *buf, size_t len,
- BOOL duplicates_allowed,
- int *n_sent)
+int messaging_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
+ struct messaging_rec **presult)
{
- struct msg_all msg_all;
+ struct messaging_read_state *state = tevent_req_data(
+ req, struct messaging_read_state);
+ int err;
- msg_all.msg_type = msg_type;
- if (msg_type < 1000)
- msg_all.msg_flag = FLAG_MSG_GENERAL;
- else if (msg_type > 1000 && msg_type < 2000)
- msg_all.msg_flag = FLAG_MSG_NMBD;
- else if (msg_type > 2000 && msg_type < 2100)
- msg_all.msg_flag = FLAG_MSG_PRINT_NOTIFY;
- else if (msg_type > 2100 && msg_type < 3000)
- msg_all.msg_flag = FLAG_MSG_PRINT_GENERAL;
- else if (msg_type > 3000 && msg_type < 4000)
- msg_all.msg_flag = FLAG_MSG_SMBD;
- else
- return False;
+ if (tevent_req_is_unix_error(req, &err)) {
+ return err;
+ }
+ if (presult != NULL) {
+ *presult = talloc_move(mem_ctx, &state->rec);
+ }
+ return 0;
+}
- msg_all.buf = buf;
- msg_all.len = len;
- msg_all.duplicates = duplicates_allowed;
- msg_all.n_sent = 0;
+static bool messaging_append_new_waiters(struct messaging_context *msg_ctx)
+{
+ if (msg_ctx->num_new_waiters == 0) {
+ return true;
+ }
- tdb_traverse(conn_tdb, traverse_fn, &msg_all);
- if (n_sent)
- *n_sent = msg_all.n_sent;
- return True;
-}
+ if (talloc_array_length(msg_ctx->waiters) <
+ (msg_ctx->num_waiters + msg_ctx->num_new_waiters)) {
+ struct tevent_req **tmp;
+ tmp = talloc_realloc(
+ msg_ctx, msg_ctx->waiters, struct tevent_req *,
+ msg_ctx->num_waiters + msg_ctx->num_new_waiters);
+ if (tmp == NULL) {
+ DEBUG(1, ("%s: talloc failed\n", __func__));
+ return false;
+ }
+ msg_ctx->waiters = tmp;
+ }
-/*
- * Block and unblock receiving of messages. Allows removal of race conditions
- * when doing a fork and changing message disposition.
- */
+ memcpy(&msg_ctx->waiters[msg_ctx->num_waiters], msg_ctx->new_waiters,
+ sizeof(struct tevent_req *) * msg_ctx->num_new_waiters);
-void message_block(void)
-{
- BlockSignals(True, SIGUSR1);
-}
+ msg_ctx->num_waiters += msg_ctx->num_new_waiters;
+ msg_ctx->num_new_waiters = 0;
-void message_unblock(void)
-{
- BlockSignals(False, SIGUSR1);
+ return true;
}
-/*
- * Samba4 API wrapper around the Samba3 implementation. Yes, I know, we could
- * import the whole Samba4 thing, but I want notify.c from Samba4 in first.
- */
-
-struct messaging_callback {
- struct messaging_callback *prev, *next;
- uint32 msg_type;
- void (*fn)(struct messaging_context *msg, void *private_data,
- uint32_t msg_type,
- struct server_id server_id, DATA_BLOB *data);
+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;
};
-struct messaging_context {
- struct server_id id;
- struct messaging_callback *callbacks;
-};
+static void messaging_defer_callback_trigger(struct tevent_context *ev,
+ struct tevent_immediate *im,
+ void *private_data);
-static int messaging_context_destructor(struct messaging_context *ctx)
+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_callback *cb;
+ struct messaging_defer_callback_state *state;
+ struct tevent_immediate *im;
- for (cb = ctx->callbacks; cb; cb = cb->next) {
- /*
- * We unconditionally remove all instances of our callback
- * from the tdb basis.
- */
- message_deregister(cb->msg_type);
+ state = talloc(msg_ctx, struct messaging_defer_callback_state);
+ if (state == NULL) {
+ DEBUG(1, ("talloc failed\n"));
+ return;
}
- return 0;
+ 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);
}
-struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx,
- struct server_id server_id,
- struct event_context *ev)
+static void messaging_defer_callback_trigger(struct tevent_context *ev,
+ struct tevent_immediate *im,
+ void *private_data)
{
- struct messaging_context *ctx;
-
- if (!(ctx = TALLOC_ZERO_P(mem_ctx, struct messaging_context))) {
- return NULL;
- }
+ struct messaging_defer_callback_state *state = talloc_get_type_abort(
+ private_data, struct messaging_defer_callback_state);
+ struct messaging_rec *rec = state->rec;
- ctx->id = server_id;
- talloc_set_destructor(ctx, messaging_context_destructor);
- return ctx;
+ state->fn(state->msg_ctx, state->private_data, rec->msg_type, rec->src,
+ &rec->buf);
+ TALLOC_FREE(state);
}
-static void messaging_callback(int msg_type, struct process_id pid,
- void *buf, size_t len, void *private_data)
+/*
+ Dispatch one messaging_rec
+*/
+void messaging_dispatch_rec(struct messaging_context *msg_ctx,
+ struct messaging_rec *rec)
{
- struct messaging_context *ctx = talloc_get_type_abort(
- private_data, struct messaging_context);
struct messaging_callback *cb, *next;
+ unsigned i;
- for (cb = ctx->callbacks; cb; cb = next) {
+ for (cb = msg_ctx->callbacks; cb != NULL; cb = next) {
+ next = cb->next;
+ if (cb->msg_type != rec->msg_type) {
+ continue;
+ }
+
+ if (messaging_is_self_send(msg_ctx, &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);
+ }
/*
- * Allow a callback to remove itself
+ * 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
*/
- next = cb->next;
+ }
- if (msg_type == cb->msg_type) {
- DATA_BLOB blob;
- struct server_id id;
+ if (!messaging_append_new_waiters(msg_ctx)) {
+ return;
+ }
- blob.data = (uint8 *)buf;
- blob.length = len;
- id.id = pid;
+ i = 0;
+ while (i < msg_ctx->num_waiters) {
+ struct tevent_req *req;
+ struct messaging_filtered_read_state *state;
+
+ req = msg_ctx->waiters[i];
+ if (req == NULL) {
+ /*
+ * This got cleaned up. In the meantime,
+ * move everything down one. We need
+ * to keep the order of waiters, as
+ * other code may depend on this.
+ */
+ if (i < msg_ctx->num_waiters - 1) {
+ memmove(&msg_ctx->waiters[i],
+ &msg_ctx->waiters[i+1],
+ sizeof(struct tevent_req *) *
+ (msg_ctx->num_waiters - i - 1));
+ }
+ msg_ctx->num_waiters -= 1;
+ continue;
+ }
- cb->fn(ctx, cb->private_data, msg_type, id, &blob);
+ state = tevent_req_data(
+ req, struct messaging_filtered_read_state);
+ if (state->filter(rec, state->private_data)) {
+ messaging_filtered_read_done(req, rec);
}
+
+ i += 1;
}
}
-/*
- * Register a dispatch function for a particular message type. Allow multiple
- * registrants
-*/
-NTSTATUS messaging_register(struct messaging_context *ctx, void *private_data,
- uint32_t msg_type,
- void (*fn)(struct messaging_context *msg,
- void *private_data,
- uint32_t msg_type,
- struct server_id server_id,
- DATA_BLOB *data))
-{
- struct messaging_callback *cb;
+static int mess_parent_dgm_cleanup(void *private_data);
+static void mess_parent_dgm_cleanup_done(struct tevent_req *req);
- if (!(cb = talloc(ctx, struct messaging_callback))) {
- return NT_STATUS_NO_MEMORY;
+bool messaging_parent_dgm_cleanup_init(struct messaging_context *msg)
+{
+ struct tevent_req *req;
+
+ req = background_job_send(
+ msg, msg->event_ctx, msg, NULL, 0,
+ lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
+ 60*15),
+ mess_parent_dgm_cleanup, msg);
+ if (req == NULL) {
+ return false;
}
-
- cb->msg_type = msg_type;
- cb->fn = fn;
- cb->private_data = private_data;
-
- DLIST_ADD(ctx->callbacks, cb);
- message_register(msg_type, messaging_callback, ctx);
- return NT_STATUS_OK;
+ tevent_req_set_callback(req, mess_parent_dgm_cleanup_done, msg);
+ return true;
}
-/*
- De-register the function for a particular message type.
-*/
-void messaging_deregister(struct messaging_context *ctx, uint32_t msg_type,
- void *private_data)
+static int mess_parent_dgm_cleanup(void *private_data)
{
- struct messaging_callback *cb, *next;
+ struct messaging_context *msg_ctx = talloc_get_type_abort(
+ private_data, struct messaging_context);
+ NTSTATUS status;
- for (cb = ctx->callbacks; cb; cb = next) {
- next = cb->next;
- if ((cb->msg_type == msg_type)
- && (cb->private_data == private_data)) {
- DLIST_REMOVE(ctx->callbacks, cb);
- TALLOC_FREE(cb);
- }
- }
+ status = messaging_dgm_wipe(msg_ctx);
+ DEBUG(10, ("messaging_dgm_wipe returned %s\n", nt_errstr(status)));
+ return lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
+ 60*15);
}
-/*
- Send a message to a particular server
-*/
-NTSTATUS messaging_send(struct messaging_context *msg,
- struct server_id server,
- uint32_t msg_type, DATA_BLOB *data)
+static void mess_parent_dgm_cleanup_done(struct tevent_req *req)
{
- return message_send_pid_internal(server.id, msg_type, data->data,
- data->length, True, 0)
- ? NT_STATUS_OK : NT_STATUS_ACCESS_DENIED;
+ struct messaging_context *msg = tevent_req_callback_data(
+ req, struct messaging_context);
+ NTSTATUS status;
+
+ status = background_job_recv(req);
+ TALLOC_FREE(req);
+ DEBUG(1, ("messaging dgm cleanup job ended with %s\n",
+ nt_errstr(status)));
+
+ req = background_job_send(
+ msg, msg->event_ctx, msg, NULL, 0,
+ lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
+ 60*15),
+ mess_parent_dgm_cleanup, msg);
+ if (req == NULL) {
+ DEBUG(1, ("background_job_send failed\n"));
+ }
+ tevent_req_set_callback(req, mess_parent_dgm_cleanup_done, msg);
}
/** @} **/
git.samba.org / samba.git / blobdiff