source/lib/messages.c

   1 /*
   2    Unix SMB/Netbios implementation.
   3    Version 3.0
   4    Samba internal messaging functions
   5    Copyright (C) Andrew Tridgell 2000
   6
   7    This program is free software; you can redistribute it and/or modify
   8    it under the terms of the GNU General Public License as published by
   9    the Free Software Foundation; either version 2 of the License, or
  10    (at your option) any later version.
  11
  12    This program is distributed in the hope that it will be useful,
  13    but WITHOUT ANY WARRANTY; without even the implied warranty of
  14    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15    GNU General Public License for more details.
  16
  17    You should have received a copy of the GNU General Public License
  18    along with this program; if not, write to the Free Software
  19    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  20 */
  21
  22 /* this module is used for internal messaging between Samba daemons.
  23
  24    The idea is that if a part of Samba wants to do communication with
  25    another Samba process then it will do a message_register() of a
  26    dispatch function, and use message_send_pid() to send messages to
  27    that process.
  28
  29    This system doesn't have any inherent size limitations but is not
  30    very efficient for large messages or when messages are sent in very
  31    quick succession.
  32
  33 */
  34
  35 #include "includes.h"
  36
  37 /* the locking database handle */
  38 static TDB_CONTEXT *tdb;
  39 static int received_signal;
  40
  41 /* change the message version with any incompatible changes in the protocol */
  42 #define MESSAGE_VERSION 1
  43
  44 struct message_rec {
  45         int msg_version;
  46         int msg_type;
  47         pid_t dest;
  48         pid_t src;
  49         size_t len;
  50 };
  51
  52 /* we have a linked list of dispatch handlers */
  53 static struct dispatch_fns {
  54         struct dispatch_fns *next, *prev;
  55         int msg_type;
  56         void (*fn)(int msg_type, pid_t pid, void *buf, size_t len);
  57 } *dispatch_fns;
  58
  59 /****************************************************************************
  60 notifications come in as signals
  61 ****************************************************************************/
  62 static void sig_usr1(void)
  63 {
  64         received_signal = 1;
  65         sys_select_signal();
  66 }
  67
  68 /****************************************************************************
  69 a useful function for testing the message system
  70 ****************************************************************************/
  71 void ping_message(int msg_type, pid_t src, void *buf, size_t len)
  72 {
  73         DEBUG(1,("INFO: Received PING message from PID %u\n",(unsigned int)src));
  74         message_send_pid(src, MSG_PONG, buf, len, True);
  75 }
  76 /****************************************************************************
  77 return current debug level
  78 ****************************************************************************/
  79 void debuglevel_message(int msg_type, pid_t src, void *buf, size_t len)
  80 {
  81         DEBUG(1,("INFO: Received REQ_DEBUGLEVEL message from PID %u\n",(unsigned int)src));
  82         message_send_pid(src, MSG_DEBUGLEVEL, DEBUGLEVEL_CLASS, sizeof(DEBUGLEVEL_CLASS), True);
  83 }
  84
  85 /****************************************************************************
  86  Initialise the messaging functions.
  87 ****************************************************************************/
  88 BOOL message_init(void)
  89 {
  90         if (tdb) return True;
  91
  92         tdb = tdb_open_log(lock_path("messages.tdb"),
  93                        0, TDB_CLEAR_IF_FIRST|TDB_DEFAULT,
  94                        O_RDWR|O_CREAT,0600);
  95
  96         if (!tdb) {
  97                 DEBUG(0,("ERROR: Failed to initialise messages database\n"));
  98                 return False;
  99         }
 100
 101         CatchSignal(SIGUSR1, SIGNAL_CAST sig_usr1);
 102
 103         message_register(MSG_PING, ping_message);
 104         message_register(MSG_REQ_DEBUGLEVEL, debuglevel_message);
 105
 106         return True;
 107 }
 108
 109
 110 /*******************************************************************
 111  form a static tdb key from a pid
 112 ******************************************************************/
 113 static TDB_DATA message_key_pid(pid_t pid)
 114 {
 115         static char key[20];
 116         TDB_DATA kbuf;
 117
 118         slprintf(key, sizeof(key)-1, "PID/%d", (int)pid);
 119
 120         kbuf.dptr = (char *)key;
 121         kbuf.dsize = strlen(key)+1;
 122         return kbuf;
 123 }
 124
 125
 126 /****************************************************************************
 127 notify a process that it has a message. If the process doesn't exist
 128 then delete its record in the database
 129 ****************************************************************************/
 130 static BOOL message_notify(pid_t pid)
 131 {
 132         if (kill(pid, SIGUSR1) == -1) {
 133                 if (errno == ESRCH) {
 134                         DEBUG(2,("pid %d doesn't exist - deleting messages record\n", (int)pid));
 135                         tdb_delete(tdb, message_key_pid(pid));
 136                 } else {
 137                         DEBUG(2,("message to process %d failed - %s\n", (int)pid, strerror(errno)));
 138                 }
 139                 return False;
 140         }
 141         return True;
 142 }
 143
 144 /****************************************************************************
 145 send a message to a particular pid
 146 ****************************************************************************/
 147 BOOL message_send_pid(pid_t pid, int msg_type, void *buf, size_t len, BOOL duplicates_allowed)
 148 {
 149         TDB_DATA kbuf;
 150         TDB_DATA dbuf;
 151         struct message_rec rec;
 152         void *p;
 153
 154         rec.msg_version = MESSAGE_VERSION;
 155         rec.msg_type = msg_type;
 156         rec.dest = pid;
 157         rec.src = sys_getpid();
 158         rec.len = len;
 159
 160         kbuf = message_key_pid(pid);
 161
 162         /* lock the record for the destination */
 163         tdb_chainlock(tdb, kbuf);
 164
 165         dbuf = tdb_fetch(tdb, kbuf);
 166
 167         if (!dbuf.dptr) {
 168                 /* its a new record */
 169                 p = (void *)malloc(len + sizeof(rec));
 170                 if (!p) goto failed;
 171
 172                 memcpy(p, &rec, sizeof(rec));
 173                 if (len > 0) memcpy((void *)((char*)p+sizeof(rec)), buf, len);
 174
 175                 dbuf.dptr = p;
 176                 dbuf.dsize = len + sizeof(rec);
 177                 tdb_store(tdb, kbuf, dbuf, TDB_REPLACE);
 178                 SAFE_FREE(p);
 179                 goto ok;
 180         }
 181
 182         if (!duplicates_allowed) {
 183                 char *ptr;
 184                 struct message_rec prec;
 185
 186                 for(ptr = (char *)dbuf.dptr; ptr < dbuf.dptr + dbuf.dsize; ) {
 187                         /*
 188                          * First check if the message header matches, then, if it's a non-zero
 189                          * sized message, check if the data matches. If so it's a duplicate and
 190                          * we can discard it. JRA.
 191                          */
 192
 193                         if (!memcmp(ptr, &rec, sizeof(rec))) {
 194                                 if (!len || (len && !memcmp( ptr + sizeof(rec), (char *)buf, len))) {
 195                                         DEBUG(10,("message_send_pid: discarding duplicate message.\n"));
 196                                         SAFE_FREE(dbuf.dptr);
 197                                         tdb_chainunlock(tdb, kbuf);
 198                                         return True;
 199                                 }
 200                         }
 201                         memcpy(&prec, ptr, sizeof(prec));
 202                         ptr += sizeof(rec) + prec.len;
 203                 }
 204         }
 205
 206         /* we're adding to an existing entry */
 207         p = (void *)malloc(dbuf.dsize + len + sizeof(rec));
 208         if (!p) goto failed;
 209
 210         memcpy(p, dbuf.dptr, dbuf.dsize);
 211         memcpy((void *)((char*)p+dbuf.dsize), &rec, sizeof(rec));
 212         if (len > 0) memcpy((void *)((char*)p+dbuf.dsize+sizeof(rec)), buf, len);
 213
 214         SAFE_FREE(dbuf.dptr);
 215         dbuf.dptr = p;
 216         dbuf.dsize += len + sizeof(rec);
 217         tdb_store(tdb, kbuf, dbuf, TDB_REPLACE);
 218         SAFE_FREE(dbuf.dptr);
 219
 220  ok:
 221         tdb_chainunlock(tdb, kbuf);
 222         errno = 0;                    /* paranoia */
 223         return message_notify(pid);
 224
 225  failed:
 226         tdb_chainunlock(tdb, kbuf);
 227         errno = 0;                    /* paranoia */
 228         return False;
 229 }
 230
 231
 232
 233 /****************************************************************************
 234 retrieve the next message for the current process
 235 ****************************************************************************/
 236 static BOOL message_recv(int *msg_type, pid_t *src, void **buf, size_t *len)
 237 {
 238         TDB_DATA kbuf;
 239         TDB_DATA dbuf;
 240         struct message_rec rec;
 241
 242         kbuf = message_key_pid(sys_getpid());
 243
 244         tdb_chainlock(tdb, kbuf);
 245
 246         dbuf = tdb_fetch(tdb, kbuf);
 247         if (dbuf.dptr == NULL || dbuf.dsize == 0) goto failed;
 248
 249         memcpy(&rec, dbuf.dptr, sizeof(rec));
 250
 251         if (rec.msg_version != MESSAGE_VERSION) {
 252                 DEBUG(0,("message version %d received (expected %d)\n", rec.msg_version, MESSAGE_VERSION));
 253                 goto failed;
 254         }
 255
 256         if (rec.len > 0) {
 257                 (*buf) = (void *)malloc(rec.len);
 258                 if (!(*buf)) goto failed;
 259
 260                 memcpy(*buf, dbuf.dptr+sizeof(rec), rec.len);
 261         } else {
 262                 *buf = NULL;
 263         }
 264
 265         *len = rec.len;
 266         *msg_type = rec.msg_type;
 267         *src = rec.src;
 268
 269         if (dbuf.dsize - (sizeof(rec)+rec.len) > 0)
 270                 memmove(dbuf.dptr, dbuf.dptr+sizeof(rec)+rec.len, dbuf.dsize - (sizeof(rec)+rec.len));
 271         dbuf.dsize -= sizeof(rec)+rec.len;
 272
 273         if (dbuf.dsize == 0)
 274                 tdb_delete(tdb, kbuf);
 275         else
 276                 tdb_store(tdb, kbuf, dbuf, TDB_REPLACE);
 277
 278         SAFE_FREE(dbuf.dptr);
 279         tdb_chainunlock(tdb, kbuf);
 280         return True;
 281
 282  failed:
 283         tdb_chainunlock(tdb, kbuf);
 284         return False;
 285 }
 286
 287
 288 /****************************************************************************
 289 receive and dispatch any messages pending for this process
 290 notice that all dispatch handlers for a particular msg_type get called,
 291 so you can register multiple handlers for a message
 292 ****************************************************************************/
 293 void message_dispatch(void)
 294 {
 295         int msg_type;
 296         pid_t src;
 297         void *buf;
 298         size_t len;
 299         struct dispatch_fns *dfn;
 300
 301         if (!received_signal)
 302                 return;
 303
 304         DEBUG(10,("message_dispatch: received_signal = %d\n", received_signal));
 305         received_signal = 0;
 306
 307         while (message_recv(&msg_type, &src, &buf, &len)) {
 308                 for (dfn = dispatch_fns; dfn; dfn = dfn->next) {
 309                         if (dfn->msg_type == msg_type) {
 310                                 DEBUG(10,("message_dispatch: processing message of type %d.\n", msg_type));
 311                                 dfn->fn(msg_type, src, buf, len);
 312                         }
 313                 }
 314                 SAFE_FREE(buf);
 315         }
 316 }
 317
 318
 319 /****************************************************************************
 320 register a dispatch function for a particular message type
 321 ****************************************************************************/
 322 void message_register(int msg_type,
 323                       void (*fn)(int msg_type, pid_t pid, void *buf, size_t len))
 324 {
 325         struct dispatch_fns *dfn;
 326
 327         dfn = (struct dispatch_fns *)malloc(sizeof(*dfn));
 328
 329         if (dfn != NULL) {
 330
 331                 ZERO_STRUCTPN(dfn);
 332
 333                 dfn->msg_type = msg_type;
 334                 dfn->fn = fn;
 335
 336                 DLIST_ADD(dispatch_fns, dfn);
 337         }
 338         else {
 339
 340                 DEBUG(0,("message_register: Not enough memory. malloc failed!\n"));
 341         }
 342 }
 343
 344 /****************************************************************************
 345 de-register the function for a particular message type
 346 ****************************************************************************/
 347 void message_deregister(int msg_type)
 348 {
 349         struct dispatch_fns *dfn, *next;
 350
 351         for (dfn = dispatch_fns; dfn; dfn = next) {
 352                 next = dfn->next;
 353                 if (dfn->msg_type == msg_type) {
 354                         DLIST_REMOVE(dispatch_fns, dfn);
 355                         SAFE_FREE(dfn);
 356                 }
 357         }
 358 }
 359
 360 struct msg_all {
 361         int msg_type;
 362         void *buf;
 363         size_t len;
 364         BOOL duplicates;
 365 };
 366
 367 /****************************************************************************
 368 send one of the messages for the broadcast
 369 ****************************************************************************/
 370 static int traverse_fn(TDB_CONTEXT *the_tdb, TDB_DATA kbuf, TDB_DATA dbuf, void *state)
 371 {
 372         struct connections_data crec;
 373         struct msg_all *msg_all = (struct msg_all *)state;
 374
 375         if (dbuf.dsize != sizeof(crec))
 376                 return 0;
 377
 378         memcpy(&crec, dbuf.dptr, sizeof(crec));
 379
 380         if (crec.cnum != -1)
 381                 return 0;
 382
 383         /* if the msg send fails because the pid was not found (i.e. smbd died),
 384          * the msg has already been deleted from the messages.tdb.*/
 385         if (!message_send_pid(crec.pid, msg_all->msg_type, msg_all->buf, msg_all->len,
 386                                                         msg_all->duplicates)) {
 387
 388                 /* if the pid was not found delete the entry from connections.tdb */
 389                 if (errno == ESRCH) {
 390                         DEBUG(2,("pid %u doesn't exist - deleting connections %d [%s]\n",
 391                                         (unsigned int)crec.pid, crec.cnum, crec.name));
 392                         tdb_delete(the_tdb, kbuf);
 393                 }
 394         }
 395         return 0;
 396 }
 397
 398 /****************************************************************************
 399 this is a useful function for sending messages to all smbd processes.
 400 It isn't very efficient, but should be OK for the sorts of applications that
 401 use it. When we need efficient broadcast we can add it.
 402 ****************************************************************************/
 403 BOOL message_send_all(TDB_CONTEXT *conn_tdb, int msg_type, void *buf, size_t len, BOOL duplicates_allowed)
 404 {
 405         struct msg_all msg_all;
 406
 407         msg_all.msg_type = msg_type;
 408         msg_all.buf = buf;
 409         msg_all.len = len;
 410         msg_all.duplicates = duplicates_allowed;
 411
 412         tdb_traverse(conn_tdb, traverse_fn, &msg_all);
 413         return True;
 414 }