2 Unix SMB/CIFS implementation.
3 Samba internal messaging functions
4 Copyright (C) Andrew Tridgell 2000
5 Copyright (C) 2001 by Martin Pool
6 Copyright (C) 2002 by Jeremy Allison
7 Copyright (C) 2007 by Volker Lendecke
9 This program is free software; you can redistribute it and/or modify
10 it under the terms of the GNU General Public License as published by
11 the Free Software Foundation; either version 3 of the License, or
12 (at your option) any later version.
14 This program is distributed in the hope that it will be useful,
15 but WITHOUT ANY WARRANTY; without even the implied warranty of
16 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17 GNU General Public License for more details.
19 You should have received a copy of the GNU General Public License
20 along with this program. If not, see <http://www.gnu.org/licenses/>.
24 @defgroup messages Internal messaging framework
28 @brief Module for internal messaging between Samba daemons.
30 The idea is that if a part of Samba wants to do communication with
31 another Samba process then it will do a message_register() of a
32 dispatch function, and use message_send_pid() to send messages to
35 The dispatch function is given the pid of the sender, and it can
36 use that to reply by message_send_pid(). See ping_message() for a
39 @caution Dispatch functions must be able to cope with incoming
40 messages on an *odd* byte boundary.
42 This system doesn't have any inherent size limitations but is not
43 very efficient for large messages or when messages are sent in very
49 #include "dbwrap/dbwrap.h"
52 #include "lib/util/tevent_unix.h"
53 #include "lib/background.h"
55 struct messaging_callback {
56 struct messaging_callback *prev, *next;
58 void (*fn)(struct messaging_context *msg, void *private_data,
60 struct server_id server_id, DATA_BLOB *data);
64 /****************************************************************************
65 A useful function for testing the message system.
66 ****************************************************************************/
68 static void ping_message(struct messaging_context *msg_ctx,
74 const char *msg = "none";
77 if (data->data != NULL) {
78 free_me = talloc_strndup(talloc_tos(), (char *)data->data,
82 DEBUG(1,("INFO: Received PING message from PID %s [%s]\n",
83 procid_str_static(&src), msg));
85 messaging_send(msg_ctx, src, MSG_PONG, data);
88 /****************************************************************************
89 Register/replace a dispatch function for a particular message type.
90 JRA changed Dec 13 2006. Only one message handler now permitted per type.
91 *NOTE*: Dispatch functions must be able to cope with incoming
92 messages on an *odd* byte boundary.
93 ****************************************************************************/
96 struct messaging_context *msg_ctx;
104 /****************************************************************************
105 Send one of the messages for the broadcast.
106 ****************************************************************************/
108 static int traverse_fn(struct db_record *rec, const struct server_id *id,
109 uint32_t msg_flags, void *state)
111 struct msg_all *msg_all = (struct msg_all *)state;
114 /* Don't send if the receiver hasn't registered an interest. */
116 if((msg_flags & msg_all->msg_flag) == 0) {
120 /* If the msg send fails because the pid was not found (i.e. smbd died),
121 * the msg has already been deleted from the messages.tdb.*/
123 status = messaging_send_buf(msg_all->msg_ctx, *id, msg_all->msg_type,
124 (const uint8_t *)msg_all->buf, msg_all->len);
126 if (NT_STATUS_EQUAL(status, NT_STATUS_INVALID_HANDLE)) {
129 * If the pid was not found delete the entry from
133 DEBUG(2, ("pid %s doesn't exist\n", procid_str_static(id)));
135 dbwrap_record_delete(rec);
142 * Send a message to all smbd processes.
144 * It isn't very efficient, but should be OK for the sorts of
145 * applications that use it. When we need efficient broadcast we can add
148 * @param n_sent Set to the number of messages sent. This should be
149 * equal to the number of processes, but be careful for races.
151 * @retval True for success.
153 bool message_send_all(struct messaging_context *msg_ctx,
155 const void *buf, size_t len,
158 struct msg_all msg_all;
160 msg_all.msg_type = msg_type;
161 if (msg_type < 0x100) {
162 msg_all.msg_flag = FLAG_MSG_GENERAL;
163 } else if (msg_type > 0x100 && msg_type < 0x200) {
164 msg_all.msg_flag = FLAG_MSG_NMBD;
165 } else if (msg_type > 0x200 && msg_type < 0x300) {
166 msg_all.msg_flag = FLAG_MSG_PRINT_GENERAL;
167 } else if (msg_type > 0x300 && msg_type < 0x400) {
168 msg_all.msg_flag = FLAG_MSG_SMBD;
169 } else if (msg_type > 0x400 && msg_type < 0x600) {
170 msg_all.msg_flag = FLAG_MSG_WINBIND;
171 } else if (msg_type > 4000 && msg_type < 5000) {
172 msg_all.msg_flag = FLAG_MSG_DBWRAP;
180 msg_all.msg_ctx = msg_ctx;
182 serverid_traverse(traverse_fn, &msg_all);
184 *n_sent = msg_all.n_sent;
188 struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx,
189 struct tevent_context *ev)
191 struct messaging_context *ctx;
194 if (!(ctx = talloc_zero(mem_ctx, struct messaging_context))) {
198 ctx->id = procid_self();
201 status = messaging_dgm_init(ctx, ctx, &ctx->local);
203 if (!NT_STATUS_IS_OK(status)) {
204 DEBUG(2, ("messaging_dgm_init failed: %s\n",
210 if (lp_clustering()) {
211 status = messaging_ctdbd_init(ctx, ctx, &ctx->remote);
213 if (!NT_STATUS_IS_OK(status)) {
214 DEBUG(2, ("messaging_ctdbd_init failed: %s\n",
220 ctx->id.vnn = get_my_vnn();
222 messaging_register(ctx, NULL, MSG_PING, ping_message);
224 /* Register some debugging related messages */
226 register_msg_pool_usage(ctx);
227 register_dmalloc_msgs(ctx);
228 debug_register_msgs(ctx);
233 struct server_id messaging_server_id(const struct messaging_context *msg_ctx)
239 * re-init after a fork
241 NTSTATUS messaging_reinit(struct messaging_context *msg_ctx)
245 TALLOC_FREE(msg_ctx->local);
247 msg_ctx->id = procid_self();
249 status = messaging_dgm_init(msg_ctx, msg_ctx, &msg_ctx->local);
250 if (!NT_STATUS_IS_OK(status)) {
251 DEBUG(0, ("messaging_dgm_init failed: %s\n",
256 TALLOC_FREE(msg_ctx->remote);
258 if (lp_clustering()) {
259 status = messaging_ctdbd_init(msg_ctx, msg_ctx,
262 if (!NT_STATUS_IS_OK(status)) {
263 DEBUG(1, ("messaging_ctdbd_init failed: %s\n",
274 * Register a dispatch function for a particular message type. Allow multiple
277 NTSTATUS messaging_register(struct messaging_context *msg_ctx,
280 void (*fn)(struct messaging_context *msg,
283 struct server_id server_id,
286 struct messaging_callback *cb;
288 DEBUG(5, ("Registering messaging pointer for type %u - "
290 (unsigned)msg_type, private_data));
293 * Only one callback per type
296 for (cb = msg_ctx->callbacks; cb != NULL; cb = cb->next) {
297 /* we allow a second registration of the same message
298 type if it has a different private pointer. This is
299 needed in, for example, the internal notify code,
300 which creates a new notify context for each tree
301 connect, and expects to receive messages to each of
303 if (cb->msg_type == msg_type && private_data == cb->private_data) {
304 DEBUG(5,("Overriding messaging pointer for type %u - private_data=%p\n",
305 (unsigned)msg_type, private_data));
307 cb->private_data = private_data;
312 if (!(cb = talloc(msg_ctx, struct messaging_callback))) {
313 return NT_STATUS_NO_MEMORY;
316 cb->msg_type = msg_type;
318 cb->private_data = private_data;
320 DLIST_ADD(msg_ctx->callbacks, cb);
325 De-register the function for a particular message type.
327 void messaging_deregister(struct messaging_context *ctx, uint32_t msg_type,
330 struct messaging_callback *cb, *next;
332 for (cb = ctx->callbacks; cb; cb = next) {
334 if ((cb->msg_type == msg_type)
335 && (cb->private_data == private_data)) {
336 DEBUG(5,("Deregistering messaging pointer for type %u - private_data=%p\n",
337 (unsigned)msg_type, private_data));
338 DLIST_REMOVE(ctx->callbacks, cb);
344 struct messaging_selfsend_state {
345 struct messaging_context *msg;
346 struct messaging_rec rec;
349 static void messaging_trigger_self(struct tevent_context *ev,
350 struct tevent_immediate *im,
354 Send a message to a particular server
356 NTSTATUS messaging_send(struct messaging_context *msg_ctx,
357 struct server_id server, uint32_t msg_type,
358 const DATA_BLOB *data)
360 if (server_id_is_disconnected(&server)) {
361 return NT_STATUS_INVALID_PARAMETER_MIX;
364 if (!procid_is_local(&server)) {
365 return msg_ctx->remote->send_fn(msg_ctx, server,
370 if (server_id_equal(&msg_ctx->id, &server)) {
371 struct messaging_selfsend_state *state;
372 struct tevent_immediate *im;
374 state = talloc_pooled_object(
375 msg_ctx, struct messaging_selfsend_state,
378 return NT_STATUS_NO_MEMORY;
380 state->msg = msg_ctx;
381 state->rec.msg_version = MESSAGE_VERSION;
382 state->rec.msg_type = msg_type & MSG_TYPE_MASK;
383 state->rec.dest = server;
384 state->rec.src = msg_ctx->id;
386 /* Can't fail, it's a pooled_object */
387 state->rec.buf = data_blob_talloc(
388 state, data->data, data->length);
390 im = tevent_create_immediate(state);
393 return NT_STATUS_NO_MEMORY;
396 tevent_schedule_immediate(im, msg_ctx->event_ctx,
397 messaging_trigger_self, state);
401 return msg_ctx->local->send_fn(msg_ctx, server, msg_type, data,
405 static void messaging_trigger_self(struct tevent_context *ev,
406 struct tevent_immediate *im,
409 struct messaging_selfsend_state *state = talloc_get_type_abort(
410 private_data, struct messaging_selfsend_state);
411 messaging_dispatch_rec(state->msg, &state->rec);
415 NTSTATUS messaging_send_buf(struct messaging_context *msg_ctx,
416 struct server_id server, uint32_t msg_type,
417 const uint8_t *buf, size_t len)
419 DATA_BLOB blob = data_blob_const(buf, len);
420 return messaging_send(msg_ctx, server, msg_type, &blob);
423 NTSTATUS messaging_send_iov(struct messaging_context *msg_ctx,
424 struct server_id server, uint32_t msg_type,
425 const struct iovec *iov, int iovlen)
430 buf = iov_buf(talloc_tos(), iov, iovlen);
432 return NT_STATUS_NO_MEMORY;
435 status = messaging_send_buf(msg_ctx, server, msg_type,
436 buf, talloc_get_size(buf));
442 static struct messaging_rec *messaging_rec_dup(TALLOC_CTX *mem_ctx,
443 struct messaging_rec *rec)
445 struct messaging_rec *result;
447 result = talloc_pooled_object(mem_ctx, struct messaging_rec,
449 if (result == NULL) {
454 /* Doesn't fail, see talloc_pooled_object */
456 result->buf.data = talloc_memdup(result, rec->buf.data,
461 struct messaging_read_state {
462 struct tevent_context *ev;
463 struct messaging_context *msg_ctx;
465 struct messaging_rec *rec;
468 static void messaging_read_cleanup(struct tevent_req *req,
469 enum tevent_req_state req_state);
471 struct tevent_req *messaging_read_send(TALLOC_CTX *mem_ctx,
472 struct tevent_context *ev,
473 struct messaging_context *msg_ctx,
476 struct tevent_req *req;
477 struct messaging_read_state *state;
480 req = tevent_req_create(mem_ctx, &state,
481 struct messaging_read_state);
486 state->msg_ctx = msg_ctx;
487 state->msg_type = msg_type;
489 waiters_len = talloc_array_length(msg_ctx->waiters);
491 if (waiters_len == msg_ctx->num_waiters) {
492 struct tevent_req **tmp;
494 tmp = talloc_realloc(msg_ctx, msg_ctx->waiters,
495 struct tevent_req *, waiters_len+1);
496 if (tevent_req_nomem(tmp, req)) {
497 return tevent_req_post(req, ev);
499 msg_ctx->waiters = tmp;
502 msg_ctx->waiters[msg_ctx->num_waiters] = req;
503 msg_ctx->num_waiters += 1;
504 tevent_req_set_cleanup_fn(req, messaging_read_cleanup);
509 static void messaging_read_cleanup(struct tevent_req *req,
510 enum tevent_req_state req_state)
512 struct messaging_read_state *state = tevent_req_data(
513 req, struct messaging_read_state);
514 struct messaging_context *msg_ctx = state->msg_ctx;
515 struct tevent_req **waiters = msg_ctx->waiters;
518 tevent_req_set_cleanup_fn(req, NULL);
520 for (i=0; i<msg_ctx->num_waiters; i++) {
521 if (waiters[i] == req) {
522 waiters[i] = waiters[msg_ctx->num_waiters-1];
523 msg_ctx->num_waiters -= 1;
529 static void messaging_read_done(struct tevent_req *req, struct messaging_rec *rec)
531 struct messaging_read_state *state = tevent_req_data(
532 req, struct messaging_read_state);
534 state->rec = messaging_rec_dup(state, rec);
535 if (tevent_req_nomem(state->rec, req)) {
538 tevent_req_done(req);
541 int messaging_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
542 struct messaging_rec **presult)
544 struct messaging_read_state *state = tevent_req_data(
545 req, struct messaging_read_state);
548 if (tevent_req_is_unix_error(req, &err)) {
549 tevent_req_received(req);
552 *presult = talloc_move(mem_ctx, &state->rec);
557 Dispatch one messaging_rec
559 void messaging_dispatch_rec(struct messaging_context *msg_ctx,
560 struct messaging_rec *rec)
562 struct messaging_callback *cb, *next;
565 for (cb = msg_ctx->callbacks; cb != NULL; cb = next) {
567 if (cb->msg_type == rec->msg_type) {
568 cb->fn(msg_ctx, cb->private_data, rec->msg_type,
569 rec->src, &rec->buf);
570 /* we continue looking for matching messages
571 after finding one. This matters for
572 subsystems like the internal notify code
573 which register more than one handler for
574 the same message type */
578 for (i=0; i<msg_ctx->num_waiters; i++) {
579 struct tevent_req *req = msg_ctx->waiters[i];
580 struct messaging_read_state *state = tevent_req_data(
581 req, struct messaging_read_state);
583 if (state->msg_type == rec->msg_type) {
584 messaging_read_done(req, rec);
590 static int mess_parent_dgm_cleanup(void *private_data);
591 static void mess_parent_dgm_cleanup_done(struct tevent_req *req);
593 bool messaging_parent_dgm_cleanup_init(struct messaging_context *msg)
595 struct tevent_req *req;
597 req = background_job_send(
598 msg, msg->event_ctx, msg, NULL, 0,
599 lp_parm_int(-1, "messaging", "messaging dgm cleanup interval", 60*15),
600 mess_parent_dgm_cleanup, msg);
604 tevent_req_set_callback(req, mess_parent_dgm_cleanup_done, msg);
608 static int mess_parent_dgm_cleanup(void *private_data)
610 struct messaging_context *msg_ctx = talloc_get_type_abort(
611 private_data, struct messaging_context);
614 status = messaging_dgm_wipe(msg_ctx);
615 DEBUG(10, ("messaging_dgm_wipe returned %s\n", nt_errstr(status)));
616 return lp_parm_int(-1, "messaging", "messaging dgm cleanup interval", 60*15);
619 static void mess_parent_dgm_cleanup_done(struct tevent_req *req)
621 struct messaging_context *msg = tevent_req_callback_data(
622 req, struct messaging_context);
625 status = background_job_recv(req);
627 DEBUG(1, ("messaging dgm cleanup job ended with %s\n", nt_errstr(status)));
629 req = background_job_send(
630 msg, msg->event_ctx, msg, NULL, 0,
631 lp_parm_int(-1, "messaging", "messaging dgm cleanup interval", 60*15),
632 mess_parent_dgm_cleanup, msg);
634 DEBUG(1, ("background_job_send failed\n"));
636 tevent_req_set_callback(req, mess_parent_dgm_cleanup_done, msg);