LCOV - code coverage report
Current view: top level - source3/lib - messages.c (source / functions) Hit Total Coverage
Test: coverage report for abartlet/fix-coverage dd10fb34 Lines: 405 515 78.6 %
Date: 2021-09-23 10:06:22 Functions: 39 43 90.7 %

          Line data    Source code
       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/messaging/messages_dgm.h"
      56             : #include "lib/util/iov_buf.h"
      57             : #include "lib/util/server_id_db.h"
      58             : #include "lib/messaging/messages_dgm_ref.h"
      59             : #include "lib/messages_ctdb.h"
      60             : #include "lib/messages_ctdb_ref.h"
      61             : #include "lib/messages_util.h"
      62             : #include "cluster_support.h"
      63             : #include "ctdbd_conn.h"
      64             : #include "ctdb_srvids.h"
      65             : 
      66             : #ifdef CLUSTER_SUPPORT
      67             : #include "ctdb_protocol.h"
      68             : #endif
      69             : 
      70             : struct messaging_callback {
      71             :         struct messaging_callback *prev, *next;
      72             :         uint32_t msg_type;
      73             :         void (*fn)(struct messaging_context *msg, void *private_data, 
      74             :                    uint32_t msg_type, 
      75             :                    struct server_id server_id, DATA_BLOB *data);
      76             :         void *private_data;
      77             : };
      78             : 
      79             : struct messaging_registered_ev {
      80             :         struct tevent_context *ev;
      81             :         struct tevent_immediate *im;
      82             :         size_t refcount;
      83             : };
      84             : 
      85             : struct messaging_context {
      86             :         struct server_id id;
      87             :         struct tevent_context *event_ctx;
      88             :         struct messaging_callback *callbacks;
      89             : 
      90             :         struct messaging_rec *posted_msgs;
      91             : 
      92             :         struct messaging_registered_ev *event_contexts;
      93             : 
      94             :         struct tevent_req **new_waiters;
      95             :         size_t num_new_waiters;
      96             : 
      97             :         struct tevent_req **waiters;
      98             :         size_t num_waiters;
      99             : 
     100             :         struct server_id_db *names_db;
     101             : 
     102             :         TALLOC_CTX *per_process_talloc_ctx;
     103             : };
     104             : 
     105             : static struct messaging_rec *messaging_rec_dup(TALLOC_CTX *mem_ctx,
     106             :                                                struct messaging_rec *rec);
     107             : static bool messaging_dispatch_classic(struct messaging_context *msg_ctx,
     108             :                                        struct messaging_rec *rec);
     109             : static bool messaging_dispatch_waiters(struct messaging_context *msg_ctx,
     110             :                                        struct tevent_context *ev,
     111             :                                        struct messaging_rec *rec);
     112             : static void messaging_dispatch_rec(struct messaging_context *msg_ctx,
     113             :                                    struct tevent_context *ev,
     114             :                                    struct messaging_rec *rec);
     115             : 
     116             : /****************************************************************************
     117             :  A useful function for testing the message system.
     118             : ****************************************************************************/
     119             : 
     120         109 : static void ping_message(struct messaging_context *msg_ctx,
     121             :                          void *private_data,
     122             :                          uint32_t msg_type,
     123             :                          struct server_id src,
     124             :                          DATA_BLOB *data)
     125             : {
     126             :         struct server_id_buf idbuf;
     127             : 
     128         109 :         DEBUG(1, ("INFO: Received PING message from PID %s [%.*s]\n",
     129             :                   server_id_str_buf(src, &idbuf), (int)data->length,
     130             :                   data->data ? (char *)data->data : ""));
     131             : 
     132         109 :         messaging_send(msg_ctx, src, MSG_PONG, data);
     133         109 : }
     134             : 
     135      186302 : struct messaging_rec *messaging_rec_create(
     136             :         TALLOC_CTX *mem_ctx, struct server_id src, struct server_id dst,
     137             :         uint32_t msg_type, const struct iovec *iov, int iovlen,
     138             :         const int *fds, size_t num_fds)
     139             : {
     140             :         ssize_t buflen;
     141             :         uint8_t *buf;
     142             :         struct messaging_rec *result;
     143             : 
     144      186302 :         if (num_fds > INT8_MAX) {
     145           0 :                 return NULL;
     146             :         }
     147             : 
     148      186302 :         buflen = iov_buflen(iov, iovlen);
     149      186302 :         if (buflen == -1) {
     150           0 :                 return NULL;
     151             :         }
     152      186302 :         buf = talloc_array(mem_ctx, uint8_t, buflen);
     153      186302 :         if (buf == NULL) {
     154           0 :                 return NULL;
     155             :         }
     156      186302 :         iov_buf(iov, iovlen, buf, buflen);
     157             : 
     158      186302 :         {
     159             :                 struct messaging_rec rec;
     160      186302 :                 int64_t fds64[MAX(1, num_fds)];
     161             :                 size_t i;
     162             : 
     163      186303 :                 for (i=0; i<num_fds; i++) {
     164           1 :                         fds64[i] = fds[i];
     165             :                 }
     166             : 
     167      344760 :                 rec = (struct messaging_rec) {
     168             :                         .msg_version = MESSAGE_VERSION, .msg_type = msg_type,
     169             :                         .src = src, .dest = dst,
     170             :                         .buf.data = buf, .buf.length = buflen,
     171      158458 :                         .num_fds = num_fds, .fds = fds64,
     172             :                 };
     173             : 
     174      186302 :                 result = messaging_rec_dup(mem_ctx, &rec);
     175             :         }
     176             : 
     177      186302 :         TALLOC_FREE(buf);
     178             : 
     179      186302 :         return result;
     180             : }
     181             : 
     182       96264 : static bool messaging_register_event_context(struct messaging_context *ctx,
     183             :                                              struct tevent_context *ev)
     184             : {
     185             :         size_t i, num_event_contexts;
     186       96264 :         struct messaging_registered_ev *free_reg = NULL;
     187             :         struct messaging_registered_ev *tmp;
     188             : 
     189       96264 :         num_event_contexts = talloc_array_length(ctx->event_contexts);
     190             : 
     191       96534 :         for (i=0; i<num_event_contexts; i++) {
     192       91979 :                 struct messaging_registered_ev *reg = &ctx->event_contexts[i];
     193             : 
     194       91979 :                 if (reg->refcount == 0) {
     195         114 :                         if (reg->ev != NULL) {
     196           0 :                                 abort();
     197             :                         }
     198         114 :                         free_reg = reg;
     199             :                         /*
     200             :                          * We continue here and may find another
     201             :                          * free_req, but the important thing is
     202             :                          * that we continue to search for an
     203             :                          * existing registration in the loop.
     204             :                          */
     205         114 :                         continue;
     206             :                 }
     207             : 
     208       91865 :                 if (reg->ev == ev) {
     209       91709 :                         reg->refcount += 1;
     210       91709 :                         return true;
     211             :                 }
     212             :         }
     213             : 
     214        4555 :         if (free_reg == NULL) {
     215        4441 :                 struct tevent_immediate *im = NULL;
     216             : 
     217        4441 :                 im = tevent_create_immediate(ctx);
     218        4441 :                 if (im == NULL) {
     219           0 :                         return false;
     220             :                 }
     221             : 
     222        4441 :                 tmp = talloc_realloc(ctx, ctx->event_contexts,
     223             :                                      struct messaging_registered_ev,
     224             :                                      num_event_contexts+1);
     225        4441 :                 if (tmp == NULL) {
     226           0 :                         return false;
     227             :                 }
     228        4441 :                 ctx->event_contexts = tmp;
     229             : 
     230        4441 :                 free_reg = &ctx->event_contexts[num_event_contexts];
     231        4441 :                 free_reg->im = talloc_move(ctx->event_contexts, &im);
     232             :         }
     233             : 
     234             :         /*
     235             :          * free_reg->im might be cached
     236             :          */
     237        4555 :         free_reg->ev = ev;
     238        4555 :         free_reg->refcount = 1;
     239             : 
     240        4555 :         return true;
     241             : }
     242             : 
     243       87388 : static bool messaging_deregister_event_context(struct messaging_context *ctx,
     244             :                                                struct tevent_context *ev)
     245             : {
     246             :         size_t i, num_event_contexts;
     247             : 
     248       87388 :         num_event_contexts = talloc_array_length(ctx->event_contexts);
     249             : 
     250       87544 :         for (i=0; i<num_event_contexts; i++) {
     251       87544 :                 struct messaging_registered_ev *reg = &ctx->event_contexts[i];
     252             : 
     253       87544 :                 if (reg->refcount == 0) {
     254           0 :                         continue;
     255             :                 }
     256             : 
     257       87544 :                 if (reg->ev == ev) {
     258       87388 :                         reg->refcount -= 1;
     259             : 
     260       87388 :                         if (reg->refcount == 0) {
     261             :                                 /*
     262             :                                  * The primary event context
     263             :                                  * is never unregistered using
     264             :                                  * messaging_deregister_event_context()
     265             :                                  * it's only registered using
     266             :                                  * messaging_register_event_context().
     267             :                                  */
     268         156 :                                 SMB_ASSERT(ev != ctx->event_ctx);
     269         150 :                                 SMB_ASSERT(reg->ev != ctx->event_ctx);
     270             : 
     271             :                                 /*
     272             :                                  * Not strictly necessary, just
     273             :                                  * paranoia
     274             :                                  */
     275         156 :                                 reg->ev = NULL;
     276             : 
     277             :                                 /*
     278             :                                  * Do not talloc_free(reg->im),
     279             :                                  * recycle immediates events.
     280             :                                  *
     281             :                                  * We just invalidate it using
     282             :                                  * the primary event context,
     283             :                                  * which is never unregistered.
     284             :                                  */
     285         156 :                                 tevent_schedule_immediate(reg->im,
     286             :                                                           ctx->event_ctx,
     287             :                                                           NULL, NULL);
     288             :                         }
     289       85660 :                         return true;
     290             :                 }
     291             :         }
     292           0 :         return false;
     293             : }
     294             : 
     295      173808 : static void messaging_post_main_event_context(struct tevent_context *ev,
     296             :                                               struct tevent_immediate *im,
     297             :                                               void *private_data)
     298             : {
     299      173808 :         struct messaging_context *ctx = talloc_get_type_abort(
     300             :                 private_data, struct messaging_context);
     301             : 
     302      508808 :         while (ctx->posted_msgs != NULL) {
     303      186248 :                 struct messaging_rec *rec = ctx->posted_msgs;
     304             :                 bool consumed;
     305             : 
     306      186248 :                 DLIST_REMOVE(ctx->posted_msgs, rec);
     307             : 
     308      186248 :                 consumed = messaging_dispatch_classic(ctx, rec);
     309      186240 :                 if (!consumed) {
     310       41972 :                         consumed = messaging_dispatch_waiters(
     311             :                                 ctx, ctx->event_ctx, rec);
     312             :                 }
     313             : 
     314      186240 :                 if (!consumed) {
     315             :                         uint8_t i;
     316             : 
     317       40199 :                         for (i=0; i<rec->num_fds; i++) {
     318           0 :                                 close(rec->fds[i]);
     319             :                         }
     320             :                 }
     321             : 
     322      186240 :                 TALLOC_FREE(rec);
     323             :         }
     324      173800 : }
     325             : 
     326           0 : static void messaging_post_sub_event_context(struct tevent_context *ev,
     327             :                                              struct tevent_immediate *im,
     328             :                                              void *private_data)
     329             : {
     330           0 :         struct messaging_context *ctx = talloc_get_type_abort(
     331             :                 private_data, struct messaging_context);
     332             :         struct messaging_rec *rec, *next;
     333             : 
     334           0 :         for (rec = ctx->posted_msgs; rec != NULL; rec = next) {
     335             :                 bool consumed;
     336             : 
     337           0 :                 next = rec->next;
     338             : 
     339           0 :                 consumed = messaging_dispatch_waiters(ctx, ev, rec);
     340           0 :                 if (consumed) {
     341           0 :                         DLIST_REMOVE(ctx->posted_msgs, rec);
     342           0 :                         TALLOC_FREE(rec);
     343             :                 }
     344             :         }
     345           0 : }
     346             : 
     347      186302 : static bool messaging_alert_event_contexts(struct messaging_context *ctx)
     348             : {
     349             :         size_t i, num_event_contexts;
     350             : 
     351      186302 :         num_event_contexts = talloc_array_length(ctx->event_contexts);
     352             : 
     353      372629 :         for (i=0; i<num_event_contexts; i++) {
     354      186327 :                 struct messaging_registered_ev *reg = &ctx->event_contexts[i];
     355             : 
     356      186327 :                 if (reg->refcount == 0) {
     357          21 :                         continue;
     358             :                 }
     359             : 
     360             :                 /*
     361             :                  * We depend on schedule_immediate to work
     362             :                  * multiple times. Might be a bit inefficient,
     363             :                  * but this needs to be proven in tests. The
     364             :                  * alternatively would be to track whether the
     365             :                  * immediate has already been scheduled. For
     366             :                  * now, avoid that complexity here.
     367             :                  */
     368             : 
     369      186306 :                 if (reg->ev == ctx->event_ctx) {
     370      186302 :                         tevent_schedule_immediate(
     371             :                                 reg->im, reg->ev,
     372             :                                 messaging_post_main_event_context,
     373             :                                 ctx);
     374             :                 } else {
     375           4 :                         tevent_schedule_immediate(
     376             :                                 reg->im, reg->ev,
     377             :                                 messaging_post_sub_event_context,
     378             :                                 ctx);
     379             :                 }
     380             : 
     381             :         }
     382      186302 :         return true;
     383             : }
     384             : 
     385       85366 : static void messaging_recv_cb(struct tevent_context *ev,
     386             :                               const uint8_t *msg, size_t msg_len,
     387             :                               int *fds, size_t num_fds,
     388             :                               void *private_data)
     389       85366 : {
     390       85366 :         struct messaging_context *msg_ctx = talloc_get_type_abort(
     391             :                 private_data, struct messaging_context);
     392             :         struct server_id_buf idbuf;
     393             :         struct messaging_rec rec;
     394       85366 :         int64_t fds64[MAX(1, MIN(num_fds, INT8_MAX))];
     395             :         size_t i;
     396             : 
     397       85366 :         if (msg_len < MESSAGE_HDR_LENGTH) {
     398           0 :                 DBG_WARNING("message too short: %zu\n", msg_len);
     399           0 :                 return;
     400             :         }
     401             : 
     402       85366 :         if (num_fds > INT8_MAX) {
     403           0 :                 DBG_WARNING("too many fds: %zu\n", num_fds);
     404           0 :                 return;
     405             :         }
     406             : 
     407       85931 :         for (i=0; i < num_fds; i++) {
     408         936 :                 fds64[i] = fds[i];
     409             :         }
     410             : 
     411      166606 :         rec = (struct messaging_rec) {
     412             :                 .msg_version = MESSAGE_VERSION,
     413       85366 :                 .buf.data = discard_const_p(uint8_t, msg) + MESSAGE_HDR_LENGTH,
     414       85366 :                 .buf.length = msg_len - MESSAGE_HDR_LENGTH,
     415             :                 .num_fds = num_fds,
     416       81240 :                 .fds = fds64,
     417             :         };
     418             : 
     419       85366 :         message_hdr_get(&rec.msg_type, &rec.src, &rec.dest, msg);
     420             : 
     421       85366 :         DBG_DEBUG("Received message 0x%x len %zu (num_fds:%zu) from %s\n",
     422             :                   (unsigned)rec.msg_type, rec.buf.length, num_fds,
     423             :                   server_id_str_buf(rec.src, &idbuf));
     424             : 
     425       85366 :         if (server_id_same_process(&rec.src, &msg_ctx->id)) {
     426           0 :                 DBG_DEBUG("Ignoring self-send\n");
     427           0 :                 return;
     428             :         }
     429             : 
     430       85366 :         messaging_dispatch_rec(msg_ctx, ev, &rec);
     431             : 
     432       86287 :         for (i=0; i<num_fds; i++) {
     433         936 :                 fds[i] = fds64[i];
     434             :         }
     435             : }
     436             : 
     437       31437 : static int messaging_context_destructor(struct messaging_context *ctx)
     438             : {
     439             :         size_t i;
     440             : 
     441      128185 :         for (i=0; i<ctx->num_new_waiters; i++) {
     442       96748 :                 if (ctx->new_waiters[i] != NULL) {
     443       29215 :                         tevent_req_set_cleanup_fn(ctx->new_waiters[i], NULL);
     444       29215 :                         ctx->new_waiters[i] = NULL;
     445             :                 }
     446             :         }
     447       56407 :         for (i=0; i<ctx->num_waiters; i++) {
     448       25626 :                 if (ctx->waiters[i] != NULL) {
     449        2271 :                         tevent_req_set_cleanup_fn(ctx->waiters[i], NULL);
     450        2271 :                         ctx->waiters[i] = NULL;
     451             :                 }
     452             :         }
     453             : 
     454             :         /*
     455             :          * The immediates from messaging_alert_event_contexts
     456             :          * reference "ctx". Don't let them outlive the
     457             :          * messaging_context we're destroying here.
     458             :          */
     459       31437 :         TALLOC_FREE(ctx->event_contexts);
     460             : 
     461       31437 :         return 0;
     462             : }
     463             : 
     464       31917 : static const char *private_path(const char *name)
     465             : {
     466       31917 :         return talloc_asprintf(talloc_tos(), "%s/%s", lp_private_dir(), name);
     467             : }
     468             : 
     469        4399 : static NTSTATUS messaging_init_internal(TALLOC_CTX *mem_ctx,
     470             :                                         struct tevent_context *ev,
     471             :                                         struct messaging_context **pmsg_ctx)
     472             : {
     473             :         TALLOC_CTX *frame;
     474             :         struct messaging_context *ctx;
     475             :         NTSTATUS status;
     476             :         int ret;
     477             :         const char *lck_path;
     478             :         const char *priv_path;
     479             :         void *ref;
     480             :         bool ok;
     481             : 
     482             :         /*
     483             :          * sec_init() *must* be called before any other
     484             :          * functions that use sec_XXX(). e.g. sec_initial_uid().
     485             :          */
     486             : 
     487        4399 :         sec_init();
     488             : 
     489        4399 :         lck_path = lock_path(talloc_tos(), "msg.lock");
     490        4399 :         if (lck_path == NULL) {
     491           0 :                 return NT_STATUS_NO_MEMORY;
     492             :         }
     493             : 
     494        4399 :         ok = directory_create_or_exist_strict(lck_path,
     495             :                                               sec_initial_uid(),
     496             :                                               0755);
     497        4399 :         if (!ok) {
     498           0 :                 DBG_DEBUG("Could not create lock directory: %s\n",
     499             :                           strerror(errno));
     500           0 :                 return NT_STATUS_ACCESS_DENIED;
     501             :         }
     502             : 
     503        4399 :         priv_path = private_path("msg.sock");
     504        4399 :         if (priv_path == NULL) {
     505           0 :                 return NT_STATUS_NO_MEMORY;
     506             :         }
     507             : 
     508        4399 :         ok = directory_create_or_exist_strict(priv_path, sec_initial_uid(),
     509             :                                               0700);
     510        4399 :         if (!ok) {
     511           0 :                 DBG_DEBUG("Could not create msg directory: %s\n",
     512             :                           strerror(errno));
     513           0 :                 return NT_STATUS_ACCESS_DENIED;
     514             :         }
     515             : 
     516        4399 :         frame = talloc_stackframe();
     517        4399 :         if (frame == NULL) {
     518           0 :                 return NT_STATUS_NO_MEMORY;
     519             :         }
     520             : 
     521        4399 :         ctx = talloc_zero(frame, struct messaging_context);
     522        4399 :         if (ctx == NULL) {
     523           0 :                 status = NT_STATUS_NO_MEMORY;
     524           0 :                 goto done;
     525             :         }
     526             : 
     527        4399 :         ctx->id = (struct server_id) {
     528        4399 :                 .pid = getpid(), .vnn = NONCLUSTER_VNN
     529             :         };
     530             : 
     531        4399 :         ctx->event_ctx = ev;
     532             : 
     533        4399 :         ctx->per_process_talloc_ctx = talloc_new(ctx);
     534        4399 :         if (ctx->per_process_talloc_ctx == NULL) {
     535           0 :                 status = NT_STATUS_NO_MEMORY;
     536           0 :                 goto done;
     537             :         }
     538             : 
     539        4399 :         ok = messaging_register_event_context(ctx, ev);
     540        4399 :         if (!ok) {
     541           0 :                 status = NT_STATUS_NO_MEMORY;
     542           0 :                 goto done;
     543             :         }
     544             : 
     545        4399 :         ref = messaging_dgm_ref(
     546             :                 ctx->per_process_talloc_ctx,
     547             :                 ctx->event_ctx,
     548             :                 &ctx->id.unique_id,
     549             :                 priv_path,
     550             :                 lck_path,
     551             :                 messaging_recv_cb,
     552             :                 ctx,
     553             :                 &ret);
     554        4399 :         if (ref == NULL) {
     555           5 :                 DEBUG(2, ("messaging_dgm_ref failed: %s\n", strerror(ret)));
     556           5 :                 status = map_nt_error_from_unix(ret);
     557           5 :                 goto done;
     558             :         }
     559        4394 :         talloc_set_destructor(ctx, messaging_context_destructor);
     560             : 
     561             : #ifdef CLUSTER_SUPPORT
     562             :         if (lp_clustering()) {
     563             :                 ref = messaging_ctdb_ref(
     564             :                         ctx->per_process_talloc_ctx,
     565             :                         ctx->event_ctx,
     566             :                         lp_ctdbd_socket(),
     567             :                         lp_ctdb_timeout(),
     568             :                         ctx->id.unique_id,
     569             :                         messaging_recv_cb,
     570             :                         ctx,
     571             :                         &ret);
     572             :                 if (ref == NULL) {
     573             :                         DBG_NOTICE("messaging_ctdb_ref failed: %s\n",
     574             :                                    strerror(ret));
     575             :                         status = map_nt_error_from_unix(ret);
     576             :                         goto done;
     577             :                 }
     578             :         }
     579             : #endif
     580             : 
     581        4394 :         ctx->id.vnn = get_my_vnn();
     582             : 
     583        4394 :         ctx->names_db = server_id_db_init(ctx,
     584             :                                           ctx->id,
     585             :                                           lp_lock_directory(),
     586             :                                           0,
     587             :                                           TDB_INCOMPATIBLE_HASH|TDB_CLEAR_IF_FIRST);
     588        4394 :         if (ctx->names_db == NULL) {
     589           0 :                 DBG_DEBUG("server_id_db_init failed\n");
     590           0 :                 status = NT_STATUS_NO_MEMORY;
     591           0 :                 goto done;
     592             :         }
     593             : 
     594        4394 :         messaging_register(ctx, NULL, MSG_PING, ping_message);
     595             : 
     596             :         /* Register some debugging related messages */
     597             : 
     598        4394 :         register_msg_pool_usage(ctx->per_process_talloc_ctx, ctx);
     599        4394 :         register_dmalloc_msgs(ctx);
     600        4394 :         debug_register_msgs(ctx);
     601             : 
     602             :         {
     603             :                 struct server_id_buf tmp;
     604        4394 :                 DBG_DEBUG("my id: %s\n", server_id_str_buf(ctx->id, &tmp));
     605             :         }
     606             : 
     607        4394 :         *pmsg_ctx = talloc_steal(mem_ctx, ctx);
     608             : 
     609        4394 :         status = NT_STATUS_OK;
     610        4399 : done:
     611        4399 :         TALLOC_FREE(frame);
     612             : 
     613        4399 :         return status;
     614             : }
     615             : 
     616        4399 : struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx,
     617             :                                          struct tevent_context *ev)
     618             : {
     619        4399 :         struct messaging_context *ctx = NULL;
     620             :         NTSTATUS status;
     621             : 
     622        4399 :         status = messaging_init_internal(mem_ctx,
     623             :                                          ev,
     624             :                                          &ctx);
     625        4399 :         if (!NT_STATUS_IS_OK(status)) {
     626           0 :                 return NULL;
     627             :         }
     628             : 
     629        4394 :         return ctx;
     630             : }
     631             : 
     632     6316018 : struct server_id messaging_server_id(const struct messaging_context *msg_ctx)
     633             : {
     634     6316018 :         return msg_ctx->id;
     635             : }
     636             : 
     637             : /*
     638             :  * re-init after a fork
     639             :  */
     640       27518 : NTSTATUS messaging_reinit(struct messaging_context *msg_ctx)
     641             : {
     642             :         int ret;
     643             :         char *lck_path;
     644             :         void *ref;
     645             : 
     646       27518 :         TALLOC_FREE(msg_ctx->per_process_talloc_ctx);
     647             : 
     648       27518 :         msg_ctx->per_process_talloc_ctx = talloc_new(msg_ctx);
     649       27518 :         if (msg_ctx->per_process_talloc_ctx == NULL) {
     650           0 :                 return NT_STATUS_NO_MEMORY;
     651             :         }
     652             : 
     653       27518 :         msg_ctx->id = (struct server_id) {
     654       27518 :                 .pid = getpid(), .vnn = msg_ctx->id.vnn
     655             :         };
     656             : 
     657       27518 :         lck_path = lock_path(talloc_tos(), "msg.lock");
     658       27518 :         if (lck_path == NULL) {
     659           0 :                 return NT_STATUS_NO_MEMORY;
     660             :         }
     661             : 
     662       27518 :         ref = messaging_dgm_ref(
     663             :                 msg_ctx->per_process_talloc_ctx,
     664             :                 msg_ctx->event_ctx,
     665             :                 &msg_ctx->id.unique_id,
     666             :                 private_path("msg.sock"),
     667             :                 lck_path,
     668             :                 messaging_recv_cb,
     669             :                 msg_ctx,
     670             :                 &ret);
     671             : 
     672       27518 :         if (ref == NULL) {
     673           0 :                 DEBUG(2, ("messaging_dgm_ref failed: %s\n", strerror(ret)));
     674           0 :                 return map_nt_error_from_unix(ret);
     675             :         }
     676             : 
     677       27518 :         if (lp_clustering()) {
     678           0 :                 ref = messaging_ctdb_ref(
     679             :                         msg_ctx->per_process_talloc_ctx,
     680             :                         msg_ctx->event_ctx,
     681             :                         lp_ctdbd_socket(),
     682             :                         lp_ctdb_timeout(),
     683             :                         msg_ctx->id.unique_id,
     684             :                         messaging_recv_cb,
     685             :                         msg_ctx,
     686             :                         &ret);
     687           0 :                 if (ref == NULL) {
     688           0 :                         DBG_NOTICE("messaging_ctdb_ref failed: %s\n",
     689             :                                    strerror(ret));
     690           0 :                         return map_nt_error_from_unix(ret);
     691             :                 }
     692             :         }
     693             : 
     694       27518 :         server_id_db_reinit(msg_ctx->names_db, msg_ctx->id);
     695       27518 :         register_msg_pool_usage(msg_ctx->per_process_talloc_ctx, msg_ctx);
     696             : 
     697       27518 :         return NT_STATUS_OK;
     698             : }
     699             : 
     700             : 
     701             : /*
     702             :  * Register a dispatch function for a particular message type. Allow multiple
     703             :  * registrants
     704             : */
     705      380678 : NTSTATUS messaging_register(struct messaging_context *msg_ctx,
     706             :                             void *private_data,
     707             :                             uint32_t msg_type,
     708             :                             void (*fn)(struct messaging_context *msg,
     709             :                                        void *private_data, 
     710             :                                        uint32_t msg_type, 
     711             :                                        struct server_id server_id,
     712             :                                        DATA_BLOB *data))
     713             : {
     714             :         struct messaging_callback *cb;
     715             : 
     716      380678 :         DEBUG(5, ("Registering messaging pointer for type %u - "
     717             :                   "private_data=%p\n",
     718             :                   (unsigned)msg_type, private_data));
     719             : 
     720             :         /*
     721             :          * Only one callback per type
     722             :          */
     723             : 
     724     9124739 :         for (cb = msg_ctx->callbacks; cb != NULL; cb = cb->next) {
     725             :                 /* we allow a second registration of the same message
     726             :                    type if it has a different private pointer. This is
     727             :                    needed in, for example, the internal notify code,
     728             :                    which creates a new notify context for each tree
     729             :                    connect, and expects to receive messages to each of
     730             :                    them. */
     731     8771496 :                 if (cb->msg_type == msg_type && private_data == cb->private_data) {
     732       27435 :                         DEBUG(5,("Overriding messaging pointer for type %u - private_data=%p\n",
     733             :                                   (unsigned)msg_type, private_data));
     734       27435 :                         cb->fn = fn;
     735       27435 :                         cb->private_data = private_data;
     736       27435 :                         return NT_STATUS_OK;
     737             :                 }
     738             :         }
     739             : 
     740      353243 :         if (!(cb = talloc(msg_ctx, struct messaging_callback))) {
     741           0 :                 return NT_STATUS_NO_MEMORY;
     742             :         }
     743             : 
     744      353243 :         cb->msg_type = msg_type;
     745      353243 :         cb->fn = fn;
     746      353243 :         cb->private_data = private_data;
     747             : 
     748      353243 :         DLIST_ADD(msg_ctx->callbacks, cb);
     749      353243 :         return NT_STATUS_OK;
     750             : }
     751             : 
     752             : /*
     753             :   De-register the function for a particular message type.
     754             : */
     755      153907 : void messaging_deregister(struct messaging_context *ctx, uint32_t msg_type,
     756             :                           void *private_data)
     757             : {
     758             :         struct messaging_callback *cb, *next;
     759             : 
     760     4466982 :         for (cb = ctx->callbacks; cb; cb = next) {
     761     4310295 :                 next = cb->next;
     762     4310295 :                 if ((cb->msg_type == msg_type)
     763      153907 :                     && (cb->private_data == private_data)) {
     764      153907 :                         DEBUG(5,("Deregistering messaging pointer for type %u - private_data=%p\n",
     765             :                                   (unsigned)msg_type, private_data));
     766      153907 :                         DLIST_REMOVE(ctx->callbacks, cb);
     767      153907 :                         TALLOC_FREE(cb);
     768             :                 }
     769             :         }
     770      153907 : }
     771             : 
     772             : /*
     773             :   Send a message to a particular server
     774             : */
     775      552217 : NTSTATUS messaging_send(struct messaging_context *msg_ctx,
     776             :                         struct server_id server, uint32_t msg_type,
     777             :                         const DATA_BLOB *data)
     778             : {
     779      552217 :         struct iovec iov = {0};
     780             : 
     781      552217 :         if (data != NULL) {
     782      551556 :                 iov.iov_base = data->data;
     783      551556 :                 iov.iov_len = data->length;
     784             :         };
     785             : 
     786      552217 :         return messaging_send_iov(msg_ctx, server, msg_type, &iov, 1, NULL, 0);
     787             : }
     788             : 
     789      340303 : NTSTATUS messaging_send_buf(struct messaging_context *msg_ctx,
     790             :                             struct server_id server, uint32_t msg_type,
     791             :                             const uint8_t *buf, size_t len)
     792             : {
     793      340303 :         DATA_BLOB blob = data_blob_const(buf, len);
     794      340303 :         return messaging_send(msg_ctx, server, msg_type, &blob);
     795             : }
     796             : 
     797      186302 : static int messaging_post_self(struct messaging_context *msg_ctx,
     798             :                                struct server_id src, struct server_id dst,
     799             :                                uint32_t msg_type,
     800             :                                const struct iovec *iov, int iovlen,
     801             :                                const int *fds, size_t num_fds)
     802             : {
     803             :         struct messaging_rec *rec;
     804             :         bool ok;
     805             : 
     806      186302 :         rec = messaging_rec_create(
     807             :                 msg_ctx, src, dst, msg_type, iov, iovlen, fds, num_fds);
     808      186302 :         if (rec == NULL) {
     809           0 :                 return ENOMEM;
     810             :         }
     811             : 
     812      186302 :         ok = messaging_alert_event_contexts(msg_ctx);
     813      186302 :         if (!ok) {
     814           0 :                 TALLOC_FREE(rec);
     815           0 :                 return ENOMEM;
     816             :         }
     817             : 
     818      186302 :         DLIST_ADD_END(msg_ctx->posted_msgs, rec);
     819             : 
     820      186065 :         return 0;
     821             : }
     822             : 
     823      862440 : int messaging_send_iov_from(struct messaging_context *msg_ctx,
     824             :                             struct server_id src, struct server_id dst,
     825             :                             uint32_t msg_type,
     826             :                             const struct iovec *iov, int iovlen,
     827             :                             const int *fds, size_t num_fds)
     828      862440 : {
     829             :         int ret;
     830             :         uint8_t hdr[MESSAGE_HDR_LENGTH];
     831      862440 :         struct iovec iov2[iovlen+1];
     832             : 
     833      862440 :         if (server_id_is_disconnected(&dst)) {
     834           0 :                 return EINVAL;
     835             :         }
     836             : 
     837      862440 :         if (num_fds > INT8_MAX) {
     838           0 :                 return EINVAL;
     839             :         }
     840             : 
     841      862440 :         if (server_id_equal(&dst, &msg_ctx->id)) {
     842      186211 :                 ret = messaging_post_self(msg_ctx, src, dst, msg_type,
     843             :                                           iov, iovlen, fds, num_fds);
     844      186211 :                 return ret;
     845             :         }
     846             : 
     847      676229 :         message_hdr_put(hdr, msg_type, src, dst);
     848      676229 :         iov2[0] = (struct iovec){ .iov_base = hdr, .iov_len = sizeof(hdr) };
     849      680252 :         memcpy(&iov2[1], iov, iovlen * sizeof(*iov));
     850             : 
     851      676229 :         if (dst.vnn != msg_ctx->id.vnn) {
     852           0 :                 if (num_fds > 0) {
     853           0 :                         return ENOSYS;
     854             :                 }
     855             : 
     856           0 :                 ret = messaging_ctdb_send(dst.vnn, dst.pid, iov2, iovlen+1);
     857           0 :                 return ret;
     858             :         }
     859             : 
     860      676229 :         ret = messaging_dgm_send(dst.pid, iov2, iovlen+1, fds, num_fds);
     861             : 
     862      676229 :         if (ret == EACCES) {
     863           0 :                 become_root();
     864           0 :                 ret = messaging_dgm_send(dst.pid, iov2, iovlen+1,
     865             :                                          fds, num_fds);
     866           0 :                 unbecome_root();
     867             :         }
     868             : 
     869      676229 :         if (ret == ECONNREFUSED) {
     870             :                 /*
     871             :                  * Linux returns this when a socket exists in the file
     872             :                  * system without a listening process. This is not
     873             :                  * documented in susv4 or the linux manpages, but it's
     874             :                  * easily testable. For the higher levels this is the
     875             :                  * same as "destination does not exist"
     876             :                  */
     877        7467 :                 ret = ENOENT;
     878             :         }
     879             : 
     880      672206 :         return ret;
     881             : }
     882             : 
     883      862440 : NTSTATUS messaging_send_iov(struct messaging_context *msg_ctx,
     884             :                             struct server_id dst, uint32_t msg_type,
     885             :                             const struct iovec *iov, int iovlen,
     886             :                             const int *fds, size_t num_fds)
     887             : {
     888             :         int ret;
     889             : 
     890      862440 :         ret = messaging_send_iov_from(msg_ctx, msg_ctx->id, dst, msg_type,
     891             :                                       iov, iovlen, fds, num_fds);
     892      862440 :         if (ret != 0) {
     893        7489 :                 return map_nt_error_from_unix(ret);
     894             :         }
     895      854951 :         return NT_STATUS_OK;
     896             : }
     897             : 
     898             : struct send_all_state {
     899             :         struct messaging_context *msg_ctx;
     900             :         int msg_type;
     901             :         const void *buf;
     902             :         size_t len;
     903             : };
     904             : 
     905      347255 : static int send_all_fn(pid_t pid, void *private_data)
     906             : {
     907      347255 :         struct send_all_state *state = private_data;
     908             :         NTSTATUS status;
     909             : 
     910      347255 :         if (pid == getpid()) {
     911       10506 :                 DBG_DEBUG("Skip ourselves in messaging_send_all\n");
     912       10440 :                 return 0;
     913             :         }
     914             : 
     915      601944 :         status = messaging_send_buf(state->msg_ctx, pid_to_procid(pid),
     916      336749 :                                     state->msg_type, state->buf, state->len);
     917      336749 :         if (!NT_STATUS_IS_OK(status)) {
     918        7469 :                 DBG_NOTICE("messaging_send_buf to %ju failed: %s\n",
     919             :                             (uintmax_t)pid, nt_errstr(status));
     920             :         }
     921             : 
     922      334191 :         return 0;
     923             : }
     924             : 
     925       10506 : void messaging_send_all(struct messaging_context *msg_ctx,
     926             :                         int msg_type, const void *buf, size_t len)
     927             : {
     928       10506 :         struct send_all_state state = {
     929             :                 .msg_ctx = msg_ctx, .msg_type = msg_type,
     930             :                 .buf = buf, .len = len
     931             :         };
     932             :         int ret;
     933             : 
     934             : #ifdef CLUSTER_SUPPORT
     935             :         if (lp_clustering()) {
     936             :                 struct ctdbd_connection *conn = messaging_ctdb_connection();
     937             :                 uint8_t msghdr[MESSAGE_HDR_LENGTH];
     938             :                 struct iovec iov[] = {
     939             :                         { .iov_base = msghdr,
     940             :                           .iov_len = sizeof(msghdr) },
     941             :                         { .iov_base = discard_const_p(void, buf),
     942             :                           .iov_len = len }
     943             :                 };
     944             : 
     945             :                 message_hdr_put(msghdr, msg_type, messaging_server_id(msg_ctx),
     946             :                                 (struct server_id) {0});
     947             : 
     948             :                 ret = ctdbd_messaging_send_iov(
     949             :                         conn, CTDB_BROADCAST_CONNECTED,
     950             :                         CTDB_SRVID_SAMBA_PROCESS,
     951             :                         iov, ARRAY_SIZE(iov));
     952             :                 if (ret != 0) {
     953             :                         DBG_WARNING("ctdbd_messaging_send_iov failed: %s\n",
     954             :                                     strerror(ret));
     955             :                 }
     956             : 
     957             :                 return;
     958             :         }
     959             : #endif
     960             : 
     961       10506 :         ret = messaging_dgm_forall(send_all_fn, &state);
     962       10506 :         if (ret != 0) {
     963           0 :                 DBG_WARNING("messaging_dgm_forall failed: %s\n",
     964             :                             strerror(ret));
     965             :         }
     966       10506 : }
     967             : 
     968      190631 : static struct messaging_rec *messaging_rec_dup(TALLOC_CTX *mem_ctx,
     969             :                                                struct messaging_rec *rec)
     970             : {
     971             :         struct messaging_rec *result;
     972      190631 :         size_t fds_size = sizeof(int64_t) * rec->num_fds;
     973             :         size_t payload_len;
     974             : 
     975      190631 :         payload_len = rec->buf.length + fds_size;
     976      190631 :         if (payload_len < rec->buf.length) {
     977             :                 /* overflow */
     978           0 :                 return NULL;
     979             :         }
     980             : 
     981      190631 :         result = talloc_pooled_object(mem_ctx, struct messaging_rec, 2,
     982             :                                       payload_len);
     983      190631 :         if (result == NULL) {
     984           0 :                 return NULL;
     985             :         }
     986      190631 :         *result = *rec;
     987             : 
     988             :         /* Doesn't fail, see talloc_pooled_object */
     989             : 
     990      190631 :         result->buf.data = talloc_memdup(result, rec->buf.data,
     991             :                                          rec->buf.length);
     992             : 
     993      190631 :         result->fds = NULL;
     994      190631 :         if (result->num_fds > 0) {
     995             :                 size_t i;
     996             : 
     997         934 :                 result->fds = talloc_memdup(result, rec->fds, fds_size);
     998             : 
     999        1871 :                 for (i=0; i<rec->num_fds; i++) {
    1000             :                         /*
    1001             :                          * fd's can only exist once
    1002             :                          */
    1003         937 :                         rec->fds[i] = -1;
    1004             :                 }
    1005             :         }
    1006             : 
    1007      190278 :         return result;
    1008             : }
    1009             : 
    1010             : struct messaging_filtered_read_state {
    1011             :         struct tevent_context *ev;
    1012             :         struct messaging_context *msg_ctx;
    1013             :         struct messaging_dgm_fde *fde;
    1014             :         struct messaging_ctdb_fde *cluster_fde;
    1015             : 
    1016             :         bool (*filter)(struct messaging_rec *rec, void *private_data);
    1017             :         void *private_data;
    1018             : 
    1019             :         struct messaging_rec *rec;
    1020             : };
    1021             : 
    1022             : static void messaging_filtered_read_cleanup(struct tevent_req *req,
    1023             :                                             enum tevent_req_state req_state);
    1024             : 
    1025       91865 : struct tevent_req *messaging_filtered_read_send(
    1026             :         TALLOC_CTX *mem_ctx, struct tevent_context *ev,
    1027             :         struct messaging_context *msg_ctx,
    1028             :         bool (*filter)(struct messaging_rec *rec, void *private_data),
    1029             :         void *private_data)
    1030             : {
    1031             :         struct tevent_req *req;
    1032             :         struct messaging_filtered_read_state *state;
    1033             :         size_t new_waiters_len;
    1034             :         bool ok;
    1035             : 
    1036       91865 :         req = tevent_req_create(mem_ctx, &state,
    1037             :                                 struct messaging_filtered_read_state);
    1038       91865 :         if (req == NULL) {
    1039           0 :                 return NULL;
    1040             :         }
    1041       91865 :         state->ev = ev;
    1042       91865 :         state->msg_ctx = msg_ctx;
    1043       91865 :         state->filter = filter;
    1044       91865 :         state->private_data = private_data;
    1045             : 
    1046             :         /*
    1047             :          * We have to defer the callback here, as we might be called from
    1048             :          * within a different tevent_context than state->ev
    1049             :          */
    1050       91865 :         tevent_req_defer_callback(req, state->ev);
    1051             : 
    1052       91865 :         state->fde = messaging_dgm_register_tevent_context(state, ev);
    1053       91865 :         if (tevent_req_nomem(state->fde, req)) {
    1054           0 :                 return tevent_req_post(req, ev);
    1055             :         }
    1056             : 
    1057       91865 :         if (lp_clustering()) {
    1058           0 :                 state->cluster_fde =
    1059           0 :                         messaging_ctdb_register_tevent_context(state, ev);
    1060           0 :                 if (tevent_req_nomem(state->cluster_fde, req)) {
    1061           0 :                         return tevent_req_post(req, ev);
    1062             :                 }
    1063             :         }
    1064             : 
    1065             :         /*
    1066             :          * We add ourselves to the "new_waiters" array, not the "waiters"
    1067             :          * array. If we are called from within messaging_read_done,
    1068             :          * messaging_dispatch_rec will be in an active for-loop on
    1069             :          * "waiters". We must be careful not to mess with this array, because
    1070             :          * it could mean that a single event is being delivered twice.
    1071             :          */
    1072             : 
    1073       91865 :         new_waiters_len = talloc_array_length(msg_ctx->new_waiters);
    1074             : 
    1075       91865 :         if (new_waiters_len == msg_ctx->num_new_waiters) {
    1076             :                 struct tevent_req **tmp;
    1077             : 
    1078       52801 :                 tmp = talloc_realloc(msg_ctx, msg_ctx->new_waiters,
    1079             :                                      struct tevent_req *, new_waiters_len+1);
    1080       52801 :                 if (tevent_req_nomem(tmp, req)) {
    1081           0 :                         return tevent_req_post(req, ev);
    1082             :                 }
    1083       52801 :                 msg_ctx->new_waiters = tmp;
    1084             :         }
    1085             : 
    1086       91865 :         msg_ctx->new_waiters[msg_ctx->num_new_waiters] = req;
    1087       91865 :         msg_ctx->num_new_waiters += 1;
    1088       91865 :         tevent_req_set_cleanup_fn(req, messaging_filtered_read_cleanup);
    1089             : 
    1090       91865 :         ok = messaging_register_event_context(msg_ctx, ev);
    1091       91865 :         if (!ok) {
    1092           0 :                 tevent_req_oom(req);
    1093           0 :                 return tevent_req_post(req, ev);
    1094             :         }
    1095             : 
    1096       89994 :         return req;
    1097             : }
    1098             : 
    1099       87388 : static void messaging_filtered_read_cleanup(struct tevent_req *req,
    1100             :                                             enum tevent_req_state req_state)
    1101             : {
    1102       87388 :         struct messaging_filtered_read_state *state = tevent_req_data(
    1103             :                 req, struct messaging_filtered_read_state);
    1104       87388 :         struct messaging_context *msg_ctx = state->msg_ctx;
    1105             :         size_t i;
    1106             :         bool ok;
    1107             : 
    1108       87388 :         tevent_req_set_cleanup_fn(req, NULL);
    1109             : 
    1110       87388 :         TALLOC_FREE(state->fde);
    1111       87388 :         TALLOC_FREE(state->cluster_fde);
    1112             : 
    1113       87388 :         ok = messaging_deregister_event_context(msg_ctx, state->ev);
    1114       87388 :         if (!ok) {
    1115           0 :                 abort();
    1116             :         }
    1117             : 
    1118             :         /*
    1119             :          * Just set the [new_]waiters entry to NULL, be careful not to mess
    1120             :          * with the other "waiters" array contents. We are often called from
    1121             :          * within "messaging_dispatch_rec", which loops over
    1122             :          * "waiters". Messing with the "waiters" array will mess up that
    1123             :          * for-loop.
    1124             :          */
    1125             : 
    1126      143144 :         for (i=0; i<msg_ctx->num_waiters; i++) {
    1127       84677 :                 if (msg_ctx->waiters[i] == req) {
    1128       27193 :                         msg_ctx->waiters[i] = NULL;
    1129       27193 :                         return;
    1130             :                 }
    1131             :         }
    1132             : 
    1133      176230 :         for (i=0; i<msg_ctx->num_new_waiters; i++) {
    1134      177837 :                 if (msg_ctx->new_waiters[i] == req) {
    1135       60195 :                         msg_ctx->new_waiters[i] = NULL;
    1136       60195 :                         return;
    1137             :                 }
    1138             :         }
    1139             : }
    1140             : 
    1141        4329 : static void messaging_filtered_read_done(struct tevent_req *req,
    1142             :                                          struct messaging_rec *rec)
    1143             : {
    1144        4329 :         struct messaging_filtered_read_state *state = tevent_req_data(
    1145             :                 req, struct messaging_filtered_read_state);
    1146             : 
    1147        4329 :         state->rec = messaging_rec_dup(state, rec);
    1148        4329 :         if (tevent_req_nomem(state->rec, req)) {
    1149           0 :                 return;
    1150             :         }
    1151        4329 :         tevent_req_done(req);
    1152             : }
    1153             : 
    1154        4329 : int messaging_filtered_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
    1155             :                                  struct messaging_rec **presult)
    1156             : {
    1157        4329 :         struct messaging_filtered_read_state *state = tevent_req_data(
    1158             :                 req, struct messaging_filtered_read_state);
    1159             :         int err;
    1160             : 
    1161        4329 :         if (tevent_req_is_unix_error(req, &err)) {
    1162           0 :                 tevent_req_received(req);
    1163           0 :                 return err;
    1164             :         }
    1165        4329 :         if (presult != NULL) {
    1166        4329 :                 *presult = talloc_move(mem_ctx, &state->rec);
    1167             :         }
    1168        4329 :         tevent_req_received(req);
    1169        4329 :         return 0;
    1170             : }
    1171             : 
    1172             : struct messaging_read_state {
    1173             :         uint32_t msg_type;
    1174             :         struct messaging_rec *rec;
    1175             : };
    1176             : 
    1177             : static bool messaging_read_filter(struct messaging_rec *rec,
    1178             :                                   void *private_data);
    1179             : static void messaging_read_done(struct tevent_req *subreq);
    1180             : 
    1181       26854 : struct tevent_req *messaging_read_send(TALLOC_CTX *mem_ctx,
    1182             :                                        struct tevent_context *ev,
    1183             :                                        struct messaging_context *msg,
    1184             :                                        uint32_t msg_type)
    1185             : {
    1186             :         struct tevent_req *req, *subreq;
    1187             :         struct messaging_read_state *state;
    1188             : 
    1189       26854 :         req = tevent_req_create(mem_ctx, &state,
    1190             :                                 struct messaging_read_state);
    1191       26854 :         if (req == NULL) {
    1192           0 :                 return NULL;
    1193             :         }
    1194       26854 :         state->msg_type = msg_type;
    1195             : 
    1196       26854 :         subreq = messaging_filtered_read_send(state, ev, msg,
    1197             :                                               messaging_read_filter, state);
    1198       26854 :         if (tevent_req_nomem(subreq, req)) {
    1199           0 :                 return tevent_req_post(req, ev);
    1200             :         }
    1201       26854 :         tevent_req_set_callback(subreq, messaging_read_done, req);
    1202       26854 :         return req;
    1203             : }
    1204             : 
    1205       44675 : static bool messaging_read_filter(struct messaging_rec *rec,
    1206             :                                   void *private_data)
    1207             : {
    1208       44675 :         struct messaging_read_state *state = talloc_get_type_abort(
    1209             :                 private_data, struct messaging_read_state);
    1210             : 
    1211       44675 :         if (rec->num_fds != 0) {
    1212         576 :                 return false;
    1213             :         }
    1214             : 
    1215       44099 :         return rec->msg_type == state->msg_type;
    1216             : }
    1217             : 
    1218         162 : static void messaging_read_done(struct tevent_req *subreq)
    1219             : {
    1220         162 :         struct tevent_req *req = tevent_req_callback_data(
    1221             :                 subreq, struct tevent_req);
    1222         162 :         struct messaging_read_state *state = tevent_req_data(
    1223             :                 req, struct messaging_read_state);
    1224             :         int ret;
    1225             : 
    1226         162 :         ret = messaging_filtered_read_recv(subreq, state, &state->rec);
    1227         162 :         TALLOC_FREE(subreq);
    1228         162 :         if (tevent_req_error(req, ret)) {
    1229           0 :                 return;
    1230             :         }
    1231         162 :         tevent_req_done(req);
    1232             : }
    1233             : 
    1234         162 : int messaging_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
    1235             :                         struct messaging_rec **presult)
    1236             : {
    1237         162 :         struct messaging_read_state *state = tevent_req_data(
    1238             :                 req, struct messaging_read_state);
    1239             :         int err;
    1240             : 
    1241         162 :         if (tevent_req_is_unix_error(req, &err)) {
    1242           0 :                 return err;
    1243             :         }
    1244         162 :         if (presult != NULL) {
    1245          60 :                 *presult = talloc_move(mem_ctx, &state->rec);
    1246             :         }
    1247          54 :         return 0;
    1248             : }
    1249             : 
    1250      104537 : static bool messaging_append_new_waiters(struct messaging_context *msg_ctx)
    1251             : {
    1252      104537 :         if (msg_ctx->num_new_waiters == 0) {
    1253       99926 :                 return true;
    1254             :         }
    1255             : 
    1256        8431 :         if (talloc_array_length(msg_ctx->waiters) <
    1257        4547 :             (msg_ctx->num_waiters + msg_ctx->num_new_waiters)) {
    1258             :                 struct tevent_req **tmp;
    1259        2529 :                 tmp = talloc_realloc(
    1260             :                         msg_ctx, msg_ctx->waiters, struct tevent_req *,
    1261             :                         msg_ctx->num_waiters + msg_ctx->num_new_waiters);
    1262        2529 :                 if (tmp == NULL) {
    1263           0 :                         DEBUG(1, ("%s: talloc failed\n", __func__));
    1264           0 :                         return false;
    1265             :                 }
    1266        2529 :                 msg_ctx->waiters = tmp;
    1267             :         }
    1268             : 
    1269        4665 :         memcpy(&msg_ctx->waiters[msg_ctx->num_waiters], msg_ctx->new_waiters,
    1270        4547 :                sizeof(struct tevent_req *) * msg_ctx->num_new_waiters);
    1271             : 
    1272        4547 :         msg_ctx->num_waiters += msg_ctx->num_new_waiters;
    1273        4547 :         msg_ctx->num_new_waiters = 0;
    1274             : 
    1275        4547 :         return true;
    1276             : }
    1277             : 
    1278      271375 : static bool messaging_dispatch_classic(struct messaging_context *msg_ctx,
    1279             :                                        struct messaging_rec *rec)
    1280             : {
    1281             :         struct messaging_callback *cb, *next;
    1282             : 
    1283     5059745 :         for (cb = msg_ctx->callbacks; cb != NULL; cb = next) {
    1284             :                 size_t j;
    1285             : 
    1286     2481033 :                 next = cb->next;
    1287     2481033 :                 if (cb->msg_type != rec->msg_type) {
    1288     2313956 :                         continue;
    1289             :                 }
    1290             : 
    1291             :                 /*
    1292             :                  * the old style callbacks don't support fd passing
    1293             :                  */
    1294      166653 :                 for (j=0; j < rec->num_fds; j++) {
    1295           0 :                         int fd = rec->fds[j];
    1296           0 :                         close(fd);
    1297             :                 }
    1298      167077 :                 rec->num_fds = 0;
    1299      167077 :                 rec->fds = NULL;
    1300             : 
    1301      167077 :                 cb->fn(msg_ctx, cb->private_data, rec->msg_type,
    1302             :                        rec->src, &rec->buf);
    1303             : 
    1304      167054 :                 return true;
    1305             :         }
    1306             : 
    1307      104116 :         return false;
    1308             : }
    1309             : 
    1310      104537 : static bool messaging_dispatch_waiters(struct messaging_context *msg_ctx,
    1311             :                                        struct tevent_context *ev,
    1312             :                                        struct messaging_rec *rec)
    1313             : {
    1314             :         size_t i;
    1315             : 
    1316      104537 :         if (!messaging_append_new_waiters(msg_ctx)) {
    1317           0 :                 return false;
    1318             :         }
    1319             : 
    1320      104355 :         i = 0;
    1321      414545 :         while (i < msg_ctx->num_waiters) {
    1322             :                 struct tevent_req *req;
    1323             :                 struct messaging_filtered_read_state *state;
    1324             : 
    1325      212752 :                 req = msg_ctx->waiters[i];
    1326      212752 :                 if (req == NULL) {
    1327             :                         /*
    1328             :                          * This got cleaned up. In the meantime,
    1329             :                          * move everything down one. We need
    1330             :                          * to keep the order of waiters, as
    1331             :                          * other code may depend on this.
    1332             :                          */
    1333        5815 :                         ARRAY_DEL_ELEMENT(
    1334             :                                 msg_ctx->waiters, i, msg_ctx->num_waiters);
    1335        5815 :                         msg_ctx->num_waiters -= 1;
    1336        5815 :                         continue;
    1337             :                 }
    1338             : 
    1339      206937 :                 state = tevent_req_data(
    1340             :                         req, struct messaging_filtered_read_state);
    1341      413175 :                 if ((ev == state->ev) &&
    1342      206238 :                     state->filter(rec, state->private_data)) {
    1343        4329 :                         messaging_filtered_read_done(req, rec);
    1344        4329 :                         return true;
    1345             :                 }
    1346             : 
    1347      202608 :                 i += 1;
    1348             :         }
    1349             : 
    1350      100142 :         return false;
    1351             : }
    1352             : 
    1353             : /*
    1354             :   Dispatch one messaging_rec
    1355             : */
    1356       85366 : static void messaging_dispatch_rec(struct messaging_context *msg_ctx,
    1357             :                                    struct tevent_context *ev,
    1358             :                                    struct messaging_rec *rec)
    1359             : {
    1360             :         bool consumed;
    1361             :         size_t i;
    1362             : 
    1363       85366 :         if (ev == msg_ctx->event_ctx) {
    1364       85127 :                 consumed = messaging_dispatch_classic(msg_ctx, rec);
    1365       85112 :                 if (consumed) {
    1366       22589 :                         return;
    1367             :                 }
    1368             :         }
    1369             : 
    1370       62565 :         consumed = messaging_dispatch_waiters(msg_ctx, ev, rec);
    1371       62565 :         if (consumed) {
    1372        2448 :                 return;
    1373             :         }
    1374             : 
    1375       60008 :         if (ev != msg_ctx->event_ctx) {
    1376             :                 struct iovec iov;
    1377          91 :                 int fds[MAX(1, rec->num_fds)];
    1378             :                 int ret;
    1379             : 
    1380             :                 /*
    1381             :                  * We've been listening on a nested event
    1382             :                  * context. Messages need to be handled in the main
    1383             :                  * event context, so post to ourselves
    1384             :                  */
    1385             : 
    1386          91 :                 iov.iov_base = rec->buf.data;
    1387          91 :                 iov.iov_len = rec->buf.length;
    1388             : 
    1389          91 :                 for (i=0; i<rec->num_fds; i++) {
    1390           0 :                         fds[i] = rec->fds[i];
    1391             :                 }
    1392             : 
    1393         146 :                 ret = messaging_post_self(
    1394          91 :                         msg_ctx, rec->src, rec->dest, rec->msg_type,
    1395          91 :                         &iov, 1, fds, rec->num_fds);
    1396          91 :                 if (ret == 0) {
    1397          91 :                         return;
    1398             :                 }
    1399             :         }
    1400             : }
    1401             : 
    1402             : static int mess_parent_dgm_cleanup(void *private_data);
    1403             : static void mess_parent_dgm_cleanup_done(struct tevent_req *req);
    1404             : 
    1405         166 : bool messaging_parent_dgm_cleanup_init(struct messaging_context *msg)
    1406             : {
    1407             :         struct tevent_req *req;
    1408             : 
    1409         166 :         req = background_job_send(
    1410             :                 msg, msg->event_ctx, msg, NULL, 0,
    1411         166 :                 lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
    1412             :                             60*15),
    1413             :                 mess_parent_dgm_cleanup, msg);
    1414         166 :         if (req == NULL) {
    1415           0 :                 DBG_WARNING("background_job_send failed\n");
    1416           0 :                 return false;
    1417             :         }
    1418         166 :         tevent_req_set_callback(req, mess_parent_dgm_cleanup_done, msg);
    1419         166 :         return true;
    1420             : }
    1421             : 
    1422           0 : static int mess_parent_dgm_cleanup(void *private_data)
    1423             : {
    1424             :         int ret;
    1425             : 
    1426           0 :         ret = messaging_dgm_wipe();
    1427           0 :         DEBUG(10, ("messaging_dgm_wipe returned %s\n",
    1428             :                    ret ? strerror(ret) : "ok"));
    1429           0 :         return lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
    1430             :                            60*15);
    1431             : }
    1432             : 
    1433           0 : static void mess_parent_dgm_cleanup_done(struct tevent_req *req)
    1434             : {
    1435           0 :         struct messaging_context *msg = tevent_req_callback_data(
    1436             :                 req, struct messaging_context);
    1437             :         NTSTATUS status;
    1438             : 
    1439           0 :         status = background_job_recv(req);
    1440           0 :         TALLOC_FREE(req);
    1441           0 :         DEBUG(1, ("messaging dgm cleanup job ended with %s\n",
    1442             :                   nt_errstr(status)));
    1443             : 
    1444           0 :         req = background_job_send(
    1445             :                 msg, msg->event_ctx, msg, NULL, 0,
    1446           0 :                 lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
    1447             :                             60*15),
    1448             :                 mess_parent_dgm_cleanup, msg);
    1449           0 :         if (req == NULL) {
    1450           0 :                 DEBUG(1, ("background_job_send failed\n"));
    1451           0 :                 return;
    1452             :         }
    1453           0 :         tevent_req_set_callback(req, mess_parent_dgm_cleanup_done, msg);
    1454             : }
    1455             : 
    1456           0 : int messaging_cleanup(struct messaging_context *msg_ctx, pid_t pid)
    1457             : {
    1458             :         int ret;
    1459             : 
    1460           0 :         if (pid == 0) {
    1461           0 :                 ret = messaging_dgm_wipe();
    1462             :         } else {
    1463           0 :                 ret = messaging_dgm_cleanup(pid);
    1464             :         }
    1465             : 
    1466           0 :         return ret;
    1467             : }
    1468             : 
    1469       32331 : struct tevent_context *messaging_tevent_context(
    1470             :         struct messaging_context *msg_ctx)
    1471             : {
    1472       32331 :         return msg_ctx->event_ctx;
    1473             : }
    1474             : 
    1475       17314 : struct server_id_db *messaging_names_db(struct messaging_context *msg_ctx)
    1476             : {
    1477       17314 :         return msg_ctx->names_db;
    1478             : }
    1479             : 
    1480             : /** @} **/

Generated by: LCOV version 1.13