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 = sizeof(key);
 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         memmove(dbuf.dptr, dbuf.dptr+sizeof(rec)+rec.len, dbuf.dsize - (sizeof(rec)+rec.len));
 282         dbuf.dsize -= sizeof(rec)+rec.len;
 283         tdb_store(tdb, kbuf, dbuf, TDB_REPLACE);
 284
 285         free(dbuf.dptr);
 286         tdb_unlockchain(tdb, kbuf);
 287         return True;
 288
 289  failed:
 290         tdb_unlockchain(tdb, kbuf);
 291         return False;
 292 }
 293
 294
 295 /****************************************************************************
 296 receive and dispatch any messages pending for this process
 297 notice that all dispatch handlers for a particular msg_type get called,
 298 so you can register multiple handlers for a message
 299 ****************************************************************************/
 300 void message_dispatch(void)
 301 {
 302         int msg_type;
 303         pid_t src;
 304         void *buf;
 305         size_t len;
 306         struct dispatch_fns *dfn;
 307
 308         if (!received_signal) return;
 309         received_signal = 0;
 310
 311         while (message_recv(&msg_type, &src, &buf, &len)) {
 312                 for (dfn = dispatch_fns; dfn; dfn = dfn->next) {
 313                         if (dfn->msg_type == msg_type) {
 314                                 dfn->fn(msg_type, src, buf, len);
 315                         }
 316                 }
 317                 if (buf) free(buf);
 318         }
 319 }
 320
 321
 322 /****************************************************************************
 323 register a dispatch function for a particular message type
 324 ****************************************************************************/
 325 void message_register(int msg_type,
 326                       void (*fn)(int msg_type, pid_t pid, void *buf, size_t len))
 327 {
 328         struct dispatch_fns *dfn;
 329
 330         dfn = (struct dispatch_fns *)malloc(sizeof(*dfn));
 331
 332         ZERO_STRUCTP(dfn);
 333
 334         dfn->msg_type = msg_type;
 335         dfn->fn = fn;
 336
 337         DLIST_ADD(dispatch_fns, dfn);
 338 }
 339
 340 /****************************************************************************
 341 de-register the function for a particular message type
 342 ****************************************************************************/
 343 void message_deregister(int msg_type)
 344 {
 345         struct dispatch_fns *dfn, *next;
 346
 347         for (dfn = dispatch_fns; dfn; dfn = next) {
 348                 next = dfn->next;
 349                 if (dfn->msg_type == msg_type) {
 350                         DLIST_REMOVE(dispatch_fns, dfn);
 351                         free(dfn);
 352                 }
 353         }
 354 }
 355
 356 static struct {
 357         int msg_type;
 358         void *buf;
 359         size_t len;
 360         BOOL duplicates;
 361 } msg_all;
 362
 363 /****************************************************************************
 364 send one of the messages for the broadcast
 365 ****************************************************************************/
 366 static int traverse_fn(TDB_CONTEXT *the_tdb, TDB_DATA kbuf, TDB_DATA dbuf, void *state)
 367 {
 368         struct connections_data crec;
 369
 370         memcpy(&crec, dbuf.dptr, sizeof(crec));
 371
 372         if (crec.cnum == -1) return 0;
 373         message_send_pid(crec.pid, msg_all.msg_type, msg_all.buf, msg_all.len, msg_all.duplicates);
 374         return 0;
 375 }
 376
 377 /****************************************************************************
 378 this is a useful function for sending messages to all smbd processes.
 379 It isn't very efficient, but should be OK for the sorts of applications that
 380 use it. When we need efficient broadcast we can add it.
 381 ****************************************************************************/
 382 BOOL message_send_all(int msg_type, void *buf, size_t len, BOOL duplicates_allowed)
 383 {
 384         TDB_CONTEXT *the_tdb;
 385
 386         the_tdb = tdb_open(lock_path("connections.tdb"), 0, 0, O_RDONLY, 0);
 387         if (!the_tdb) {
 388                 DEBUG(2,("Failed to open connections database in message_send_all\n"));
 389                 return False;
 390         }
 391
 392         msg_all.msg_type = msg_type;
 393         msg_all.buf = buf;
 394         msg_all.len = len;
 395         msg_all.duplicates = duplicates_allowed;
 396
 397         tdb_traverse(the_tdb, traverse_fn, NULL);
 398         tdb_close(the_tdb);
 399         return True;
 400 }