bc6f20993810a9fa6723e430e621becbe39ffc55
[ira/wip.git] / 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                 tdb_delete(tdb, message_key_pid(pid));
165                 DEBUG(2,("message_send_pid: pid %d doesn't exist\n", (int)pid));
166                 return False;
167         }
168
169         rec.msg_version = MESSAGE_VERSION;
170         rec.msg_type = msg_type;
171         rec.dest = pid;
172         rec.src = sys_getpid();
173         rec.len = len;
174
175         kbuf = message_key_pid(pid);
176
177         /* lock the record for the destination */
178         tdb_chainlock(tdb, kbuf);
179
180         dbuf = tdb_fetch(tdb, kbuf);
181
182         if (!dbuf.dptr) {
183                 /* its a new record */
184                 p = (void *)malloc(len + sizeof(rec));
185                 if (!p) goto failed;
186
187                 memcpy(p, &rec, sizeof(rec));
188                 if (len > 0) memcpy((void *)((char*)p+sizeof(rec)), buf, len);
189
190                 dbuf.dptr = p;
191                 dbuf.dsize = len + sizeof(rec);
192                 tdb_store(tdb, kbuf, dbuf, TDB_REPLACE);
193                 free(p);
194                 goto ok;
195         }
196
197         if (!duplicates_allowed) {
198                 char *ptr;
199                 struct message_rec prec;
200                 
201                 for(ptr = (char *)dbuf.dptr; ptr < dbuf.dptr + dbuf.dsize; ) {
202                         /*
203                          * First check if the message header matches, then, if it's a non-zero
204                          * sized message, check if the data matches. If so it's a duplicate and
205                          * we can discard it. JRA.
206                          */
207
208                         if (!memcmp(ptr, &rec, sizeof(rec))) {
209                                 if (!len || (len && !memcmp( ptr + sizeof(rec), (char *)buf, len))) {
210                                         DEBUG(10,("message_send_pid: discarding duplicate message.\n"));
211                                         free(dbuf.dptr);
212                                         tdb_chainunlock(tdb, kbuf);
213                                         return True;
214                                 }
215                         }
216                         memcpy(&prec, ptr, sizeof(prec));
217                         ptr += sizeof(rec) + prec.len;
218                 }
219         }
220
221         /* we're adding to an existing entry */
222         p = (void *)malloc(dbuf.dsize + len + sizeof(rec));
223         if (!p) goto failed;
224
225         memcpy(p, dbuf.dptr, dbuf.dsize);
226         memcpy((void *)((char*)p+dbuf.dsize), &rec, sizeof(rec));
227         if (len > 0) memcpy((void *)((char*)p+dbuf.dsize+sizeof(rec)), buf, len);
228
229         free(dbuf.dptr);
230         dbuf.dptr = p;
231         dbuf.dsize += len + sizeof(rec);
232         tdb_store(tdb, kbuf, dbuf, TDB_REPLACE);
233         free(dbuf.dptr);
234
235  ok:
236         tdb_chainunlock(tdb, kbuf);
237         return message_notify(pid);
238
239  failed:
240         tdb_chainunlock(tdb, kbuf);
241         return False;
242 }
243
244
245
246 /****************************************************************************
247 retrieve the next message for the current process
248 ****************************************************************************/
249 static BOOL message_recv(int *msg_type, pid_t *src, void **buf, size_t *len)
250 {
251         TDB_DATA kbuf;
252         TDB_DATA dbuf;
253         struct message_rec rec;
254
255         kbuf = message_key_pid(sys_getpid());
256
257         tdb_chainlock(tdb, kbuf);
258         
259         dbuf = tdb_fetch(tdb, kbuf);
260         if (dbuf.dptr == NULL || dbuf.dsize == 0) goto failed;
261
262         memcpy(&rec, dbuf.dptr, sizeof(rec));
263
264         if (rec.msg_version != MESSAGE_VERSION) {
265                 DEBUG(0,("message version %d received (expected %d)\n", rec.msg_version, MESSAGE_VERSION));
266                 goto failed;
267         }
268
269         if (rec.len > 0) {
270                 (*buf) = (void *)malloc(rec.len);
271                 if (!(*buf)) goto failed;
272
273                 memcpy(*buf, dbuf.dptr+sizeof(rec), rec.len);
274         } else {
275                 *buf = NULL;
276         }
277
278         *len = rec.len;
279         *msg_type = rec.msg_type;
280         *src = rec.src;
281
282         if (dbuf.dsize - (sizeof(rec)+rec.len) > 0)
283                 memmove(dbuf.dptr, dbuf.dptr+sizeof(rec)+rec.len, dbuf.dsize - (sizeof(rec)+rec.len));
284         dbuf.dsize -= sizeof(rec)+rec.len;
285
286         if (dbuf.dsize == 0)
287                 tdb_delete(tdb, kbuf);
288         else
289                 tdb_store(tdb, kbuf, dbuf, TDB_REPLACE);
290
291         free(dbuf.dptr);
292         tdb_chainunlock(tdb, kbuf);
293         return True;
294
295  failed:
296         tdb_chainunlock(tdb, kbuf);
297         return False;
298 }
299
300
301 /****************************************************************************
302 receive and dispatch any messages pending for this process
303 notice that all dispatch handlers for a particular msg_type get called,
304 so you can register multiple handlers for a message
305 ****************************************************************************/
306 void message_dispatch(void)
307 {
308         int msg_type;
309         pid_t src;
310         void *buf;
311         size_t len;
312         struct dispatch_fns *dfn;
313
314         if (!received_signal) return;
315         received_signal = 0;
316
317         while (message_recv(&msg_type, &src, &buf, &len)) {
318                 for (dfn = dispatch_fns; dfn; dfn = dfn->next) {
319                         if (dfn->msg_type == msg_type) {
320                                 dfn->fn(msg_type, src, buf, len);
321                         }
322                 }
323                 if (buf) free(buf);
324         }
325 }
326
327
328 /****************************************************************************
329 register a dispatch function for a particular message type
330 ****************************************************************************/
331 void message_register(int msg_type, 
332                       void (*fn)(int msg_type, pid_t pid, void *buf, size_t len))
333 {
334         struct dispatch_fns *dfn;
335
336         dfn = (struct dispatch_fns *)malloc(sizeof(*dfn));
337
338         ZERO_STRUCTP(dfn);
339
340         dfn->msg_type = msg_type;
341         dfn->fn = fn;
342
343         DLIST_ADD(dispatch_fns, dfn);
344 }
345
346 /****************************************************************************
347 de-register the function for a particular message type
348 ****************************************************************************/
349 void message_deregister(int msg_type)
350 {
351         struct dispatch_fns *dfn, *next;
352
353         for (dfn = dispatch_fns; dfn; dfn = next) {
354                 next = dfn->next;
355                 if (dfn->msg_type == msg_type) {
356                         DLIST_REMOVE(dispatch_fns, dfn);
357                         free(dfn);
358                 }
359         }       
360 }
361
362 static struct {
363         int msg_type;
364         void *buf;
365         size_t len;
366         BOOL duplicates;
367 } msg_all;
368
369 /****************************************************************************
370 send one of the messages for the broadcast
371 ****************************************************************************/
372 static int traverse_fn(TDB_CONTEXT *the_tdb, TDB_DATA kbuf, TDB_DATA dbuf, void *state)
373 {
374         struct connections_data crec;
375
376         memcpy(&crec, dbuf.dptr, sizeof(crec));
377
378         if (crec.cnum != -1) return 0;
379         message_send_pid(crec.pid, msg_all.msg_type, msg_all.buf, msg_all.len, msg_all.duplicates);
380         return 0;
381 }
382
383 /****************************************************************************
384 this is a useful function for sending messages to all smbd processes.
385 It isn't very efficient, but should be OK for the sorts of applications that 
386 use it. When we need efficient broadcast we can add it.
387 ****************************************************************************/
388 BOOL message_send_all(TDB_CONTEXT *conn_tdb, int msg_type, void *buf, size_t len, BOOL duplicates_allowed)
389 {
390         msg_all.msg_type = msg_type;
391         msg_all.buf = buf;
392         msg_all.len = len;
393         msg_all.duplicates = duplicates_allowed;
394
395         tdb_traverse(conn_tdb, traverse_fn, NULL);
396         return True;
397 }