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
  63 static void sig_usr1(void)
  64 {
  65         received_signal = 1;
  66         sys_select_signal();
  67 }
  68
  69 /****************************************************************************
  70  A useful function for testing the message system.
  71 ****************************************************************************/
  72
  73 void ping_message(int msg_type, pid_t src, void *buf, size_t len)
  74 {
  75         char *msg = buf ? buf : "none";
  76         DEBUG(1,("INFO: Received PING message from PID %u [%s]\n",(unsigned int)src, msg));
  77         message_send_pid(src, MSG_PONG, buf, len, True);
  78 }
  79
  80 /****************************************************************************
  81  Return current debug level.
  82 ****************************************************************************/
  83
  84 void debuglevel_message(int msg_type, pid_t src, void *buf, size_t len)
  85 {
  86         DEBUG(1,("INFO: Received REQ_DEBUGLEVEL message from PID %u\n",(unsigned int)src));
  87         message_send_pid(src, MSG_DEBUGLEVEL, DEBUGLEVEL_CLASS, sizeof(DEBUGLEVEL_CLASS), True);
  88 }
  89
  90 /****************************************************************************
  91  Initialise the messaging functions.
  92 ****************************************************************************/
  93
  94 BOOL message_init(void)
  95 {
  96         if (tdb) return True;
  97
  98         tdb = tdb_open_log(lock_path("messages.tdb"),
  99                        0, TDB_CLEAR_IF_FIRST|TDB_DEFAULT,
 100                        O_RDWR|O_CREAT,0600);
 101
 102         if (!tdb) {
 103                 DEBUG(0,("ERROR: Failed to initialise messages database\n"));
 104                 return False;
 105         }
 106
 107         CatchSignal(SIGUSR1, SIGNAL_CAST sig_usr1);
 108
 109         message_register(MSG_PING, ping_message);
 110         message_register(MSG_REQ_DEBUGLEVEL, debuglevel_message);
 111
 112         return True;
 113 }
 114
 115 /*******************************************************************
 116  Form a static tdb key from a pid.
 117 ******************************************************************/
 118
 119 static TDB_DATA message_key_pid(pid_t pid)
 120 {
 121         static char key[20];
 122         TDB_DATA kbuf;
 123
 124         slprintf(key, sizeof(key)-1, "PID/%d", (int)pid);
 125
 126         kbuf.dptr = (char *)key;
 127         kbuf.dsize = strlen(key)+1;
 128         return kbuf;
 129 }
 130
 131 /****************************************************************************
 132  Notify a process that it has a message. If the process doesn't exist
 133  then delete its record in the database.
 134 ****************************************************************************/
 135
 136 static BOOL message_notify(pid_t pid)
 137 {
 138         if (kill(pid, SIGUSR1) == -1) {
 139                 if (errno == ESRCH) {
 140                         DEBUG(2,("pid %d doesn't exist - deleting messages record\n", (int)pid));
 141                         tdb_delete(tdb, message_key_pid(pid));
 142                 } else {
 143                         DEBUG(2,("message to process %d failed - %s\n", (int)pid, strerror(errno)));
 144                 }
 145                 return False;
 146         }
 147         return True;
 148 }
 149
 150 /****************************************************************************
 151  Send a message to a particular pid.
 152 ****************************************************************************/
 153
 154 BOOL message_send_pid(pid_t pid, int msg_type, void *buf, size_t len, BOOL duplicates_allowed)
 155 {
 156         TDB_DATA kbuf;
 157         TDB_DATA dbuf;
 158         struct message_rec rec;
 159         void *p;
 160
 161         rec.msg_version = MESSAGE_VERSION;
 162         rec.msg_type = msg_type;
 163         rec.dest = pid;
 164         rec.src = sys_getpid();
 165         rec.len = len;
 166
 167         kbuf = message_key_pid(pid);
 168
 169         /* lock the record for the destination */
 170         tdb_chainlock(tdb, kbuf);
 171
 172         dbuf = tdb_fetch(tdb, kbuf);
 173
 174         if (!dbuf.dptr) {
 175                 /* its a new record */
 176                 p = (void *)malloc(len + sizeof(rec));
 177                 if (!p) goto failed;
 178
 179                 memcpy(p, &rec, sizeof(rec));
 180                 if (len > 0) memcpy((void *)((char*)p+sizeof(rec)), buf, len);
 181
 182                 dbuf.dptr = p;
 183                 dbuf.dsize = len + sizeof(rec);
 184                 tdb_store(tdb, kbuf, dbuf, TDB_REPLACE);
 185                 SAFE_FREE(p);
 186                 goto ok;
 187         }
 188
 189         if (!duplicates_allowed) {
 190                 char *ptr;
 191                 struct message_rec prec;
 192
 193                 for(ptr = (char *)dbuf.dptr; ptr < dbuf.dptr + dbuf.dsize; ) {
 194                         /*
 195                          * First check if the message header matches, then, if it's a non-zero
 196                          * sized message, check if the data matches. If so it's a duplicate and
 197                          * we can discard it. JRA.
 198                          */
 199
 200                         if (!memcmp(ptr, &rec, sizeof(rec))) {
 201                                 if (!len || (len && !memcmp( ptr + sizeof(rec), (char *)buf, len))) {
 202                                         DEBUG(10,("message_send_pid: discarding duplicate message.\n"));
 203                                         SAFE_FREE(dbuf.dptr);
 204                                         tdb_chainunlock(tdb, kbuf);
 205                                         return True;
 206                                 }
 207                         }
 208                         memcpy(&prec, ptr, sizeof(prec));
 209                         ptr += sizeof(rec) + prec.len;
 210                 }
 211         }
 212
 213         /* we're adding to an existing entry */
 214         p = (void *)malloc(dbuf.dsize + len + sizeof(rec));
 215         if (!p) goto failed;
 216
 217         memcpy(p, dbuf.dptr, dbuf.dsize);
 218         memcpy((void *)((char*)p+dbuf.dsize), &rec, sizeof(rec));
 219         if (len > 0) memcpy((void *)((char*)p+dbuf.dsize+sizeof(rec)), buf, len);
 220
 221         SAFE_FREE(dbuf.dptr);
 222         dbuf.dptr = p;
 223         dbuf.dsize += len + sizeof(rec);
 224         tdb_store(tdb, kbuf, dbuf, TDB_REPLACE);
 225         SAFE_FREE(dbuf.dptr);
 226
 227  ok:
 228         tdb_chainunlock(tdb, kbuf);
 229         errno = 0;                    /* paranoia */
 230         return message_notify(pid);
 231
 232  failed:
 233         tdb_chainunlock(tdb, kbuf);
 234         errno = 0;                    /* paranoia */
 235         return False;
 236 }
 237
 238 /****************************************************************************
 239  Retrieve the next message for the current process.
 240 ****************************************************************************/
 241
 242 static BOOL message_recv(int *msg_type, pid_t *src, void **buf, size_t *len)
 243 {
 244         TDB_DATA kbuf;
 245         TDB_DATA dbuf;
 246         struct message_rec rec;
 247
 248         kbuf = message_key_pid(sys_getpid());
 249
 250         tdb_chainlock(tdb, kbuf);
 251
 252         dbuf = tdb_fetch(tdb, kbuf);
 253         if (dbuf.dptr == NULL || dbuf.dsize == 0) goto failed;
 254
 255         memcpy(&rec, dbuf.dptr, sizeof(rec));
 256
 257         if (rec.msg_version != MESSAGE_VERSION) {
 258                 DEBUG(0,("message version %d received (expected %d)\n", rec.msg_version, MESSAGE_VERSION));
 259                 goto failed;
 260         }
 261
 262         if (rec.len > 0) {
 263                 (*buf) = (void *)malloc(rec.len);
 264                 if (!(*buf)) goto failed;
 265
 266                 memcpy(*buf, dbuf.dptr+sizeof(rec), rec.len);
 267         } else {
 268                 *buf = NULL;
 269         }
 270
 271         *len = rec.len;
 272         *msg_type = rec.msg_type;
 273         *src = rec.src;
 274
 275         if (dbuf.dsize - (sizeof(rec)+rec.len) > 0)
 276                 memmove(dbuf.dptr, dbuf.dptr+sizeof(rec)+rec.len, dbuf.dsize - (sizeof(rec)+rec.len));
 277         dbuf.dsize -= sizeof(rec)+rec.len;
 278
 279         if (dbuf.dsize == 0)
 280                 tdb_delete(tdb, kbuf);
 281         else
 282                 tdb_store(tdb, kbuf, dbuf, TDB_REPLACE);
 283
 284         SAFE_FREE(dbuf.dptr);
 285         tdb_chainunlock(tdb, kbuf);
 286         return True;
 287
 288  failed:
 289         tdb_chainunlock(tdb, kbuf);
 290         return False;
 291 }
 292
 293 /****************************************************************************
 294  Receive and dispatch any messages pending for this process.
 295  Notice that all dispatch handlers for a particular msg_type get called,
 296  so you can register multiple handlers for a message.
 297 ****************************************************************************/
 298
 299 void message_dispatch(void)
 300 {
 301         int msg_type;
 302         pid_t src;
 303         void *buf;
 304         size_t len;
 305         struct dispatch_fns *dfn;
 306
 307         if (!received_signal) return;
 308
 309         DEBUG(10,("message_dispatch: received_signal = %d\n", received_signal));
 310
 311         received_signal = 0;
 312
 313         while (message_recv(&msg_type, &src, &buf, &len)) {
 314                 for (dfn = dispatch_fns; dfn; dfn = dfn->next) {
 315                         if (dfn->msg_type == msg_type) {
 316                                 DEBUG(10,("message_dispatch: processing message of type %d.\n", msg_type));
 317                                 dfn->fn(msg_type, src, buf, len);
 318                         }
 319                 }
 320                 SAFE_FREE(buf);
 321         }
 322 }
 323
 324 /****************************************************************************
 325  Register a dispatch function for a particular message type.
 326 ****************************************************************************/
 327
 328 void message_register(int msg_type,
 329                       void (*fn)(int msg_type, pid_t pid, void *buf, size_t len))
 330 {
 331         struct dispatch_fns *dfn;
 332
 333         dfn = (struct dispatch_fns *)malloc(sizeof(*dfn));
 334
 335         if (dfn != NULL) {
 336
 337                 ZERO_STRUCTPN(dfn);
 338
 339                 dfn->msg_type = msg_type;
 340                 dfn->fn = fn;
 341
 342                 DLIST_ADD(dispatch_fns, dfn);
 343         }
 344         else {
 345
 346                 DEBUG(0,("message_register: Not enough memory. malloc failed!\n"));
 347         }
 348 }
 349
 350 /****************************************************************************
 351  De-register the function for a particular message type.
 352 ****************************************************************************/
 353
 354 void message_deregister(int msg_type)
 355 {
 356         struct dispatch_fns *dfn, *next;
 357
 358         for (dfn = dispatch_fns; dfn; dfn = next) {
 359                 next = dfn->next;
 360                 if (dfn->msg_type == msg_type) {
 361                         DLIST_REMOVE(dispatch_fns, dfn);
 362                         SAFE_FREE(dfn);
 363                 }
 364         }
 365 }
 366
 367 struct msg_all {
 368         int msg_type;
 369         void *buf;
 370         size_t len;
 371         BOOL duplicates;
 372 };
 373
 374 /****************************************************************************
 375  Send one of the messages for the broadcast.
 376 ****************************************************************************/
 377
 378 static int traverse_fn(TDB_CONTEXT *the_tdb, TDB_DATA kbuf, TDB_DATA dbuf, void *state)
 379 {
 380         struct connections_data crec;
 381         struct msg_all *msg_all = (struct msg_all *)state;
 382
 383         if (dbuf.dsize != sizeof(crec))
 384                 return 0;
 385
 386         memcpy(&crec, dbuf.dptr, sizeof(crec));
 387
 388         if (crec.cnum != -1)
 389                 return 0;
 390
 391         /* if the msg send fails because the pid was not found (i.e. smbd died),
 392          * the msg has already been deleted from the messages.tdb.*/
 393         if (!message_send_pid(crec.pid, msg_all->msg_type, msg_all->buf, msg_all->len,
 394                                                         msg_all->duplicates)) {
 395
 396                 /* if the pid was not found delete the entry from connections.tdb */
 397                 if (errno == ESRCH) {
 398                         DEBUG(2,("pid %u doesn't exist - deleting connections %d [%s]\n",
 399                                         (unsigned int)crec.pid, crec.cnum, crec.name));
 400                         tdb_delete(the_tdb, kbuf);
 401                 }
 402         }
 403         return 0;
 404 }
 405
 406 /****************************************************************************
 407  This is a useful function for sending messages to all smbd processes.
 408  It isn't very efficient, but should be OK for the sorts of applications that
 409  use it. When we need efficient broadcast we can add it.
 410 ****************************************************************************/
 411
 412 BOOL message_send_all(TDB_CONTEXT *conn_tdb, int msg_type, void *buf, size_t len, BOOL duplicates_allowed)
 413 {
 414         struct msg_all msg_all;
 415
 416         msg_all.msg_type = msg_type;
 417         msg_all.buf = buf;
 418         msg_all.len = len;
 419         msg_all.duplicates = duplicates_allowed;
 420
 421         tdb_traverse(conn_tdb, traverse_fn, &msg_all);
 422         return True;
 423 }