?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
dsdb: Ignore errors from search in dns_notify module
[Samba.git] / source3 / lib / messages.c
blob7eadb0209b9679e27075c574d5e90e9dede6b282
1 /*
2 Unix SMB/CIFS implementation.
3 Samba internal messaging functions
4 Copyright (C) Andrew Tridgell 2000
5 Copyright (C) 2001 by Martin Pool
6 Copyright (C) 2002 by Jeremy Allison
7 Copyright (C) 2007 by Volker Lendecke
8
9 This program is free software; you can redistribute it and/or modify
10 it under the terms of the GNU General Public License as published by
11 the Free Software Foundation; either version 3 of the License, or
12 (at your option) any later version.
13
14 This program is distributed in the hope that it will be useful,
15 but WITHOUT ANY WARRANTY; without even the implied warranty of
16 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17 GNU General Public License for more details.
18
19 You should have received a copy of the GNU General Public License
20 along with this program. If not, see <http://www.gnu.org/licenses/>.
21 */
22
23 /**
24 @defgroup messages Internal messaging framework
25 @{
26 @file messages.c
27
28 @brief Module for internal messaging between Samba daemons.
29
30 The idea is that if a part of Samba wants to do communication with
31 another Samba process then it will do a message_register() of a
32 dispatch function, and use message_send_pid() to send messages to
33 that process.
34
35 The dispatch function is given the pid of the sender, and it can
36 use that to reply by message_send_pid(). See ping_message() for a
37 simple example.
38
39 @caution Dispatch functions must be able to cope with incoming
40 messages on an *odd* byte boundary.
41
42 This system doesn't have any inherent size limitations but is not
43 very efficient for large messages or when messages are sent in very
44 quick succession.
45
46 */
47
48 #include "includes.h"
49 #include "dbwrap/dbwrap.h"
50 #include "serverid.h"
51 #include "messages.h"
52 #include "lib/util/tevent_unix.h"
53 #include "lib/background.h"
54 #include "lib/messages_dgm.h"
55 #include "lib/iov_buf.h"
56 #include "lib/util/server_id_db.h"
57
58 struct messaging_callback {
59 struct messaging_callback *prev, *next;
60 uint32 msg_type;
61 void (*fn)(struct messaging_context *msg, void *private_data,
62 uint32_t msg_type,
63 struct server_id server_id, DATA_BLOB *data);
64 void *private_data;
65 };
66
67 struct messaging_context {
68 struct server_id id;
69 struct tevent_context *event_ctx;
70 struct messaging_callback *callbacks;
71
72 struct tevent_req **new_waiters;
73 unsigned num_new_waiters;
74
75 struct tevent_req **waiters;
76 unsigned num_waiters;
77
78 struct messaging_backend *remote;
79
80 struct server_id_db *names_db;
81 };
82
83 struct messaging_hdr {
84 uint32_t msg_type;
85 struct server_id dst;
86 struct server_id src;
87 };
88
89 /****************************************************************************
90 A useful function for testing the message system.
91 ****************************************************************************/
92
93 static void ping_message(struct messaging_context *msg_ctx,
94 void *private_data,
95 uint32_t msg_type,
96 struct server_id src,
97 DATA_BLOB *data)
98 {
99 struct server_id_buf idbuf;
100
101 DEBUG(1, ("INFO: Received PING message from PID %s [%.*s]\n",
102 server_id_str_buf(src, &idbuf), (int)data->length,
103 data->data ? (char *)data->data : ""));
104
105 messaging_send(msg_ctx, src, MSG_PONG, data);
106 }
107
108 /****************************************************************************
109 Register/replace a dispatch function for a particular message type.
110 JRA changed Dec 13 2006. Only one message handler now permitted per type.
111 *NOTE*: Dispatch functions must be able to cope with incoming
112 messages on an *odd* byte boundary.
113 ****************************************************************************/
114
115 struct msg_all {
116 struct messaging_context *msg_ctx;
117 int msg_type;
118 uint32 msg_flag;
119 const void *buf;
120 size_t len;
121 int n_sent;
122 };
123
124 /****************************************************************************
125 Send one of the messages for the broadcast.
126 ****************************************************************************/
127
128 static int traverse_fn(struct db_record *rec, const struct server_id *id,
129 uint32_t msg_flags, void *state)
130 {
131 struct msg_all *msg_all = (struct msg_all *)state;
132 NTSTATUS status;
133
134 /* Don't send if the receiver hasn't registered an interest. */
135
136 if((msg_flags & msg_all->msg_flag) == 0) {
137 return 0;
138 }
139
140 /* If the msg send fails because the pid was not found (i.e. smbd died),
141 * the msg has already been deleted from the messages.tdb.*/
142
143 status = messaging_send_buf(msg_all->msg_ctx, *id, msg_all->msg_type,
144 (const uint8_t *)msg_all->buf, msg_all->len);
145
146 if (NT_STATUS_EQUAL(status, NT_STATUS_INVALID_HANDLE)) {
147 struct server_id_buf idbuf;
148
149 /*
150 * If the pid was not found delete the entry from
151 * serverid.tdb
152 */
153
154 DEBUG(2, ("pid %s doesn't exist\n",
155 server_id_str_buf(*id, &idbuf)));
156
157 dbwrap_record_delete(rec);
158 }
159 msg_all->n_sent++;
160 return 0;
161 }
162
163 /**
164 * Send a message to all smbd processes.
165 *
166 * It isn't very efficient, but should be OK for the sorts of
167 * applications that use it. When we need efficient broadcast we can add
168 * it.
169 *
170 * @param n_sent Set to the number of messages sent. This should be
171 * equal to the number of processes, but be careful for races.
172 *
173 * @retval True for success.
174 **/
175 bool message_send_all(struct messaging_context *msg_ctx,
176 int msg_type,
177 const void *buf, size_t len,
178 int *n_sent)
179 {
180 struct msg_all msg_all;
181
182 msg_all.msg_type = msg_type;
183 if (msg_type < 0x100) {
184 msg_all.msg_flag = FLAG_MSG_GENERAL;
185 } else if (msg_type > 0x100 && msg_type < 0x200) {
186 msg_all.msg_flag = FLAG_MSG_NMBD;
187 } else if (msg_type > 0x200 && msg_type < 0x300) {
188 msg_all.msg_flag = FLAG_MSG_PRINT_GENERAL;
189 } else if (msg_type > 0x300 && msg_type < 0x400) {
190 msg_all.msg_flag = FLAG_MSG_SMBD;
191 } else if (msg_type > 0x400 && msg_type < 0x600) {
192 msg_all.msg_flag = FLAG_MSG_WINBIND;
193 } else if (msg_type > 4000 && msg_type < 5000) {
194 msg_all.msg_flag = FLAG_MSG_DBWRAP;
195 } else {
196 return false;
197 }
198
199 msg_all.buf = buf;
200 msg_all.len = len;
201 msg_all.n_sent = 0;
202 msg_all.msg_ctx = msg_ctx;
203
204 serverid_traverse(traverse_fn, &msg_all);
205 if (n_sent)
206 *n_sent = msg_all.n_sent;
207 return true;
208 }
209
210 static void messaging_recv_cb(const uint8_t *msg, size_t msg_len,
211 int *fds, size_t num_fds,
212 void *private_data)
213 {
214 struct messaging_context *msg_ctx = talloc_get_type_abort(
215 private_data, struct messaging_context);
216 const struct messaging_hdr *hdr;
217 struct server_id_buf idbuf;
218 struct messaging_rec rec;
219 int64_t fds64[MIN(num_fds, INT8_MAX)];
220 size_t i;
221
222 if (msg_len < sizeof(*hdr)) {
223 for (i=0; i < num_fds; i++) {
224 close(fds[i]);
225 }
226 DEBUG(1, ("message too short: %u\n", (unsigned)msg_len));
227 return;
228 }
229
230 if (num_fds > INT8_MAX) {
231 for (i=0; i < num_fds; i++) {
232 close(fds[i]);
233 }
234 DEBUG(1, ("too many fds: %u\n", (unsigned)num_fds));
235 return;
236 }
237
238 /*
239 * "consume" the fds by copying them and setting
240 * the original variable to -1
241 */
242 for (i=0; i < num_fds; i++) {
243 fds64[i] = fds[i];
244 fds[i] = -1;
245 }
246
247 /*
248 * messages_dgm guarantees alignment, so we can cast here
249 */
250 hdr = (const struct messaging_hdr *)msg;
251
252 DEBUG(10, ("%s: Received message 0x%x len %u (num_fds:%u) from %s\n",
253 __func__, (unsigned)hdr->msg_type,
254 (unsigned)(msg_len - sizeof(*hdr)),
255 (unsigned)num_fds,
256 server_id_str_buf(hdr->src, &idbuf)));
257
258 rec = (struct messaging_rec) {
259 .msg_version = MESSAGE_VERSION,
260 .msg_type = hdr->msg_type,
261 .src = hdr->src,
262 .dest = hdr->dst,
263 .buf.data = discard_const_p(uint8, msg) + sizeof(*hdr),
264 .buf.length = msg_len - sizeof(*hdr),
265 .num_fds = num_fds,
266 .fds = fds64,
267 };
268
269 messaging_dispatch_rec(msg_ctx, &rec);
270 }
271
272 static int messaging_context_destructor(struct messaging_context *ctx)
273 {
274 unsigned i;
275
276 messaging_dgm_destroy();
277
278 for (i=0; i<ctx->num_new_waiters; i++) {
279 if (ctx->new_waiters[i] != NULL) {
280 tevent_req_set_cleanup_fn(ctx->new_waiters[i], NULL);
281 ctx->new_waiters[i] = NULL;
282 }
283 }
284 for (i=0; i<ctx->num_waiters; i++) {
285 if (ctx->waiters[i] != NULL) {
286 tevent_req_set_cleanup_fn(ctx->waiters[i], NULL);
287 ctx->waiters[i] = NULL;
288 }
289 }
290
291 return 0;
292 }
293
294 struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx,
295 struct tevent_context *ev)
296 {
297 struct messaging_context *ctx;
298 NTSTATUS status;
299 int ret;
300
301 if (!(ctx = talloc_zero(mem_ctx, struct messaging_context))) {
302 return NULL;
303 }
304
305 ctx->id = procid_self();
306 ctx->event_ctx = ev;
307
308 sec_init();
309
310 ret = messaging_dgm_init(ctx->event_ctx, ctx->id,
311 lp_cache_directory(), sec_initial_uid(),
312 messaging_recv_cb, ctx);
313
314 if (ret != 0) {
315 DEBUG(2, ("messaging_dgm_init failed: %s\n", strerror(ret)));
316 TALLOC_FREE(ctx);
317 return NULL;
318 }
319
320 ctx->names_db = server_id_db_init(
321 ctx, ctx->id, lp_cache_directory(), 0,
322 TDB_INCOMPATIBLE_HASH|TDB_CLEAR_IF_FIRST);
323 if (ctx->names_db == NULL) {
324 DEBUG(10, ("%s: server_id_db_init failed\n", __func__));
325 TALLOC_FREE(ctx);
326 return NULL;
327 }
328
329 talloc_set_destructor(ctx, messaging_context_destructor);
330
331 if (lp_clustering()) {
332 status = messaging_ctdbd_init(ctx, ctx, &ctx->remote);
333
334 if (!NT_STATUS_IS_OK(status)) {
335 DEBUG(2, ("messaging_ctdbd_init failed: %s\n",
336 nt_errstr(status)));
337 TALLOC_FREE(ctx);
338 return NULL;
339 }
340 }
341 ctx->id.vnn = get_my_vnn();
342
343 messaging_register(ctx, NULL, MSG_PING, ping_message);
344
345 /* Register some debugging related messages */
346
347 register_msg_pool_usage(ctx);
348 register_dmalloc_msgs(ctx);
349 debug_register_msgs(ctx);
350
351 return ctx;
352 }
353
354 struct server_id messaging_server_id(const struct messaging_context *msg_ctx)
355 {
356 return msg_ctx->id;
357 }
358
359 /*
360 * re-init after a fork
361 */
362 NTSTATUS messaging_reinit(struct messaging_context *msg_ctx)
363 {
364 NTSTATUS status;
365 int ret;
366
367 messaging_dgm_destroy();
368
369 msg_ctx->id = procid_self();
370
371 ret = messaging_dgm_init(msg_ctx->event_ctx, msg_ctx->id,
372 lp_cache_directory(), sec_initial_uid(),
373 messaging_recv_cb, msg_ctx);
374 if (ret != 0) {
375 DEBUG(0, ("messaging_dgm_init failed: %s\n", strerror(errno)));
376 return map_nt_error_from_unix(ret);
377 }
378
379 TALLOC_FREE(msg_ctx->remote);
380
381 if (lp_clustering()) {
382 status = messaging_ctdbd_init(msg_ctx, msg_ctx,
383 &msg_ctx->remote);
384
385 if (!NT_STATUS_IS_OK(status)) {
386 DEBUG(1, ("messaging_ctdbd_init failed: %s\n",
387 nt_errstr(status)));
388 return status;
389 }
390 }
391
392 server_id_db_reinit(msg_ctx->names_db, msg_ctx->id);
393
394 return NT_STATUS_OK;
395 }
396
397
398 /*
399 * Register a dispatch function for a particular message type. Allow multiple
400 * registrants
401 */
402 NTSTATUS messaging_register(struct messaging_context *msg_ctx,
403 void *private_data,
404 uint32_t msg_type,
405 void (*fn)(struct messaging_context *msg,
406 void *private_data,
407 uint32_t msg_type,
408 struct server_id server_id,
409 DATA_BLOB *data))
410 {
411 struct messaging_callback *cb;
412
413 DEBUG(5, ("Registering messaging pointer for type %u - "
414 "private_data=%p\n",
415 (unsigned)msg_type, private_data));
416
417 /*
418 * Only one callback per type
419 */
420
421 for (cb = msg_ctx->callbacks; cb != NULL; cb = cb->next) {
422 /* we allow a second registration of the same message
423 type if it has a different private pointer. This is
424 needed in, for example, the internal notify code,
425 which creates a new notify context for each tree
426 connect, and expects to receive messages to each of
427 them. */
428 if (cb->msg_type == msg_type && private_data == cb->private_data) {
429 DEBUG(5,("Overriding messaging pointer for type %u - private_data=%p\n",
430 (unsigned)msg_type, private_data));
431 cb->fn = fn;
432 cb->private_data = private_data;
433 return NT_STATUS_OK;
434 }
435 }
436
437 if (!(cb = talloc(msg_ctx, struct messaging_callback))) {
438 return NT_STATUS_NO_MEMORY;
439 }
440
441 cb->msg_type = msg_type;
442 cb->fn = fn;
443 cb->private_data = private_data;
444
445 DLIST_ADD(msg_ctx->callbacks, cb);
446 return NT_STATUS_OK;
447 }
448
449 /*
450 De-register the function for a particular message type.
451 */
452 void messaging_deregister(struct messaging_context *ctx, uint32_t msg_type,
453 void *private_data)
454 {
455 struct messaging_callback *cb, *next;
456
457 for (cb = ctx->callbacks; cb; cb = next) {
458 next = cb->next;
459 if ((cb->msg_type == msg_type)
460 && (cb->private_data == private_data)) {
461 DEBUG(5,("Deregistering messaging pointer for type %u - private_data=%p\n",
462 (unsigned)msg_type, private_data));
463 DLIST_REMOVE(ctx->callbacks, cb);
464 TALLOC_FREE(cb);
465 }
466 }
467 }
468
469 /*
470 Send a message to a particular server
471 */
472 NTSTATUS messaging_send(struct messaging_context *msg_ctx,
473 struct server_id server, uint32_t msg_type,
474 const DATA_BLOB *data)
475 {
476 struct iovec iov;
477
478 iov.iov_base = data->data;
479 iov.iov_len = data->length;
480
481 return messaging_send_iov(msg_ctx, server, msg_type, &iov, 1, NULL, 0);
482 }
483
484 NTSTATUS messaging_send_buf(struct messaging_context *msg_ctx,
485 struct server_id server, uint32_t msg_type,
486 const uint8_t *buf, size_t len)
487 {
488 DATA_BLOB blob = data_blob_const(buf, len);
489 return messaging_send(msg_ctx, server, msg_type, &blob);
490 }
491
492 NTSTATUS messaging_send_iov_from(struct messaging_context *msg_ctx,
493 struct server_id src, struct server_id dst,
494 uint32_t msg_type,
495 const struct iovec *iov, int iovlen,
496 const int *fds, size_t num_fds)
497 {
498 int ret;
499 struct messaging_hdr hdr;
500 struct iovec iov2[iovlen+1];
501
502 if (server_id_is_disconnected(&dst)) {
503 return NT_STATUS_INVALID_PARAMETER_MIX;
504 }
505
506 if (num_fds > INT8_MAX) {
507 return NT_STATUS_INVALID_PARAMETER_MIX;
508 }
509
510 if (!procid_is_local(&dst)) {
511 if (num_fds > 0) {
512 return NT_STATUS_NOT_SUPPORTED;
513 }
514
515 ret = msg_ctx->remote->send_fn(src, dst,
516 msg_type, iov, iovlen,
517 NULL, 0,
518 msg_ctx->remote);
519 if (ret != 0) {
520 return map_nt_error_from_unix(ret);
521 }
522 return NT_STATUS_OK;
523 }
524
525 ZERO_STRUCT(hdr);
526 hdr = (struct messaging_hdr) {
527 .msg_type = msg_type,
528 .dst = dst,
529 .src = src
530 };
531 iov2[0] = (struct iovec){ .iov_base = &hdr, .iov_len = sizeof(hdr) };
532 memcpy(&iov2[1], iov, iovlen * sizeof(*iov));
533
534 become_root();
535 ret = messaging_dgm_send(dst.pid, iov2, iovlen+1, fds, num_fds);
536 unbecome_root();
537
538 if (ret != 0) {
539 return map_nt_error_from_unix(ret);
540 }
541 return NT_STATUS_OK;
542 }
543
544 NTSTATUS messaging_send_iov(struct messaging_context *msg_ctx,
545 struct server_id dst, uint32_t msg_type,
546 const struct iovec *iov, int iovlen,
547 const int *fds, size_t num_fds)
548 {
549 return messaging_send_iov_from(msg_ctx, msg_ctx->id, dst, msg_type,
550 iov, iovlen, fds, num_fds);
551 }
552
553 static struct messaging_rec *messaging_rec_dup(TALLOC_CTX *mem_ctx,
554 struct messaging_rec *rec)
555 {
556 struct messaging_rec *result;
557 size_t fds_size = sizeof(int64_t) * rec->num_fds;
558
559 result = talloc_pooled_object(mem_ctx, struct messaging_rec, 2,
560 rec->buf.length + fds_size);
561 if (result == NULL) {
562 return NULL;
563 }
564 *result = *rec;
565
566 /* Doesn't fail, see talloc_pooled_object */
567
568 result->buf.data = talloc_memdup(result, rec->buf.data,
569 rec->buf.length);
570
571 result->fds = NULL;
572 if (result->num_fds > 0) {
573 result->fds = talloc_array(result, int64_t, result->num_fds);
574 memcpy(result->fds, rec->fds, fds_size);
575 }
576
577 return result;
578 }
579
580 struct messaging_filtered_read_state {
581 struct tevent_context *ev;
582 struct messaging_context *msg_ctx;
583 void *tevent_handle;
584
585 bool (*filter)(struct messaging_rec *rec, void *private_data);
586 void *private_data;
587
588 struct messaging_rec *rec;
589 };
590
591 static void messaging_filtered_read_cleanup(struct tevent_req *req,
592 enum tevent_req_state req_state);
593
594 struct tevent_req *messaging_filtered_read_send(
595 TALLOC_CTX *mem_ctx, struct tevent_context *ev,
596 struct messaging_context *msg_ctx,
597 bool (*filter)(struct messaging_rec *rec, void *private_data),
598 void *private_data)
599 {
600 struct tevent_req *req;
601 struct messaging_filtered_read_state *state;
602 size_t new_waiters_len;
603
604 req = tevent_req_create(mem_ctx, &state,
605 struct messaging_filtered_read_state);
606 if (req == NULL) {
607 return NULL;
608 }
609 state->ev = ev;
610 state->msg_ctx = msg_ctx;
611 state->filter = filter;
612 state->private_data = private_data;
613
614 /*
615 * We have to defer the callback here, as we might be called from
616 * within a different tevent_context than state->ev
617 */
618 tevent_req_defer_callback(req, state->ev);
619
620 state->tevent_handle = messaging_dgm_register_tevent_context(
621 state, ev);
622 if (tevent_req_nomem(state, req)) {
623 return tevent_req_post(req, ev);
624 }
625
626 /*
627 * We add ourselves to the "new_waiters" array, not the "waiters"
628 * array. If we are called from within messaging_read_done,
629 * messaging_dispatch_rec will be in an active for-loop on
630 * "waiters". We must be careful not to mess with this array, because
631 * it could mean that a single event is being delivered twice.
632 */
633
634 new_waiters_len = talloc_array_length(msg_ctx->new_waiters);
635
636 if (new_waiters_len == msg_ctx->num_new_waiters) {
637 struct tevent_req **tmp;
638
639 tmp = talloc_realloc(msg_ctx, msg_ctx->new_waiters,
640 struct tevent_req *, new_waiters_len+1);
641 if (tevent_req_nomem(tmp, req)) {
642 return tevent_req_post(req, ev);
643 }
644 msg_ctx->new_waiters = tmp;
645 }
646
647 msg_ctx->new_waiters[msg_ctx->num_new_waiters] = req;
648 msg_ctx->num_new_waiters += 1;
649 tevent_req_set_cleanup_fn(req, messaging_filtered_read_cleanup);
650
651 return req;
652 }
653
654 static void messaging_filtered_read_cleanup(struct tevent_req *req,
655 enum tevent_req_state req_state)
656 {
657 struct messaging_filtered_read_state *state = tevent_req_data(
658 req, struct messaging_filtered_read_state);
659 struct messaging_context *msg_ctx = state->msg_ctx;
660 unsigned i;
661
662 tevent_req_set_cleanup_fn(req, NULL);
663
664 TALLOC_FREE(state->tevent_handle);
665
666 /*
667 * Just set the [new_]waiters entry to NULL, be careful not to mess
668 * with the other "waiters" array contents. We are often called from
669 * within "messaging_dispatch_rec", which loops over
670 * "waiters". Messing with the "waiters" array will mess up that
671 * for-loop.
672 */
673
674 for (i=0; i<msg_ctx->num_waiters; i++) {
675 if (msg_ctx->waiters[i] == req) {
676 msg_ctx->waiters[i] = NULL;
677 return;
678 }
679 }
680
681 for (i=0; i<msg_ctx->num_new_waiters; i++) {
682 if (msg_ctx->new_waiters[i] == req) {
683 msg_ctx->new_waiters[i] = NULL;
684 return;
685 }
686 }
687 }
688
689 static void messaging_filtered_read_done(struct tevent_req *req,
690 struct messaging_rec *rec)
691 {
692 struct messaging_filtered_read_state *state = tevent_req_data(
693 req, struct messaging_filtered_read_state);
694
695 state->rec = messaging_rec_dup(state, rec);
696 if (tevent_req_nomem(state->rec, req)) {
697 return;
698 }
699 tevent_req_done(req);
700 }
701
702 int messaging_filtered_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
703 struct messaging_rec **presult)
704 {
705 struct messaging_filtered_read_state *state = tevent_req_data(
706 req, struct messaging_filtered_read_state);
707 int err;
708
709 if (tevent_req_is_unix_error(req, &err)) {
710 tevent_req_received(req);
711 return err;
712 }
713 *presult = talloc_move(mem_ctx, &state->rec);
714 return 0;
715 }
716
717 struct messaging_read_state {
718 uint32_t msg_type;
719 struct messaging_rec *rec;
720 };
721
722 static bool messaging_read_filter(struct messaging_rec *rec,
723 void *private_data);
724 static void messaging_read_done(struct tevent_req *subreq);
725
726 struct tevent_req *messaging_read_send(TALLOC_CTX *mem_ctx,
727 struct tevent_context *ev,
728 struct messaging_context *msg,
729 uint32_t msg_type)
730 {
731 struct tevent_req *req, *subreq;
732 struct messaging_read_state *state;
733
734 req = tevent_req_create(mem_ctx, &state,
735 struct messaging_read_state);
736 if (req == NULL) {
737 return NULL;
738 }
739 state->msg_type = msg_type;
740
741 subreq = messaging_filtered_read_send(state, ev, msg,
742 messaging_read_filter, state);
743 if (tevent_req_nomem(subreq, req)) {
744 return tevent_req_post(req, ev);
745 }
746 tevent_req_set_callback(subreq, messaging_read_done, req);
747 return req;
748 }
749
750 static bool messaging_read_filter(struct messaging_rec *rec,
751 void *private_data)
752 {
753 struct messaging_read_state *state = talloc_get_type_abort(
754 private_data, struct messaging_read_state);
755
756 if (rec->num_fds != 0) {
757 return false;
758 }
759
760 return rec->msg_type == state->msg_type;
761 }
762
763 static void messaging_read_done(struct tevent_req *subreq)
764 {
765 struct tevent_req *req = tevent_req_callback_data(
766 subreq, struct tevent_req);
767 struct messaging_read_state *state = tevent_req_data(
768 req, struct messaging_read_state);
769 int ret;
770
771 ret = messaging_filtered_read_recv(subreq, state, &state->rec);
772 TALLOC_FREE(subreq);
773 if (tevent_req_error(req, ret)) {
774 return;
775 }
776 tevent_req_done(req);
777 }
778
779 int messaging_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
780 struct messaging_rec **presult)
781 {
782 struct messaging_read_state *state = tevent_req_data(
783 req, struct messaging_read_state);
784 int err;
785
786 if (tevent_req_is_unix_error(req, &err)) {
787 return err;
788 }
789 if (presult != NULL) {
790 *presult = talloc_move(mem_ctx, &state->rec);
791 }
792 return 0;
793 }
794
795 struct messaging_handler_state {
796 struct tevent_context *ev;
797 struct messaging_context *msg_ctx;
798 uint32_t msg_type;
799 bool (*handler)(struct messaging_context *msg_ctx,
800 struct messaging_rec **rec, void *private_data);
801 void *private_data;
802 };
803
804 static void messaging_handler_got_msg(struct tevent_req *subreq);
805
806 struct tevent_req *messaging_handler_send(
807 TALLOC_CTX *mem_ctx, struct tevent_context *ev,
808 struct messaging_context *msg_ctx, uint32_t msg_type,
809 bool (*handler)(struct messaging_context *msg_ctx,
810 struct messaging_rec **rec, void *private_data),
811 void *private_data)
812 {
813 struct tevent_req *req, *subreq;
814 struct messaging_handler_state *state;
815
816 req = tevent_req_create(mem_ctx, &state,
817 struct messaging_handler_state);
818 if (req == NULL) {
819 return NULL;
820 }
821 state->ev = ev;
822 state->msg_ctx = msg_ctx;
823 state->msg_type = msg_type;
824 state->handler = handler;
825 state->private_data = private_data;
826
827 subreq = messaging_read_send(state, state->ev, state->msg_ctx,
828 state->msg_type);
829 if (tevent_req_nomem(subreq, req)) {
830 return tevent_req_post(req, ev);
831 }
832 tevent_req_set_callback(subreq, messaging_handler_got_msg, req);
833 return req;
834 }
835
836 static void messaging_handler_got_msg(struct tevent_req *subreq)
837 {
838 struct tevent_req *req = tevent_req_callback_data(
839 subreq, struct tevent_req);
840 struct messaging_handler_state *state = tevent_req_data(
841 req, struct messaging_handler_state);
842 struct messaging_rec *rec;
843 int ret;
844 bool ok;
845
846 ret = messaging_read_recv(subreq, state, &rec);
847 TALLOC_FREE(subreq);
848 if (tevent_req_error(req, ret)) {
849 return;
850 }
851
852 subreq = messaging_read_send(state, state->ev, state->msg_ctx,
853 state->msg_type);
854 if (tevent_req_nomem(subreq, req)) {
855 return;
856 }
857 tevent_req_set_callback(subreq, messaging_handler_got_msg, req);
858
859 ok = state->handler(state->msg_ctx, &rec, state->private_data);
860 TALLOC_FREE(rec);
861 if (ok) {
862 /*
863 * Next round
864 */
865 return;
866 }
867 TALLOC_FREE(subreq);
868 tevent_req_done(req);
869 }
870
871 int messaging_handler_recv(struct tevent_req *req)
872 {
873 return tevent_req_simple_recv_unix(req);
874 }
875
876 static bool messaging_append_new_waiters(struct messaging_context *msg_ctx)
877 {
878 if (msg_ctx->num_new_waiters == 0) {
879 return true;
880 }
881
882 if (talloc_array_length(msg_ctx->waiters) <
883 (msg_ctx->num_waiters + msg_ctx->num_new_waiters)) {
884 struct tevent_req **tmp;
885 tmp = talloc_realloc(
886 msg_ctx, msg_ctx->waiters, struct tevent_req *,
887 msg_ctx->num_waiters + msg_ctx->num_new_waiters);
888 if (tmp == NULL) {
889 DEBUG(1, ("%s: talloc failed\n", __func__));
890 return false;
891 }
892 msg_ctx->waiters = tmp;
893 }
894
895 memcpy(&msg_ctx->waiters[msg_ctx->num_waiters], msg_ctx->new_waiters,
896 sizeof(struct tevent_req *) * msg_ctx->num_new_waiters);
897
898 msg_ctx->num_waiters += msg_ctx->num_new_waiters;
899 msg_ctx->num_new_waiters = 0;
900
901 return true;
902 }
903
904 /*
905 Dispatch one messaging_rec
906 */
907 void messaging_dispatch_rec(struct messaging_context *msg_ctx,
908 struct messaging_rec *rec)
909 {
910 struct messaging_callback *cb, *next;
911 unsigned i;
912 size_t j;
913
914 for (cb = msg_ctx->callbacks; cb != NULL; cb = next) {
915 next = cb->next;
916 if (cb->msg_type != rec->msg_type) {
917 continue;
918 }
919
920 /*
921 * the old style callbacks don't support fd passing
922 */
923 for (j=0; j < rec->num_fds; j++) {
924 int fd = rec->fds[j];
925 close(fd);
926 }
927 rec->num_fds = 0;
928 rec->fds = NULL;
929
930 cb->fn(msg_ctx, cb->private_data, rec->msg_type,
931 rec->src, &rec->buf);
932
933 /*
934 * we continue looking for matching messages after finding
935 * one. This matters for subsystems like the internal notify
936 * code which register more than one handler for the same
937 * message type
938 */
939 }
940
941 if (!messaging_append_new_waiters(msg_ctx)) {
942 for (j=0; j < rec->num_fds; j++) {
943 int fd = rec->fds[j];
944 close(fd);
945 }
946 rec->num_fds = 0;
947 rec->fds = NULL;
948 return;
949 }
950
951 i = 0;
952 while (i < msg_ctx->num_waiters) {
953 struct tevent_req *req;
954 struct messaging_filtered_read_state *state;
955
956 req = msg_ctx->waiters[i];
957 if (req == NULL) {
958 /*
959 * This got cleaned up. In the meantime,
960 * move everything down one. We need
961 * to keep the order of waiters, as
962 * other code may depend on this.
963 */
964 if (i < msg_ctx->num_waiters - 1) {
965 memmove(&msg_ctx->waiters[i],
966 &msg_ctx->waiters[i+1],
967 sizeof(struct tevent_req *) *
968 (msg_ctx->num_waiters - i - 1));
969 }
970 msg_ctx->num_waiters -= 1;
971 continue;
972 }
973
974 state = tevent_req_data(
975 req, struct messaging_filtered_read_state);
976 if (state->filter(rec, state->private_data)) {
977 messaging_filtered_read_done(req, rec);
978
979 /*
980 * Only the first one gets the fd-array
981 */
982 rec->num_fds = 0;
983 rec->fds = NULL;
984 }
985
986 i += 1;
987 }
988
989 /*
990 * If the fd-array isn't used, just close it.
991 */
992 for (j=0; j < rec->num_fds; j++) {
993 int fd = rec->fds[j];
994 close(fd);
995 }
996 rec->num_fds = 0;
997 rec->fds = NULL;
998 }
999
1000 static int mess_parent_dgm_cleanup(void *private_data);
1001 static void mess_parent_dgm_cleanup_done(struct tevent_req *req);
1002
1003 bool messaging_parent_dgm_cleanup_init(struct messaging_context *msg)
1004 {
1005 struct tevent_req *req;
1006
1007 req = background_job_send(
1008 msg, msg->event_ctx, msg, NULL, 0,
1009 lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
1010 60*15),
1011 mess_parent_dgm_cleanup, msg);
1012 if (req == NULL) {
1013 return false;
1014 }
1015 tevent_req_set_callback(req, mess_parent_dgm_cleanup_done, msg);
1016 return true;
1017 }
1018
1019 static int mess_parent_dgm_cleanup(void *private_data)
1020 {
1021 int ret;
1022
1023 ret = messaging_dgm_wipe();
1024 DEBUG(10, ("messaging_dgm_wipe returned %s\n",
1025 ret ? strerror(ret) : "ok"));
1026 return lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
1027 60*15);
1028 }
1029
1030 static void mess_parent_dgm_cleanup_done(struct tevent_req *req)
1031 {
1032 struct messaging_context *msg = tevent_req_callback_data(
1033 req, struct messaging_context);
1034 NTSTATUS status;
1035
1036 status = background_job_recv(req);
1037 TALLOC_FREE(req);
1038 DEBUG(1, ("messaging dgm cleanup job ended with %s\n",
1039 nt_errstr(status)));
1040
1041 req = background_job_send(
1042 msg, msg->event_ctx, msg, NULL, 0,
1043 lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
1044 60*15),
1045 mess_parent_dgm_cleanup, msg);
1046 if (req == NULL) {
1047 DEBUG(1, ("background_job_send failed\n"));
1048 }
1049 tevent_req_set_callback(req, mess_parent_dgm_cleanup_done, msg);
1050 }
1051
1052 int messaging_cleanup(struct messaging_context *msg_ctx, pid_t pid)
1053 {
1054 int ret;
1055
1056 if (pid == 0) {
1057 ret = messaging_dgm_wipe();
1058 } else {
1059 ret = messaging_dgm_cleanup(pid);
1060 }
1061
1062 return ret;
1063 }
1064
1065 struct tevent_context *messaging_tevent_context(
1066 struct messaging_context *msg_ctx)
1067 {
1068 return msg_ctx->event_ctx;
1069 }
1070
1071 struct server_id_db *messaging_names_db(struct messaging_context *msg_ctx)
1072 {
1073 return msg_ctx->names_db;
1074 }
1075
1076 /** @} **/
Samba GIT mirror