source3/lib/messages.c

   1 /*
   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
   8
   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.
  13
  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.
  18
  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/>.
  21 */
  22
  23 /**
  24   @defgroup messages Internal messaging framework
  25   @{
  26   @file messages.c
  27
  28   @brief  Module for internal messaging between Samba daemons.
  29
  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
  33    that process.
  34
  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
  37    simple example.
  38
  39    @caution Dispatch functions must be able to cope with incoming
  40    messages on an *odd* byte boundary.
  41
  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
  44    quick succession.
  45
  46 */
  47
  48 #include "includes.h"
  49 #include "dbwrap/dbwrap.h"
  50 #include "serverid.h"
  51 #include "messages.h"
  52 #include "lib/util/tevent_unix.h"
  53 #include "lib/background.h"
  54
  55 struct messaging_callback {
  56         struct messaging_callback *prev, *next;
  57         uint32 msg_type;
  58         void (*fn)(struct messaging_context *msg, void *private_data,
  59                    uint32_t msg_type,
  60                    struct server_id server_id, DATA_BLOB *data);
  61         void *private_data;
  62 };
  63
  64 /****************************************************************************
  65  A useful function for testing the message system.
  66 ****************************************************************************/
  67
  68 static void ping_message(struct messaging_context *msg_ctx,
  69                          void *private_data,
  70                          uint32_t msg_type,
  71                          struct server_id src,
  72                          DATA_BLOB *data)
  73 {
  74         const char *msg = "none";
  75         char *free_me = NULL;
  76
  77         if (data->data != NULL) {
  78                 free_me = talloc_strndup(talloc_tos(), (char *)data->data,
  79                                          data->length);
  80                 msg = free_me;
  81         }
  82         DEBUG(1,("INFO: Received PING message from PID %s [%s]\n",
  83                  procid_str_static(&src), msg));
  84         TALLOC_FREE(free_me);
  85         messaging_send(msg_ctx, src, MSG_PONG, data);
  86 }
  87
  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 ****************************************************************************/
  94
  95 struct msg_all {
  96         struct messaging_context *msg_ctx;
  97         int msg_type;
  98         uint32 msg_flag;
  99         const void *buf;
 100         size_t len;
 101         int n_sent;
 102 };
 103
 104 /****************************************************************************
 105  Send one of the messages for the broadcast.
 106 ****************************************************************************/
 107
 108 static int traverse_fn(struct db_record *rec, const struct server_id *id,
 109                        uint32_t msg_flags, void *state)
 110 {
 111         struct msg_all *msg_all = (struct msg_all *)state;
 112         NTSTATUS status;
 113
 114         /* Don't send if the receiver hasn't registered an interest. */
 115
 116         if((msg_flags & msg_all->msg_flag) == 0) {
 117                 return 0;
 118         }
 119
 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.*/
 122
 123         status = messaging_send_buf(msg_all->msg_ctx, *id, msg_all->msg_type,
 124                                     (const uint8_t *)msg_all->buf, msg_all->len);
 125
 126         if (NT_STATUS_EQUAL(status, NT_STATUS_INVALID_HANDLE)) {
 127
 128                 /*
 129                  * If the pid was not found delete the entry from
 130                  * serverid.tdb
 131                  */
 132
 133                 DEBUG(2, ("pid %s doesn't exist\n", procid_str_static(id)));
 134
 135                 dbwrap_record_delete(rec);
 136         }
 137         msg_all->n_sent++;
 138         return 0;
 139 }
 140
 141 /**
 142  * Send a message to all smbd processes.
 143  *
 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
 146  * it.
 147  *
 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.
 150  *
 151  * @retval True for success.
 152  **/
 153 bool message_send_all(struct messaging_context *msg_ctx,
 154                       int msg_type,
 155                       const void *buf, size_t len,
 156                       int *n_sent)
 157 {
 158         struct msg_all msg_all;
 159
 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;
 173         } else {
 174                 return false;
 175         }
 176
 177         msg_all.buf = buf;
 178         msg_all.len = len;
 179         msg_all.n_sent = 0;
 180         msg_all.msg_ctx = msg_ctx;
 181
 182         serverid_traverse(traverse_fn, &msg_all);
 183         if (n_sent)
 184                 *n_sent = msg_all.n_sent;
 185         return true;
 186 }
 187
 188 struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx,
 189                                          struct tevent_context *ev)
 190 {
 191         struct messaging_context *ctx;
 192         NTSTATUS status;
 193
 194         if (!(ctx = talloc_zero(mem_ctx, struct messaging_context))) {
 195                 return NULL;
 196         }
 197
 198         ctx->id = procid_self();
 199         ctx->event_ctx = ev;
 200
 201         status = messaging_dgm_init(ctx, ctx, &ctx->local);
 202
 203         if (!NT_STATUS_IS_OK(status)) {
 204                 DEBUG(2, ("messaging_dgm_init failed: %s\n",
 205                           nt_errstr(status)));
 206                 TALLOC_FREE(ctx);
 207                 return NULL;
 208         }
 209
 210         if (lp_clustering()) {
 211                 status = messaging_ctdbd_init(ctx, ctx, &ctx->remote);
 212
 213                 if (!NT_STATUS_IS_OK(status)) {
 214                         DEBUG(2, ("messaging_ctdbd_init failed: %s\n",
 215                                   nt_errstr(status)));
 216                         TALLOC_FREE(ctx);
 217                         return NULL;
 218                 }
 219         }
 220         ctx->id.vnn = get_my_vnn();
 221
 222         messaging_register(ctx, NULL, MSG_PING, ping_message);
 223
 224         /* Register some debugging related messages */
 225
 226         register_msg_pool_usage(ctx);
 227         register_dmalloc_msgs(ctx);
 228         debug_register_msgs(ctx);
 229
 230         return ctx;
 231 }
 232
 233 struct server_id messaging_server_id(const struct messaging_context *msg_ctx)
 234 {
 235         return msg_ctx->id;
 236 }
 237
 238 /*
 239  * re-init after a fork
 240  */
 241 NTSTATUS messaging_reinit(struct messaging_context *msg_ctx)
 242 {
 243         NTSTATUS status;
 244
 245         TALLOC_FREE(msg_ctx->local);
 246
 247         msg_ctx->id = procid_self();
 248
 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",
 252                           nt_errstr(status)));
 253                 return status;
 254         }
 255
 256         TALLOC_FREE(msg_ctx->remote);
 257
 258         if (lp_clustering()) {
 259                 status = messaging_ctdbd_init(msg_ctx, msg_ctx,
 260                                               &msg_ctx->remote);
 261
 262                 if (!NT_STATUS_IS_OK(status)) {
 263                         DEBUG(1, ("messaging_ctdbd_init failed: %s\n",
 264                                   nt_errstr(status)));
 265                         return status;
 266                 }
 267         }
 268
 269         return NT_STATUS_OK;
 270 }
 271
 272
 273 /*
 274  * Register a dispatch function for a particular message type. Allow multiple
 275  * registrants
 276 */
 277 NTSTATUS messaging_register(struct messaging_context *msg_ctx,
 278                             void *private_data,
 279                             uint32_t msg_type,
 280                             void (*fn)(struct messaging_context *msg,
 281                                        void *private_data,
 282                                        uint32_t msg_type,
 283                                        struct server_id server_id,
 284                                        DATA_BLOB *data))
 285 {
 286         struct messaging_callback *cb;
 287
 288         DEBUG(5, ("Registering messaging pointer for type %u - "
 289                   "private_data=%p\n",
 290                   (unsigned)msg_type, private_data));
 291
 292         /*
 293          * Only one callback per type
 294          */
 295
 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
 302                    them. */
 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));
 306                         cb->fn = fn;
 307                         cb->private_data = private_data;
 308                         return NT_STATUS_OK;
 309                 }
 310         }
 311
 312         if (!(cb = talloc(msg_ctx, struct messaging_callback))) {
 313                 return NT_STATUS_NO_MEMORY;
 314         }
 315
 316         cb->msg_type = msg_type;
 317         cb->fn = fn;
 318         cb->private_data = private_data;
 319
 320         DLIST_ADD(msg_ctx->callbacks, cb);
 321         return NT_STATUS_OK;
 322 }
 323
 324 /*
 325   De-register the function for a particular message type.
 326 */
 327 void messaging_deregister(struct messaging_context *ctx, uint32_t msg_type,
 328                           void *private_data)
 329 {
 330         struct messaging_callback *cb, *next;
 331
 332         for (cb = ctx->callbacks; cb; cb = next) {
 333                 next = 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);
 339                         TALLOC_FREE(cb);
 340                 }
 341         }
 342 }
 343
 344 struct messaging_selfsend_state {
 345         struct messaging_context *msg;
 346         struct messaging_rec rec;
 347 };
 348
 349 static void messaging_trigger_self(struct tevent_context *ev,
 350                                    struct tevent_immediate *im,
 351                                    void *private_data);
 352
 353 /*
 354   Send a message to a particular server
 355 */
 356 NTSTATUS messaging_send(struct messaging_context *msg_ctx,
 357                         struct server_id server, uint32_t msg_type,
 358                         const DATA_BLOB *data)
 359 {
 360         if (server_id_is_disconnected(&server)) {
 361                 return NT_STATUS_INVALID_PARAMETER_MIX;
 362         }
 363
 364         if (!procid_is_local(&server)) {
 365                 return msg_ctx->remote->send_fn(msg_ctx, server,
 366                                                 msg_type, data,
 367                                                 msg_ctx->remote);
 368         }
 369
 370         if (server_id_equal(&msg_ctx->id, &server)) {
 371                 struct messaging_selfsend_state *state;
 372                 struct tevent_immediate *im;
 373
 374                 state = talloc_pooled_object(
 375                         msg_ctx, struct messaging_selfsend_state,
 376                         1, data->length);
 377                 if (state == NULL) {
 378                         return NT_STATUS_NO_MEMORY;
 379                 }
 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;
 385
 386                 /* Can't fail, it's a pooled_object */
 387                 state->rec.buf = data_blob_talloc(
 388                         state, data->data, data->length);
 389
 390                 im = tevent_create_immediate(state);
 391                 if (im == NULL) {
 392                         TALLOC_FREE(state);
 393                         return NT_STATUS_NO_MEMORY;
 394                 }
 395
 396                 tevent_schedule_immediate(im, msg_ctx->event_ctx,
 397                                           messaging_trigger_self, state);
 398                 return NT_STATUS_OK;
 399         }
 400
 401         return msg_ctx->local->send_fn(msg_ctx, server, msg_type, data,
 402                                        msg_ctx->local);
 403 }
 404
 405 static void messaging_trigger_self(struct tevent_context *ev,
 406                                    struct tevent_immediate *im,
 407                                    void *private_data)
 408 {
 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);
 412         TALLOC_FREE(state);
 413 }
 414
 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)
 418 {
 419         DATA_BLOB blob = data_blob_const(buf, len);
 420         return messaging_send(msg_ctx, server, msg_type, &blob);
 421 }
 422
 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)
 426 {
 427         uint8_t *buf;
 428         NTSTATUS status;
 429
 430         buf = iov_buf(talloc_tos(), iov, iovlen);
 431         if (buf == NULL) {
 432                 return NT_STATUS_NO_MEMORY;
 433         }
 434
 435         status = messaging_send_buf(msg_ctx, server, msg_type,
 436                                     buf, talloc_get_size(buf));
 437
 438         TALLOC_FREE(buf);
 439         return status;
 440 }
 441
 442 static struct messaging_rec *messaging_rec_dup(TALLOC_CTX *mem_ctx,
 443                                                struct messaging_rec *rec)
 444 {
 445         struct messaging_rec *result;
 446
 447         result = talloc_pooled_object(mem_ctx, struct messaging_rec,
 448                                       1, rec->buf.length);
 449         if (result == NULL) {
 450                 return NULL;
 451         }
 452         *result = *rec;
 453
 454         /* Doesn't fail, see talloc_pooled_object */
 455
 456         result->buf.data = talloc_memdup(result, rec->buf.data,
 457                                          rec->buf.length);
 458         return result;
 459 }
 460
 461 struct messaging_filtered_read_state {
 462         struct tevent_context *ev;
 463         struct messaging_context *msg_ctx;
 464         void *tevent_handle;
 465
 466         bool (*filter)(struct messaging_rec *rec, void *private_data);
 467         void *private_data;
 468
 469         struct messaging_rec *rec;
 470 };
 471
 472 static void messaging_filtered_read_cleanup(struct tevent_req *req,
 473                                             enum tevent_req_state req_state);
 474
 475 struct tevent_req *messaging_filtered_read_send(
 476         TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 477         struct messaging_context *msg_ctx,
 478         bool (*filter)(struct messaging_rec *rec, void *private_data),
 479         void *private_data)
 480 {
 481         struct tevent_req *req;
 482         struct messaging_filtered_read_state *state;
 483         size_t new_waiters_len;
 484
 485         req = tevent_req_create(mem_ctx, &state,
 486                                 struct messaging_filtered_read_state);
 487         if (req == NULL) {
 488                 return NULL;
 489         }
 490         state->ev = ev;
 491         state->msg_ctx = msg_ctx;
 492         state->filter = filter;
 493         state->private_data = private_data;
 494
 495         /*
 496          * We have to defer the callback here, as we might be called from
 497          * within a different tevent_context than state->ev
 498          */
 499         tevent_req_defer_callback(req, state->ev);
 500
 501         state->tevent_handle = messaging_dgm_register_tevent_context(
 502                 state, msg_ctx, ev);
 503         if (tevent_req_nomem(state, req)) {
 504                 return tevent_req_post(req, ev);
 505         }
 506
 507         /*
 508          * We add ourselves to the "new_waiters" array, not the "waiters"
 509          * array. If we are called from within messaging_read_done,
 510          * messaging_dispatch_rec will be in an active for-loop on
 511          * "waiters". We must be careful not to mess with this array, because
 512          * it could mean that a single event is being delivered twice.
 513          */
 514
 515         new_waiters_len = talloc_array_length(msg_ctx->new_waiters);
 516
 517         if (new_waiters_len == msg_ctx->num_new_waiters) {
 518                 struct tevent_req **tmp;
 519
 520                 tmp = talloc_realloc(msg_ctx, msg_ctx->new_waiters,
 521                                      struct tevent_req *, new_waiters_len+1);
 522                 if (tevent_req_nomem(tmp, req)) {
 523                         return tevent_req_post(req, ev);
 524                 }
 525                 msg_ctx->new_waiters = tmp;
 526         }
 527
 528         msg_ctx->new_waiters[msg_ctx->num_new_waiters] = req;
 529         msg_ctx->num_new_waiters += 1;
 530         tevent_req_set_cleanup_fn(req, messaging_filtered_read_cleanup);
 531
 532         return req;
 533 }
 534
 535 static void messaging_filtered_read_cleanup(struct tevent_req *req,
 536                                             enum tevent_req_state req_state)
 537 {
 538         struct messaging_filtered_read_state *state = tevent_req_data(
 539                 req, struct messaging_filtered_read_state);
 540         struct messaging_context *msg_ctx = state->msg_ctx;
 541         unsigned i;
 542
 543         tevent_req_set_cleanup_fn(req, NULL);
 544
 545         TALLOC_FREE(state->tevent_handle);
 546
 547         /*
 548          * Just set the [new_]waiters entry to NULL, be careful not to mess
 549          * with the other "waiters" array contents. We are often called from
 550          * within "messaging_dispatch_rec", which loops over
 551          * "waiters". Messing with the "waiters" array will mess up that
 552          * for-loop.
 553          */
 554
 555         for (i=0; i<msg_ctx->num_waiters; i++) {
 556                 if (msg_ctx->waiters[i] == req) {
 557                         msg_ctx->waiters[i] = NULL;
 558                         return;
 559                 }
 560         }
 561
 562         for (i=0; i<msg_ctx->num_new_waiters; i++) {
 563                 if (msg_ctx->new_waiters[i] == req) {
 564                         msg_ctx->new_waiters[i] = NULL;
 565                         return;
 566                 }
 567         }
 568 }
 569
 570 static void messaging_filtered_read_done(struct tevent_req *req,
 571                                          struct messaging_rec *rec)
 572 {
 573         struct messaging_filtered_read_state *state = tevent_req_data(
 574                 req, struct messaging_filtered_read_state);
 575
 576         state->rec = messaging_rec_dup(state, rec);
 577         if (tevent_req_nomem(state->rec, req)) {
 578                 return;
 579         }
 580         tevent_req_done(req);
 581 }
 582
 583 int messaging_filtered_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
 584                                  struct messaging_rec **presult)
 585 {
 586         struct messaging_filtered_read_state *state = tevent_req_data(
 587                 req, struct messaging_filtered_read_state);
 588         int err;
 589
 590         if (tevent_req_is_unix_error(req, &err)) {
 591                 tevent_req_received(req);
 592                 return err;
 593         }
 594         *presult = talloc_move(mem_ctx, &state->rec);
 595         return 0;
 596 }
 597
 598 struct messaging_read_state {
 599         uint32_t msg_type;
 600         struct messaging_rec *rec;
 601 };
 602
 603 static bool messaging_read_filter(struct messaging_rec *rec,
 604                                   void *private_data);
 605 static void messaging_read_done(struct tevent_req *subreq);
 606
 607 struct tevent_req *messaging_read_send(TALLOC_CTX *mem_ctx,
 608                                        struct tevent_context *ev,
 609                                        struct messaging_context *msg,
 610                                        uint32_t msg_type)
 611 {
 612         struct tevent_req *req, *subreq;
 613         struct messaging_read_state *state;
 614
 615         req = tevent_req_create(mem_ctx, &state,
 616                                 struct messaging_read_state);
 617         if (req == NULL) {
 618                 return NULL;
 619         }
 620         state->msg_type = msg_type;
 621
 622         subreq = messaging_filtered_read_send(state, ev, msg,
 623                                               messaging_read_filter, state);
 624         if (tevent_req_nomem(subreq, req)) {
 625                 return tevent_req_post(req, ev);
 626         }
 627         tevent_req_set_callback(subreq, messaging_read_done, req);
 628         return req;
 629 }
 630
 631 static bool messaging_read_filter(struct messaging_rec *rec,
 632                                   void *private_data)
 633 {
 634         struct messaging_read_state *state = talloc_get_type_abort(
 635                 private_data, struct messaging_read_state);
 636
 637         return rec->msg_type == state->msg_type;
 638 }
 639
 640 static void messaging_read_done(struct tevent_req *subreq)
 641 {
 642         struct tevent_req *req = tevent_req_callback_data(
 643                 subreq, struct tevent_req);
 644         struct messaging_read_state *state = tevent_req_data(
 645                 req, struct messaging_read_state);
 646         int ret;
 647
 648         ret = messaging_filtered_read_recv(subreq, state, &state->rec);
 649         TALLOC_FREE(subreq);
 650         if (tevent_req_error(req, ret)) {
 651                 return;
 652         }
 653         tevent_req_done(req);
 654 }
 655
 656 int messaging_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
 657                         struct messaging_rec **presult)
 658 {
 659         struct messaging_read_state *state = tevent_req_data(
 660                 req, struct messaging_read_state);
 661         int err;
 662
 663         if (tevent_req_is_unix_error(req, &err)) {
 664                 return err;
 665         }
 666         if (presult != NULL) {
 667                 *presult = talloc_move(mem_ctx, &state->rec);
 668         }
 669         return 0;
 670 }
 671
 672 static bool messaging_append_new_waiters(struct messaging_context *msg_ctx)
 673 {
 674         if (msg_ctx->num_new_waiters == 0) {
 675                 return true;
 676         }
 677
 678         if (talloc_array_length(msg_ctx->waiters) <
 679             (msg_ctx->num_waiters + msg_ctx->num_new_waiters)) {
 680                 struct tevent_req **tmp;
 681                 tmp = talloc_realloc(
 682                         msg_ctx, msg_ctx->waiters, struct tevent_req *,
 683                         msg_ctx->num_waiters + msg_ctx->num_new_waiters);
 684                 if (tmp == NULL) {
 685                         DEBUG(1, ("%s: talloc failed\n", __func__));
 686                         return false;
 687                 }
 688                 msg_ctx->waiters = tmp;
 689         }
 690
 691         memcpy(&msg_ctx->waiters[msg_ctx->num_waiters], msg_ctx->new_waiters,
 692                sizeof(struct tevent_req *) * msg_ctx->num_new_waiters);
 693
 694         msg_ctx->num_waiters += msg_ctx->num_new_waiters;
 695         msg_ctx->num_new_waiters = 0;
 696
 697         return true;
 698 }
 699
 700 /*
 701   Dispatch one messaging_rec
 702 */
 703 void messaging_dispatch_rec(struct messaging_context *msg_ctx,
 704                             struct messaging_rec *rec)
 705 {
 706         struct messaging_callback *cb, *next;
 707         unsigned i;
 708
 709         for (cb = msg_ctx->callbacks; cb != NULL; cb = next) {
 710                 next = cb->next;
 711                 if (cb->msg_type == rec->msg_type) {
 712                         cb->fn(msg_ctx, cb->private_data, rec->msg_type,
 713                                rec->src, &rec->buf);
 714                         /* we continue looking for matching messages
 715                            after finding one. This matters for
 716                            subsystems like the internal notify code
 717                            which register more than one handler for
 718                            the same message type */
 719                 }
 720         }
 721
 722         if (!messaging_append_new_waiters(msg_ctx)) {
 723                 return;
 724         }
 725
 726         i = 0;
 727         while (i < msg_ctx->num_waiters) {
 728                 struct tevent_req *req;
 729                 struct messaging_filtered_read_state *state;
 730
 731                 req = msg_ctx->waiters[i];
 732                 if (req == NULL) {
 733                         /*
 734                          * This got cleaned up. In the meantime,
 735                          * move everything down one. We need
 736                          * to keep the order of waiters, as
 737                          * other code may depend on this.
 738                          */
 739                         if (i <  msg_ctx->num_waiters - 1) {
 740                                 memmove(&msg_ctx->waiters[i],
 741                                         &msg_ctx->waiters[i+1],
 742                                         sizeof(struct tevent_req *) *
 743                                             (msg_ctx->num_waiters - i - 1));
 744                         }
 745                         msg_ctx->num_waiters -= 1;
 746                         continue;
 747                 }
 748
 749                 state = tevent_req_data(
 750                         req, struct messaging_filtered_read_state);
 751                 if (state->filter(rec, state->private_data)) {
 752                         messaging_filtered_read_done(req, rec);
 753                 }
 754
 755                 i += 1;
 756         }
 757         return;
 758 }
 759
 760 static int mess_parent_dgm_cleanup(void *private_data);
 761 static void mess_parent_dgm_cleanup_done(struct tevent_req *req);
 762
 763 bool messaging_parent_dgm_cleanup_init(struct messaging_context *msg)
 764 {
 765         struct tevent_req *req;
 766
 767         req = background_job_send(
 768                 msg, msg->event_ctx, msg, NULL, 0,
 769                 lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
 770                             60*15),
 771                 mess_parent_dgm_cleanup, msg);
 772         if (req == NULL) {
 773                 return false;
 774         }
 775         tevent_req_set_callback(req, mess_parent_dgm_cleanup_done, msg);
 776         return true;
 777 }
 778
 779 static int mess_parent_dgm_cleanup(void *private_data)
 780 {
 781         struct messaging_context *msg_ctx = talloc_get_type_abort(
 782                 private_data, struct messaging_context);
 783         NTSTATUS status;
 784
 785         status = messaging_dgm_wipe(msg_ctx);
 786         DEBUG(10, ("messaging_dgm_wipe returned %s\n", nt_errstr(status)));
 787         return lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
 788                            60*15);
 789 }
 790
 791 static void mess_parent_dgm_cleanup_done(struct tevent_req *req)
 792 {
 793         struct messaging_context *msg = tevent_req_callback_data(
 794                 req, struct messaging_context);
 795         NTSTATUS status;
 796
 797         status = background_job_recv(req);
 798         TALLOC_FREE(req);
 799         DEBUG(1, ("messaging dgm cleanup job ended with %s\n",
 800                   nt_errstr(status)));
 801
 802         req = background_job_send(
 803                 msg, msg->event_ctx, msg, NULL, 0,
 804                 lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
 805                             60*15),
 806                 mess_parent_dgm_cleanup, msg);
 807         if (req == NULL) {
 808                 DEBUG(1, ("background_job_send failed\n"));
 809         }
 810         tevent_req_set_callback(req, mess_parent_dgm_cleanup_done, msg);
 811 }
 812
 813 /** @} **/