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
   8    This program is free software; you can redistribute it and/or modify
   9    it under the terms of the GNU General Public License as published by
  10    the Free Software Foundation; either version 2 of the License, or
  11    (at your option) any later version.
  12
  13    This program is distributed in the hope that it will be useful,
  14    but WITHOUT ANY WARRANTY; without even the implied warranty of
  15    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16    GNU General Public License for more details.
  17
  18    You should have received a copy of the GNU General Public License
  19    along with this program; if not, write to the Free Software
  20    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  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
  50 /* the locking database handle */
  51 static TDB_CONTEXT *tdb;
  52 static int received_signal;
  53
  54 /* change the message version with any incompatible changes in the protocol */
  55 #define MESSAGE_VERSION 1
  56
  57 struct message_rec {
  58         int msg_version;
  59         int msg_type;
  60         struct process_id dest;
  61         struct process_id src;
  62         size_t len;
  63 };
  64
  65 /* we have a linked list of dispatch handlers */
  66 static struct dispatch_fns {
  67         struct dispatch_fns *next, *prev;
  68         int msg_type;
  69         void (*fn)(int msg_type, struct process_id pid, void *buf, size_t len);
  70 } *dispatch_fns;
  71
  72 /****************************************************************************
  73  Free global objects.
  74 ****************************************************************************/
  75
  76 void gfree_messsges(void)
  77 {
  78         struct dispatch_fns *dfn, *next;
  79
  80         /* delete the dispatch_fns list */
  81         dfn = dispatch_fns;
  82         while( dfn ) {
  83                 next = dfn->next;
  84                 DLIST_REMOVE(dispatch_fns, dfn);
  85                 SAFE_FREE(dfn);
  86                 dfn = next;
  87         }
  88 }
  89
  90 /****************************************************************************
  91  Notifications come in as signals.
  92 ****************************************************************************/
  93
  94 static void sig_usr1(void)
  95 {
  96         received_signal = 1;
  97         sys_select_signal(SIGUSR1);
  98 }
  99
 100 /****************************************************************************
 101  A useful function for testing the message system.
 102 ****************************************************************************/
 103
 104 static void ping_message(int msg_type, struct process_id src,
 105                          void *buf, size_t len)
 106 {
 107         const char *msg = buf ? buf : "none";
 108
 109         DEBUG(1,("INFO: Received PING message from PID %s [%s]\n",
 110                  procid_str_static(&src), msg));
 111         message_send_pid(src, MSG_PONG, buf, len, True);
 112 }
 113
 114 /****************************************************************************
 115  Initialise the messaging functions.
 116 ****************************************************************************/
 117
 118 BOOL message_init(void)
 119 {
 120         if (tdb) return True;
 121
 122         tdb = tdb_open_log(lock_path("messages.tdb"),
 123                        0, TDB_CLEAR_IF_FIRST|TDB_DEFAULT,
 124                        O_RDWR|O_CREAT,0600);
 125
 126         if (!tdb) {
 127                 DEBUG(0,("ERROR: Failed to initialise messages database\n"));
 128                 return False;
 129         }
 130
 131         CatchSignal(SIGUSR1, SIGNAL_CAST sig_usr1);
 132
 133         message_register(MSG_PING, ping_message);
 134
 135         /* Register some debugging related messages */
 136
 137         register_msg_pool_usage();
 138         register_dmalloc_msgs();
 139
 140         return True;
 141 }
 142
 143 /*******************************************************************
 144  Form a static tdb key from a pid.
 145 ******************************************************************/
 146
 147 static TDB_DATA message_key_pid(struct process_id pid)
 148 {
 149         static char key[20];
 150         TDB_DATA kbuf;
 151
 152         slprintf(key, sizeof(key)-1, "PID/%s", procid_str_static(&pid));
 153
 154         kbuf.dptr = (char *)key;
 155         kbuf.dsize = strlen(key)+1;
 156         return kbuf;
 157 }
 158
 159 /****************************************************************************
 160  Notify a process that it has a message. If the process doesn't exist
 161  then delete its record in the database.
 162 ****************************************************************************/
 163
 164 static BOOL message_notify(struct process_id procid)
 165 {
 166         pid_t pid = procid.pid;
 167         /*
 168          * Doing kill with a non-positive pid causes messages to be
 169          * sent to places we don't want.
 170          */
 171
 172         SMB_ASSERT(pid > 0);
 173
 174         if (kill(pid, SIGUSR1) == -1) {
 175                 if (errno == ESRCH) {
 176                         DEBUG(2,("pid %d doesn't exist - deleting messages record\n", (int)pid));
 177                         tdb_delete(tdb, message_key_pid(procid));
 178                 } else {
 179                         DEBUG(2,("message to process %d failed - %s\n", (int)pid, strerror(errno)));
 180                 }
 181                 return False;
 182         }
 183         return True;
 184 }
 185
 186 /****************************************************************************
 187  Send a message to a particular pid.
 188 ****************************************************************************/
 189
 190 static BOOL message_send_pid_internal(struct process_id pid, int msg_type,
 191                                       const void *buf, size_t len,
 192                                       BOOL duplicates_allowed,
 193                                       unsigned int timeout)
 194 {
 195         TDB_DATA kbuf;
 196         TDB_DATA dbuf;
 197         TDB_DATA old_dbuf;
 198         struct message_rec rec;
 199         char *ptr;
 200         struct message_rec prec;
 201
 202         /* NULL pointer means implicit length zero. */
 203         if (!buf) {
 204                 SMB_ASSERT(len == 0);
 205         }
 206
 207         /*
 208          * Doing kill with a non-positive pid causes messages to be
 209          * sent to places we don't want.
 210          */
 211
 212         SMB_ASSERT(procid_to_pid(&pid) > 0);
 213
 214         rec.msg_version = MESSAGE_VERSION;
 215         rec.msg_type = msg_type;
 216         rec.dest = pid;
 217         rec.src = procid_self();
 218         rec.len = buf ? len : 0;
 219
 220         kbuf = message_key_pid(pid);
 221
 222         dbuf.dptr = (void *)SMB_MALLOC(len + sizeof(rec));
 223         if (!dbuf.dptr)
 224                 return False;
 225
 226         memcpy(dbuf.dptr, &rec, sizeof(rec));
 227         if (len > 0 && buf)
 228                 memcpy((void *)((char*)dbuf.dptr+sizeof(rec)), buf, len);
 229
 230         dbuf.dsize = len + sizeof(rec);
 231
 232         if (duplicates_allowed) {
 233
 234                 /* If duplicates are allowed we can just append the message and return. */
 235
 236                 /* lock the record for the destination */
 237                 if (timeout) {
 238                         if (tdb_chainlock_with_timeout(tdb, kbuf, timeout) == -1) {
 239                                 DEBUG(0,("message_send_pid_internal: failed to get chainlock with timeout %ul.\n", timeout));
 240                                 return False;
 241                         }
 242                 } else {
 243                         if (tdb_chainlock(tdb, kbuf) == -1) {
 244                                 DEBUG(0,("message_send_pid_internal: failed to get chainlock.\n"));
 245                                 return False;
 246                         }
 247                 }
 248                 tdb_append(tdb, kbuf, dbuf);
 249                 tdb_chainunlock(tdb, kbuf);
 250
 251                 SAFE_FREE(dbuf.dptr);
 252                 errno = 0;                    /* paranoia */
 253                 return message_notify(pid);
 254         }
 255
 256         /* lock the record for the destination */
 257         if (timeout) {
 258                 if (tdb_chainlock_with_timeout(tdb, kbuf, timeout) == -1) {
 259                         DEBUG(0,("message_send_pid_internal: failed to get chainlock with timeout %ul.\n", timeout));
 260                         return False;
 261                 }
 262         } else {
 263                 if (tdb_chainlock(tdb, kbuf) == -1) {
 264                         DEBUG(0,("message_send_pid_internal: failed to get chainlock.\n"));
 265                         return False;
 266                 }
 267         }
 268
 269         old_dbuf = tdb_fetch(tdb, kbuf);
 270
 271         if (!old_dbuf.dptr) {
 272                 /* its a new record */
 273
 274                 tdb_store(tdb, kbuf, dbuf, TDB_REPLACE);
 275                 tdb_chainunlock(tdb, kbuf);
 276
 277                 SAFE_FREE(dbuf.dptr);
 278                 errno = 0;                    /* paranoia */
 279                 return message_notify(pid);
 280         }
 281
 282         /* Not a new record. Check for duplicates. */
 283
 284         for(ptr = (char *)old_dbuf.dptr; ptr < old_dbuf.dptr + old_dbuf.dsize; ) {
 285                 /*
 286                  * First check if the message header matches, then, if it's a non-zero
 287                  * sized message, check if the data matches. If so it's a duplicate and
 288                  * we can discard it. JRA.
 289                  */
 290
 291                 if (!memcmp(ptr, &rec, sizeof(rec))) {
 292                         if (!len || (len && !memcmp( ptr + sizeof(rec), buf, len))) {
 293                                 tdb_chainunlock(tdb, kbuf);
 294                                 DEBUG(10,("message_send_pid_internal: discarding duplicate message.\n"));
 295                                 SAFE_FREE(dbuf.dptr);
 296                                 SAFE_FREE(old_dbuf.dptr);
 297                                 return True;
 298                         }
 299                 }
 300                 memcpy(&prec, ptr, sizeof(prec));
 301                 ptr += sizeof(rec) + prec.len;
 302         }
 303
 304         /* we're adding to an existing entry */
 305
 306         tdb_append(tdb, kbuf, dbuf);
 307         tdb_chainunlock(tdb, kbuf);
 308
 309         SAFE_FREE(old_dbuf.dptr);
 310         SAFE_FREE(dbuf.dptr);
 311
 312         errno = 0;                    /* paranoia */
 313         return message_notify(pid);
 314 }
 315
 316 /****************************************************************************
 317  Send a message to a particular pid - no timeout.
 318 ****************************************************************************/
 319
 320 BOOL message_send_pid(struct process_id pid, int msg_type, const void *buf, size_t len, BOOL duplicates_allowed)
 321 {
 322         return message_send_pid_internal(pid, msg_type, buf, len, duplicates_allowed, 0);
 323 }
 324
 325 /****************************************************************************
 326  Send a message to a particular pid, with timeout in seconds.
 327 ****************************************************************************/
 328
 329 BOOL message_send_pid_with_timeout(struct process_id pid, int msg_type, const void *buf, size_t len,
 330                 BOOL duplicates_allowed, unsigned int timeout)
 331 {
 332         return message_send_pid_internal(pid, msg_type, buf, len, duplicates_allowed, timeout);
 333 }
 334
 335 /****************************************************************************
 336  Count the messages pending for a particular pid. Expensive....
 337 ****************************************************************************/
 338
 339 unsigned int messages_pending_for_pid(struct process_id pid)
 340 {
 341         TDB_DATA kbuf;
 342         TDB_DATA dbuf;
 343         char *buf;
 344         unsigned int message_count = 0;
 345
 346         kbuf = message_key_pid(pid);
 347
 348         dbuf = tdb_fetch(tdb, kbuf);
 349         if (dbuf.dptr == NULL || dbuf.dsize == 0) {
 350                 SAFE_FREE(dbuf.dptr);
 351                 return 0;
 352         }
 353
 354         for (buf = dbuf.dptr; dbuf.dsize > sizeof(struct message_rec);) {
 355                 struct message_rec rec;
 356                 memcpy(&rec, buf, sizeof(rec));
 357                 buf += (sizeof(rec) + rec.len);
 358                 dbuf.dsize -= (sizeof(rec) + rec.len);
 359                 message_count++;
 360         }
 361
 362         SAFE_FREE(dbuf.dptr);
 363         return message_count;
 364 }
 365
 366 /****************************************************************************
 367  Retrieve all messages for the current process.
 368 ****************************************************************************/
 369
 370 static BOOL retrieve_all_messages(char **msgs_buf, size_t *total_len)
 371 {
 372         TDB_DATA kbuf;
 373         TDB_DATA dbuf;
 374         TDB_DATA null_dbuf;
 375
 376         ZERO_STRUCT(null_dbuf);
 377
 378         *msgs_buf = NULL;
 379         *total_len = 0;
 380
 381         kbuf = message_key_pid(pid_to_procid(sys_getpid()));
 382
 383         if (tdb_chainlock(tdb, kbuf) == -1)
 384                 return False;
 385
 386         dbuf = tdb_fetch(tdb, kbuf);
 387         /*
 388          * Replace with an empty record to keep the allocated
 389          * space in the tdb.
 390          */
 391         tdb_store(tdb, kbuf, null_dbuf, TDB_REPLACE);
 392         tdb_chainunlock(tdb, kbuf);
 393
 394         if (dbuf.dptr == NULL || dbuf.dsize == 0) {
 395                 SAFE_FREE(dbuf.dptr);
 396                 return False;
 397         }
 398
 399         *msgs_buf = dbuf.dptr;
 400         *total_len = dbuf.dsize;
 401
 402         return True;
 403 }
 404
 405 /****************************************************************************
 406  Parse out the next message for the current process.
 407 ****************************************************************************/
 408
 409 static BOOL message_recv(char *msgs_buf, size_t total_len, int *msg_type,
 410                          struct process_id *src, char **buf, size_t *len)
 411 {
 412         struct message_rec rec;
 413         char *ret_buf = *buf;
 414
 415         *buf = NULL;
 416         *len = 0;
 417
 418         if (total_len - (ret_buf - msgs_buf) < sizeof(rec))
 419                 return False;
 420
 421         memcpy(&rec, ret_buf, sizeof(rec));
 422         ret_buf += sizeof(rec);
 423
 424         if (rec.msg_version != MESSAGE_VERSION) {
 425                 DEBUG(0,("message version %d received (expected %d)\n", rec.msg_version, MESSAGE_VERSION));
 426                 return False;
 427         }
 428
 429         if (rec.len > 0) {
 430                 if (total_len - (ret_buf - msgs_buf) < rec.len)
 431                         return False;
 432         }
 433
 434         *len = rec.len;
 435         *msg_type = rec.msg_type;
 436         *src = rec.src;
 437         *buf = ret_buf;
 438
 439         return True;
 440 }
 441
 442 /****************************************************************************
 443  Receive and dispatch any messages pending for this process.
 444  Notice that all dispatch handlers for a particular msg_type get called,
 445  so you can register multiple handlers for a message.
 446  *NOTE*: Dispatch functions must be able to cope with incoming
 447  messages on an *odd* byte boundary.
 448 ****************************************************************************/
 449
 450 void message_dispatch(void)
 451 {
 452         int msg_type;
 453         struct process_id src;
 454         char *buf;
 455         char *msgs_buf;
 456         size_t len, total_len;
 457         struct dispatch_fns *dfn;
 458         int n_handled;
 459
 460         if (!received_signal)
 461                 return;
 462
 463         DEBUG(10,("message_dispatch: received_signal = %d\n", received_signal));
 464
 465         received_signal = 0;
 466
 467         if (!retrieve_all_messages(&msgs_buf, &total_len))
 468                 return;
 469
 470         for (buf = msgs_buf; message_recv(msgs_buf, total_len, &msg_type, &src, &buf, &len); buf += len) {
 471                 DEBUG(10,("message_dispatch: received msg_type=%d "
 472                           "src_pid=%u\n", msg_type,
 473                           (unsigned int) procid_to_pid(&src)));
 474                 n_handled = 0;
 475                 for (dfn = dispatch_fns; dfn; dfn = dfn->next) {
 476                         if (dfn->msg_type == msg_type) {
 477                                 DEBUG(10,("message_dispatch: processing message of type %d.\n", msg_type));
 478                                 dfn->fn(msg_type, src, len ? (void *)buf : NULL, len);
 479                                 n_handled++;
 480                         }
 481                 }
 482                 if (!n_handled) {
 483                         DEBUG(5,("message_dispatch: warning: no handlers registed for "
 484                                  "msg_type %d in pid %u\n",
 485                                  msg_type, (unsigned int)sys_getpid()));
 486                 }
 487         }
 488         SAFE_FREE(msgs_buf);
 489 }
 490
 491 /****************************************************************************
 492  Register a dispatch function for a particular message type.
 493  *NOTE*: Dispatch functions must be able to cope with incoming
 494  messages on an *odd* byte boundary.
 495 ****************************************************************************/
 496
 497 void message_register(int msg_type,
 498                       void (*fn)(int msg_type, struct process_id pid,
 499                                  void *buf, size_t len))
 500 {
 501         struct dispatch_fns *dfn;
 502
 503         dfn = SMB_MALLOC_P(struct dispatch_fns);
 504
 505         if (dfn != NULL) {
 506
 507                 ZERO_STRUCTPN(dfn);
 508
 509                 dfn->msg_type = msg_type;
 510                 dfn->fn = fn;
 511
 512                 DLIST_ADD(dispatch_fns, dfn);
 513         }
 514         else {
 515
 516                 DEBUG(0,("message_register: Not enough memory. malloc failed!\n"));
 517         }
 518 }
 519
 520 /****************************************************************************
 521  De-register the function for a particular message type.
 522 ****************************************************************************/
 523
 524 void message_deregister(int msg_type)
 525 {
 526         struct dispatch_fns *dfn, *next;
 527
 528         for (dfn = dispatch_fns; dfn; dfn = next) {
 529                 next = dfn->next;
 530                 if (dfn->msg_type == msg_type) {
 531                         DLIST_REMOVE(dispatch_fns, dfn);
 532                         SAFE_FREE(dfn);
 533                 }
 534         }
 535 }
 536
 537 struct msg_all {
 538         int msg_type;
 539         uint32 msg_flag;
 540         const void *buf;
 541         size_t len;
 542         BOOL duplicates;
 543         int n_sent;
 544 };
 545
 546 /****************************************************************************
 547  Send one of the messages for the broadcast.
 548 ****************************************************************************/
 549
 550 static int traverse_fn(TDB_CONTEXT *the_tdb, TDB_DATA kbuf, TDB_DATA dbuf, void *state)
 551 {
 552         struct connections_data crec;
 553         struct msg_all *msg_all = (struct msg_all *)state;
 554
 555         if (dbuf.dsize != sizeof(crec))
 556                 return 0;
 557
 558         memcpy(&crec, dbuf.dptr, sizeof(crec));
 559
 560         if (crec.cnum != -1)
 561                 return 0;
 562
 563         /* Don't send if the receiver hasn't registered an interest. */
 564
 565         if(!(crec.bcast_msg_flags & msg_all->msg_flag))
 566                 return 0;
 567
 568         /* If the msg send fails because the pid was not found (i.e. smbd died),
 569          * the msg has already been deleted from the messages.tdb.*/
 570
 571         if (!message_send_pid(crec.pid, msg_all->msg_type,
 572                               msg_all->buf, msg_all->len,
 573                               msg_all->duplicates)) {
 574
 575                 /* If the pid was not found delete the entry from connections.tdb */
 576
 577                 if (errno == ESRCH) {
 578                         DEBUG(2,("pid %s doesn't exist - deleting connections %d [%s]\n",
 579                                  procid_str_static(&crec.pid),
 580                                  crec.cnum, crec.name));
 581                         tdb_delete(the_tdb, kbuf);
 582                 }
 583         }
 584         msg_all->n_sent++;
 585         return 0;
 586 }
 587
 588 /**
 589  * Send a message to all smbd processes.
 590  *
 591  * It isn't very efficient, but should be OK for the sorts of
 592  * applications that use it. When we need efficient broadcast we can add
 593  * it.
 594  *
 595  * @param n_sent Set to the number of messages sent.  This should be
 596  * equal to the number of processes, but be careful for races.
 597  *
 598  * @retval True for success.
 599  **/
 600 BOOL message_send_all(TDB_CONTEXT *conn_tdb, int msg_type,
 601                       const void *buf, size_t len,
 602                       BOOL duplicates_allowed,
 603                       int *n_sent)
 604 {
 605         struct msg_all msg_all;
 606
 607         msg_all.msg_type = msg_type;
 608         if (msg_type < 1000)
 609                 msg_all.msg_flag = FLAG_MSG_GENERAL;
 610         else if (msg_type > 1000 && msg_type < 2000)
 611                 msg_all.msg_flag = FLAG_MSG_NMBD;
 612         else if (msg_type > 2000 && msg_type < 2100)
 613                 msg_all.msg_flag = FLAG_MSG_PRINT_NOTIFY;
 614         else if (msg_type > 2100 && msg_type < 3000)
 615                 msg_all.msg_flag = FLAG_MSG_PRINT_GENERAL;
 616         else if (msg_type > 3000 && msg_type < 4000)
 617                 msg_all.msg_flag = FLAG_MSG_SMBD;
 618         else
 619                 return False;
 620
 621         msg_all.buf = buf;
 622         msg_all.len = len;
 623         msg_all.duplicates = duplicates_allowed;
 624         msg_all.n_sent = 0;
 625
 626         tdb_traverse(conn_tdb, traverse_fn, &msg_all);
 627         if (n_sent)
 628                 *n_sent = msg_all.n_sent;
 629         return True;
 630 }
 631
 632 /*
 633  * Block and unblock receiving of messages. Allows removal of race conditions
 634  * when doing a fork and changing message disposition.
 635  */
 636
 637 void message_block(void)
 638 {
 639         BlockSignals(True, SIGUSR1);
 640 }
 641
 642 void message_unblock(void)
 643 {
 644         BlockSignals(False, SIGUSR1);
 645 }
 646 /** @} **/