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 static bool messaging_is_self_send(const struct messaging_context *msg_ctx,
 345                                    const struct server_id *dst)
 346 {
 347         return ((msg_ctx->id.vnn == dst->vnn) &&
 348                 (msg_ctx->id.pid == dst->pid));
 349 }
 350
 351 /*
 352   Send a message to a particular server
 353 */
 354 NTSTATUS messaging_send(struct messaging_context *msg_ctx,
 355                         struct server_id server, uint32_t msg_type,
 356                         const DATA_BLOB *data)
 357 {
 358         struct iovec iov;
 359
 360         iov.iov_base = data->data;
 361         iov.iov_len = data->length;
 362
 363         return messaging_send_iov(msg_ctx, server, msg_type, &iov, 1);
 364 }
 365
 366 NTSTATUS messaging_send_buf(struct messaging_context *msg_ctx,
 367                             struct server_id server, uint32_t msg_type,
 368                             const uint8_t *buf, size_t len)
 369 {
 370         DATA_BLOB blob = data_blob_const(buf, len);
 371         return messaging_send(msg_ctx, server, msg_type, &blob);
 372 }
 373
 374 NTSTATUS messaging_send_iov(struct messaging_context *msg_ctx,
 375                             struct server_id server, uint32_t msg_type,
 376                             const struct iovec *iov, int iovlen)
 377 {
 378         if (server_id_is_disconnected(&server)) {
 379                 return NT_STATUS_INVALID_PARAMETER_MIX;
 380         }
 381
 382         if (!procid_is_local(&server)) {
 383                 return msg_ctx->remote->send_fn(msg_ctx, server,
 384                                                 msg_type, iov, iovlen,
 385                                                 msg_ctx->remote);
 386         }
 387
 388         if (messaging_is_self_send(msg_ctx, &server)) {
 389                 struct messaging_rec rec;
 390                 uint8_t *buf;
 391                 DATA_BLOB data;
 392
 393                 buf = iov_buf(talloc_tos(), iov, iovlen);
 394                 if (buf == NULL) {
 395                         return NT_STATUS_NO_MEMORY;
 396                 }
 397
 398                 data = data_blob_const(buf, talloc_get_size(buf));
 399
 400                 rec.msg_version = MESSAGE_VERSION;
 401                 rec.msg_type = msg_type & MSG_TYPE_MASK;
 402                 rec.dest = server;
 403                 rec.src = msg_ctx->id;
 404                 rec.buf = data;
 405                 messaging_dispatch_rec(msg_ctx, &rec);
 406                 TALLOC_FREE(buf);
 407                 return NT_STATUS_OK;
 408         }
 409
 410         return msg_ctx->local->send_fn(msg_ctx, server, msg_type, iov, iovlen,
 411                                        msg_ctx->local);
 412 }
 413
 414 static struct messaging_rec *messaging_rec_dup(TALLOC_CTX *mem_ctx,
 415                                                struct messaging_rec *rec)
 416 {
 417         struct messaging_rec *result;
 418
 419         result = talloc_pooled_object(mem_ctx, struct messaging_rec,
 420                                       1, rec->buf.length);
 421         if (result == NULL) {
 422                 return NULL;
 423         }
 424         *result = *rec;
 425
 426         /* Doesn't fail, see talloc_pooled_object */
 427
 428         result->buf.data = talloc_memdup(result, rec->buf.data,
 429                                          rec->buf.length);
 430         return result;
 431 }
 432
 433 struct messaging_filtered_read_state {
 434         struct tevent_context *ev;
 435         struct messaging_context *msg_ctx;
 436         void *tevent_handle;
 437
 438         bool (*filter)(struct messaging_rec *rec, void *private_data);
 439         void *private_data;
 440
 441         struct messaging_rec *rec;
 442 };
 443
 444 static void messaging_filtered_read_cleanup(struct tevent_req *req,
 445                                             enum tevent_req_state req_state);
 446
 447 struct tevent_req *messaging_filtered_read_send(
 448         TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 449         struct messaging_context *msg_ctx,
 450         bool (*filter)(struct messaging_rec *rec, void *private_data),
 451         void *private_data)
 452 {
 453         struct tevent_req *req;
 454         struct messaging_filtered_read_state *state;
 455         size_t new_waiters_len;
 456
 457         req = tevent_req_create(mem_ctx, &state,
 458                                 struct messaging_filtered_read_state);
 459         if (req == NULL) {
 460                 return NULL;
 461         }
 462         state->ev = ev;
 463         state->msg_ctx = msg_ctx;
 464         state->filter = filter;
 465         state->private_data = private_data;
 466
 467         /*
 468          * We have to defer the callback here, as we might be called from
 469          * within a different tevent_context than state->ev
 470          */
 471         tevent_req_defer_callback(req, state->ev);
 472
 473         state->tevent_handle = messaging_dgm_register_tevent_context(
 474                 state, msg_ctx, ev);
 475         if (tevent_req_nomem(state, req)) {
 476                 return tevent_req_post(req, ev);
 477         }
 478
 479         /*
 480          * We add ourselves to the "new_waiters" array, not the "waiters"
 481          * array. If we are called from within messaging_read_done,
 482          * messaging_dispatch_rec will be in an active for-loop on
 483          * "waiters". We must be careful not to mess with this array, because
 484          * it could mean that a single event is being delivered twice.
 485          */
 486
 487         new_waiters_len = talloc_array_length(msg_ctx->new_waiters);
 488
 489         if (new_waiters_len == msg_ctx->num_new_waiters) {
 490                 struct tevent_req **tmp;
 491
 492                 tmp = talloc_realloc(msg_ctx, msg_ctx->new_waiters,
 493                                      struct tevent_req *, new_waiters_len+1);
 494                 if (tevent_req_nomem(tmp, req)) {
 495                         return tevent_req_post(req, ev);
 496                 }
 497                 msg_ctx->new_waiters = tmp;
 498         }
 499
 500         msg_ctx->new_waiters[msg_ctx->num_new_waiters] = req;
 501         msg_ctx->num_new_waiters += 1;
 502         tevent_req_set_cleanup_fn(req, messaging_filtered_read_cleanup);
 503
 504         return req;
 505 }
 506
 507 static void messaging_filtered_read_cleanup(struct tevent_req *req,
 508                                             enum tevent_req_state req_state)
 509 {
 510         struct messaging_filtered_read_state *state = tevent_req_data(
 511                 req, struct messaging_filtered_read_state);
 512         struct messaging_context *msg_ctx = state->msg_ctx;
 513         unsigned i;
 514
 515         tevent_req_set_cleanup_fn(req, NULL);
 516
 517         TALLOC_FREE(state->tevent_handle);
 518
 519         /*
 520          * Just set the [new_]waiters entry to NULL, be careful not to mess
 521          * with the other "waiters" array contents. We are often called from
 522          * within "messaging_dispatch_rec", which loops over
 523          * "waiters". Messing with the "waiters" array will mess up that
 524          * for-loop.
 525          */
 526
 527         for (i=0; i<msg_ctx->num_waiters; i++) {
 528                 if (msg_ctx->waiters[i] == req) {
 529                         msg_ctx->waiters[i] = NULL;
 530                         return;
 531                 }
 532         }
 533
 534         for (i=0; i<msg_ctx->num_new_waiters; i++) {
 535                 if (msg_ctx->new_waiters[i] == req) {
 536                         msg_ctx->new_waiters[i] = NULL;
 537                         return;
 538                 }
 539         }
 540 }
 541
 542 static void messaging_filtered_read_done(struct tevent_req *req,
 543                                          struct messaging_rec *rec)
 544 {
 545         struct messaging_filtered_read_state *state = tevent_req_data(
 546                 req, struct messaging_filtered_read_state);
 547
 548         state->rec = messaging_rec_dup(state, rec);
 549         if (tevent_req_nomem(state->rec, req)) {
 550                 return;
 551         }
 552         tevent_req_done(req);
 553 }
 554
 555 int messaging_filtered_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
 556                                  struct messaging_rec **presult)
 557 {
 558         struct messaging_filtered_read_state *state = tevent_req_data(
 559                 req, struct messaging_filtered_read_state);
 560         int err;
 561
 562         if (tevent_req_is_unix_error(req, &err)) {
 563                 tevent_req_received(req);
 564                 return err;
 565         }
 566         *presult = talloc_move(mem_ctx, &state->rec);
 567         return 0;
 568 }
 569
 570 struct messaging_read_state {
 571         uint32_t msg_type;
 572         struct messaging_rec *rec;
 573 };
 574
 575 static bool messaging_read_filter(struct messaging_rec *rec,
 576                                   void *private_data);
 577 static void messaging_read_done(struct tevent_req *subreq);
 578
 579 struct tevent_req *messaging_read_send(TALLOC_CTX *mem_ctx,
 580                                        struct tevent_context *ev,
 581                                        struct messaging_context *msg,
 582                                        uint32_t msg_type)
 583 {
 584         struct tevent_req *req, *subreq;
 585         struct messaging_read_state *state;
 586
 587         req = tevent_req_create(mem_ctx, &state,
 588                                 struct messaging_read_state);
 589         if (req == NULL) {
 590                 return NULL;
 591         }
 592         state->msg_type = msg_type;
 593
 594         subreq = messaging_filtered_read_send(state, ev, msg,
 595                                               messaging_read_filter, state);
 596         if (tevent_req_nomem(subreq, req)) {
 597                 return tevent_req_post(req, ev);
 598         }
 599         tevent_req_set_callback(subreq, messaging_read_done, req);
 600         return req;
 601 }
 602
 603 static bool messaging_read_filter(struct messaging_rec *rec,
 604                                   void *private_data)
 605 {
 606         struct messaging_read_state *state = talloc_get_type_abort(
 607                 private_data, struct messaging_read_state);
 608
 609         return rec->msg_type == state->msg_type;
 610 }
 611
 612 static void messaging_read_done(struct tevent_req *subreq)
 613 {
 614         struct tevent_req *req = tevent_req_callback_data(
 615                 subreq, struct tevent_req);
 616         struct messaging_read_state *state = tevent_req_data(
 617                 req, struct messaging_read_state);
 618         int ret;
 619
 620         ret = messaging_filtered_read_recv(subreq, state, &state->rec);
 621         TALLOC_FREE(subreq);
 622         if (tevent_req_error(req, ret)) {
 623                 return;
 624         }
 625         tevent_req_done(req);
 626 }
 627
 628 int messaging_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
 629                         struct messaging_rec **presult)
 630 {
 631         struct messaging_read_state *state = tevent_req_data(
 632                 req, struct messaging_read_state);
 633         int err;
 634
 635         if (tevent_req_is_unix_error(req, &err)) {
 636                 return err;
 637         }
 638         if (presult != NULL) {
 639                 *presult = talloc_move(mem_ctx, &state->rec);
 640         }
 641         return 0;
 642 }
 643
 644 static bool messaging_append_new_waiters(struct messaging_context *msg_ctx)
 645 {
 646         if (msg_ctx->num_new_waiters == 0) {
 647                 return true;
 648         }
 649
 650         if (talloc_array_length(msg_ctx->waiters) <
 651             (msg_ctx->num_waiters + msg_ctx->num_new_waiters)) {
 652                 struct tevent_req **tmp;
 653                 tmp = talloc_realloc(
 654                         msg_ctx, msg_ctx->waiters, struct tevent_req *,
 655                         msg_ctx->num_waiters + msg_ctx->num_new_waiters);
 656                 if (tmp == NULL) {
 657                         DEBUG(1, ("%s: talloc failed\n", __func__));
 658                         return false;
 659                 }
 660                 msg_ctx->waiters = tmp;
 661         }
 662
 663         memcpy(&msg_ctx->waiters[msg_ctx->num_waiters], msg_ctx->new_waiters,
 664                sizeof(struct tevent_req *) * msg_ctx->num_new_waiters);
 665
 666         msg_ctx->num_waiters += msg_ctx->num_new_waiters;
 667         msg_ctx->num_new_waiters = 0;
 668
 669         return true;
 670 }
 671
 672 struct messaging_defer_callback_state {
 673         struct messaging_context *msg_ctx;
 674         struct messaging_rec *rec;
 675         void (*fn)(struct messaging_context *msg, void *private_data,
 676                    uint32_t msg_type, struct server_id server_id,
 677                    DATA_BLOB *data);
 678         void *private_data;
 679 };
 680
 681 static void messaging_defer_callback_trigger(struct tevent_context *ev,
 682                                              struct tevent_immediate *im,
 683                                              void *private_data);
 684
 685 static void messaging_defer_callback(
 686         struct messaging_context *msg_ctx, struct messaging_rec *rec,
 687         void (*fn)(struct messaging_context *msg, void *private_data,
 688                    uint32_t msg_type, struct server_id server_id,
 689                    DATA_BLOB *data),
 690         void *private_data)
 691 {
 692         struct messaging_defer_callback_state *state;
 693         struct tevent_immediate *im;
 694
 695         state = talloc(msg_ctx, struct messaging_defer_callback_state);
 696         if (state == NULL) {
 697                 DEBUG(1, ("talloc failed\n"));
 698                 return;
 699         }
 700         state->msg_ctx = msg_ctx;
 701         state->fn = fn;
 702         state->private_data = private_data;
 703
 704         state->rec = messaging_rec_dup(state, rec);
 705         if (state->rec == NULL) {
 706                 DEBUG(1, ("talloc failed\n"));
 707                 TALLOC_FREE(state);
 708                 return;
 709         }
 710
 711         im = tevent_create_immediate(state);
 712         if (im == NULL) {
 713                 DEBUG(1, ("tevent_create_immediate failed\n"));
 714                 TALLOC_FREE(state);
 715                 return;
 716         }
 717         tevent_schedule_immediate(im, msg_ctx->event_ctx,
 718                                   messaging_defer_callback_trigger, state);
 719 }
 720
 721 static void messaging_defer_callback_trigger(struct tevent_context *ev,
 722                                              struct tevent_immediate *im,
 723                                              void *private_data)
 724 {
 725         struct messaging_defer_callback_state *state = talloc_get_type_abort(
 726                 private_data, struct messaging_defer_callback_state);
 727         struct messaging_rec *rec = state->rec;
 728
 729         state->fn(state->msg_ctx, state->private_data, rec->msg_type, rec->src,
 730                   &rec->buf);
 731         TALLOC_FREE(state);
 732 }
 733
 734 /*
 735   Dispatch one messaging_rec
 736 */
 737 void messaging_dispatch_rec(struct messaging_context *msg_ctx,
 738                             struct messaging_rec *rec)
 739 {
 740         struct messaging_callback *cb, *next;
 741         unsigned i;
 742
 743         for (cb = msg_ctx->callbacks; cb != NULL; cb = next) {
 744                 next = cb->next;
 745                 if (cb->msg_type != rec->msg_type) {
 746                         continue;
 747                 }
 748
 749                 if (messaging_is_self_send(msg_ctx, &rec->dest)) {
 750                         /*
 751                          * This is a self-send. We are called here from
 752                          * messaging_send(), and we don't want to directly
 753                          * recurse into the callback but go via a
 754                          * tevent_loop_once
 755                          */
 756                         messaging_defer_callback(msg_ctx, rec, cb->fn,
 757                                                  cb->private_data);
 758                 } else {
 759                         /*
 760                          * This comes from a different process. we are called
 761                          * from the event loop, so we should call back
 762                          * directly.
 763                          */
 764                         cb->fn(msg_ctx, cb->private_data, rec->msg_type,
 765                                rec->src, &rec->buf);
 766                 }
 767                 /*
 768                  * we continue looking for matching messages after finding
 769                  * one. This matters for subsystems like the internal notify
 770                  * code which register more than one handler for the same
 771                  * message type
 772                  */
 773         }
 774
 775         if (!messaging_append_new_waiters(msg_ctx)) {
 776                 return;
 777         }
 778
 779         i = 0;
 780         while (i < msg_ctx->num_waiters) {
 781                 struct tevent_req *req;
 782                 struct messaging_filtered_read_state *state;
 783
 784                 req = msg_ctx->waiters[i];
 785                 if (req == NULL) {
 786                         /*
 787                          * This got cleaned up. In the meantime,
 788                          * move everything down one. We need
 789                          * to keep the order of waiters, as
 790                          * other code may depend on this.
 791                          */
 792                         if (i < msg_ctx->num_waiters - 1) {
 793                                 memmove(&msg_ctx->waiters[i],
 794                                         &msg_ctx->waiters[i+1],
 795                                         sizeof(struct tevent_req *) *
 796                                             (msg_ctx->num_waiters - i - 1));
 797                         }
 798                         msg_ctx->num_waiters -= 1;
 799                         continue;
 800                 }
 801
 802                 state = tevent_req_data(
 803                         req, struct messaging_filtered_read_state);
 804                 if (state->filter(rec, state->private_data)) {
 805                         messaging_filtered_read_done(req, rec);
 806                 }
 807
 808                 i += 1;
 809         }
 810 }
 811
 812 static int mess_parent_dgm_cleanup(void *private_data);
 813 static void mess_parent_dgm_cleanup_done(struct tevent_req *req);
 814
 815 bool messaging_parent_dgm_cleanup_init(struct messaging_context *msg)
 816 {
 817         struct tevent_req *req;
 818
 819         req = background_job_send(
 820                 msg, msg->event_ctx, msg, NULL, 0,
 821                 lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
 822                             60*15),
 823                 mess_parent_dgm_cleanup, msg);
 824         if (req == NULL) {
 825                 return false;
 826         }
 827         tevent_req_set_callback(req, mess_parent_dgm_cleanup_done, msg);
 828         return true;
 829 }
 830
 831 static int mess_parent_dgm_cleanup(void *private_data)
 832 {
 833         struct messaging_context *msg_ctx = talloc_get_type_abort(
 834                 private_data, struct messaging_context);
 835         NTSTATUS status;
 836
 837         status = messaging_dgm_wipe(msg_ctx);
 838         DEBUG(10, ("messaging_dgm_wipe returned %s\n", nt_errstr(status)));
 839         return lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
 840                            60*15);
 841 }
 842
 843 static void mess_parent_dgm_cleanup_done(struct tevent_req *req)
 844 {
 845         struct messaging_context *msg = tevent_req_callback_data(
 846                 req, struct messaging_context);
 847         NTSTATUS status;
 848
 849         status = background_job_recv(req);
 850         TALLOC_FREE(req);
 851         DEBUG(1, ("messaging dgm cleanup job ended with %s\n",
 852                   nt_errstr(status)));
 853
 854         req = background_job_send(
 855                 msg, msg->event_ctx, msg, NULL, 0,
 856                 lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
 857                             60*15),
 858                 mess_parent_dgm_cleanup, msg);
 859         if (req == NULL) {
 860                 DEBUG(1, ("background_job_send failed\n"));
 861         }
 862         tevent_req_set_callback(req, mess_parent_dgm_cleanup_done, msg);
 863 }
 864
 865 /** @} **/