source/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 2 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, write to the Free Software
  21    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  22 */
  23
  24 /**
  25   @defgroup messages Internal messaging framework
  26   @{
  27   @file messages.c
  28
  29   @brief  Module for internal messaging between Samba daemons.
  30
  31    The idea is that if a part of Samba wants to do communication with
  32    another Samba process then it will do a message_register() of a
  33    dispatch function, and use message_send_pid() to send messages to
  34    that process.
  35
  36    The dispatch function is given the pid of the sender, and it can
  37    use that to reply by message_send_pid().  See ping_message() for a
  38    simple example.
  39
  40    @caution Dispatch functions must be able to cope with incoming
  41    messages on an *odd* byte boundary.
  42
  43    This system doesn't have any inherent size limitations but is not
  44    very efficient for large messages or when messages are sent in very
  45    quick succession.
  46
  47 */
  48
  49 #include "includes.h"
  50
  51 /* the locking database handle */
  52 static TDB_CONTEXT *tdb;
  53 static int received_signal;
  54
  55 /* change the message version with any incompatible changes in the protocol */
  56 #define MESSAGE_VERSION 1
  57
  58 struct message_rec {
  59         int msg_version;
  60         int msg_type;
  61         struct server_id dest;
  62         struct server_id src;
  63         size_t len;
  64 };
  65
  66 /* we have a linked list of dispatch handlers */
  67 static struct dispatch_fns {
  68         struct dispatch_fns *next, *prev;
  69         int msg_type;
  70         void (*fn)(int msg_type, struct server_id pid, void *buf, size_t len,
  71                    void *private_data);
  72         void *private_data;
  73 } *dispatch_fns;
  74
  75 static void message_register(int msg_type,
  76                              void (*fn)(int msg_type, struct server_id pid,
  77                                         void *buf, size_t len,
  78                                         void *private_data),
  79                              void *private_data);
  80
  81 /****************************************************************************
  82  Free global objects.
  83 ****************************************************************************/
  84
  85 void gfree_messages(void)
  86 {
  87         struct dispatch_fns *dfn, *next;
  88
  89         /* delete the dispatch_fns list */
  90         dfn = dispatch_fns;
  91         while( dfn ) {
  92                 next = dfn->next;
  93                 DLIST_REMOVE(dispatch_fns, dfn);
  94                 SAFE_FREE(dfn);
  95                 dfn = next;
  96         }
  97 }
  98
  99 /****************************************************************************
 100  Notifications come in as signals.
 101 ****************************************************************************/
 102
 103 static void sig_usr1(void)
 104 {
 105         received_signal = 1;
 106         sys_select_signal(SIGUSR1);
 107 }
 108
 109 static NTSTATUS message_send_pid(struct server_id pid, int msg_type,
 110                                  const void *buf, size_t len);
 111
 112 /****************************************************************************
 113  A useful function for testing the message system.
 114 ****************************************************************************/
 115
 116 static void ping_message(int msg_type, struct server_id src,
 117                          void *buf, size_t len, void *private_data)
 118 {
 119         const char *msg = buf ? (const char *)buf : "none";
 120
 121         DEBUG(1,("INFO: Received PING message from PID %s [%s]\n",
 122                  procid_str_static(&src), msg));
 123         message_send_pid(src, MSG_PONG, buf, len);
 124 }
 125
 126 /****************************************************************************
 127  Initialise the messaging functions.
 128 ****************************************************************************/
 129
 130 static BOOL message_init(struct messaging_context *msg_ctx)
 131 {
 132         sec_init();
 133
 134         if (tdb)
 135                 return True;
 136
 137         tdb = tdb_open_log(lock_path("messages.tdb"),
 138                        0, TDB_CLEAR_IF_FIRST|TDB_DEFAULT,
 139                        O_RDWR|O_CREAT,0600);
 140
 141         if (!tdb) {
 142                 DEBUG(0,("ERROR: Failed to initialise messages database\n"));
 143                 return False;
 144         }
 145
 146         /* Activate the per-hashchain freelist */
 147         tdb_set_max_dead(tdb, 5);
 148
 149         CatchSignal(SIGUSR1, SIGNAL_CAST sig_usr1);
 150
 151         message_register(MSG_PING, ping_message, NULL);
 152
 153         /* Register some debugging related messages */
 154
 155         register_msg_pool_usage(msg_ctx);
 156         register_dmalloc_msgs(msg_ctx);
 157         debug_register_msgs(msg_ctx);
 158
 159         return True;
 160 }
 161
 162 /*******************************************************************
 163  Form a static tdb key from a pid.
 164 ******************************************************************/
 165
 166 static TDB_DATA message_key_pid(struct server_id pid)
 167 {
 168         static char key[20];
 169         TDB_DATA kbuf;
 170
 171         slprintf(key, sizeof(key)-1, "PID/%s", procid_str_static(&pid));
 172
 173         kbuf.dptr = (uint8 *)key;
 174         kbuf.dsize = strlen(key)+1;
 175         return kbuf;
 176 }
 177
 178 /****************************************************************************
 179  Notify a process that it has a message. If the process doesn't exist
 180  then delete its record in the database.
 181 ****************************************************************************/
 182
 183 static NTSTATUS message_notify(struct server_id procid)
 184 {
 185         pid_t pid = procid.pid;
 186         int ret;
 187         uid_t euid = geteuid();
 188
 189         /*
 190          * Doing kill with a non-positive pid causes messages to be
 191          * sent to places we don't want.
 192          */
 193
 194         SMB_ASSERT(pid > 0);
 195
 196         if (euid != 0) {
 197                 /* If we're not root become so to send the message. */
 198                 save_re_uid();
 199                 set_effective_uid(0);
 200         }
 201
 202         ret = kill(pid, SIGUSR1);
 203
 204         if (euid != 0) {
 205                 /* Go back to who we were. */
 206                 int saved_errno = errno;
 207                 restore_re_uid_fromroot();
 208                 errno = saved_errno;
 209         }
 210
 211         if (ret == 0) {
 212                 return NT_STATUS_OK;
 213         }
 214
 215         /*
 216          * Something has gone wrong
 217          */
 218
 219         if (errno == ESRCH) {
 220                 DEBUG(2,("pid %d doesn't exist - deleting messages record\n",
 221                          (int)pid));
 222                 tdb_delete(tdb, message_key_pid(procid));
 223
 224                 /*
 225                  * INVALID_HANDLE is the closest I can think of -- vl
 226                  */
 227                 return NT_STATUS_INVALID_HANDLE;
 228         }
 229
 230         DEBUG(2,("message to process %d failed - %s\n", (int)pid,
 231                  strerror(errno)));
 232
 233         /*
 234          * No call to map_nt_error_from_unix -- don't want to link in
 235          * errormap.o into lots of utils.
 236          */
 237
 238         if (errno == EINVAL) return NT_STATUS_INVALID_PARAMETER;
 239         if (errno == EPERM)  return NT_STATUS_ACCESS_DENIED;
 240         return NT_STATUS_UNSUCCESSFUL;
 241 }
 242
 243 /****************************************************************************
 244  Send a message to a particular pid.
 245 ****************************************************************************/
 246
 247 static NTSTATUS message_send_pid(struct server_id pid, int msg_type,
 248                                  const void *buf, size_t len)
 249 {
 250         TDB_DATA dbuf;
 251         struct message_rec rec;
 252         int ret;
 253
 254         /* NULL pointer means implicit length zero. */
 255         if (!buf) {
 256                 SMB_ASSERT(len == 0);
 257         }
 258
 259         /*
 260          * Doing kill with a non-positive pid causes messages to be
 261          * sent to places we don't want.
 262          */
 263
 264         SMB_ASSERT(procid_to_pid(&pid) > 0);
 265
 266         rec.msg_version = MESSAGE_VERSION;
 267         rec.msg_type = msg_type;
 268         rec.dest = pid;
 269         rec.src = procid_self();
 270         rec.len = buf ? len : 0;
 271
 272         dbuf.dptr = (uint8 *)SMB_MALLOC(len + sizeof(rec));
 273         if (!dbuf.dptr) {
 274                 return NT_STATUS_NO_MEMORY;
 275         }
 276
 277         memcpy(dbuf.dptr, &rec, sizeof(rec));
 278         if (len > 0 && buf)
 279                 memcpy((void *)((char*)dbuf.dptr+sizeof(rec)), buf, len);
 280
 281         dbuf.dsize = len + sizeof(rec);
 282
 283         ret = tdb_append(tdb, message_key_pid(pid), dbuf);
 284
 285         SAFE_FREE(dbuf.dptr);
 286
 287         if (ret == -1) {
 288                 return NT_STATUS_INTERNAL_ERROR;
 289         }
 290
 291         errno = 0;                    /* paranoia */
 292         return message_notify(pid);
 293 }
 294
 295 /****************************************************************************
 296  Count the messages pending for a particular pid. Expensive....
 297 ****************************************************************************/
 298
 299 unsigned int messages_pending_for_pid(struct server_id pid)
 300 {
 301         TDB_DATA dbuf;
 302         uint8 *buf;
 303         unsigned int message_count = 0;
 304
 305         dbuf = tdb_fetch(tdb, message_key_pid(pid));
 306         if (dbuf.dptr == NULL || dbuf.dsize == 0) {
 307                 SAFE_FREE(dbuf.dptr);
 308                 return 0;
 309         }
 310
 311         for (buf = dbuf.dptr; dbuf.dsize > sizeof(struct message_rec);) {
 312                 struct message_rec rec;
 313                 memcpy(&rec, buf, sizeof(rec));
 314                 buf += (sizeof(rec) + rec.len);
 315                 dbuf.dsize -= (sizeof(rec) + rec.len);
 316                 message_count++;
 317         }
 318
 319         SAFE_FREE(dbuf.dptr);
 320         return message_count;
 321 }
 322
 323 /****************************************************************************
 324  Retrieve all messages for the current process.
 325 ****************************************************************************/
 326
 327 static BOOL retrieve_all_messages(char **msgs_buf, size_t *total_len)
 328 {
 329         TDB_DATA kbuf;
 330         TDB_DATA dbuf;
 331         TDB_DATA null_dbuf;
 332
 333         ZERO_STRUCT(null_dbuf);
 334
 335         *msgs_buf = NULL;
 336         *total_len = 0;
 337
 338         kbuf = message_key_pid(procid_self());
 339
 340         if (tdb_chainlock(tdb, kbuf) == -1)
 341                 return False;
 342
 343         dbuf = tdb_fetch(tdb, kbuf);
 344         /*
 345          * Replace with an empty record to keep the allocated
 346          * space in the tdb.
 347          */
 348         tdb_store(tdb, kbuf, null_dbuf, TDB_REPLACE);
 349         tdb_chainunlock(tdb, kbuf);
 350
 351         if (dbuf.dptr == NULL || dbuf.dsize == 0) {
 352                 SAFE_FREE(dbuf.dptr);
 353                 return False;
 354         }
 355
 356         *msgs_buf = (char *)dbuf.dptr;
 357         *total_len = dbuf.dsize;
 358
 359         return True;
 360 }
 361
 362 /****************************************************************************
 363  Parse out the next message for the current process.
 364 ****************************************************************************/
 365
 366 static BOOL message_recv(char *msgs_buf, size_t total_len, int *msg_type,
 367                          struct server_id *src, char **buf, size_t *len)
 368 {
 369         struct message_rec rec;
 370         char *ret_buf = *buf;
 371
 372         *buf = NULL;
 373         *len = 0;
 374
 375         if (total_len - (ret_buf - msgs_buf) < sizeof(rec))
 376                 return False;
 377
 378         memcpy(&rec, ret_buf, sizeof(rec));
 379         ret_buf += sizeof(rec);
 380
 381         if (rec.msg_version != MESSAGE_VERSION) {
 382                 DEBUG(0,("message version %d received (expected %d)\n",
 383                          rec.msg_version, MESSAGE_VERSION));
 384                 return False;
 385         }
 386
 387         if (rec.len > 0) {
 388                 if (total_len - (ret_buf - msgs_buf) < rec.len)
 389                         return False;
 390         }
 391
 392         *len = rec.len;
 393         *msg_type = rec.msg_type;
 394         *src = rec.src;
 395         *buf = ret_buf;
 396
 397         return True;
 398 }
 399
 400 /****************************************************************************
 401  Receive and dispatch any messages pending for this process.
 402  JRA changed Dec 13 2006. Only one message handler now permitted per type.
 403  *NOTE*: Dispatch functions must be able to cope with incoming
 404  messages on an *odd* byte boundary.
 405 ****************************************************************************/
 406
 407 void message_dispatch(void)
 408 {
 409         int msg_type;
 410         struct server_id src;
 411         char *buf;
 412         char *msgs_buf;
 413         size_t len, total_len;
 414         int n_handled;
 415
 416         if (!received_signal)
 417                 return;
 418
 419         DEBUG(10, ("message_dispatch: received_signal = %d\n",
 420                    received_signal));
 421
 422         received_signal = 0;
 423
 424         if (!retrieve_all_messages(&msgs_buf, &total_len))
 425                 return;
 426
 427         for (buf = msgs_buf;
 428              message_recv(msgs_buf, total_len, &msg_type, &src, &buf, &len);
 429              buf += len) {
 430                 struct dispatch_fns *dfn;
 431
 432                 DEBUG(10,("message_dispatch: received msg_type=%d "
 433                           "src_pid=%u\n", msg_type,
 434                           (unsigned int) procid_to_pid(&src)));
 435
 436                 n_handled = 0;
 437                 for (dfn = dispatch_fns; dfn; dfn = dfn->next) {
 438                         if (dfn->msg_type == msg_type) {
 439                                 DEBUG(10,("message_dispatch: processing "
 440                                           "message of type %d.\n", msg_type));
 441                                 dfn->fn(msg_type, src,
 442                                         len ? (void *)buf : NULL, len,
 443                                         dfn->private_data);
 444                                 n_handled++;
 445                                 break;
 446                         }
 447                 }
 448                 if (!n_handled) {
 449                         DEBUG(5,("message_dispatch: warning: no handler "
 450                                  "registed for msg_type %d in pid %u\n",
 451                                  msg_type, (unsigned int)sys_getpid()));
 452                 }
 453         }
 454         SAFE_FREE(msgs_buf);
 455 }
 456
 457 /****************************************************************************
 458  Register/replace a dispatch function for a particular message type.
 459  JRA changed Dec 13 2006. Only one message handler now permitted per type.
 460  *NOTE*: Dispatch functions must be able to cope with incoming
 461  messages on an *odd* byte boundary.
 462 ****************************************************************************/
 463
 464 static void message_register(int msg_type,
 465                              void (*fn)(int msg_type, struct server_id pid,
 466                                         void *buf, size_t len,
 467                                         void *private_data),
 468                              void *private_data)
 469 {
 470         struct dispatch_fns *dfn;
 471
 472         for (dfn = dispatch_fns; dfn; dfn = dfn->next) {
 473                 if (dfn->msg_type == msg_type) {
 474                         dfn->fn = fn;
 475                         return;
 476                 }
 477         }
 478
 479         if (!(dfn = SMB_MALLOC_P(struct dispatch_fns))) {
 480                 DEBUG(0,("message_register: Not enough memory. malloc "
 481                          "failed!\n"));
 482                 return;
 483         }
 484
 485         ZERO_STRUCTPN(dfn);
 486
 487         dfn->msg_type = msg_type;
 488         dfn->fn = fn;
 489         dfn->private_data = private_data;
 490
 491         DLIST_ADD(dispatch_fns, dfn);
 492 }
 493
 494 /****************************************************************************
 495  De-register the function for a particular message type.
 496 ****************************************************************************/
 497
 498 static void message_deregister(int msg_type)
 499 {
 500         struct dispatch_fns *dfn, *next;
 501
 502         for (dfn = dispatch_fns; dfn; dfn = next) {
 503                 next = dfn->next;
 504                 if (dfn->msg_type == msg_type) {
 505                         DLIST_REMOVE(dispatch_fns, dfn);
 506                         SAFE_FREE(dfn);
 507                         return;
 508                 }
 509         }
 510 }
 511
 512 struct msg_all {
 513         int msg_type;
 514         uint32 msg_flag;
 515         const void *buf;
 516         size_t len;
 517         BOOL duplicates;
 518         int n_sent;
 519 };
 520
 521 /****************************************************************************
 522  Send one of the messages for the broadcast.
 523 ****************************************************************************/
 524
 525 static int traverse_fn(TDB_CONTEXT *the_tdb, TDB_DATA kbuf, TDB_DATA dbuf,
 526                        void *state)
 527 {
 528         struct connections_data crec;
 529         struct msg_all *msg_all = (struct msg_all *)state;
 530         NTSTATUS status;
 531
 532         if (dbuf.dsize != sizeof(crec))
 533                 return 0;
 534
 535         memcpy(&crec, dbuf.dptr, sizeof(crec));
 536
 537         if (crec.cnum != -1)
 538                 return 0;
 539
 540         /* Don't send if the receiver hasn't registered an interest. */
 541
 542         if(!(crec.bcast_msg_flags & msg_all->msg_flag))
 543                 return 0;
 544
 545         /* If the msg send fails because the pid was not found (i.e. smbd died),
 546          * the msg has already been deleted from the messages.tdb.*/
 547
 548         status = message_send_pid(crec.pid, msg_all->msg_type,
 549                                   msg_all->buf, msg_all->len);
 550
 551         if (NT_STATUS_EQUAL(status, NT_STATUS_INVALID_HANDLE)) {
 552
 553                 /* If the pid was not found delete the entry from
 554                  * connections.tdb */
 555
 556                 DEBUG(2,("pid %s doesn't exist - deleting connections "
 557                          "%d [%s]\n", procid_str_static(&crec.pid), crec.cnum,
 558                          crec.servicename));
 559                 tdb_delete(the_tdb, kbuf);
 560         }
 561         msg_all->n_sent++;
 562         return 0;
 563 }
 564
 565 /**
 566  * Send a message to all smbd processes.
 567  *
 568  * It isn't very efficient, but should be OK for the sorts of
 569  * applications that use it. When we need efficient broadcast we can add
 570  * it.
 571  *
 572  * @param n_sent Set to the number of messages sent.  This should be
 573  * equal to the number of processes, but be careful for races.
 574  *
 575  * @retval True for success.
 576  **/
 577 BOOL message_send_all(struct messaging_context *msg_ctx,
 578                       int msg_type,
 579                       const void *buf, size_t len,
 580                       BOOL duplicates_allowed,
 581                       int *n_sent)
 582 {
 583         struct msg_all msg_all;
 584
 585         msg_all.msg_type = msg_type;
 586         if (msg_type < 1000)
 587                 msg_all.msg_flag = FLAG_MSG_GENERAL;
 588         else if (msg_type > 1000 && msg_type < 2000)
 589                 msg_all.msg_flag = FLAG_MSG_NMBD;
 590         else if (msg_type > 2000 && msg_type < 2100)
 591                 msg_all.msg_flag = FLAG_MSG_PRINT_NOTIFY;
 592         else if (msg_type > 2100 && msg_type < 3000)
 593                 msg_all.msg_flag = FLAG_MSG_PRINT_GENERAL;
 594         else if (msg_type > 3000 && msg_type < 4000)
 595                 msg_all.msg_flag = FLAG_MSG_SMBD;
 596         else
 597                 return False;
 598
 599         msg_all.buf = buf;
 600         msg_all.len = len;
 601         msg_all.duplicates = duplicates_allowed;
 602         msg_all.n_sent = 0;
 603
 604         connections_traverse(traverse_fn, &msg_all);
 605         if (n_sent)
 606                 *n_sent = msg_all.n_sent;
 607         return True;
 608 }
 609
 610 /*
 611  * Block and unblock receiving of messages. Allows removal of race conditions
 612  * when doing a fork and changing message disposition.
 613  */
 614
 615 void message_block(void)
 616 {
 617         BlockSignals(True, SIGUSR1);
 618 }
 619
 620 void message_unblock(void)
 621 {
 622         BlockSignals(False, SIGUSR1);
 623 }
 624
 625 /*
 626  * Samba4 API wrapper around the Samba3 implementation. Yes, I know, we could
 627  * import the whole Samba4 thing, but I want notify.c from Samba4 in first.
 628  */
 629
 630 struct messaging_callback {
 631         struct messaging_callback *prev, *next;
 632         uint32 msg_type;
 633         void (*fn)(struct messaging_context *msg, void *private_data,
 634                    uint32_t msg_type,
 635                    struct server_id server_id, DATA_BLOB *data);
 636         void *private_data;
 637 };
 638
 639 struct messaging_context {
 640         struct server_id id;
 641         struct event_context *event_ctx;
 642         struct messaging_callback *callbacks;
 643 };
 644
 645 static int messaging_context_destructor(struct messaging_context *ctx)
 646 {
 647         struct messaging_callback *cb;
 648
 649         for (cb = ctx->callbacks; cb; cb = cb->next) {
 650                 /*
 651                  * We unconditionally remove all instances of our callback
 652                  * from the tdb basis.
 653                  */
 654                 message_deregister(cb->msg_type);
 655         }
 656         return 0;
 657 }
 658
 659 struct event_context *messaging_event_context(struct messaging_context *msg_ctx)
 660 {
 661         return msg_ctx->event_ctx;
 662 }
 663
 664 struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx,
 665                                          struct server_id server_id,
 666                                          struct event_context *ev)
 667 {
 668         struct messaging_context *ctx;
 669
 670         if (!(ctx = TALLOC_ZERO_P(mem_ctx, struct messaging_context))) {
 671                 return NULL;
 672         }
 673
 674         ctx->id = server_id;
 675         ctx->event_ctx = ev;
 676         talloc_set_destructor(ctx, messaging_context_destructor);
 677
 678         if (!message_init(ctx)) {
 679                 DEBUG(0, ("message_init failed: %s\n", strerror(errno)));
 680                 TALLOC_FREE(ctx);
 681         }
 682
 683         return ctx;
 684 }
 685
 686 static void messaging_callback(int msg_type, struct server_id pid,
 687                                void *buf, size_t len, void *private_data)
 688 {
 689         struct messaging_context *ctx = talloc_get_type_abort(
 690                 private_data, struct messaging_context);
 691         struct messaging_callback *cb, *next;
 692
 693         for (cb = ctx->callbacks; cb; cb = next) {
 694                 /*
 695                  * Allow a callback to remove itself
 696                  */
 697                 next = cb->next;
 698
 699                 if (msg_type == cb->msg_type) {
 700                         DATA_BLOB blob;
 701
 702                         blob.data = (uint8 *)buf;
 703                         blob.length = len;
 704
 705                         cb->fn(ctx, cb->private_data, msg_type, pid, &blob);
 706                 }
 707         }
 708 }
 709
 710 /*
 711  * Register a dispatch function for a particular message type. Allow multiple
 712  * registrants
 713 */
 714 NTSTATUS messaging_register(struct messaging_context *ctx, void *private_data,
 715                             uint32_t msg_type,
 716                             void (*fn)(struct messaging_context *msg,
 717                                        void *private_data,
 718                                        uint32_t msg_type,
 719                                        struct server_id server_id,
 720                                        DATA_BLOB *data))
 721 {
 722         struct messaging_callback *cb;
 723
 724         if (!(cb = talloc(ctx, struct messaging_callback))) {
 725                 return NT_STATUS_NO_MEMORY;
 726         }
 727
 728         cb->msg_type = msg_type;
 729         cb->fn = fn;
 730         cb->private_data = private_data;
 731
 732         DLIST_ADD(ctx->callbacks, cb);
 733         message_register(msg_type, messaging_callback, ctx);
 734         return NT_STATUS_OK;
 735 }
 736
 737 /*
 738   De-register the function for a particular message type.
 739 */
 740 void messaging_deregister(struct messaging_context *ctx, uint32_t msg_type,
 741                           void *private_data)
 742 {
 743         struct messaging_callback *cb, *next;
 744
 745         for (cb = ctx->callbacks; cb; cb = next) {
 746                 next = cb->next;
 747                 if ((cb->msg_type == msg_type)
 748                     && (cb->private_data == private_data)) {
 749                         DLIST_REMOVE(ctx->callbacks, cb);
 750                         TALLOC_FREE(cb);
 751                 }
 752         }
 753 }
 754
 755 /*
 756   Send a message to a particular server
 757 */
 758 NTSTATUS messaging_send(struct messaging_context *msg_ctx,
 759                         struct server_id server,
 760                         uint32_t msg_type, const DATA_BLOB *data)
 761 {
 762         return message_send_pid(server, msg_type, data->data, data->length);
 763 }
 764
 765 NTSTATUS messaging_send_buf(struct messaging_context *msg_ctx,
 766                             struct server_id server, uint32_t msg_type,
 767                             const uint8 *buf, size_t len)
 768 {
 769         DATA_BLOB blob = data_blob_const(buf, len);
 770         return messaging_send(msg_ctx, server, msg_type, &blob);
 771 }
 772
 773 /** @} **/