source3/lib/messages.c

   1 /*
   2    Unix SMB/CIFS implementation.
   3    Samba internal messaging functions
   4    Copyright (C) Andrew Tridgell 2000
   5    Copyright (C) 2001 by Martin Pool
   6    Copyright (C) 2002 by Jeremy Allison
   7    Copyright (C) 2007 by Volker Lendecke
   8
   9    This program is free software; you can redistribute it and/or modify
  10    it under the terms of the GNU General Public License as published by
  11    the Free Software Foundation; either version 3 of the License, or
  12    (at your option) any later version.
  13
  14    This program is distributed in the hope that it will be useful,
  15    but WITHOUT ANY WARRANTY; without even the implied warranty of
  16    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  17    GNU General Public License for more details.
  18
  19    You should have received a copy of the GNU General Public License
  20    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  21 */
  22
  23 /**
  24   @defgroup messages Internal messaging framework
  25   @{
  26   @file messages.c
  27
  28   @brief  Module for internal messaging between Samba daemons.
  29
  30    The idea is that if a part of Samba wants to do communication with
  31    another Samba process then it will do a message_register() of a
  32    dispatch function, and use message_send_pid() to send messages to
  33    that process.
  34
  35    The dispatch function is given the pid of the sender, and it can
  36    use that to reply by message_send_pid().  See ping_message() for a
  37    simple example.
  38
  39    @caution Dispatch functions must be able to cope with incoming
  40    messages on an *odd* byte boundary.
  41
  42    This system doesn't have any inherent size limitations but is not
  43    very efficient for large messages or when messages are sent in very
  44    quick succession.
  45
  46 */
  47
  48 #include "includes.h"
  49 #include "dbwrap/dbwrap.h"
  50 #include "serverid.h"
  51 #include "messages.h"
  52 #include "lib/util/tevent_unix.h"
  53
  54 struct messaging_callback {
  55         struct messaging_callback *prev, *next;
  56         uint32 msg_type;
  57         void (*fn)(struct messaging_context *msg, void *private_data,
  58                    uint32_t msg_type,
  59                    struct server_id server_id, DATA_BLOB *data);
  60         void *private_data;
  61 };
  62
  63 /****************************************************************************
  64  A useful function for testing the message system.
  65 ****************************************************************************/
  66
  67 static void ping_message(struct messaging_context *msg_ctx,
  68                          void *private_data,
  69                          uint32_t msg_type,
  70                          struct server_id src,
  71                          DATA_BLOB *data)
  72 {
  73         const char *msg = "none";
  74         char *free_me = NULL;
  75
  76         if (data->data != NULL) {
  77                 free_me = talloc_strndup(talloc_tos(), (char *)data->data,
  78                                          data->length);
  79                 msg = free_me;
  80         }
  81         DEBUG(1,("INFO: Received PING message from PID %s [%s]\n",
  82                  procid_str_static(&src), msg));
  83         TALLOC_FREE(free_me);
  84         messaging_send(msg_ctx, src, MSG_PONG, data);
  85 }
  86
  87 /****************************************************************************
  88  Register/replace a dispatch function for a particular message type.
  89  JRA changed Dec 13 2006. Only one message handler now permitted per type.
  90  *NOTE*: Dispatch functions must be able to cope with incoming
  91  messages on an *odd* byte boundary.
  92 ****************************************************************************/
  93
  94 struct msg_all {
  95         struct messaging_context *msg_ctx;
  96         int msg_type;
  97         uint32 msg_flag;
  98         const void *buf;
  99         size_t len;
 100         int n_sent;
 101 };
 102
 103 /****************************************************************************
 104  Send one of the messages for the broadcast.
 105 ****************************************************************************/
 106
 107 static int traverse_fn(struct db_record *rec, const struct server_id *id,
 108                        uint32_t msg_flags, void *state)
 109 {
 110         struct msg_all *msg_all = (struct msg_all *)state;
 111         NTSTATUS status;
 112
 113         /* Don't send if the receiver hasn't registered an interest. */
 114
 115         if((msg_flags & msg_all->msg_flag) == 0) {
 116                 return 0;
 117         }
 118
 119         /* If the msg send fails because the pid was not found (i.e. smbd died),
 120          * the msg has already been deleted from the messages.tdb.*/
 121
 122         status = messaging_send_buf(msg_all->msg_ctx, *id, msg_all->msg_type,
 123                                     (const uint8_t *)msg_all->buf, msg_all->len);
 124
 125         if (NT_STATUS_EQUAL(status, NT_STATUS_INVALID_HANDLE)) {
 126
 127                 /*
 128                  * If the pid was not found delete the entry from
 129                  * serverid.tdb
 130                  */
 131
 132                 DEBUG(2, ("pid %s doesn't exist\n", procid_str_static(id)));
 133
 134                 dbwrap_record_delete(rec);
 135         }
 136         msg_all->n_sent++;
 137         return 0;
 138 }
 139
 140 /**
 141  * Send a message to all smbd processes.
 142  *
 143  * It isn't very efficient, but should be OK for the sorts of
 144  * applications that use it. When we need efficient broadcast we can add
 145  * it.
 146  *
 147  * @param n_sent Set to the number of messages sent.  This should be
 148  * equal to the number of processes, but be careful for races.
 149  *
 150  * @retval True for success.
 151  **/
 152 bool message_send_all(struct messaging_context *msg_ctx,
 153                       int msg_type,
 154                       const void *buf, size_t len,
 155                       int *n_sent)
 156 {
 157         struct msg_all msg_all;
 158
 159         msg_all.msg_type = msg_type;
 160         if (msg_type < 0x100) {
 161                 msg_all.msg_flag = FLAG_MSG_GENERAL;
 162         } else if (msg_type > 0x100 && msg_type < 0x200) {
 163                 msg_all.msg_flag = FLAG_MSG_NMBD;
 164         } else if (msg_type > 0x200 && msg_type < 0x300) {
 165                 msg_all.msg_flag = FLAG_MSG_PRINT_GENERAL;
 166         } else if (msg_type > 0x300 && msg_type < 0x400) {
 167                 msg_all.msg_flag = FLAG_MSG_SMBD;
 168         } else if (msg_type > 0x400 && msg_type < 0x600) {
 169                 msg_all.msg_flag = FLAG_MSG_WINBIND;
 170         } else if (msg_type > 4000 && msg_type < 5000) {
 171                 msg_all.msg_flag = FLAG_MSG_DBWRAP;
 172         } else {
 173                 return false;
 174         }
 175
 176         msg_all.buf = buf;
 177         msg_all.len = len;
 178         msg_all.n_sent = 0;
 179         msg_all.msg_ctx = msg_ctx;
 180
 181         serverid_traverse(traverse_fn, &msg_all);
 182         if (n_sent)
 183                 *n_sent = msg_all.n_sent;
 184         return true;
 185 }
 186
 187 struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx,
 188                                          struct tevent_context *ev)
 189 {
 190         struct messaging_context *ctx;
 191         NTSTATUS status;
 192
 193         if (!(ctx = talloc_zero(mem_ctx, struct messaging_context))) {
 194                 return NULL;
 195         }
 196
 197         ctx->id = procid_self();
 198         ctx->event_ctx = ev;
 199
 200         status = messaging_tdb_init(ctx, ctx, &ctx->local);
 201
 202         if (!NT_STATUS_IS_OK(status)) {
 203                 DEBUG(2, ("messaging_tdb_init failed: %s\n",
 204                           nt_errstr(status)));
 205                 TALLOC_FREE(ctx);
 206                 return NULL;
 207         }
 208
 209 #ifdef CLUSTER_SUPPORT
 210         if (lp_clustering()) {
 211                 status = messaging_ctdbd_init(ctx, ctx, &ctx->remote);
 212
 213                 if (!NT_STATUS_IS_OK(status)) {
 214                         DEBUG(2, ("messaging_ctdbd_init failed: %s\n",
 215                                   nt_errstr(status)));
 216                         TALLOC_FREE(ctx);
 217                         return NULL;
 218                 }
 219         }
 220         ctx->id.vnn = get_my_vnn();
 221 #endif
 222
 223         messaging_register(ctx, NULL, MSG_PING, ping_message);
 224
 225         /* Register some debugging related messages */
 226
 227         register_msg_pool_usage(ctx);
 228         register_dmalloc_msgs(ctx);
 229         debug_register_msgs(ctx);
 230
 231         return ctx;
 232 }
 233
 234 struct server_id messaging_server_id(const struct messaging_context *msg_ctx)
 235 {
 236         return msg_ctx->id;
 237 }
 238
 239 /*
 240  * re-init after a fork
 241  */
 242 NTSTATUS messaging_reinit(struct messaging_context *msg_ctx)
 243 {
 244         NTSTATUS status;
 245
 246         TALLOC_FREE(msg_ctx->local);
 247
 248         msg_ctx->id = procid_self();
 249
 250         status = messaging_tdb_init(msg_ctx, msg_ctx, &msg_ctx->local);
 251         if (!NT_STATUS_IS_OK(status)) {
 252                 DEBUG(0, ("messaging_tdb_init failed: %s\n",
 253                           nt_errstr(status)));
 254                 return status;
 255         }
 256
 257 #ifdef CLUSTER_SUPPORT
 258         TALLOC_FREE(msg_ctx->remote);
 259
 260         if (lp_clustering()) {
 261                 status = messaging_ctdbd_init(msg_ctx, msg_ctx,
 262                                               &msg_ctx->remote);
 263
 264                 if (!NT_STATUS_IS_OK(status)) {
 265                         DEBUG(1, ("messaging_ctdbd_init failed: %s\n",
 266                                   nt_errstr(status)));
 267                         return status;
 268                 }
 269         }
 270
 271 #endif
 272
 273         return NT_STATUS_OK;
 274 }
 275
 276
 277 /*
 278  * Register a dispatch function for a particular message type. Allow multiple
 279  * registrants
 280 */
 281 NTSTATUS messaging_register(struct messaging_context *msg_ctx,
 282                             void *private_data,
 283                             uint32_t msg_type,
 284                             void (*fn)(struct messaging_context *msg,
 285                                        void *private_data,
 286                                        uint32_t msg_type,
 287                                        struct server_id server_id,
 288                                        DATA_BLOB *data))
 289 {
 290         struct messaging_callback *cb;
 291
 292         DEBUG(5, ("Registering messaging pointer for type %u - "
 293                   "private_data=%p\n",
 294                   (unsigned)msg_type, private_data));
 295
 296         /*
 297          * Only one callback per type
 298          */
 299
 300         for (cb = msg_ctx->callbacks; cb != NULL; cb = cb->next) {
 301                 /* we allow a second registration of the same message
 302                    type if it has a different private pointer. This is
 303                    needed in, for example, the internal notify code,
 304                    which creates a new notify context for each tree
 305                    connect, and expects to receive messages to each of
 306                    them. */
 307                 if (cb->msg_type == msg_type && private_data == cb->private_data) {
 308                         DEBUG(5,("Overriding messaging pointer for type %u - private_data=%p\n",
 309                                   (unsigned)msg_type, private_data));
 310                         cb->fn = fn;
 311                         cb->private_data = private_data;
 312                         return NT_STATUS_OK;
 313                 }
 314         }
 315
 316         if (!(cb = talloc(msg_ctx, struct messaging_callback))) {
 317                 return NT_STATUS_NO_MEMORY;
 318         }
 319
 320         cb->msg_type = msg_type;
 321         cb->fn = fn;
 322         cb->private_data = private_data;
 323
 324         DLIST_ADD(msg_ctx->callbacks, cb);
 325         return NT_STATUS_OK;
 326 }
 327
 328 /*
 329   De-register the function for a particular message type.
 330 */
 331 void messaging_deregister(struct messaging_context *ctx, uint32_t msg_type,
 332                           void *private_data)
 333 {
 334         struct messaging_callback *cb, *next;
 335
 336         for (cb = ctx->callbacks; cb; cb = next) {
 337                 next = cb->next;
 338                 if ((cb->msg_type == msg_type)
 339                     && (cb->private_data == private_data)) {
 340                         DEBUG(5,("Deregistering messaging pointer for type %u - private_data=%p\n",
 341                                   (unsigned)msg_type, private_data));
 342                         DLIST_REMOVE(ctx->callbacks, cb);
 343                         TALLOC_FREE(cb);
 344                 }
 345         }
 346 }
 347
 348 struct messaging_selfsend_state {
 349         struct messaging_context *msg;
 350         struct messaging_rec rec;
 351 };
 352
 353 static void messaging_trigger_self(struct tevent_context *ev,
 354                                    struct tevent_immediate *im,
 355                                    void *private_data);
 356
 357 /*
 358   Send a message to a particular server
 359 */
 360 NTSTATUS messaging_send(struct messaging_context *msg_ctx,
 361                         struct server_id server, uint32_t msg_type,
 362                         const DATA_BLOB *data)
 363 {
 364         if (server_id_is_disconnected(&server)) {
 365                 return NT_STATUS_INVALID_PARAMETER_MIX;
 366         }
 367
 368 #ifdef CLUSTER_SUPPORT
 369         if (!procid_is_local(&server)) {
 370                 return msg_ctx->remote->send_fn(msg_ctx, server,
 371                                                 msg_type, data,
 372                                                 msg_ctx->remote);
 373         }
 374 #endif
 375
 376         if (server_id_equal(&msg_ctx->id, &server)) {
 377                 struct messaging_selfsend_state *state;
 378                 struct tevent_immediate *im;
 379
 380                 state = talloc_pooled_object(
 381                         msg_ctx, struct messaging_selfsend_state,
 382                         1, data->length);
 383                 if (state == NULL) {
 384                         return NT_STATUS_NO_MEMORY;
 385                 }
 386                 state->msg = msg_ctx;
 387                 state->rec.msg_version = MESSAGE_VERSION;
 388                 state->rec.msg_type = msg_type & MSG_TYPE_MASK;
 389                 state->rec.dest = server;
 390                 state->rec.src = msg_ctx->id;
 391
 392                 /* Can't fail, it's a pooled_object */
 393                 state->rec.buf = data_blob_talloc(
 394                         state, data->data, data->length);
 395
 396                 im = tevent_create_immediate(state);
 397                 if (im == NULL) {
 398                         TALLOC_FREE(state);
 399                         return NT_STATUS_NO_MEMORY;
 400                 }
 401
 402                 tevent_schedule_immediate(im, msg_ctx->event_ctx,
 403                                           messaging_trigger_self, state);
 404                 return NT_STATUS_OK;
 405         }
 406
 407         return msg_ctx->local->send_fn(msg_ctx, server, msg_type, data,
 408                                        msg_ctx->local);
 409 }
 410
 411 static void messaging_trigger_self(struct tevent_context *ev,
 412                                    struct tevent_immediate *im,
 413                                    void *private_data)
 414 {
 415         struct messaging_selfsend_state *state = talloc_get_type_abort(
 416                 private_data, struct messaging_selfsend_state);
 417         messaging_dispatch_rec(state->msg, &state->rec);
 418         TALLOC_FREE(state);
 419 }
 420
 421 NTSTATUS messaging_send_buf(struct messaging_context *msg_ctx,
 422                             struct server_id server, uint32_t msg_type,
 423                             const uint8_t *buf, size_t len)
 424 {
 425         DATA_BLOB blob = data_blob_const(buf, len);
 426         return messaging_send(msg_ctx, server, msg_type, &blob);
 427 }
 428
 429 static struct messaging_rec *messaging_rec_dup(TALLOC_CTX *mem_ctx,
 430                                                struct messaging_rec *rec)
 431 {
 432         struct messaging_rec *result;
 433
 434         result = talloc_pooled_object(mem_ctx, struct messaging_rec,
 435                                       1, rec->buf.length);
 436         if (result == NULL) {
 437                 return NULL;
 438         }
 439         *result = *rec;
 440
 441         /* Doesn't fail, see talloc_pooled_object */
 442
 443         result->buf.data = talloc_memdup(result, rec->buf.data,
 444                                          rec->buf.length);
 445         return result;
 446 }
 447
 448 struct messaging_read_state {
 449         struct tevent_context *ev;
 450         struct messaging_context *msg_ctx;
 451         uint32_t msg_type;
 452         struct messaging_rec *rec;
 453 };
 454
 455 static void messaging_read_cleanup(struct tevent_req *req,
 456                                    enum tevent_req_state req_state);
 457
 458 struct tevent_req *messaging_read_send(TALLOC_CTX *mem_ctx,
 459                                        struct tevent_context *ev,
 460                                        struct messaging_context *msg_ctx,
 461                                        uint32_t msg_type)
 462 {
 463         struct tevent_req *req;
 464         struct messaging_read_state *state;
 465         size_t waiters_len;
 466
 467         req = tevent_req_create(mem_ctx, &state,
 468                                 struct messaging_read_state);
 469         if (req == NULL) {
 470                 return NULL;
 471         }
 472         state->ev = ev;
 473         state->msg_ctx = msg_ctx;
 474         state->msg_type = msg_type;
 475
 476         waiters_len = talloc_array_length(msg_ctx->waiters);
 477
 478         if (waiters_len == msg_ctx->num_waiters) {
 479                 struct tevent_req **tmp;
 480
 481                 tmp = talloc_realloc(msg_ctx, msg_ctx->waiters,
 482                                      struct tevent_req *, waiters_len+1);
 483                 if (tevent_req_nomem(tmp, req)) {
 484                         return tevent_req_post(req, ev);
 485                 }
 486                 msg_ctx->waiters = tmp;
 487         }
 488
 489         msg_ctx->waiters[msg_ctx->num_waiters] = req;
 490         msg_ctx->num_waiters += 1;
 491         tevent_req_set_cleanup_fn(req, messaging_read_cleanup);
 492
 493         return req;
 494 }
 495
 496 static void messaging_read_cleanup(struct tevent_req *req,
 497                                    enum tevent_req_state req_state)
 498 {
 499         struct messaging_read_state *state = tevent_req_data(
 500                 req, struct messaging_read_state);
 501         struct messaging_context *msg_ctx = state->msg_ctx;
 502         struct tevent_req **waiters = msg_ctx->waiters;
 503         unsigned i;
 504
 505         tevent_req_set_cleanup_fn(req, NULL);
 506
 507         for (i=0; i<msg_ctx->num_waiters; i++) {
 508                 if (waiters[i] == req) {
 509                         waiters[i] = waiters[msg_ctx->num_waiters-1];
 510                         msg_ctx->num_waiters -= 1;
 511                         return;
 512                 }
 513         }
 514 }
 515
 516 static void messaging_read_done(struct tevent_req *req, struct messaging_rec *rec)
 517 {
 518         struct messaging_read_state *state = tevent_req_data(
 519                 req, struct messaging_read_state);
 520
 521         state->rec = messaging_rec_dup(state, rec);
 522         if (tevent_req_nomem(state->rec, req)) {
 523                 return;
 524         }
 525         tevent_req_done(req);
 526 }
 527
 528 int messaging_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
 529                         struct messaging_rec **presult)
 530 {
 531         struct messaging_read_state *state = tevent_req_data(
 532                 req, struct messaging_read_state);
 533         int err;
 534
 535         if (tevent_req_is_unix_error(req, &err)) {
 536                 tevent_req_received(req);
 537                 return err;
 538         }
 539         *presult = talloc_move(mem_ctx, &state->rec);
 540         return 0;
 541 }
 542
 543 /*
 544   Dispatch one messaging_rec
 545 */
 546 void messaging_dispatch_rec(struct messaging_context *msg_ctx,
 547                             struct messaging_rec *rec)
 548 {
 549         struct messaging_callback *cb, *next;
 550         unsigned i;
 551
 552         for (cb = msg_ctx->callbacks; cb != NULL; cb = next) {
 553                 next = cb->next;
 554                 if (cb->msg_type == rec->msg_type) {
 555                         cb->fn(msg_ctx, cb->private_data, rec->msg_type,
 556                                rec->src, &rec->buf);
 557                         /* we continue looking for matching messages
 558                            after finding one. This matters for
 559                            subsystems like the internal notify code
 560                            which register more than one handler for
 561                            the same message type */
 562                 }
 563         }
 564
 565         for (i=0; i<msg_ctx->num_waiters; i++) {
 566                 struct tevent_req *req = msg_ctx->waiters[i];
 567                 struct messaging_read_state *state = tevent_req_data(
 568                         req, struct messaging_read_state);
 569
 570                 if (state->msg_type == rec->msg_type) {
 571                         messaging_read_done(req, rec);
 572                 }
 573         }
 574         return;
 575 }
 576
 577 /** @} **/