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 #include "lib/messages_dgm.h"
  55
  56 struct messaging_callback {
  57         struct messaging_callback *prev, *next;
  58         uint32 msg_type;
  59         void (*fn)(struct messaging_context *msg, void *private_data,
  60                    uint32_t msg_type,
  61                    struct server_id server_id, DATA_BLOB *data);
  62         void *private_data;
  63 };
  64
  65 struct messaging_context {
  66         struct server_id id;
  67         struct tevent_context *event_ctx;
  68         struct messaging_callback *callbacks;
  69
  70         struct tevent_req **new_waiters;
  71         unsigned num_new_waiters;
  72
  73         struct tevent_req **waiters;
  74         unsigned num_waiters;
  75
  76         struct messaging_backend *remote;
  77 };
  78
  79 struct messaging_hdr {
  80         int msg_type;
  81         struct server_id dst;
  82         struct server_id src;
  83 };
  84
  85 /****************************************************************************
  86  A useful function for testing the message system.
  87 ****************************************************************************/
  88
  89 static void ping_message(struct messaging_context *msg_ctx,
  90                          void *private_data,
  91                          uint32_t msg_type,
  92                          struct server_id src,
  93                          DATA_BLOB *data)
  94 {
  95         struct server_id_buf idbuf;
  96
  97         DEBUG(1, ("INFO: Received PING message from PID %s [%.*s]\n",
  98                   server_id_str_buf(src, &idbuf), (int)data->length,
  99                   data->data ? (char *)data->data : ""));
 100
 101         messaging_send(msg_ctx, src, MSG_PONG, data);
 102 }
 103
 104 /****************************************************************************
 105  Register/replace a dispatch function for a particular message type.
 106  JRA changed Dec 13 2006. Only one message handler now permitted per type.
 107  *NOTE*: Dispatch functions must be able to cope with incoming
 108  messages on an *odd* byte boundary.
 109 ****************************************************************************/
 110
 111 struct msg_all {
 112         struct messaging_context *msg_ctx;
 113         int msg_type;
 114         uint32 msg_flag;
 115         const void *buf;
 116         size_t len;
 117         int n_sent;
 118 };
 119
 120 /****************************************************************************
 121  Send one of the messages for the broadcast.
 122 ****************************************************************************/
 123
 124 static int traverse_fn(struct db_record *rec, const struct server_id *id,
 125                        uint32_t msg_flags, void *state)
 126 {
 127         struct msg_all *msg_all = (struct msg_all *)state;
 128         NTSTATUS status;
 129
 130         /* Don't send if the receiver hasn't registered an interest. */
 131
 132         if((msg_flags & msg_all->msg_flag) == 0) {
 133                 return 0;
 134         }
 135
 136         /* If the msg send fails because the pid was not found (i.e. smbd died),
 137          * the msg has already been deleted from the messages.tdb.*/
 138
 139         status = messaging_send_buf(msg_all->msg_ctx, *id, msg_all->msg_type,
 140                                     (const uint8_t *)msg_all->buf, msg_all->len);
 141
 142         if (NT_STATUS_EQUAL(status, NT_STATUS_INVALID_HANDLE)) {
 143                 struct server_id_buf idbuf;
 144
 145                 /*
 146                  * If the pid was not found delete the entry from
 147                  * serverid.tdb
 148                  */
 149
 150                 DEBUG(2, ("pid %s doesn't exist\n",
 151                           server_id_str_buf(*id, &idbuf)));
 152
 153                 dbwrap_record_delete(rec);
 154         }
 155         msg_all->n_sent++;
 156         return 0;
 157 }
 158
 159 /**
 160  * Send a message to all smbd processes.
 161  *
 162  * It isn't very efficient, but should be OK for the sorts of
 163  * applications that use it. When we need efficient broadcast we can add
 164  * it.
 165  *
 166  * @param n_sent Set to the number of messages sent.  This should be
 167  * equal to the number of processes, but be careful for races.
 168  *
 169  * @retval True for success.
 170  **/
 171 bool message_send_all(struct messaging_context *msg_ctx,
 172                       int msg_type,
 173                       const void *buf, size_t len,
 174                       int *n_sent)
 175 {
 176         struct msg_all msg_all;
 177
 178         msg_all.msg_type = msg_type;
 179         if (msg_type < 0x100) {
 180                 msg_all.msg_flag = FLAG_MSG_GENERAL;
 181         } else if (msg_type > 0x100 && msg_type < 0x200) {
 182                 msg_all.msg_flag = FLAG_MSG_NMBD;
 183         } else if (msg_type > 0x200 && msg_type < 0x300) {
 184                 msg_all.msg_flag = FLAG_MSG_PRINT_GENERAL;
 185         } else if (msg_type > 0x300 && msg_type < 0x400) {
 186                 msg_all.msg_flag = FLAG_MSG_SMBD;
 187         } else if (msg_type > 0x400 && msg_type < 0x600) {
 188                 msg_all.msg_flag = FLAG_MSG_WINBIND;
 189         } else if (msg_type > 4000 && msg_type < 5000) {
 190                 msg_all.msg_flag = FLAG_MSG_DBWRAP;
 191         } else {
 192                 return false;
 193         }
 194
 195         msg_all.buf = buf;
 196         msg_all.len = len;
 197         msg_all.n_sent = 0;
 198         msg_all.msg_ctx = msg_ctx;
 199
 200         serverid_traverse(traverse_fn, &msg_all);
 201         if (n_sent)
 202                 *n_sent = msg_all.n_sent;
 203         return true;
 204 }
 205
 206 static void messaging_recv_cb(const uint8_t *msg, size_t msg_len,
 207                               void *private_data)
 208 {
 209         struct messaging_context *msg_ctx = talloc_get_type_abort(
 210                 private_data, struct messaging_context);
 211         const struct messaging_hdr *hdr;
 212         struct server_id_buf idbuf;
 213         struct messaging_rec rec;
 214
 215         if (msg_len < sizeof(*hdr)) {
 216                 DEBUG(1, ("message too short: %u\n", (unsigned)msg_len));
 217                 return;
 218         }
 219
 220         /*
 221          * messages_dgm guarantees alignment, so we can cast here
 222          */
 223         hdr = (const struct messaging_hdr *)msg;
 224
 225         DEBUG(10, ("%s: Received message 0x%x len %u from %s\n", __func__,
 226                    (unsigned)hdr->msg_type, (unsigned)(msg_len - sizeof(*hdr)),
 227                    server_id_str_buf(hdr->src, &idbuf)));
 228
 229         rec = (struct messaging_rec) {
 230                 .msg_version = MESSAGE_VERSION,
 231                 .msg_type = hdr->msg_type,
 232                 .src = hdr->src,
 233                 .dest = hdr->dst,
 234                 .buf.data = discard_const_p(uint8, msg) + sizeof(*hdr),
 235                 .buf.length = msg_len - sizeof(*hdr)
 236         };
 237
 238         messaging_dispatch_rec(msg_ctx, &rec);
 239 }
 240
 241 static int messaging_context_destructor(struct messaging_context *ctx)
 242 {
 243         messaging_dgm_destroy();
 244         return 0;
 245 }
 246
 247 struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx,
 248                                          struct tevent_context *ev)
 249 {
 250         struct messaging_context *ctx;
 251         NTSTATUS status;
 252         int ret;
 253
 254         if (!(ctx = talloc_zero(mem_ctx, struct messaging_context))) {
 255                 return NULL;
 256         }
 257
 258         ctx->id = procid_self();
 259         ctx->event_ctx = ev;
 260
 261         sec_init();
 262
 263         ret = messaging_dgm_init(ctx->event_ctx, ctx->id,
 264                                  lp_cache_directory(), sec_initial_uid(),
 265                                  messaging_recv_cb, ctx);
 266
 267         if (ret != 0) {
 268                 DEBUG(2, ("messaging_dgm_init failed: %s\n", strerror(ret)));
 269                 TALLOC_FREE(ctx);
 270                 return NULL;
 271         }
 272
 273         talloc_set_destructor(ctx, messaging_context_destructor);
 274
 275         if (lp_clustering()) {
 276                 status = messaging_ctdbd_init(ctx, ctx, &ctx->remote);
 277
 278                 if (!NT_STATUS_IS_OK(status)) {
 279                         DEBUG(2, ("messaging_ctdbd_init failed: %s\n",
 280                                   nt_errstr(status)));
 281                         TALLOC_FREE(ctx);
 282                         return NULL;
 283                 }
 284         }
 285         ctx->id.vnn = get_my_vnn();
 286
 287         messaging_register(ctx, NULL, MSG_PING, ping_message);
 288
 289         /* Register some debugging related messages */
 290
 291         register_msg_pool_usage(ctx);
 292         register_dmalloc_msgs(ctx);
 293         debug_register_msgs(ctx);
 294
 295         return ctx;
 296 }
 297
 298 struct server_id messaging_server_id(const struct messaging_context *msg_ctx)
 299 {
 300         return msg_ctx->id;
 301 }
 302
 303 /*
 304  * re-init after a fork
 305  */
 306 NTSTATUS messaging_reinit(struct messaging_context *msg_ctx)
 307 {
 308         NTSTATUS status;
 309         int ret;
 310
 311         messaging_dgm_destroy();
 312
 313         msg_ctx->id = procid_self();
 314
 315         ret = messaging_dgm_init(msg_ctx->event_ctx, msg_ctx->id,
 316                                  lp_cache_directory(), sec_initial_uid(),
 317                                  messaging_recv_cb, msg_ctx);
 318         if (ret != 0) {
 319                 DEBUG(0, ("messaging_dgm_init failed: %s\n", strerror(errno)));
 320                 return map_nt_error_from_unix(ret);
 321         }
 322
 323         TALLOC_FREE(msg_ctx->remote);
 324
 325         if (lp_clustering()) {
 326                 status = messaging_ctdbd_init(msg_ctx, msg_ctx,
 327                                               &msg_ctx->remote);
 328
 329                 if (!NT_STATUS_IS_OK(status)) {
 330                         DEBUG(1, ("messaging_ctdbd_init failed: %s\n",
 331                                   nt_errstr(status)));
 332                         return status;
 333                 }
 334         }
 335
 336         return NT_STATUS_OK;
 337 }
 338
 339
 340 /*
 341  * Register a dispatch function for a particular message type. Allow multiple
 342  * registrants
 343 */
 344 NTSTATUS messaging_register(struct messaging_context *msg_ctx,
 345                             void *private_data,
 346                             uint32_t msg_type,
 347                             void (*fn)(struct messaging_context *msg,
 348                                        void *private_data,
 349                                        uint32_t msg_type,
 350                                        struct server_id server_id,
 351                                        DATA_BLOB *data))
 352 {
 353         struct messaging_callback *cb;
 354
 355         DEBUG(5, ("Registering messaging pointer for type %u - "
 356                   "private_data=%p\n",
 357                   (unsigned)msg_type, private_data));
 358
 359         /*
 360          * Only one callback per type
 361          */
 362
 363         for (cb = msg_ctx->callbacks; cb != NULL; cb = cb->next) {
 364                 /* we allow a second registration of the same message
 365                    type if it has a different private pointer. This is
 366                    needed in, for example, the internal notify code,
 367                    which creates a new notify context for each tree
 368                    connect, and expects to receive messages to each of
 369                    them. */
 370                 if (cb->msg_type == msg_type && private_data == cb->private_data) {
 371                         DEBUG(5,("Overriding messaging pointer for type %u - private_data=%p\n",
 372                                   (unsigned)msg_type, private_data));
 373                         cb->fn = fn;
 374                         cb->private_data = private_data;
 375                         return NT_STATUS_OK;
 376                 }
 377         }
 378
 379         if (!(cb = talloc(msg_ctx, struct messaging_callback))) {
 380                 return NT_STATUS_NO_MEMORY;
 381         }
 382
 383         cb->msg_type = msg_type;
 384         cb->fn = fn;
 385         cb->private_data = private_data;
 386
 387         DLIST_ADD(msg_ctx->callbacks, cb);
 388         return NT_STATUS_OK;
 389 }
 390
 391 /*
 392   De-register the function for a particular message type.
 393 */
 394 void messaging_deregister(struct messaging_context *ctx, uint32_t msg_type,
 395                           void *private_data)
 396 {
 397         struct messaging_callback *cb, *next;
 398
 399         for (cb = ctx->callbacks; cb; cb = next) {
 400                 next = cb->next;
 401                 if ((cb->msg_type == msg_type)
 402                     && (cb->private_data == private_data)) {
 403                         DEBUG(5,("Deregistering messaging pointer for type %u - private_data=%p\n",
 404                                   (unsigned)msg_type, private_data));
 405                         DLIST_REMOVE(ctx->callbacks, cb);
 406                         TALLOC_FREE(cb);
 407                 }
 408         }
 409 }
 410
 411 /*
 412   Send a message to a particular server
 413 */
 414 NTSTATUS messaging_send(struct messaging_context *msg_ctx,
 415                         struct server_id server, uint32_t msg_type,
 416                         const DATA_BLOB *data)
 417 {
 418         struct iovec iov;
 419
 420         iov.iov_base = data->data;
 421         iov.iov_len = data->length;
 422
 423         return messaging_send_iov(msg_ctx, server, msg_type, &iov, 1);
 424 }
 425
 426 NTSTATUS messaging_send_buf(struct messaging_context *msg_ctx,
 427                             struct server_id server, uint32_t msg_type,
 428                             const uint8_t *buf, size_t len)
 429 {
 430         DATA_BLOB blob = data_blob_const(buf, len);
 431         return messaging_send(msg_ctx, server, msg_type, &blob);
 432 }
 433
 434 NTSTATUS messaging_send_iov(struct messaging_context *msg_ctx,
 435                             struct server_id server, uint32_t msg_type,
 436                             const struct iovec *iov, int iovlen)
 437 {
 438         int ret;
 439         struct messaging_hdr hdr;
 440         struct iovec iov2[iovlen+1];
 441
 442         if (server_id_is_disconnected(&server)) {
 443                 return NT_STATUS_INVALID_PARAMETER_MIX;
 444         }
 445
 446         if (!procid_is_local(&server)) {
 447                 ret = msg_ctx->remote->send_fn(msg_ctx->id, server,
 448                                                msg_type, iov, iovlen,
 449                                                msg_ctx->remote);
 450                 if (ret != 0) {
 451                         return map_nt_error_from_unix(ret);
 452                 }
 453                 return NT_STATUS_OK;
 454         }
 455
 456         if (server_id_same_process(&msg_ctx->id, &server)) {
 457                 struct messaging_rec rec;
 458                 uint8_t *buf;
 459
 460                 /*
 461                  * Self-send, directly dispatch
 462                  */
 463
 464                 buf = iov_buf(talloc_tos(), iov, iovlen);
 465                 if (buf == NULL) {
 466                         return NT_STATUS_NO_MEMORY;
 467                 }
 468
 469                 rec.msg_version = MESSAGE_VERSION;
 470                 rec.msg_type = msg_type & MSG_TYPE_MASK;
 471                 rec.dest = server;
 472                 rec.src = msg_ctx->id;
 473                 rec.buf = data_blob_const(buf, talloc_get_size(buf));
 474                 messaging_dispatch_rec(msg_ctx, &rec);
 475                 TALLOC_FREE(buf);
 476                 return NT_STATUS_OK;
 477         }
 478
 479         hdr = (struct messaging_hdr) {
 480                 .msg_type = msg_type,
 481                 .dst = server,
 482                 .src = msg_ctx->id
 483         };
 484         iov2[0] = (struct iovec){ .iov_base = &hdr, .iov_len = sizeof(hdr) };
 485         memcpy(&iov2[1], iov, iovlen * sizeof(*iov));
 486
 487         become_root();
 488         ret = messaging_dgm_send(server.pid, iov2, iovlen+1);
 489         unbecome_root();
 490
 491         if (ret != 0) {
 492                 return map_nt_error_from_unix(ret);
 493         }
 494         return NT_STATUS_OK;
 495 }
 496
 497 static struct messaging_rec *messaging_rec_dup(TALLOC_CTX *mem_ctx,
 498                                                struct messaging_rec *rec)
 499 {
 500         struct messaging_rec *result;
 501
 502         result = talloc_pooled_object(mem_ctx, struct messaging_rec,
 503                                       1, rec->buf.length);
 504         if (result == NULL) {
 505                 return NULL;
 506         }
 507         *result = *rec;
 508
 509         /* Doesn't fail, see talloc_pooled_object */
 510
 511         result->buf.data = talloc_memdup(result, rec->buf.data,
 512                                          rec->buf.length);
 513         return result;
 514 }
 515
 516 struct messaging_filtered_read_state {
 517         struct tevent_context *ev;
 518         struct messaging_context *msg_ctx;
 519         void *tevent_handle;
 520
 521         bool (*filter)(struct messaging_rec *rec, void *private_data);
 522         void *private_data;
 523
 524         struct messaging_rec *rec;
 525 };
 526
 527 static void messaging_filtered_read_cleanup(struct tevent_req *req,
 528                                             enum tevent_req_state req_state);
 529
 530 struct tevent_req *messaging_filtered_read_send(
 531         TALLOC_CTX *mem_ctx, struct tevent_context *ev,
 532         struct messaging_context *msg_ctx,
 533         bool (*filter)(struct messaging_rec *rec, void *private_data),
 534         void *private_data)
 535 {
 536         struct tevent_req *req;
 537         struct messaging_filtered_read_state *state;
 538         size_t new_waiters_len;
 539
 540         req = tevent_req_create(mem_ctx, &state,
 541                                 struct messaging_filtered_read_state);
 542         if (req == NULL) {
 543                 return NULL;
 544         }
 545         state->ev = ev;
 546         state->msg_ctx = msg_ctx;
 547         state->filter = filter;
 548         state->private_data = private_data;
 549
 550         /*
 551          * We have to defer the callback here, as we might be called from
 552          * within a different tevent_context than state->ev
 553          */
 554         tevent_req_defer_callback(req, state->ev);
 555
 556         state->tevent_handle = messaging_dgm_register_tevent_context(
 557                 state, ev);
 558         if (tevent_req_nomem(state, req)) {
 559                 return tevent_req_post(req, ev);
 560         }
 561
 562         /*
 563          * We add ourselves to the "new_waiters" array, not the "waiters"
 564          * array. If we are called from within messaging_read_done,
 565          * messaging_dispatch_rec will be in an active for-loop on
 566          * "waiters". We must be careful not to mess with this array, because
 567          * it could mean that a single event is being delivered twice.
 568          */
 569
 570         new_waiters_len = talloc_array_length(msg_ctx->new_waiters);
 571
 572         if (new_waiters_len == msg_ctx->num_new_waiters) {
 573                 struct tevent_req **tmp;
 574
 575                 tmp = talloc_realloc(msg_ctx, msg_ctx->new_waiters,
 576                                      struct tevent_req *, new_waiters_len+1);
 577                 if (tevent_req_nomem(tmp, req)) {
 578                         return tevent_req_post(req, ev);
 579                 }
 580                 msg_ctx->new_waiters = tmp;
 581         }
 582
 583         msg_ctx->new_waiters[msg_ctx->num_new_waiters] = req;
 584         msg_ctx->num_new_waiters += 1;
 585         tevent_req_set_cleanup_fn(req, messaging_filtered_read_cleanup);
 586
 587         return req;
 588 }
 589
 590 static void messaging_filtered_read_cleanup(struct tevent_req *req,
 591                                             enum tevent_req_state req_state)
 592 {
 593         struct messaging_filtered_read_state *state = tevent_req_data(
 594                 req, struct messaging_filtered_read_state);
 595         struct messaging_context *msg_ctx = state->msg_ctx;
 596         unsigned i;
 597
 598         tevent_req_set_cleanup_fn(req, NULL);
 599
 600         TALLOC_FREE(state->tevent_handle);
 601
 602         /*
 603          * Just set the [new_]waiters entry to NULL, be careful not to mess
 604          * with the other "waiters" array contents. We are often called from
 605          * within "messaging_dispatch_rec", which loops over
 606          * "waiters". Messing with the "waiters" array will mess up that
 607          * for-loop.
 608          */
 609
 610         for (i=0; i<msg_ctx->num_waiters; i++) {
 611                 if (msg_ctx->waiters[i] == req) {
 612                         msg_ctx->waiters[i] = NULL;
 613                         return;
 614                 }
 615         }
 616
 617         for (i=0; i<msg_ctx->num_new_waiters; i++) {
 618                 if (msg_ctx->new_waiters[i] == req) {
 619                         msg_ctx->new_waiters[i] = NULL;
 620                         return;
 621                 }
 622         }
 623 }
 624
 625 static void messaging_filtered_read_done(struct tevent_req *req,
 626                                          struct messaging_rec *rec)
 627 {
 628         struct messaging_filtered_read_state *state = tevent_req_data(
 629                 req, struct messaging_filtered_read_state);
 630
 631         state->rec = messaging_rec_dup(state, rec);
 632         if (tevent_req_nomem(state->rec, req)) {
 633                 return;
 634         }
 635         tevent_req_done(req);
 636 }
 637
 638 int messaging_filtered_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
 639                                  struct messaging_rec **presult)
 640 {
 641         struct messaging_filtered_read_state *state = tevent_req_data(
 642                 req, struct messaging_filtered_read_state);
 643         int err;
 644
 645         if (tevent_req_is_unix_error(req, &err)) {
 646                 tevent_req_received(req);
 647                 return err;
 648         }
 649         *presult = talloc_move(mem_ctx, &state->rec);
 650         return 0;
 651 }
 652
 653 struct messaging_read_state {
 654         uint32_t msg_type;
 655         struct messaging_rec *rec;
 656 };
 657
 658 static bool messaging_read_filter(struct messaging_rec *rec,
 659                                   void *private_data);
 660 static void messaging_read_done(struct tevent_req *subreq);
 661
 662 struct tevent_req *messaging_read_send(TALLOC_CTX *mem_ctx,
 663                                        struct tevent_context *ev,
 664                                        struct messaging_context *msg,
 665                                        uint32_t msg_type)
 666 {
 667         struct tevent_req *req, *subreq;
 668         struct messaging_read_state *state;
 669
 670         req = tevent_req_create(mem_ctx, &state,
 671                                 struct messaging_read_state);
 672         if (req == NULL) {
 673                 return NULL;
 674         }
 675         state->msg_type = msg_type;
 676
 677         subreq = messaging_filtered_read_send(state, ev, msg,
 678                                               messaging_read_filter, state);
 679         if (tevent_req_nomem(subreq, req)) {
 680                 return tevent_req_post(req, ev);
 681         }
 682         tevent_req_set_callback(subreq, messaging_read_done, req);
 683         return req;
 684 }
 685
 686 static bool messaging_read_filter(struct messaging_rec *rec,
 687                                   void *private_data)
 688 {
 689         struct messaging_read_state *state = talloc_get_type_abort(
 690                 private_data, struct messaging_read_state);
 691
 692         return rec->msg_type == state->msg_type;
 693 }
 694
 695 static void messaging_read_done(struct tevent_req *subreq)
 696 {
 697         struct tevent_req *req = tevent_req_callback_data(
 698                 subreq, struct tevent_req);
 699         struct messaging_read_state *state = tevent_req_data(
 700                 req, struct messaging_read_state);
 701         int ret;
 702
 703         ret = messaging_filtered_read_recv(subreq, state, &state->rec);
 704         TALLOC_FREE(subreq);
 705         if (tevent_req_error(req, ret)) {
 706                 return;
 707         }
 708         tevent_req_done(req);
 709 }
 710
 711 int messaging_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
 712                         struct messaging_rec **presult)
 713 {
 714         struct messaging_read_state *state = tevent_req_data(
 715                 req, struct messaging_read_state);
 716         int err;
 717
 718         if (tevent_req_is_unix_error(req, &err)) {
 719                 return err;
 720         }
 721         if (presult != NULL) {
 722                 *presult = talloc_move(mem_ctx, &state->rec);
 723         }
 724         return 0;
 725 }
 726
 727 static bool messaging_append_new_waiters(struct messaging_context *msg_ctx)
 728 {
 729         if (msg_ctx->num_new_waiters == 0) {
 730                 return true;
 731         }
 732
 733         if (talloc_array_length(msg_ctx->waiters) <
 734             (msg_ctx->num_waiters + msg_ctx->num_new_waiters)) {
 735                 struct tevent_req **tmp;
 736                 tmp = talloc_realloc(
 737                         msg_ctx, msg_ctx->waiters, struct tevent_req *,
 738                         msg_ctx->num_waiters + msg_ctx->num_new_waiters);
 739                 if (tmp == NULL) {
 740                         DEBUG(1, ("%s: talloc failed\n", __func__));
 741                         return false;
 742                 }
 743                 msg_ctx->waiters = tmp;
 744         }
 745
 746         memcpy(&msg_ctx->waiters[msg_ctx->num_waiters], msg_ctx->new_waiters,
 747                sizeof(struct tevent_req *) * msg_ctx->num_new_waiters);
 748
 749         msg_ctx->num_waiters += msg_ctx->num_new_waiters;
 750         msg_ctx->num_new_waiters = 0;
 751
 752         return true;
 753 }
 754
 755 struct messaging_defer_callback_state {
 756         struct messaging_context *msg_ctx;
 757         struct messaging_rec *rec;
 758         void (*fn)(struct messaging_context *msg, void *private_data,
 759                    uint32_t msg_type, struct server_id server_id,
 760                    DATA_BLOB *data);
 761         void *private_data;
 762 };
 763
 764 static void messaging_defer_callback_trigger(struct tevent_context *ev,
 765                                              struct tevent_immediate *im,
 766                                              void *private_data);
 767
 768 static void messaging_defer_callback(
 769         struct messaging_context *msg_ctx, struct messaging_rec *rec,
 770         void (*fn)(struct messaging_context *msg, void *private_data,
 771                    uint32_t msg_type, struct server_id server_id,
 772                    DATA_BLOB *data),
 773         void *private_data)
 774 {
 775         struct messaging_defer_callback_state *state;
 776         struct tevent_immediate *im;
 777
 778         state = talloc(msg_ctx, struct messaging_defer_callback_state);
 779         if (state == NULL) {
 780                 DEBUG(1, ("talloc failed\n"));
 781                 return;
 782         }
 783         state->msg_ctx = msg_ctx;
 784         state->fn = fn;
 785         state->private_data = private_data;
 786
 787         state->rec = messaging_rec_dup(state, rec);
 788         if (state->rec == NULL) {
 789                 DEBUG(1, ("talloc failed\n"));
 790                 TALLOC_FREE(state);
 791                 return;
 792         }
 793
 794         im = tevent_create_immediate(state);
 795         if (im == NULL) {
 796                 DEBUG(1, ("tevent_create_immediate failed\n"));
 797                 TALLOC_FREE(state);
 798                 return;
 799         }
 800         tevent_schedule_immediate(im, msg_ctx->event_ctx,
 801                                   messaging_defer_callback_trigger, state);
 802 }
 803
 804 static void messaging_defer_callback_trigger(struct tevent_context *ev,
 805                                              struct tevent_immediate *im,
 806                                              void *private_data)
 807 {
 808         struct messaging_defer_callback_state *state = talloc_get_type_abort(
 809                 private_data, struct messaging_defer_callback_state);
 810         struct messaging_rec *rec = state->rec;
 811
 812         state->fn(state->msg_ctx, state->private_data, rec->msg_type, rec->src,
 813                   &rec->buf);
 814         TALLOC_FREE(state);
 815 }
 816
 817 /*
 818   Dispatch one messaging_rec
 819 */
 820 void messaging_dispatch_rec(struct messaging_context *msg_ctx,
 821                             struct messaging_rec *rec)
 822 {
 823         struct messaging_callback *cb, *next;
 824         unsigned i;
 825
 826         for (cb = msg_ctx->callbacks; cb != NULL; cb = next) {
 827                 next = cb->next;
 828                 if (cb->msg_type != rec->msg_type) {
 829                         continue;
 830                 }
 831
 832                 if (server_id_same_process(&rec->src, &rec->dest)) {
 833                         /*
 834                          * This is a self-send. We are called here from
 835                          * messaging_send(), and we don't want to directly
 836                          * recurse into the callback but go via a
 837                          * tevent_loop_once
 838                          */
 839                         messaging_defer_callback(msg_ctx, rec, cb->fn,
 840                                                  cb->private_data);
 841                 } else {
 842                         /*
 843                          * This comes from a different process. we are called
 844                          * from the event loop, so we should call back
 845                          * directly.
 846                          */
 847                         cb->fn(msg_ctx, cb->private_data, rec->msg_type,
 848                                rec->src, &rec->buf);
 849                 }
 850                 /*
 851                  * we continue looking for matching messages after finding
 852                  * one. This matters for subsystems like the internal notify
 853                  * code which register more than one handler for the same
 854                  * message type
 855                  */
 856         }
 857
 858         if (!messaging_append_new_waiters(msg_ctx)) {
 859                 return;
 860         }
 861
 862         i = 0;
 863         while (i < msg_ctx->num_waiters) {
 864                 struct tevent_req *req;
 865                 struct messaging_filtered_read_state *state;
 866
 867                 req = msg_ctx->waiters[i];
 868                 if (req == NULL) {
 869                         /*
 870                          * This got cleaned up. In the meantime,
 871                          * move everything down one. We need
 872                          * to keep the order of waiters, as
 873                          * other code may depend on this.
 874                          */
 875                         if (i < msg_ctx->num_waiters - 1) {
 876                                 memmove(&msg_ctx->waiters[i],
 877                                         &msg_ctx->waiters[i+1],
 878                                         sizeof(struct tevent_req *) *
 879                                             (msg_ctx->num_waiters - i - 1));
 880                         }
 881                         msg_ctx->num_waiters -= 1;
 882                         continue;
 883                 }
 884
 885                 state = tevent_req_data(
 886                         req, struct messaging_filtered_read_state);
 887                 if (state->filter(rec, state->private_data)) {
 888                         messaging_filtered_read_done(req, rec);
 889                 }
 890
 891                 i += 1;
 892         }
 893 }
 894
 895 static int mess_parent_dgm_cleanup(void *private_data);
 896 static void mess_parent_dgm_cleanup_done(struct tevent_req *req);
 897
 898 bool messaging_parent_dgm_cleanup_init(struct messaging_context *msg)
 899 {
 900         struct tevent_req *req;
 901
 902         req = background_job_send(
 903                 msg, msg->event_ctx, msg, NULL, 0,
 904                 lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
 905                             60*15),
 906                 mess_parent_dgm_cleanup, msg);
 907         if (req == NULL) {
 908                 return false;
 909         }
 910         tevent_req_set_callback(req, mess_parent_dgm_cleanup_done, msg);
 911         return true;
 912 }
 913
 914 static int mess_parent_dgm_cleanup(void *private_data)
 915 {
 916         int ret;
 917
 918         ret = messaging_dgm_wipe();
 919         DEBUG(10, ("messaging_dgm_wipe returned %s\n",
 920                    ret ? strerror(ret) : "ok"));
 921         return lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
 922                            60*15);
 923 }
 924
 925 static void mess_parent_dgm_cleanup_done(struct tevent_req *req)
 926 {
 927         struct messaging_context *msg = tevent_req_callback_data(
 928                 req, struct messaging_context);
 929         NTSTATUS status;
 930
 931         status = background_job_recv(req);
 932         TALLOC_FREE(req);
 933         DEBUG(1, ("messaging dgm cleanup job ended with %s\n",
 934                   nt_errstr(status)));
 935
 936         req = background_job_send(
 937                 msg, msg->event_ctx, msg, NULL, 0,
 938                 lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
 939                             60*15),
 940                 mess_parent_dgm_cleanup, msg);
 941         if (req == NULL) {
 942                 DEBUG(1, ("background_job_send failed\n"));
 943         }
 944         tevent_req_set_callback(req, mess_parent_dgm_cleanup_done, msg);
 945 }
 946
 947 int messaging_cleanup(struct messaging_context *msg_ctx, pid_t pid)
 948 {
 949         int ret;
 950
 951         if (pid == 0) {
 952                 ret = messaging_dgm_wipe();
 953         } else {
 954                 ret = messaging_dgm_cleanup(pid);
 955         }
 956
 957         return ret;
 958 }
 959
 960 struct tevent_context *messaging_tevent_context(
 961         struct messaging_context *msg_ctx)
 962 {
 963         return msg_ctx->event_ctx;
 964 }
 965
 966 /** @} **/