r3314: added a option "socket:testnonblock" to the generic socket code. If
[bbaumbach/samba-autobuild/.git] / source4 / lib / messaging / messaging.c
1 /* 
2    Unix SMB/CIFS implementation.
3
4    Samba internal messaging functions
5
6    Copyright (C) Andrew Tridgell 2004
7    
8    This program is free software; you can redistribute it and/or modify
9    it under the terms of the GNU General Public License as published by
10    the Free Software Foundation; either version 2 of the License, or
11    (at your option) any later version.
12    
13    This program is distributed in the hope that it will be useful,
14    but WITHOUT ANY WARRANTY; without even the implied warranty of
15    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
16    GNU General Public License for more details.
17    
18    You should have received a copy of the GNU General Public License
19    along with this program; if not, write to the Free Software
20    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
21 */
22
23 #include "includes.h"
24
25 /* change the message version with any incompatible changes in the protocol */
26 #define MESSAGING_VERSION 1
27
28 struct messaging_context {
29         servid_t server_id;
30         struct socket_context *sock;
31         char *path;
32         struct dispatch_fn *dispatch;
33
34         struct {
35                 struct event_context *ev;
36                 struct fd_event *fde;
37         } event;
38 };
39
40 /* we have a linked list of dispatch handlers that this messaging
41    server can deal with */
42 struct dispatch_fn {
43         struct dispatch_fn *next, *prev;
44         uint32_t msg_type;
45         void *private;
46         void (*fn)(struct messaging_context *msg, void *private, 
47                    uint32_t msg_type, servid_t server_id, DATA_BLOB *data);
48 };
49
50 /* an individual message */
51 struct messaging_rec {
52         struct messaging_context *msg;
53         struct socket_context *sock;
54         struct fd_event *fde;
55         const char *path;
56
57         struct {
58                 uint32_t version;
59                 uint32_t msg_type;
60                 servid_t from;
61                 servid_t to;
62                 uint32_t length;
63         } header;
64
65         DATA_BLOB data;
66
67         uint32_t ndone;
68 };
69
70 /*
71  A useful function for testing the message system.
72 */
73 static void ping_message(struct messaging_context *msg, void *private, 
74                          uint32_t msg_type, servid_t src, DATA_BLOB *data)
75 {
76         DEBUG(1,("INFO: Received PING message from server %u [%.*s]\n",
77                  (uint_t)src, data->length, data->data?(const char *)data->data:""));
78         messaging_send(msg, src, MSG_PONG, data);
79 }
80
81 /* 
82    return the path to a messaging socket
83 */
84 static char *messaging_path(TALLOC_CTX *mem_ctx, servid_t server_id)
85 {
86         char *name = talloc_asprintf(mem_ctx, "messaging/msg.%u", (unsigned)server_id);
87         char *ret;
88         ret = lock_path(mem_ctx, name);
89         talloc_free(name);
90         return ret;
91 }
92
93 /*
94   dispatch a fully received message
95 */
96 static void messaging_dispatch(struct messaging_context *msg, struct messaging_rec *rec)
97 {
98         struct dispatch_fn *d, *next;
99         for (d=msg->dispatch;d;d=next) {
100                 next = d->next;
101                 if (d->msg_type == rec->header.msg_type) {
102                         d->fn(msg, d->private, d->msg_type, rec->header.from, &rec->data);
103                 }
104         }
105
106         /* we don't free the record itself here as there may
107            be more messages from this client */
108         data_blob_free(&rec->data);
109         rec->header.length = 0;
110         rec->ndone = 0;
111 }
112
113
114 /*
115   handle IO for a single message
116 */
117 static void messaging_recv_handler(struct event_context *ev, struct fd_event *fde, 
118                                  time_t t, uint16_t flags)
119 {
120         struct messaging_rec *rec = fde->private;
121         struct messaging_context *msg = rec->msg;
122         NTSTATUS status;
123
124         if (rec->ndone < sizeof(rec->header)) {
125                 /* receive the header */
126                 size_t nread;
127
128                 status = socket_recv(rec->sock, 
129                                      rec->ndone + (char *)&rec->header,
130                                      sizeof(rec->header) - rec->ndone, &nread, 0);
131                 if (NT_STATUS_IS_ERR(status)) {
132                         talloc_free(rec);
133                         return;
134                 }
135
136                 if (nread == 0) {
137                         return;
138                 }
139
140                 rec->ndone += nread;
141
142                 if (rec->ndone == sizeof(rec->header)) {
143                         if (rec->header.version != MESSAGING_VERSION) {
144                                 DEBUG(0,("meessage with wrong version %u\n",
145                                          rec->header.version));
146                                 talloc_free(rec);
147                         }
148                         rec->data = data_blob_talloc(rec, NULL, rec->header.length);
149                         if (rec->data.length != rec->header.length) {
150                                 DEBUG(0,("Unable to allocate message of size %u\n",
151                                          rec->header.length));
152                                 talloc_free(rec);
153                         }
154                 }
155         }
156
157         if (rec->ndone >= sizeof(rec->header) && 
158             rec->ndone < sizeof(rec->header) + rec->header.length) {
159                 /* receive the body, if any */
160                 size_t nread;
161
162                 status = socket_recv(rec->sock, 
163                                      rec->data.data + (rec->ndone - sizeof(rec->header)),
164                                      sizeof(rec->header) + rec->header.length - rec->ndone, 
165                                      &nread, 0);
166                 if (NT_STATUS_IS_ERR(status)) {
167                         talloc_free(rec);
168                         return;
169                 }
170
171                 if (nread == 0) {
172                         return;
173                 }
174
175                 rec->ndone += nread;
176         }
177
178         if (rec->ndone == sizeof(rec->header) + rec->header.length) {
179                 /* we've got the whole message */
180                 messaging_dispatch(msg, rec);
181         }
182 }
183
184 /*
185   destroy a messaging record
186 */
187 static int rec_destructor(void *ptr)
188 {
189         struct messaging_rec *rec = ptr;
190         struct messaging_context *msg = rec->msg;
191         event_remove_fd(msg->event.ev, rec->fde);
192         return 0;
193 }
194
195 /*
196   handle a new incoming connection
197 */
198 static void messaging_listen_handler(struct event_context *ev, struct fd_event *fde, 
199                                      time_t t, uint16_t flags)
200 {
201         struct messaging_context *msg = fde->private;
202         struct messaging_rec *rec;
203         NTSTATUS status;
204         struct fd_event fde2;
205
206         rec = talloc_p(msg, struct messaging_rec);
207         if (rec == NULL) {
208                 smb_panic("Unable to allocate messaging_rec");
209         }
210
211         status = socket_accept(msg->sock, &rec->sock);
212         if (!NT_STATUS_IS_OK(status)) {
213                 smb_panic("Unable to accept messaging_rec");
214         }
215         talloc_steal(rec, rec->sock);
216
217         rec->msg = msg;
218         rec->ndone = 0;
219         rec->header.length = 0;
220         rec->path = msg->path;
221
222         fde2.private    = rec;
223         fde2.fd         = socket_get_fd(rec->sock);
224         fde2.flags      = EVENT_FD_READ;
225         fde2.handler    = messaging_recv_handler;
226
227         rec->fde        = event_add_fd(msg->event.ev, &fde2);
228
229         talloc_set_destructor(rec, rec_destructor);
230 }
231
232 /*
233   Register a dispatch function for a particular message type.
234 */
235 void messaging_register(struct messaging_context *msg, void *private,
236                         uint32_t msg_type, 
237                         void (*fn)(struct messaging_context *, void *, uint32_t, servid_t, DATA_BLOB *))
238 {
239         struct dispatch_fn *d;
240
241         d = talloc_p(msg, struct dispatch_fn);
242         d->msg_type = msg_type;
243         d->private = private;
244         d->fn = fn;
245         DLIST_ADD(msg->dispatch, d);
246 }
247
248 /*
249   De-register the function for a particular message type.
250 */
251 void messaging_deregister(struct messaging_context *msg, uint32_t msg_type, void *private)
252 {
253         struct dispatch_fn *d, *next;
254
255         for (d = msg->dispatch; d; d = next) {
256                 next = d->next;
257                 if (d->msg_type == msg_type && 
258                     d->private == private) {
259                         DLIST_REMOVE(msg->dispatch, d);
260                         talloc_free(d);
261                 }
262         }       
263 }
264
265
266
267 /*
268   handle IO for sending a message
269 */
270 static void messaging_send_handler(struct event_context *ev, struct fd_event *fde, 
271                                    time_t t, uint16_t flags)
272 {
273         struct messaging_rec *rec = fde->private;
274         NTSTATUS status;
275
276         if (rec->ndone < sizeof(rec->header)) {
277                 /* send the header */
278                 size_t nsent;
279                 DATA_BLOB blob;
280
281                 blob.data = rec->ndone + (char *)&rec->header;
282                 blob.length = sizeof(rec->header) - rec->ndone;
283
284                 status = socket_send(rec->sock, &blob, &nsent, 0);
285                 if (NT_STATUS_IS_ERR(status)) {
286                         talloc_free(rec);
287                         return;
288                 }
289
290                 if (nsent == 0) {
291                         return;
292                 }
293
294                 rec->ndone += nsent;
295         }
296
297         if (rec->ndone >= sizeof(rec->header) && 
298             rec->ndone < sizeof(rec->header) + rec->header.length) {
299                 /* send the body, if any */
300                 DATA_BLOB blob;
301                 size_t nsent;
302
303                 blob.data = rec->data.data + (rec->ndone - sizeof(rec->header));
304                 blob.length = rec->header.length - (rec->ndone - sizeof(rec->header));
305
306                 status = socket_send(rec->sock, &blob, &nsent, 0);
307                 if (NT_STATUS_IS_ERR(status)) {
308                         talloc_free(rec);
309                         return;
310                 }
311
312                 rec->ndone += nsent;
313         }
314
315         if (rec->ndone == sizeof(rec->header) + rec->header.length) {
316                 /* we've done the whole message */
317                 talloc_free(rec);
318         }
319 }
320
321
322 /*
323   when the servers listen queue is full we use this to backoff the message
324 */
325 static void messaging_backoff_handler(struct event_context *ev, struct timed_event *te, time_t t)
326 {
327         struct messaging_rec *rec = te->private;
328         struct messaging_context *msg = rec->msg;
329         NTSTATUS status;
330         struct fd_event fde;
331
332         status = socket_connect(rec->sock, NULL, 0, rec->path, 0, 0);
333         if (NT_STATUS_EQUAL(status, STATUS_MORE_ENTRIES)) {
334                 /* backoff again */
335                 te->next_event = t+1;
336                 return;
337         }
338
339         if (!NT_STATUS_IS_OK(status)) {
340                 DEBUG(1,("messaging: Lost message from %u to %u of type %u after backoff - %s\n", 
341                          rec->header.from, rec->header.to, rec->header.msg_type, nt_errstr(status)));
342                 talloc_free(rec);
343                 return;
344         }
345
346         fde.private     = rec;
347         fde.fd          = socket_get_fd(rec->sock);
348         fde.flags       = EVENT_FD_WRITE;
349         fde.handler     = messaging_send_handler;
350
351         rec->fde        = event_add_fd(msg->event.ev, &fde);
352
353         talloc_set_destructor(rec, rec_destructor);
354
355         messaging_send_handler(msg->event.ev, rec->fde, 0, EVENT_FD_WRITE);
356 }
357
358
359 /*
360   Send a message to a particular server
361 */
362 NTSTATUS messaging_send(struct messaging_context *msg, servid_t server, uint32_t msg_type, DATA_BLOB *data)
363 {
364         struct messaging_rec *rec;
365         NTSTATUS status;
366         struct fd_event fde;
367
368         rec = talloc_p(msg, struct messaging_rec);
369         if (rec == NULL) {
370                 return NT_STATUS_NO_MEMORY;
371         }
372
373         rec->msg = msg;
374         rec->header.version = MESSAGING_VERSION;
375         rec->header.msg_type = msg_type;
376         rec->header.from = msg->server_id;
377         rec->header.to = server;
378         rec->header.length = data?data->length:0;
379         if (rec->header.length != 0) {
380                 rec->data = data_blob_talloc(rec, data->data, data->length);
381         } else {
382                 rec->data = data_blob(NULL, 0);
383         }
384         rec->ndone = 0;
385
386         status = socket_create("unix", SOCKET_TYPE_STREAM, &rec->sock, 0);
387         if (!NT_STATUS_IS_OK(status)) {
388                 talloc_free(rec);
389                 return status;
390         }
391         talloc_steal(rec, rec->sock);
392
393         rec->path = messaging_path(rec, server);
394
395         status = socket_connect(rec->sock, NULL, 0, rec->path, 0, 0);
396         if (NT_STATUS_EQUAL(status, STATUS_MORE_ENTRIES)) {
397                 /* backoff on this message - the servers listen queue is full */
398                 struct timed_event te;
399                 te.next_event = time(NULL)+1;
400                 te.handler = messaging_backoff_handler;
401                 te.private = rec;
402                 event_add_timed(msg->event.ev, &te);
403                 return NT_STATUS_OK;
404         }
405
406         if (!NT_STATUS_IS_OK(status)) {
407                 talloc_free(rec);
408                 return status;
409         }
410
411         fde.private     = rec;
412         fde.fd          = socket_get_fd(rec->sock);
413         fde.flags       = EVENT_FD_WRITE;
414         fde.handler     = messaging_send_handler;
415
416         rec->fde        = event_add_fd(msg->event.ev, &fde);
417
418         talloc_set_destructor(rec, rec_destructor);
419
420         messaging_send_handler(msg->event.ev, rec->fde, 0, EVENT_FD_WRITE);
421
422         return NT_STATUS_OK;
423 }
424
425
426 /*
427   destroy the messaging context
428 */
429 static int messaging_destructor(void *ptr)
430 {
431         struct messaging_context *msg = ptr;
432         event_remove_fd(msg->event.ev, msg->event.fde);
433         unlink(msg->path);
434         return 0;
435 }
436
437 /*
438   create the listening socket and setup the dispatcher
439 */
440 struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx, servid_t server_id, struct event_context *ev)
441 {
442         struct messaging_context *msg;
443         NTSTATUS status;
444         struct fd_event fde;
445
446         msg = talloc_p(mem_ctx, struct messaging_context);
447         if (msg == NULL) {
448                 return NULL;
449         }
450
451         /* create the messaging directory if needed */
452         msg->path = lock_path(msg, "messaging");
453         mkdir(msg->path, 0700);
454         talloc_free(msg->path);
455
456         msg->server_id = server_id;
457         msg->dispatch = NULL;
458         msg->path = messaging_path(msg, server_id);
459
460         status = socket_create("unix", SOCKET_TYPE_STREAM, &msg->sock, 0);
461         if (!NT_STATUS_IS_OK(status)) {
462                 talloc_free(msg);
463                 return NULL;
464         }
465
466         /* by stealing here we ensure that the socket is cleaned up (and even 
467            deleted) on exit */
468         talloc_steal(msg, msg->sock);
469
470         status = socket_listen(msg->sock, msg->path, 0, 50, 0);
471         if (!NT_STATUS_IS_OK(status)) {
472                 DEBUG(0,("Unable to setup messaging listener for '%s'\n", msg->path));
473                 talloc_free(msg);
474                 return NULL;
475         }
476
477         fde.private     = msg;
478         fde.fd          = socket_get_fd(msg->sock);
479         fde.flags       = EVENT_FD_READ;
480         fde.handler     = messaging_listen_handler;
481
482         msg->event.ev   = ev;
483         msg->event.fde  = event_add_fd(ev, &fde);
484
485         talloc_set_destructor(msg, messaging_destructor);
486         
487         messaging_register(msg, NULL, MSG_PING, ping_message);
488
489         return msg;
490 }
491
492