?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
gssapi: check for gss_acquire_cred_from
[Samba.git] / source3 / lib / messages.c
blobd7ad49d7c309c5a07e02b979ce7182994c6d2b91
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 "lib/util/server_id.h"
50 #include "dbwrap/dbwrap.h"
51 #include "serverid.h"
52 #include "messages.h"
53 #include "lib/util/tevent_unix.h"
54 #include "lib/background.h"
55 #include "lib/messages_dgm.h"
56 #include "lib/util/iov_buf.h"
57 #include "lib/util/server_id_db.h"
58 #include "lib/messages_dgm_ref.h"
59 #include "lib/messages_util.h"
60
61 struct messaging_callback {
62 struct messaging_callback *prev, *next;
63 uint32_t msg_type;
64 void (*fn)(struct messaging_context *msg, void *private_data,
65 uint32_t msg_type,
66 struct server_id server_id, DATA_BLOB *data);
67 void *private_data;
68 };
69
70 struct messaging_context {
71 struct server_id id;
72 struct tevent_context *event_ctx;
73 struct messaging_callback *callbacks;
74
75 struct tevent_req **new_waiters;
76 unsigned num_new_waiters;
77
78 struct tevent_req **waiters;
79 unsigned num_waiters;
80
81 void *msg_dgm_ref;
82 struct messaging_backend *remote;
83
84 struct server_id_db *names_db;
85 };
86
87 static struct messaging_rec *messaging_rec_dup(TALLOC_CTX *mem_ctx,
88 struct messaging_rec *rec);
89 static void messaging_dispatch_rec(struct messaging_context *msg_ctx,
90 struct tevent_context *ev,
91 struct messaging_rec *rec);
92
93 /****************************************************************************
94 A useful function for testing the message system.
95 ****************************************************************************/
96
97 static void ping_message(struct messaging_context *msg_ctx,
98 void *private_data,
99 uint32_t msg_type,
100 struct server_id src,
101 DATA_BLOB *data)
102 {
103 struct server_id_buf idbuf;
104
105 DEBUG(1, ("INFO: Received PING message from PID %s [%.*s]\n",
106 server_id_str_buf(src, &idbuf), (int)data->length,
107 data->data ? (char *)data->data : ""));
108
109 messaging_send(msg_ctx, src, MSG_PONG, data);
110 }
111
112 static struct messaging_rec *messaging_rec_create(
113 TALLOC_CTX *mem_ctx, struct server_id src, struct server_id dst,
114 uint32_t msg_type, const struct iovec *iov, int iovlen,
115 const int *fds, size_t num_fds)
116 {
117 ssize_t buflen;
118 uint8_t *buf;
119 struct messaging_rec *result;
120
121 if (num_fds > INT8_MAX) {
122 return NULL;
123 }
124
125 buflen = iov_buflen(iov, iovlen);
126 if (buflen == -1) {
127 return NULL;
128 }
129 buf = talloc_array(mem_ctx, uint8_t, buflen);
130 if (buf == NULL) {
131 return NULL;
132 }
133 iov_buf(iov, iovlen, buf, buflen);
134
135 {
136 struct messaging_rec rec;
137 int64_t fds64[num_fds];
138 size_t i;
139
140 for (i=0; i<num_fds; i++) {
141 fds64[i] = fds[i];
142 }
143
144 rec = (struct messaging_rec) {
145 .msg_version = MESSAGE_VERSION, .msg_type = msg_type,
146 .src = src, .dest = dst,
147 .buf.data = buf, .buf.length = buflen,
148 .num_fds = num_fds, .fds = fds64,
149 };
150
151 result = messaging_rec_dup(mem_ctx, &rec);
152 }
153
154 TALLOC_FREE(buf);
155
156 return result;
157 }
158
159 static void messaging_recv_cb(struct tevent_context *ev,
160 const uint8_t *msg, size_t msg_len,
161 int *fds, size_t num_fds,
162 void *private_data)
163 {
164 struct messaging_context *msg_ctx = talloc_get_type_abort(
165 private_data, struct messaging_context);
166 struct server_id_buf idbuf;
167 struct messaging_rec rec;
168 int64_t fds64[MIN(num_fds, INT8_MAX)];
169 size_t i;
170
171 if (msg_len < MESSAGE_HDR_LENGTH) {
172 DBG_WARNING("message too short: %zu\n", msg_len);
173 goto close_fail;
174 }
175
176 if (num_fds > INT8_MAX) {
177 DBG_WARNING("too many fds: %zu\n", num_fds);
178 goto close_fail;
179 }
180
181 /*
182 * "consume" the fds by copying them and setting
183 * the original variable to -1
184 */
185 for (i=0; i < num_fds; i++) {
186 fds64[i] = fds[i];
187 fds[i] = -1;
188 }
189
190 rec = (struct messaging_rec) {
191 .msg_version = MESSAGE_VERSION,
192 .buf.data = discard_const_p(uint8_t, msg) + MESSAGE_HDR_LENGTH,
193 .buf.length = msg_len - MESSAGE_HDR_LENGTH,
194 .num_fds = num_fds,
195 .fds = fds64,
196 };
197
198 message_hdr_get(&rec.msg_type, &rec.src, &rec.dest, msg);
199
200 DBG_DEBUG("Received message 0x%x len %zu (num_fds:%zu) from %s\n",
201 (unsigned)rec.msg_type, rec.buf.length, num_fds,
202 server_id_str_buf(rec.src, &idbuf));
203
204 messaging_dispatch_rec(msg_ctx, ev, &rec);
205 return;
206
207 close_fail:
208 for (i=0; i < num_fds; i++) {
209 close(fds[i]);
210 }
211 }
212
213 static int messaging_context_destructor(struct messaging_context *ctx)
214 {
215 unsigned i;
216
217 for (i=0; i<ctx->num_new_waiters; i++) {
218 if (ctx->new_waiters[i] != NULL) {
219 tevent_req_set_cleanup_fn(ctx->new_waiters[i], NULL);
220 ctx->new_waiters[i] = NULL;
221 }
222 }
223 for (i=0; i<ctx->num_waiters; i++) {
224 if (ctx->waiters[i] != NULL) {
225 tevent_req_set_cleanup_fn(ctx->waiters[i], NULL);
226 ctx->waiters[i] = NULL;
227 }
228 }
229
230 return 0;
231 }
232
233 static const char *private_path(const char *name)
234 {
235 return talloc_asprintf(talloc_tos(), "%s/%s", lp_private_dir(), name);
236 }
237
238 static NTSTATUS messaging_init_internal(TALLOC_CTX *mem_ctx,
239 struct tevent_context *ev,
240 struct messaging_context **pmsg_ctx)
241 {
242 TALLOC_CTX *frame;
243 struct messaging_context *ctx;
244 NTSTATUS status = NT_STATUS_UNSUCCESSFUL;
245 int ret;
246 const char *lck_path;
247 const char *priv_path;
248 bool ok;
249
250 lck_path = lock_path("msg.lock");
251 if (lck_path == NULL) {
252 return NT_STATUS_NO_MEMORY;
253 }
254
255 ok = directory_create_or_exist_strict(lck_path,
256 sec_initial_uid(),
257 0755);
258 if (!ok) {
259 DBG_DEBUG("Could not create lock directory: %s\n",
260 strerror(errno));
261 return NT_STATUS_ACCESS_DENIED;
262 }
263
264 priv_path = private_path("msg.sock");
265 if (priv_path == NULL) {
266 return NT_STATUS_NO_MEMORY;
267 }
268
269 ok = directory_create_or_exist_strict(priv_path, sec_initial_uid(),
270 0700);
271 if (!ok) {
272 DBG_DEBUG("Could not create msg directory: %s\n",
273 strerror(errno));
274 return NT_STATUS_ACCESS_DENIED;
275 }
276
277 frame = talloc_stackframe();
278 if (frame == NULL) {
279 return NT_STATUS_NO_MEMORY;
280 }
281
282 ctx = talloc_zero(frame, struct messaging_context);
283 if (ctx == NULL) {
284 status = NT_STATUS_NO_MEMORY;
285 goto done;
286 }
287
288 ctx->id = (struct server_id) {
289 .pid = getpid(), .vnn = NONCLUSTER_VNN
290 };
291
292 ctx->event_ctx = ev;
293
294 sec_init();
295
296 ctx->msg_dgm_ref = messaging_dgm_ref(ctx,
297 ctx->event_ctx,
298 &ctx->id.unique_id,
299 priv_path,
300 lck_path,
301 messaging_recv_cb,
302 ctx,
303 &ret);
304 if (ctx->msg_dgm_ref == NULL) {
305 DEBUG(2, ("messaging_dgm_ref failed: %s\n", strerror(ret)));
306 status = NT_STATUS_INTERNAL_ERROR;
307 goto done;
308 }
309 talloc_set_destructor(ctx, messaging_context_destructor);
310
311 if (lp_clustering()) {
312 ret = messaging_ctdbd_init(ctx, ctx, &ctx->remote);
313
314 if (ret != 0) {
315 DEBUG(2, ("messaging_ctdbd_init failed: %s\n",
316 strerror(ret)));
317 status = NT_STATUS_INTERNAL_ERROR;
318 goto done;
319 }
320 }
321 ctx->id.vnn = get_my_vnn();
322
323 ctx->names_db = server_id_db_init(ctx,
324 ctx->id,
325 lp_lock_directory(),
326 0,
327 TDB_INCOMPATIBLE_HASH|TDB_CLEAR_IF_FIRST);
328 if (ctx->names_db == NULL) {
329 DBG_DEBUG("server_id_db_init failed\n");
330 status = NT_STATUS_INTERNAL_ERROR;
331 goto done;
332 }
333
334 messaging_register(ctx, NULL, MSG_PING, ping_message);
335
336 /* Register some debugging related messages */
337
338 register_msg_pool_usage(ctx);
339 register_dmalloc_msgs(ctx);
340 debug_register_msgs(ctx);
341
342 {
343 struct server_id_buf tmp;
344 DBG_DEBUG("my id: %s\n", server_id_str_buf(ctx->id, &tmp));
345 }
346
347 *pmsg_ctx = talloc_steal(mem_ctx, ctx);
348
349 status = NT_STATUS_OK;
350 done:
351 TALLOC_FREE(frame);
352
353 return status;
354 }
355
356 struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx,
357 struct tevent_context *ev)
358 {
359 struct messaging_context *ctx = NULL;
360 NTSTATUS status;
361
362 status = messaging_init_internal(mem_ctx,
363 ev,
364 &ctx);
365 if (!NT_STATUS_IS_OK(status)) {
366 return NULL;
367 }
368
369 return ctx;
370 }
371
372 NTSTATUS messaging_init_client(TALLOC_CTX *mem_ctx,
373 struct tevent_context *ev,
374 struct messaging_context **pmsg_ctx)
375 {
376 return messaging_init_internal(mem_ctx,
377 ev,
378 pmsg_ctx);
379 }
380
381 struct server_id messaging_server_id(const struct messaging_context *msg_ctx)
382 {
383 return msg_ctx->id;
384 }
385
386 /*
387 * re-init after a fork
388 */
389 NTSTATUS messaging_reinit(struct messaging_context *msg_ctx)
390 {
391 int ret;
392 char *lck_path;
393
394 TALLOC_FREE(msg_ctx->msg_dgm_ref);
395
396 msg_ctx->id = (struct server_id) {
397 .pid = getpid(), .vnn = msg_ctx->id.vnn
398 };
399
400 lck_path = lock_path("msg.lock");
401 if (lck_path == NULL) {
402 return NT_STATUS_NO_MEMORY;
403 }
404
405 msg_ctx->msg_dgm_ref = messaging_dgm_ref(
406 msg_ctx, msg_ctx->event_ctx, &msg_ctx->id.unique_id,
407 private_path("msg.sock"), lck_path,
408 messaging_recv_cb, msg_ctx, &ret);
409
410 if (msg_ctx->msg_dgm_ref == NULL) {
411 DEBUG(2, ("messaging_dgm_ref failed: %s\n", strerror(ret)));
412 return map_nt_error_from_unix(ret);
413 }
414
415 if (lp_clustering()) {
416 ret = messaging_ctdbd_reinit(msg_ctx, msg_ctx,
417 msg_ctx->remote);
418
419 if (ret != 0) {
420 DEBUG(1, ("messaging_ctdbd_init failed: %s\n",
421 strerror(ret)));
422 return map_nt_error_from_unix(ret);
423 }
424 }
425
426 server_id_db_reinit(msg_ctx->names_db, msg_ctx->id);
427
428 return NT_STATUS_OK;
429 }
430
431
432 /*
433 * Register a dispatch function for a particular message type. Allow multiple
434 * registrants
435 */
436 NTSTATUS messaging_register(struct messaging_context *msg_ctx,
437 void *private_data,
438 uint32_t msg_type,
439 void (*fn)(struct messaging_context *msg,
440 void *private_data,
441 uint32_t msg_type,
442 struct server_id server_id,
443 DATA_BLOB *data))
444 {
445 struct messaging_callback *cb;
446
447 DEBUG(5, ("Registering messaging pointer for type %u - "
448 "private_data=%p\n",
449 (unsigned)msg_type, private_data));
450
451 /*
452 * Only one callback per type
453 */
454
455 for (cb = msg_ctx->callbacks; cb != NULL; cb = cb->next) {
456 /* we allow a second registration of the same message
457 type if it has a different private pointer. This is
458 needed in, for example, the internal notify code,
459 which creates a new notify context for each tree
460 connect, and expects to receive messages to each of
461 them. */
462 if (cb->msg_type == msg_type && private_data == cb->private_data) {
463 DEBUG(5,("Overriding messaging pointer for type %u - private_data=%p\n",
464 (unsigned)msg_type, private_data));
465 cb->fn = fn;
466 cb->private_data = private_data;
467 return NT_STATUS_OK;
468 }
469 }
470
471 if (!(cb = talloc(msg_ctx, struct messaging_callback))) {
472 return NT_STATUS_NO_MEMORY;
473 }
474
475 cb->msg_type = msg_type;
476 cb->fn = fn;
477 cb->private_data = private_data;
478
479 DLIST_ADD(msg_ctx->callbacks, cb);
480 return NT_STATUS_OK;
481 }
482
483 /*
484 De-register the function for a particular message type.
485 */
486 void messaging_deregister(struct messaging_context *ctx, uint32_t msg_type,
487 void *private_data)
488 {
489 struct messaging_callback *cb, *next;
490
491 for (cb = ctx->callbacks; cb; cb = next) {
492 next = cb->next;
493 if ((cb->msg_type == msg_type)
494 && (cb->private_data == private_data)) {
495 DEBUG(5,("Deregistering messaging pointer for type %u - private_data=%p\n",
496 (unsigned)msg_type, private_data));
497 DLIST_REMOVE(ctx->callbacks, cb);
498 TALLOC_FREE(cb);
499 }
500 }
501 }
502
503 /*
504 Send a message to a particular server
505 */
506 NTSTATUS messaging_send(struct messaging_context *msg_ctx,
507 struct server_id server, uint32_t msg_type,
508 const DATA_BLOB *data)
509 {
510 struct iovec iov = {0};
511
512 if (data != NULL) {
513 iov.iov_base = data->data;
514 iov.iov_len = data->length;
515 };
516
517 return messaging_send_iov(msg_ctx, server, msg_type, &iov, 1, NULL, 0);
518 }
519
520 NTSTATUS messaging_send_buf(struct messaging_context *msg_ctx,
521 struct server_id server, uint32_t msg_type,
522 const uint8_t *buf, size_t len)
523 {
524 DATA_BLOB blob = data_blob_const(buf, len);
525 return messaging_send(msg_ctx, server, msg_type, &blob);
526 }
527
528 struct messaging_post_state {
529 struct messaging_context *msg_ctx;
530 struct messaging_rec *rec;
531 };
532
533 static void messaging_post_handler(struct tevent_context *ev,
534 struct tevent_immediate *ti,
535 void *private_data);
536
537 static int messaging_post_self(struct messaging_context *msg_ctx,
538 struct server_id src, struct server_id dst,
539 uint32_t msg_type,
540 const struct iovec *iov, int iovlen,
541 const int *fds, size_t num_fds)
542 {
543 struct tevent_immediate *ti;
544 struct messaging_post_state *state;
545
546 state = talloc(msg_ctx, struct messaging_post_state);
547 if (state == NULL) {
548 return ENOMEM;
549 }
550 state->msg_ctx = msg_ctx;
551
552 ti = tevent_create_immediate(state);
553 if (ti == NULL) {
554 goto fail;
555 }
556 state->rec = messaging_rec_create(
557 state, src, dst, msg_type, iov, iovlen, fds, num_fds);
558 if (state->rec == NULL) {
559 goto fail;
560 }
561
562 tevent_schedule_immediate(ti, msg_ctx->event_ctx,
563 messaging_post_handler, state);
564 return 0;
565
566 fail:
567 TALLOC_FREE(state);
568 return ENOMEM;
569 }
570
571 static void messaging_post_handler(struct tevent_context *ev,
572 struct tevent_immediate *ti,
573 void *private_data)
574 {
575 struct messaging_post_state *state = talloc_get_type_abort(
576 private_data, struct messaging_post_state);
577 messaging_dispatch_rec(state->msg_ctx, ev, state->rec);
578 TALLOC_FREE(state);
579 }
580
581 int messaging_send_iov_from(struct messaging_context *msg_ctx,
582 struct server_id src, struct server_id dst,
583 uint32_t msg_type,
584 const struct iovec *iov, int iovlen,
585 const int *fds, size_t num_fds)
586 {
587 int ret;
588 uint8_t hdr[MESSAGE_HDR_LENGTH];
589 struct iovec iov2[iovlen+1];
590
591 if (server_id_is_disconnected(&dst)) {
592 return EINVAL;
593 }
594
595 if (num_fds > INT8_MAX) {
596 return EINVAL;
597 }
598
599 if (dst.vnn != msg_ctx->id.vnn) {
600 if (num_fds > 0) {
601 return ENOSYS;
602 }
603
604 ret = msg_ctx->remote->send_fn(src, dst,
605 msg_type, iov, iovlen,
606 NULL, 0,
607 msg_ctx->remote);
608 return ret;
609 }
610
611 if (server_id_equal(&dst, &msg_ctx->id)) {
612 ret = messaging_post_self(msg_ctx, src, dst, msg_type,
613 iov, iovlen, fds, num_fds);
614 return ret;
615 }
616
617 message_hdr_put(hdr, msg_type, src, dst);
618 iov2[0] = (struct iovec){ .iov_base = hdr, .iov_len = sizeof(hdr) };
619 memcpy(&iov2[1], iov, iovlen * sizeof(*iov));
620
621 ret = messaging_dgm_send(dst.pid, iov2, iovlen+1, fds, num_fds);
622
623 if (ret == EACCES) {
624 become_root();
625 ret = messaging_dgm_send(dst.pid, iov2, iovlen+1,
626 fds, num_fds);
627 unbecome_root();
628 }
629
630 if (ret == ECONNREFUSED) {
631 /*
632 * Linux returns this when a socket exists in the file
633 * system without a listening process. This is not
634 * documented in susv4 or the linux manpages, but it's
635 * easily testable. For the higher levels this is the
636 * same as "destination does not exist"
637 */
638 ret = ENOENT;
639 }
640
641 return ret;
642 }
643
644 NTSTATUS messaging_send_iov(struct messaging_context *msg_ctx,
645 struct server_id dst, uint32_t msg_type,
646 const struct iovec *iov, int iovlen,
647 const int *fds, size_t num_fds)
648 {
649 int ret;
650
651 ret = messaging_send_iov_from(msg_ctx, msg_ctx->id, dst, msg_type,
652 iov, iovlen, fds, num_fds);
653 if (ret != 0) {
654 return map_nt_error_from_unix(ret);
655 }
656 return NT_STATUS_OK;
657 }
658
659 static struct messaging_rec *messaging_rec_dup(TALLOC_CTX *mem_ctx,
660 struct messaging_rec *rec)
661 {
662 struct messaging_rec *result;
663 size_t fds_size = sizeof(int64_t) * rec->num_fds;
664 size_t payload_len;
665
666 payload_len = rec->buf.length + fds_size;
667 if (payload_len < rec->buf.length) {
668 /* overflow */
669 return NULL;
670 }
671
672 result = talloc_pooled_object(mem_ctx, struct messaging_rec, 2,
673 payload_len);
674 if (result == NULL) {
675 return NULL;
676 }
677 *result = *rec;
678
679 /* Doesn't fail, see talloc_pooled_object */
680
681 result->buf.data = talloc_memdup(result, rec->buf.data,
682 rec->buf.length);
683
684 result->fds = NULL;
685 if (result->num_fds > 0) {
686 result->fds = talloc_memdup(result, rec->fds, fds_size);
687 }
688
689 return result;
690 }
691
692 struct messaging_filtered_read_state {
693 struct tevent_context *ev;
694 struct messaging_context *msg_ctx;
695 struct messaging_dgm_fde *fde;
696
697 bool (*filter)(struct messaging_rec *rec, void *private_data);
698 void *private_data;
699
700 struct messaging_rec *rec;
701 };
702
703 static void messaging_filtered_read_cleanup(struct tevent_req *req,
704 enum tevent_req_state req_state);
705
706 struct tevent_req *messaging_filtered_read_send(
707 TALLOC_CTX *mem_ctx, struct tevent_context *ev,
708 struct messaging_context *msg_ctx,
709 bool (*filter)(struct messaging_rec *rec, void *private_data),
710 void *private_data)
711 {
712 struct tevent_req *req;
713 struct messaging_filtered_read_state *state;
714 size_t new_waiters_len;
715
716 req = tevent_req_create(mem_ctx, &state,
717 struct messaging_filtered_read_state);
718 if (req == NULL) {
719 return NULL;
720 }
721 state->ev = ev;
722 state->msg_ctx = msg_ctx;
723 state->filter = filter;
724 state->private_data = private_data;
725
726 /*
727 * We have to defer the callback here, as we might be called from
728 * within a different tevent_context than state->ev
729 */
730 tevent_req_defer_callback(req, state->ev);
731
732 state->fde = messaging_dgm_register_tevent_context(state, ev);
733 if (tevent_req_nomem(state->fde, req)) {
734 return tevent_req_post(req, ev);
735 }
736
737 /*
738 * We add ourselves to the "new_waiters" array, not the "waiters"
739 * array. If we are called from within messaging_read_done,
740 * messaging_dispatch_rec will be in an active for-loop on
741 * "waiters". We must be careful not to mess with this array, because
742 * it could mean that a single event is being delivered twice.
743 */
744
745 new_waiters_len = talloc_array_length(msg_ctx->new_waiters);
746
747 if (new_waiters_len == msg_ctx->num_new_waiters) {
748 struct tevent_req **tmp;
749
750 tmp = talloc_realloc(msg_ctx, msg_ctx->new_waiters,
751 struct tevent_req *, new_waiters_len+1);
752 if (tevent_req_nomem(tmp, req)) {
753 return tevent_req_post(req, ev);
754 }
755 msg_ctx->new_waiters = tmp;
756 }
757
758 msg_ctx->new_waiters[msg_ctx->num_new_waiters] = req;
759 msg_ctx->num_new_waiters += 1;
760 tevent_req_set_cleanup_fn(req, messaging_filtered_read_cleanup);
761
762 return req;
763 }
764
765 static void messaging_filtered_read_cleanup(struct tevent_req *req,
766 enum tevent_req_state req_state)
767 {
768 struct messaging_filtered_read_state *state = tevent_req_data(
769 req, struct messaging_filtered_read_state);
770 struct messaging_context *msg_ctx = state->msg_ctx;
771 unsigned i;
772
773 tevent_req_set_cleanup_fn(req, NULL);
774
775 TALLOC_FREE(state->fde);
776
777 /*
778 * Just set the [new_]waiters entry to NULL, be careful not to mess
779 * with the other "waiters" array contents. We are often called from
780 * within "messaging_dispatch_rec", which loops over
781 * "waiters". Messing with the "waiters" array will mess up that
782 * for-loop.
783 */
784
785 for (i=0; i<msg_ctx->num_waiters; i++) {
786 if (msg_ctx->waiters[i] == req) {
787 msg_ctx->waiters[i] = NULL;
788 return;
789 }
790 }
791
792 for (i=0; i<msg_ctx->num_new_waiters; i++) {
793 if (msg_ctx->new_waiters[i] == req) {
794 msg_ctx->new_waiters[i] = NULL;
795 return;
796 }
797 }
798 }
799
800 static void messaging_filtered_read_done(struct tevent_req *req,
801 struct messaging_rec *rec)
802 {
803 struct messaging_filtered_read_state *state = tevent_req_data(
804 req, struct messaging_filtered_read_state);
805
806 state->rec = messaging_rec_dup(state, rec);
807 if (tevent_req_nomem(state->rec, req)) {
808 return;
809 }
810 tevent_req_done(req);
811 }
812
813 int messaging_filtered_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
814 struct messaging_rec **presult)
815 {
816 struct messaging_filtered_read_state *state = tevent_req_data(
817 req, struct messaging_filtered_read_state);
818 int err;
819
820 if (tevent_req_is_unix_error(req, &err)) {
821 tevent_req_received(req);
822 return err;
823 }
824 if (presult != NULL) {
825 *presult = talloc_move(mem_ctx, &state->rec);
826 }
827 return 0;
828 }
829
830 struct messaging_read_state {
831 uint32_t msg_type;
832 struct messaging_rec *rec;
833 };
834
835 static bool messaging_read_filter(struct messaging_rec *rec,
836 void *private_data);
837 static void messaging_read_done(struct tevent_req *subreq);
838
839 struct tevent_req *messaging_read_send(TALLOC_CTX *mem_ctx,
840 struct tevent_context *ev,
841 struct messaging_context *msg,
842 uint32_t msg_type)
843 {
844 struct tevent_req *req, *subreq;
845 struct messaging_read_state *state;
846
847 req = tevent_req_create(mem_ctx, &state,
848 struct messaging_read_state);
849 if (req == NULL) {
850 return NULL;
851 }
852 state->msg_type = msg_type;
853
854 subreq = messaging_filtered_read_send(state, ev, msg,
855 messaging_read_filter, state);
856 if (tevent_req_nomem(subreq, req)) {
857 return tevent_req_post(req, ev);
858 }
859 tevent_req_set_callback(subreq, messaging_read_done, req);
860 return req;
861 }
862
863 static bool messaging_read_filter(struct messaging_rec *rec,
864 void *private_data)
865 {
866 struct messaging_read_state *state = talloc_get_type_abort(
867 private_data, struct messaging_read_state);
868
869 if (rec->num_fds != 0) {
870 return false;
871 }
872
873 return rec->msg_type == state->msg_type;
874 }
875
876 static void messaging_read_done(struct tevent_req *subreq)
877 {
878 struct tevent_req *req = tevent_req_callback_data(
879 subreq, struct tevent_req);
880 struct messaging_read_state *state = tevent_req_data(
881 req, struct messaging_read_state);
882 int ret;
883
884 ret = messaging_filtered_read_recv(subreq, state, &state->rec);
885 TALLOC_FREE(subreq);
886 if (tevent_req_error(req, ret)) {
887 return;
888 }
889 tevent_req_done(req);
890 }
891
892 int messaging_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
893 struct messaging_rec **presult)
894 {
895 struct messaging_read_state *state = tevent_req_data(
896 req, struct messaging_read_state);
897 int err;
898
899 if (tevent_req_is_unix_error(req, &err)) {
900 return err;
901 }
902 if (presult != NULL) {
903 *presult = talloc_move(mem_ctx, &state->rec);
904 }
905 return 0;
906 }
907
908 struct messaging_handler_state {
909 struct tevent_context *ev;
910 struct messaging_context *msg_ctx;
911 uint32_t msg_type;
912 bool (*handler)(struct messaging_context *msg_ctx,
913 struct messaging_rec **rec, void *private_data);
914 void *private_data;
915 };
916
917 static void messaging_handler_got_msg(struct tevent_req *subreq);
918
919 struct tevent_req *messaging_handler_send(
920 TALLOC_CTX *mem_ctx, struct tevent_context *ev,
921 struct messaging_context *msg_ctx, uint32_t msg_type,
922 bool (*handler)(struct messaging_context *msg_ctx,
923 struct messaging_rec **rec, void *private_data),
924 void *private_data)
925 {
926 struct tevent_req *req, *subreq;
927 struct messaging_handler_state *state;
928
929 req = tevent_req_create(mem_ctx, &state,
930 struct messaging_handler_state);
931 if (req == NULL) {
932 return NULL;
933 }
934 state->ev = ev;
935 state->msg_ctx = msg_ctx;
936 state->msg_type = msg_type;
937 state->handler = handler;
938 state->private_data = private_data;
939
940 subreq = messaging_read_send(state, state->ev, state->msg_ctx,
941 state->msg_type);
942 if (tevent_req_nomem(subreq, req)) {
943 return tevent_req_post(req, ev);
944 }
945 tevent_req_set_callback(subreq, messaging_handler_got_msg, req);
946 return req;
947 }
948
949 static void messaging_handler_got_msg(struct tevent_req *subreq)
950 {
951 struct tevent_req *req = tevent_req_callback_data(
952 subreq, struct tevent_req);
953 struct messaging_handler_state *state = tevent_req_data(
954 req, struct messaging_handler_state);
955 struct messaging_rec *rec;
956 int ret;
957 bool ok;
958
959 ret = messaging_read_recv(subreq, state, &rec);
960 TALLOC_FREE(subreq);
961 if (tevent_req_error(req, ret)) {
962 return;
963 }
964
965 subreq = messaging_read_send(state, state->ev, state->msg_ctx,
966 state->msg_type);
967 if (tevent_req_nomem(subreq, req)) {
968 return;
969 }
970 tevent_req_set_callback(subreq, messaging_handler_got_msg, req);
971
972 ok = state->handler(state->msg_ctx, &rec, state->private_data);
973 TALLOC_FREE(rec);
974 if (ok) {
975 /*
976 * Next round
977 */
978 return;
979 }
980 TALLOC_FREE(subreq);
981 tevent_req_done(req);
982 }
983
984 int messaging_handler_recv(struct tevent_req *req)
985 {
986 return tevent_req_simple_recv_unix(req);
987 }
988
989 static bool messaging_append_new_waiters(struct messaging_context *msg_ctx)
990 {
991 if (msg_ctx->num_new_waiters == 0) {
992 return true;
993 }
994
995 if (talloc_array_length(msg_ctx->waiters) <
996 (msg_ctx->num_waiters + msg_ctx->num_new_waiters)) {
997 struct tevent_req **tmp;
998 tmp = talloc_realloc(
999 msg_ctx, msg_ctx->waiters, struct tevent_req *,
1000 msg_ctx->num_waiters + msg_ctx->num_new_waiters);
1001 if (tmp == NULL) {
1002 DEBUG(1, ("%s: talloc failed\n", __func__));
1003 return false;
1004 }
1005 msg_ctx->waiters = tmp;
1006 }
1007
1008 memcpy(&msg_ctx->waiters[msg_ctx->num_waiters], msg_ctx->new_waiters,
1009 sizeof(struct tevent_req *) * msg_ctx->num_new_waiters);
1010
1011 msg_ctx->num_waiters += msg_ctx->num_new_waiters;
1012 msg_ctx->num_new_waiters = 0;
1013
1014 return true;
1015 }
1016
1017 static void messaging_dispatch_classic(struct messaging_context *msg_ctx,
1018 struct messaging_rec *rec)
1019 {
1020 struct messaging_callback *cb, *next;
1021
1022 for (cb = msg_ctx->callbacks; cb != NULL; cb = next) {
1023 size_t j;
1024
1025 next = cb->next;
1026 if (cb->msg_type != rec->msg_type) {
1027 continue;
1028 }
1029
1030 /*
1031 * the old style callbacks don't support fd passing
1032 */
1033 for (j=0; j < rec->num_fds; j++) {
1034 int fd = rec->fds[j];
1035 close(fd);
1036 }
1037 rec->num_fds = 0;
1038 rec->fds = NULL;
1039
1040 cb->fn(msg_ctx, cb->private_data, rec->msg_type,
1041 rec->src, &rec->buf);
1042
1043 /*
1044 * we continue looking for matching messages after finding
1045 * one. This matters for subsystems like the internal notify
1046 * code which register more than one handler for the same
1047 * message type
1048 */
1049 }
1050 }
1051
1052 /*
1053 Dispatch one messaging_rec
1054 */
1055 static void messaging_dispatch_rec(struct messaging_context *msg_ctx,
1056 struct tevent_context *ev,
1057 struct messaging_rec *rec)
1058 {
1059 unsigned i;
1060 size_t j;
1061
1062 if (ev == msg_ctx->event_ctx) {
1063 messaging_dispatch_classic(msg_ctx, rec);
1064 }
1065
1066 if (!messaging_append_new_waiters(msg_ctx)) {
1067 for (j=0; j < rec->num_fds; j++) {
1068 int fd = rec->fds[j];
1069 close(fd);
1070 }
1071 rec->num_fds = 0;
1072 rec->fds = NULL;
1073 return;
1074 }
1075
1076 i = 0;
1077 while (i < msg_ctx->num_waiters) {
1078 struct tevent_req *req;
1079 struct messaging_filtered_read_state *state;
1080
1081 req = msg_ctx->waiters[i];
1082 if (req == NULL) {
1083 /*
1084 * This got cleaned up. In the meantime,
1085 * move everything down one. We need
1086 * to keep the order of waiters, as
1087 * other code may depend on this.
1088 */
1089 if (i < msg_ctx->num_waiters - 1) {
1090 memmove(&msg_ctx->waiters[i],
1091 &msg_ctx->waiters[i+1],
1092 sizeof(struct tevent_req *) *
1093 (msg_ctx->num_waiters - i - 1));
1094 }
1095 msg_ctx->num_waiters -= 1;
1096 continue;
1097 }
1098
1099 state = tevent_req_data(
1100 req, struct messaging_filtered_read_state);
1101 if ((ev == state->ev) &&
1102 state->filter(rec, state->private_data)) {
1103 messaging_filtered_read_done(req, rec);
1104
1105 /*
1106 * Only the first one gets the fd-array
1107 */
1108 rec->num_fds = 0;
1109 rec->fds = NULL;
1110 }
1111
1112 i += 1;
1113 }
1114
1115 if (ev != msg_ctx->event_ctx) {
1116 struct iovec iov;
1117 int fds[rec->num_fds];
1118 int ret;
1119
1120 /*
1121 * We've been listening on a nested event
1122 * context. Messages need to be handled in the main
1123 * event context, so post to ourselves
1124 */
1125
1126 iov.iov_base = rec->buf.data;
1127 iov.iov_len = rec->buf.length;
1128
1129 for (i=0; i<rec->num_fds; i++) {
1130 fds[i] = rec->fds[i];
1131 }
1132
1133 ret = messaging_post_self(
1134 msg_ctx, rec->src, rec->dest, rec->msg_type,
1135 &iov, 1, fds, rec->num_fds);
1136 if (ret == 0) {
1137 return;
1138 }
1139 }
1140
1141 /*
1142 * If the fd-array isn't used, just close it.
1143 */
1144 for (j=0; j < rec->num_fds; j++) {
1145 int fd = rec->fds[j];
1146 close(fd);
1147 }
1148 rec->num_fds = 0;
1149 rec->fds = NULL;
1150 }
1151
1152 static int mess_parent_dgm_cleanup(void *private_data);
1153 static void mess_parent_dgm_cleanup_done(struct tevent_req *req);
1154
1155 bool messaging_parent_dgm_cleanup_init(struct messaging_context *msg)
1156 {
1157 struct tevent_req *req;
1158
1159 req = background_job_send(
1160 msg, msg->event_ctx, msg, NULL, 0,
1161 lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
1162 60*15),
1163 mess_parent_dgm_cleanup, msg);
1164 if (req == NULL) {
1165 return false;
1166 }
1167 tevent_req_set_callback(req, mess_parent_dgm_cleanup_done, msg);
1168 return true;
1169 }
1170
1171 static int mess_parent_dgm_cleanup(void *private_data)
1172 {
1173 int ret;
1174
1175 ret = messaging_dgm_wipe();
1176 DEBUG(10, ("messaging_dgm_wipe returned %s\n",
1177 ret ? strerror(ret) : "ok"));
1178 return lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
1179 60*15);
1180 }
1181
1182 static void mess_parent_dgm_cleanup_done(struct tevent_req *req)
1183 {
1184 struct messaging_context *msg = tevent_req_callback_data(
1185 req, struct messaging_context);
1186 NTSTATUS status;
1187
1188 status = background_job_recv(req);
1189 TALLOC_FREE(req);
1190 DEBUG(1, ("messaging dgm cleanup job ended with %s\n",
1191 nt_errstr(status)));
1192
1193 req = background_job_send(
1194 msg, msg->event_ctx, msg, NULL, 0,
1195 lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
1196 60*15),
1197 mess_parent_dgm_cleanup, msg);
1198 if (req == NULL) {
1199 DEBUG(1, ("background_job_send failed\n"));
1200 return;
1201 }
1202 tevent_req_set_callback(req, mess_parent_dgm_cleanup_done, msg);
1203 }
1204
1205 int messaging_cleanup(struct messaging_context *msg_ctx, pid_t pid)
1206 {
1207 int ret;
1208
1209 if (pid == 0) {
1210 ret = messaging_dgm_wipe();
1211 } else {
1212 ret = messaging_dgm_cleanup(pid);
1213 }
1214
1215 return ret;
1216 }
1217
1218 struct tevent_context *messaging_tevent_context(
1219 struct messaging_context *msg_ctx)
1220 {
1221 return msg_ctx->event_ctx;
1222 }
1223
1224 struct server_id_db *messaging_names_db(struct messaging_context *msg_ctx)
1225 {
1226 return msg_ctx->names_db;
1227 }
1228
1229 /** @} **/
Samba GIT mirror