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