Changeset 988 for vendor/current/source4/lib/messaging

Timestamp:

Nov 24, 2016, 1:14:11 PM (9 years ago)

Author:

Silvan Scherrer

Message:

Samba Server: update vendor to version 4.4.3

Location:

vendor/current/source4/lib/messaging

Files:

• irpc.h (modified) (2 diffs)
• messaging.c (modified) (29 diffs)
• messaging.h (modified) (1 diff)
• pymessaging.c (modified) (16 diffs)
• tests/irpc.c (modified) (8 diffs)
• tests/messaging.c (modified) (8 diffs)
• wscript_build (modified) (1 diff)

vendor/current/source4/lib/messaging/irpc.h

@@ -36 +36 @@
         bool defer_reply;
         bool no_reply;
-        struct messaging_context *msg_ctx;
+        struct imessaging_context *msg_ctx;
         struct irpc_list *irpc;
         void *data;
-        struct tevent_context *ev;
 };
 
@@ -59 +58 @@
 struct ndr_interface_table;
 
-NTSTATUS irpc_register(struct messaging_context *msg_ctx, 
+NTSTATUS irpc_register(struct imessaging_context *msg_ctx,
                        const struct ndr_interface_table *table, 
                        int call, irpc_function_t fn, void *private_data);
 
 struct dcerpc_binding_handle *irpc_binding_handle(TALLOC_CTX *mem_ctx,
-                                        struct messaging_context *msg_ctx,
-                                        struct server_id server_id,
-                                        const struct ndr_interface_table *table);
+                                                  struct imessaging_context *msg_ctx,
+                                                  struct server_id server_id,
+                                                  const struct ndr_interface_table *table);
 struct dcerpc_binding_handle *irpc_binding_handle_by_name(TALLOC_CTX *mem_ctx,
-                                        struct messaging_context *msg_ctx,
-                                        const char *dest_task,
-                                        const struct ndr_interface_table *table);
+                                                          struct imessaging_context *msg_ctx,
+                                                          const char *dest_task,
+                                                          const struct ndr_interface_table *table);
 void irpc_binding_handle_add_security_token(struct dcerpc_binding_handle *h,
                                             struct security_token *token);
 
-NTSTATUS irpc_add_name(struct messaging_context *msg_ctx, const char *name);
-struct server_id *irpc_servers_byname(struct messaging_context *msg_ctx, TALLOC_CTX *mem_ctx, const char *name);
-void irpc_remove_name(struct messaging_context *msg_ctx, const char *name);
+NTSTATUS irpc_add_name(struct imessaging_context *msg_ctx, const char *name);
+NTSTATUS irpc_servers_byname(struct imessaging_context *msg_ctx,
+                             TALLOC_CTX *mem_ctx, const char *name,
+                             unsigned *num_servers,
+                             struct server_id **servers);
+struct irpc_name_records *irpc_all_servers(struct imessaging_context *msg_ctx,
+                                           TALLOC_CTX *mem_ctx);
+void irpc_remove_name(struct imessaging_context *msg_ctx, const char *name);
 NTSTATUS irpc_send_reply(struct irpc_message *m, NTSTATUS status);
 

vendor/current/source4/lib/messaging/messaging.c

@@ -1 @@
-/* 
+/*
    Unix SMB/CIFS implementation.
 
@@ -5 +5 @@
 
    Copyright (C) Andrew Tridgell 2004
-   
+
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation; either version 3 of the License, or
    (at your option) any later version.
-   
+
    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.
-   
+
    You should have received a copy of the GNU General Public License
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
@@ -28 +28 @@
 #include "librpc/gen_ndr/ndr_irpc.h"
 #include "lib/messaging/irpc.h"
-#include "lib/util/tdb_wrap.h"
 #include "../lib/util/unix_privs.h"
 #include "librpc/rpc/dcerpc.h"
-#include <tdb.h>
-#include "../lib/util/util_tdb.h"
 #include "cluster/cluster.h"
 #include "../lib/util/tevent_ntstatus.h"
+#include "lib/param/param.h"
+#include "lib/util/server_id_db.h"
+#include "lib/util/talloc_report.h"
+#include "../source3/lib/messages_dgm.h"
+#include "../source3/lib/messages_dgm_ref.h"
+#include "../source3/lib/messages_util.h"
+#include <tdb.h>
 
 /* change the message version with any incompatible changes in the protocol */
-#define MESSAGING_VERSION 1
+#define IMESSAGING_VERSION 1
 
 /*
@@ -43 +47 @@
 */
 struct irpc_request {
-        struct messaging_context *msg_ctx;
+        struct imessaging_context *msg_ctx;
         int callid;
         struct {
@@ -51 +55 @@
 };
 
-struct messaging_context {
+struct imessaging_context {
+        struct imessaging_context *prev, *next;
         struct server_id server_id;
-        struct socket_context *sock;
-        const char *base_path;
-        const char *path;
+        const char *sock_dir;
+        const char *lock_dir;
         struct dispatch_fn **dispatch;
         uint32_t num_types;
         struct idr_context *dispatch_tree;
-        struct messaging_rec *pending;
-        struct messaging_rec *retry_queue;
         struct irpc_list *irpc;
         struct idr_context *idr;
-        const char **names;
+        struct server_id_db *names;
         struct timeval start_time;
-        struct tevent_timer *retry_te;
-        struct {
-                struct tevent_context *ev;
-                struct tevent_fd *fde;
-        } event;
+        void *msg_dgm_ref;
 };
 
@@ -82 +80 @@
 
 /* an individual message */
-struct messaging_rec {
-        struct messaging_rec *next, *prev;
-        struct messaging_context *msg;
-        const char *path;
-
-        struct messaging_header {
-                uint32_t version;
-                uint32_t msg_type;
-                struct server_id from;
-                struct server_id to;
-                uint32_t length;
-        } *header;
-
-        DATA_BLOB packet;
-        uint32_t retries;
-};
-
-
-static void irpc_handler(struct messaging_context *, void *, 
+
+static void irpc_handler(struct imessaging_context *, void *,
                          uint32_t, struct server_id, DATA_BLOB *);
 
@@ -107 +88 @@
  A useful function for testing the message system.
 */
-static void ping_message(struct messaging_context *msg, void *private_data,
+static void ping_message(struct imessaging_context *msg, void *private_data,
                          uint32_t msg_type, struct server_id src, DATA_BLOB *data)
 {
-        DEBUG(1,("INFO: Received PING message from server %u.%u [%.*s]\n",
-                 (unsigned int)src.node, (unsigned int)src.id, (int)data->length,
+        struct server_id_buf idbuf;
+        DEBUG(1,("INFO: Received PING message from server %s [%.*s]\n",
+                 server_id_str_buf(src, &idbuf), (int)data->length,
                  data->data?(const char *)data->data:""));
-        messaging_send(msg, src, MSG_PONG, data);
+        imessaging_send(msg, src, MSG_PONG, data);
+}
+
+static void pool_message(struct imessaging_context *msg, void *private_data,
+                         uint32_t msg_type, struct server_id src,
+                         DATA_BLOB *data)
+{
+        char *report;
+
+        report = talloc_report_str(msg, NULL);
+
+        if (report != NULL) {
+                DATA_BLOB blob = { .data = (uint8_t *)report,
+                                   .length = talloc_get_size(report) - 1};
+                imessaging_send(msg, src, MSG_POOL_USAGE, &blob);
+        }
+        talloc_free(report);
 }
 
@@ -119 +117 @@
   return uptime of messaging server via irpc
 */
-static NTSTATUS irpc_uptime(struct irpc_message *msg, 
+static NTSTATUS irpc_uptime(struct irpc_message *msg,
                             struct irpc_uptime *r)
 {
-        struct messaging_context *ctx = talloc_get_type(msg->private_data, struct messaging_context);
+        struct imessaging_context *ctx = talloc_get_type(msg->private_data, struct imessaging_context);
         *r->out.start_time = timeval_to_nttime(&ctx->start_time);
         return NT_STATUS_OK;
 }
 
-/* 
-   return the path to a messaging socket
-*/
-static char *messaging_path(struct messaging_context *msg, struct server_id server_id)
-{
-        TALLOC_CTX *tmp_ctx = talloc_new(msg);
-        const char *id = cluster_id_string(tmp_ctx, server_id);
-        char *s;
-        if (id == NULL) {
-                return NULL;
-        }
-        s = talloc_asprintf(msg, "%s/msg.%s", msg->base_path, id);
-        talloc_steal(s, tmp_ctx);
-        return s;
-}
-
-/*
-  dispatch a fully received message
-
-  note that this deliberately can match more than one message handler
-  per message. That allows a single messasging context to register
-  (for example) a debug handler for more than one piece of code
-*/
-static void messaging_dispatch(struct messaging_context *msg, struct messaging_rec *rec)
-{
-        struct dispatch_fn *d, *next;
-
+static struct dispatch_fn *imessaging_find_dispatch(
+        struct imessaging_context *msg, uint32_t msg_type)
+{
         /* temporary IDs use an idtree, the rest use a array of pointers */
-        if (rec->header->msg_type >= MSG_TMP_BASE) {
-                d = (struct dispatch_fn *)idr_find(msg->dispatch_tree, 
-                                                   rec->header->msg_type);
-        } else if (rec->header->msg_type < msg->num_types) {
-                d = msg->dispatch[rec->header->msg_type];
-        } else {
-                d = NULL;
-        }
-
-        for (; d; d = next) {
-                DATA_BLOB data;
-                next = d->next;
-                data.data = rec->packet.data + sizeof(*rec->header);
-                data.length = rec->header->length;
-                d->fn(msg, d->private_data, d->msg_type, rec->header->from, &data);
-        }
-        rec->header->length = 0;
-}
-
-/*
-  handler for messages that arrive from other nodes in the cluster
-*/
-static void cluster_message_handler(struct messaging_context *msg, DATA_BLOB packet)
-{
-        struct messaging_rec *rec;
-
-        rec = talloc(msg, struct messaging_rec);
-        if (rec == NULL) {
-                smb_panic("Unable to allocate messaging_rec");
-        }
-
-        rec->msg           = msg;
-        rec->path          = msg->path;
-        rec->header        = (struct messaging_header *)packet.data;
-        rec->packet        = packet;
-        rec->retries       = 0;
-
-        if (packet.length != sizeof(*rec->header) + rec->header->length) {
-                DEBUG(0,("messaging: bad message header size %d should be %d\n", 
-                         rec->header->length, (int)(packet.length - sizeof(*rec->header))));
-                talloc_free(rec);
-                return;
-        }
-
-        messaging_dispatch(msg, rec);
-        talloc_free(rec);
-}
-
-
-
-/*
-  try to send the message
-*/
-static NTSTATUS try_send(struct messaging_rec *rec)
-{
-        struct messaging_context *msg = rec->msg;
-        size_t nsent;
-        void *priv;
-        NTSTATUS status;
-        struct socket_address *path;
-
-        /* rec->path is the path of the *other* socket, where we want
-         * this to end up */
-        path = socket_address_from_strings(msg, msg->sock->backend_name, 
-                                           rec->path, 0);
-        if (!path) {
-                return NT_STATUS_NO_MEMORY;
-        }
-
-        /* we send with privileges so messages work from any context */
-        priv = root_privileges();
-        status = socket_sendto(msg->sock, &rec->packet, &nsent, path);
-        talloc_free(path);
-        talloc_free(priv);
-
-        return status;
-}
-
-/*
-  retry backed off messages
-*/
-static void msg_retry_timer(struct tevent_context *ev, struct tevent_timer *te, 
-                            struct timeval t, void *private_data)
-{
-        struct messaging_context *msg = talloc_get_type(private_data,
-                                                        struct messaging_context);
-        msg->retry_te = NULL;
-
-        /* put the messages back on the main queue */
-        while (msg->retry_queue) {
-                struct messaging_rec *rec = msg->retry_queue;
-                DLIST_REMOVE(msg->retry_queue, rec);
-                DLIST_ADD_END(msg->pending, rec, struct messaging_rec *);
-        }
-
-        EVENT_FD_WRITEABLE(msg->event.fde);     
-}
-
-/*
-  handle a socket write event
-*/
-static void messaging_send_handler(struct messaging_context *msg)
-{
-        while (msg->pending) {
-                struct messaging_rec *rec = msg->pending;
-                NTSTATUS status;
-                status = try_send(rec);
-                if (NT_STATUS_EQUAL(status, STATUS_MORE_ENTRIES)) {
-                        rec->retries++;
-                        if (rec->retries > 3) {
-                                /* we're getting continuous write errors -
-                                   backoff this record */
-                                DLIST_REMOVE(msg->pending, rec);
-                                DLIST_ADD_END(msg->retry_queue, rec, 
-                                              struct messaging_rec *);
-                                if (msg->retry_te == NULL) {
-                                        msg->retry_te = 
-                                                event_add_timed(msg->event.ev, msg, 
-                                                                timeval_current_ofs(1, 0), 
-                                                                msg_retry_timer, msg);
-                                }
-                        }
-                        break;
-                }
-                rec->retries = 0;
-                if (!NT_STATUS_IS_OK(status)) {
-                        TALLOC_CTX *tmp_ctx = talloc_new(msg);
-                        DEBUG(1,("messaging: Lost message from %s to %s of type %u - %s\n", 
-                                 cluster_id_string(tmp_ctx, rec->header->from),
-                                 cluster_id_string(tmp_ctx, rec->header->to),
-                                 rec->header->msg_type, 
-                                 nt_errstr(status)));
-                        talloc_free(tmp_ctx);
-                }
-                DLIST_REMOVE(msg->pending, rec);
-                talloc_free(rec);
-        }
-        if (msg->pending == NULL) {
-                EVENT_FD_NOT_WRITEABLE(msg->event.fde);
-        }
-}
-
-/*
-  handle a new incoming packet
-*/
-static void messaging_recv_handler(struct messaging_context *msg)
-{
-        struct messaging_rec *rec;
-        NTSTATUS status;
-        DATA_BLOB packet;
-        size_t msize;
-
-        /* see how many bytes are in the next packet */
-        status = socket_pending(msg->sock, &msize);
-        if (!NT_STATUS_IS_OK(status)) {
-                DEBUG(0,("socket_pending failed in messaging - %s\n", 
-                         nt_errstr(status)));
-                return;
-        }
-        
-        packet = data_blob_talloc(msg, NULL, msize);
-        if (packet.data == NULL) {
-                /* assume this is temporary and retry */
-                return;
-        }
-            
-        status = socket_recv(msg->sock, packet.data, msize, &msize);
-        if (!NT_STATUS_IS_OK(status)) {
-                data_blob_free(&packet);
-                return;
-        }
-
-        if (msize < sizeof(*rec->header)) {
-                DEBUG(0,("messaging: bad message of size %d\n", (int)msize));
-                data_blob_free(&packet);
-                return;
-        }
-
-        rec = talloc(msg, struct messaging_rec);
-        if (rec == NULL) {
-                smb_panic("Unable to allocate messaging_rec");
-        }
-
-        talloc_steal(rec, packet.data);
-        rec->msg           = msg;
-        rec->path          = msg->path;
-        rec->header        = (struct messaging_header *)packet.data;
-        rec->packet        = packet;
-        rec->retries       = 0;
-
-        if (msize != sizeof(*rec->header) + rec->header->length) {
-                DEBUG(0,("messaging: bad message header size %d should be %d\n", 
-                         rec->header->length, (int)(msize - sizeof(*rec->header))));
-                talloc_free(rec);
-                return;
-        }
-
-        messaging_dispatch(msg, rec);
-        talloc_free(rec);
-}
-
-
-/*
-  handle a socket event
-*/
-static void messaging_handler(struct tevent_context *ev, struct tevent_fd *fde, 
-                              uint16_t flags, void *private_data)
-{
-        struct messaging_context *msg = talloc_get_type(private_data,
-                                                        struct messaging_context);
-        if (flags & EVENT_FD_WRITE) {
-                messaging_send_handler(msg);
-        }
-        if (flags & EVENT_FD_READ) {
-                messaging_recv_handler(msg);
-        }
-}
-
+        if (msg_type >= MSG_TMP_BASE) {
+                return (struct dispatch_fn *)idr_find(msg->dispatch_tree,
+                                                      msg_type);
+        }
+        if (msg_type < msg->num_types) {
+                return msg->dispatch[msg_type];
+        }
+        return NULL;
+}
 
 /*
   Register a dispatch function for a particular message type.
 */
-NTSTATUS messaging_register(struct messaging_context *msg, void *private_data,
+NTSTATUS imessaging_register(struct imessaging_context *msg, void *private_data,
                             uint32_t msg_type, msg_callback_t fn)
 {
@@ -410 +175 @@
   above MSG_TMP_BASE
 */
-NTSTATUS messaging_register_tmp(struct messaging_context *msg, void *private_data,
+NTSTATUS imessaging_register_tmp(struct imessaging_context *msg, void *private_data,
                                 msg_callback_t fn, uint32_t *msg_type)
 {
@@ -436 +201 @@
   De-register the function for a particular message type.
 */
-void messaging_deregister(struct messaging_context *msg, uint32_t msg_type, void *private_data)
+void imessaging_deregister(struct imessaging_context *msg, uint32_t msg_type, void *private_data)
 {
         struct dispatch_fn *d, *next;
 
         if (msg_type >= msg->num_types) {
-                d = (struct dispatch_fn *)idr_find(msg->dispatch_tree, 
+                d = (struct dispatch_fn *)idr_find(msg->dispatch_tree,
                                                    msg_type);
                 if (!d) return;
@@ -461 +226 @@
   Send a message to a particular server
 */
-NTSTATUS messaging_send(struct messaging_context *msg, struct server_id server, 
+NTSTATUS imessaging_send(struct imessaging_context *msg, struct server_id server,
                         uint32_t msg_type, const DATA_BLOB *data)
 {
-        struct messaging_rec *rec;
-        NTSTATUS status;
-        size_t dlength = data?data->length:0;
-
-        rec = talloc(msg, struct messaging_rec);
-        if (rec == NULL) {
-                return NT_STATUS_NO_MEMORY;
-        }
-
-        rec->packet = data_blob_talloc(rec, NULL, sizeof(*rec->header) + dlength);
-        if (rec->packet.data == NULL) {
-                talloc_free(rec);
-                return NT_STATUS_NO_MEMORY;
-        }
-
-        rec->retries       = 0;
-        rec->msg              = msg;
-        rec->header           = (struct messaging_header *)rec->packet.data;
-        /* zero padding */
-        ZERO_STRUCTP(rec->header);
-        rec->header->version  = MESSAGING_VERSION;
-        rec->header->msg_type = msg_type;
-        rec->header->from     = msg->server_id;
-        rec->header->to       = server;
-        rec->header->length   = dlength;
-        if (dlength != 0) {
-                memcpy(rec->packet.data + sizeof(*rec->header), 
-                       data->data, dlength);
-        }
+        uint8_t hdr[MESSAGE_HDR_LENGTH];
+        struct iovec iov[2];
+        int num_iov, ret;
+        pid_t pid;
+        void *priv;
 
         if (!cluster_node_equal(&msg->server_id, &server)) {
-                /* the destination is on another node - dispatch via
-                   the cluster layer */
-                status = cluster_message_send(server, &rec->packet);
-                talloc_free(rec);
-                return status;
-        }
-
-        rec->path = messaging_path(msg, server);
-        talloc_steal(rec, rec->path);
-
-        if (msg->pending != NULL) {
-                status = STATUS_MORE_ENTRIES;
-        } else {
-                status = try_send(rec);
-        }
-
-        if (NT_STATUS_EQUAL(status, STATUS_MORE_ENTRIES)) {
-                if (msg->pending == NULL) {
-                        EVENT_FD_WRITEABLE(msg->event.fde);
-                }
-                DLIST_ADD_END(msg->pending, rec, struct messaging_rec *);
+                /* No cluster in source4... */
                 return NT_STATUS_OK;
         }
 
-        talloc_free(rec);
-
-        return status;
+        message_hdr_put(hdr, msg_type, msg->server_id, server);
+
+        iov[0] = (struct iovec) { .iov_base = &hdr, .iov_len = sizeof(hdr) };
+        num_iov = 1;
+
+        if (data != NULL) {
+                iov[1] = (struct iovec) { .iov_base = data->data,
+                                          .iov_len = data->length };
+                num_iov += 1;
+        }
+
+        pid = server.pid;
+        if (pid == 0) {
+                pid = getpid();
+        }
+
+        priv = root_privileges();
+        ret = messaging_dgm_send(pid, iov, num_iov, NULL, 0);
+        TALLOC_FREE(priv);
+        if (ret != 0) {
+                return map_nt_error_from_unix_common(ret);
+        }
+        return NT_STATUS_OK;
 }
 
@@ -527 +268 @@
   Send a message to a particular server, with the message containing a single pointer
 */
-NTSTATUS messaging_send_ptr(struct messaging_context *msg, struct server_id server, 
+NTSTATUS imessaging_send_ptr(struct imessaging_context *msg, struct server_id server,
                             uint32_t msg_type, void *ptr)
 {
@@ -535 +276 @@
         blob.length = sizeof(void *);
 
-        return messaging_send(msg, server, msg_type, &blob);
-}
-
-
-/*
-  destroy the messaging context
-*/
-static int messaging_destructor(struct messaging_context *msg)
-{
-        unlink(msg->path);
-        while (msg->names && msg->names[0]) {
-                irpc_remove_name(msg, msg->names[0]);
+        return imessaging_send(msg, server, msg_type, &blob);
+}
+
+
+/*
+  remove our messaging socket and database entry
+*/
+int imessaging_cleanup(struct imessaging_context *msg)
+{
+        if (!msg) {
+                return 0;
         }
         return 0;
 }
 
+static void imessaging_dgm_recv(const uint8_t *buf, size_t buf_len,
+                                int *fds, size_t num_fds,
+                                void *private_data);
+
 /*
   create the listening socket and setup the dispatcher
-*/
-struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx, 
-                                         const char *dir,
-                                         struct server_id server_id, 
-                                         struct tevent_context *ev)
-{
-        struct messaging_context *msg;
-        NTSTATUS status;
-        struct socket_address *path;
+
+  use auto_remove=true when you want a destructor to remove the
+  associated messaging socket and database entry on talloc free. Don't
+  use this in processes that may fork and a child may talloc free this
+  memory
+*/
+struct imessaging_context *imessaging_init(TALLOC_CTX *mem_ctx,
+                                           struct loadparm_context *lp_ctx,
+                                           struct server_id server_id,
+                                           struct tevent_context *ev,
+                                           bool auto_remove)
+{
+        struct imessaging_context *msg;
+        bool ok;
+        int ret;
+        const char *lock_dir = NULL;
+        int tdb_flags = TDB_INCOMPATIBLE_HASH | TDB_CLEAR_IF_FIRST;
 
         if (ev == NULL) {
@@ -567 +319 @@
         }
 
-        msg = talloc_zero(mem_ctx, struct messaging_context);
+        msg = talloc_zero(mem_ctx, struct imessaging_context);
         if (msg == NULL) {
                 return NULL;
         }
 
-        /* setup a handler for messages from other cluster nodes, if appropriate */
-        status = cluster_message_init(msg, server_id, cluster_message_handler);
-        if (!NT_STATUS_IS_OK(status)) {
-                talloc_free(msg);
-                return NULL;
-        }
-
         /* create the messaging directory if needed */
-        mkdir(dir, 0700);
-
-        msg->base_path     = talloc_reference(msg, dir);
-        msg->path          = messaging_path(msg, server_id);
+
+        lock_dir = lpcfg_lock_directory(lp_ctx);
+        if (lock_dir == NULL) {
+                goto fail;
+        }
+
+        msg->sock_dir = lpcfg_private_path(msg, lp_ctx, "msg.sock");
+        if (msg->sock_dir == NULL) {
+                goto fail;
+        }
+        ok = directory_create_or_exist_strict(msg->sock_dir, geteuid(), 0700);
+        if (!ok) {
+                goto fail;
+        }
+
+        msg->lock_dir = lpcfg_lock_path(msg, lp_ctx, "msg.lock");
+        if (msg->lock_dir == NULL) {
+                goto fail;
+        }
+        ok = directory_create_or_exist_strict(msg->lock_dir, geteuid(), 0755);
+        if (!ok) {
+                goto fail;
+        }
+
+        msg->msg_dgm_ref = messaging_dgm_ref(
+                msg, ev, &server_id.unique_id, msg->sock_dir, msg->lock_dir,
+                imessaging_dgm_recv, msg, &ret);
+
+        if (msg->msg_dgm_ref == NULL) {
+                goto fail;
+        }
+
         msg->server_id     = server_id;
         msg->idr           = idr_init(msg);
+        if (msg->idr == NULL) {
+                goto fail;
+        }
+
         msg->dispatch_tree = idr_init(msg);
+        if (msg->dispatch_tree == NULL) {
+                goto fail;
+        }
+
         msg->start_time    = timeval_current();
 
-        status = socket_create("unix", SOCKET_TYPE_DGRAM, &msg->sock, 0);
-        if (!NT_STATUS_IS_OK(status)) {
-                talloc_free(msg);
-                return NULL;
-        }
-
-        /* by stealing here we ensure that the socket is cleaned up (and even 
-           deleted) on exit */
-        talloc_steal(msg, msg->sock);
-
-        path = socket_address_from_strings(msg, msg->sock->backend_name, 
-                                           msg->path, 0);
-        if (!path) {
-                talloc_free(msg);
-                return NULL;
-        }
-
-        status = socket_listen(msg->sock, path, 50, 0);
-        if (!NT_STATUS_IS_OK(status)) {
-                DEBUG(0,("Unable to setup messaging listener for '%s':%s\n", msg->path, nt_errstr(status)));
-                talloc_free(msg);
-                return NULL;
-        }
-
-        /* it needs to be non blocking for sends */
-        set_blocking(socket_get_fd(msg->sock), false);
-
-        msg->event.ev   = ev;
-        msg->event.fde  = event_add_fd(ev, msg, socket_get_fd(msg->sock), 
-                                       EVENT_FD_READ, messaging_handler, msg);
-        tevent_fd_set_auto_close(msg->event.fde);
-
-        talloc_set_destructor(msg, messaging_destructor);
-        
-        messaging_register(msg, NULL, MSG_PING, ping_message);
-        messaging_register(msg, NULL, MSG_IRPC, irpc_handler);
+        tdb_flags |= lpcfg_tdb_flags(lp_ctx, 0);
+
+        msg->names = server_id_db_init(msg, server_id, lock_dir, 0, tdb_flags);
+        if (msg->names == NULL) {
+                goto fail;
+        }
+
+        if (auto_remove) {
+                talloc_set_destructor(msg, imessaging_cleanup);
+        }
+
+        imessaging_register(msg, NULL, MSG_PING, ping_message);
+        imessaging_register(msg, NULL, MSG_REQ_POOL_USAGE, pool_message);
+        imessaging_register(msg, NULL, MSG_IRPC, irpc_handler);
         IRPC_REGISTER(msg, irpc, IRPC_UPTIME, irpc_uptime, msg);
 
         return msg;
-}
-
-/* 
-   A hack, for the short term until we get 'client only' messaging in place 
-*/
-struct messaging_context *messaging_client_init(TALLOC_CTX *mem_ctx, 
-                                                const char *dir,
+fail:
+        talloc_free(msg);
+        return NULL;
+}
+
+static void imessaging_dgm_recv(const uint8_t *buf, size_t buf_len,
+                                int *fds, size_t num_fds,
+                                void *private_data)
+{
+        struct imessaging_context *msg = talloc_get_type_abort(
+                private_data, struct imessaging_context);
+        uint32_t msg_type;
+        struct server_id src, dst;
+        struct server_id_buf srcbuf, dstbuf;
+        DATA_BLOB data;
+
+        if (buf_len < MESSAGE_HDR_LENGTH) {
+                /* Invalid message, ignore */
+                return;
+        }
+
+        message_hdr_get(&msg_type, &src, &dst, buf);
+
+        data.data = discard_const_p(uint8_t, buf + MESSAGE_HDR_LENGTH);
+        data.length = buf_len - MESSAGE_HDR_LENGTH;
+
+        if ((cluster_id_equal(&dst, &msg->server_id)) ||
+            ((dst.task_id == 0) && (msg->server_id.pid == 0))) {
+                struct dispatch_fn *d, *next;
+
+                DEBUG(10, ("%s: dst %s matches my id: %s, type=0x%x\n",
+                           __func__,
+                           server_id_str_buf(dst, &dstbuf),
+                           server_id_str_buf(msg->server_id, &srcbuf),
+                           (unsigned)msg_type));
+
+                d = imessaging_find_dispatch(msg, msg_type);
+
+                for (; d; d = next) {
+                        next = d->next;
+                        d->fn(msg, d->private_data, d->msg_type, src, &data);
+                }
+        } else {
+                DEBUG(10, ("%s: Ignoring type=0x%x dst %s, I am %s, \n",
+                           __func__, (unsigned)msg_type,
+                           server_id_str_buf(dst, &dstbuf),
+                           server_id_str_buf(msg->server_id, &srcbuf)));
+        }
+}
+
+/*
+   A hack, for the short term until we get 'client only' messaging in place
+*/
+struct imessaging_context *imessaging_client_init(TALLOC_CTX *mem_ctx,
+                                                  struct loadparm_context *lp_ctx,
                                                 struct tevent_context *ev)
 {
         struct server_id id;
         ZERO_STRUCT(id);
-        id.id = random() % 0x10000000;
-        return messaging_init(mem_ctx, dir, id, ev);
+        id.pid = getpid();
+        id.task_id = generate_random();
+        id.vnn = NONCLUSTER_VNN;
+
+        /* This is because we are not in the s3 serverid database */
+        id.unique_id = SERVERID_UNIQUE_ID_NOT_TO_VERIFY;
+
+        return imessaging_init(mem_ctx, lp_ctx, id, ev, true);
 }
 /*
@@ -658 +471 @@
   register a irpc server function
 */
-NTSTATUS irpc_register(struct messaging_context *msg_ctx, 
-                       const struct ndr_interface_table *table, 
+NTSTATUS irpc_register(struct imessaging_context *msg_ctx,
+                       const struct ndr_interface_table *table,
                        int callnum, irpc_function_t fn, void *private_data)
 {
@@ -689 +502 @@
   handle an incoming irpc reply message
 */
-static void irpc_handler_reply(struct messaging_context *msg_ctx, struct irpc_message *m)
+static void irpc_handler_reply(struct imessaging_context *msg_ctx, struct irpc_message *m)
 {
         struct irpc_request *irpc;
@@ -735 +548 @@
         /* send the reply message */
         packet = ndr_push_blob(push);
-        status = messaging_send(m->msg_ctx, m->from, MSG_IRPC, &packet);
+        status = imessaging_send(m->msg_ctx, m->from, MSG_IRPC, &packet);
         if (!NT_STATUS_IS_OK(status)) goto failed;
 
@@ -746 +559 @@
   handle an incoming irpc request message
 */
-static void irpc_handler_request(struct messaging_context *msg_ctx, 
+static void irpc_handler_request(struct imessaging_context *msg_ctx,
                                  struct irpc_message *m)
 {
@@ -784 +597 @@
         m->irpc        = i;
         m->data        = r;
-        m->ev          = msg_ctx->event.ev;
 
         m->header.status = i->fn(m, r);
@@ -810 +622 @@
   handle an incoming irpc message
 */
-static void irpc_handler(struct messaging_context *msg_ctx, void *private_data,
+static void irpc_handler(struct imessaging_context *msg_ctx, void *private_data,
                          uint32_t msg_type, struct server_id src, DATA_BLOB *packet)
 {
@@ -855 +667 @@
 
 /*
-  open the naming database
-*/
-static struct tdb_wrap *irpc_namedb_open(struct messaging_context *msg_ctx)
-{
-        struct tdb_wrap *t;
-        char *path = talloc_asprintf(msg_ctx, "%s/names.tdb", msg_ctx->base_path);
-        if (path == NULL) {
+  add a string name that this irpc server can be called on
+*/
+NTSTATUS irpc_add_name(struct imessaging_context *msg_ctx, const char *name)
+{
+        int ret;
+
+        ret = server_id_db_add(msg_ctx->names, name);
+        if (ret != 0) {
+                return map_nt_error_from_unix_common(ret);
+        }
+        return NT_STATUS_OK;
+}
+
+/*
+  return a list of server ids for a server name
+*/
+NTSTATUS irpc_servers_byname(struct imessaging_context *msg_ctx,
+                             TALLOC_CTX *mem_ctx, const char *name,
+                             unsigned *num_servers,
+                             struct server_id **servers)
+{
+        int ret;
+
+        ret = server_id_db_lookup(msg_ctx->names, name, mem_ctx,
+                                  num_servers, servers);
+        if (ret != 0) {
+                return map_nt_error_from_unix_common(ret);
+        }
+        return NT_STATUS_OK;
+}
+
+static int all_servers_func(const char *name, unsigned num_servers,
+                            const struct server_id *servers,
+                            void *private_data)
+{
+        struct irpc_name_records *name_records = talloc_get_type(
+                private_data, struct irpc_name_records);
+        struct irpc_name_record *name_record;
+        int i;
+
+        name_records->names
+                = talloc_realloc(name_records, name_records->names,
+                                 struct irpc_name_record *, name_records->num_records+1);
+        if (!name_records->names) {
+                return -1;
+        }
+
+        name_records->names[name_records->num_records] = name_record
+                = talloc(name_records->names,
+                         struct irpc_name_record);
+        if (!name_record) {
+                return -1;
+        }
+
+        name_records->num_records++;
+
+        name_record->name = talloc_strdup(name_record, name);
+        if (!name_record->name) {
+                return -1;
+        }
+
+        name_record->count = num_servers;
+        name_record->ids = talloc_array(name_record, struct server_id,
+                                        num_servers);
+        if (name_record->ids == NULL) {
+                return -1;
+        }
+        for (i=0;i<name_record->count;i++) {
+                name_record->ids[i] = servers[i];
+        }
+        return 0;
+}
+
+/*
+  return a list of server ids for a server name
+*/
+struct irpc_name_records *irpc_all_servers(struct imessaging_context *msg_ctx,
+                                           TALLOC_CTX *mem_ctx)
+{
+        int ret;
+        struct irpc_name_records *name_records = talloc_zero(mem_ctx, struct irpc_name_records);
+        if (name_records == NULL) {
                 return NULL;
         }
-        t = tdb_wrap_open(msg_ctx, path, 0, 0, O_RDWR|O_CREAT, 0660);
-        talloc_free(path);
-        return t;
-}
-        
-
-/*
-  add a string name that this irpc server can be called on
-*/
-NTSTATUS irpc_add_name(struct messaging_context *msg_ctx, const char *name)
-{
-        struct tdb_wrap *t;
-        TDB_DATA rec;
-        int count;
-        NTSTATUS status = NT_STATUS_OK;
-
-        t = irpc_namedb_open(msg_ctx);
-        NT_STATUS_HAVE_NO_MEMORY(t);
-
-        if (tdb_lock_bystring(t->tdb, name) != 0) {
-                talloc_free(t);
-                return NT_STATUS_LOCK_NOT_GRANTED;
-        }
-        rec = tdb_fetch_bystring(t->tdb, name);
-        count = rec.dsize / sizeof(struct server_id);
-        rec.dptr = (unsigned char *)realloc_p(rec.dptr, struct server_id, count+1);
-        rec.dsize += sizeof(struct server_id);
-        if (rec.dptr == NULL) {
-                tdb_unlock_bystring(t->tdb, name);
-                talloc_free(t);
-                return NT_STATUS_NO_MEMORY;
-        }
-        ((struct server_id *)rec.dptr)[count] = msg_ctx->server_id;
-        if (tdb_store_bystring(t->tdb, name, rec, 0) != 0) {
-                status = NT_STATUS_INTERNAL_ERROR;
-        }
-        free(rec.dptr);
-        tdb_unlock_bystring(t->tdb, name);
-        talloc_free(t);
-
-        msg_ctx->names = str_list_add(msg_ctx->names, name);
-        talloc_steal(msg_ctx, msg_ctx->names);
-
-        return status;
-}
-
-/*
-  return a list of server ids for a server name
-*/
-struct server_id *irpc_servers_byname(struct messaging_context *msg_ctx,
-                                      TALLOC_CTX *mem_ctx,
-                                      const char *name)
-{
-        struct tdb_wrap *t;
-        TDB_DATA rec;
-        int count, i;
-        struct server_id *ret;
-
-        t = irpc_namedb_open(msg_ctx);
-        if (t == NULL) {
+
+        ret = server_id_db_traverse_read(msg_ctx->names, all_servers_func,
+                                         name_records);
+        if (ret == -1) {
+                TALLOC_FREE(name_records);
                 return NULL;
         }
 
-        if (tdb_lock_bystring(t->tdb, name) != 0) {
-                talloc_free(t);
-                return NULL;
-        }
-        rec = tdb_fetch_bystring(t->tdb, name);
-        if (rec.dptr == NULL) {
-                tdb_unlock_bystring(t->tdb, name);
-                talloc_free(t);
-                return NULL;
-        }
-        count = rec.dsize / sizeof(struct server_id);
-        ret = talloc_array(mem_ctx, struct server_id, count+1);
-        if (ret == NULL) {
-                tdb_unlock_bystring(t->tdb, name);
-                talloc_free(t);
-                return NULL;
-        }
-        for (i=0;i<count;i++) {
-                ret[i] = ((struct server_id *)rec.dptr)[i];
-        }
-        ret[i] = cluster_id(0, 0);
-        free(rec.dptr);
-        tdb_unlock_bystring(t->tdb, name);
-        talloc_free(t);
-
-        return ret;
+        return name_records;
 }
 
@@ -958 +765 @@
   remove a name from a messaging context
 */
-void irpc_remove_name(struct messaging_context *msg_ctx, const char *name)
-{
-        struct tdb_wrap *t;
-        TDB_DATA rec;
-        int count, i;
-        struct server_id *ids;
-
-        str_list_remove(msg_ctx->names, name);
-
-        t = irpc_namedb_open(msg_ctx);
-        if (t == NULL) {
-                return;
-        }
-
-        if (tdb_lock_bystring(t->tdb, name) != 0) {
-                talloc_free(t);
-                return;
-        }
-        rec = tdb_fetch_bystring(t->tdb, name);
-        if (rec.dptr == NULL) {
-                tdb_unlock_bystring(t->tdb, name);
-                talloc_free(t);
-                return;
-        }
-        count = rec.dsize / sizeof(struct server_id);
-        if (count == 0) {
-                free(rec.dptr);
-                tdb_unlock_bystring(t->tdb, name);
-                talloc_free(t);
-                return;
-        }
-        ids = (struct server_id *)rec.dptr;
-        for (i=0;i<count;i++) {
-                if (cluster_id_equal(&ids[i], &msg_ctx->server_id)) {
-                        if (i < count-1) {
-                                memmove(ids+i, ids+i+1, 
-                                        sizeof(struct server_id) * (count-(i+1)));
-                        }
-                        rec.dsize -= sizeof(struct server_id);
-                        break;
-                }
-        }
-        tdb_store_bystring(t->tdb, name, rec, 0);
-        free(rec.dptr);
-        tdb_unlock_bystring(t->tdb, name);
-        talloc_free(t);
-}
-
-struct server_id messaging_get_server_id(struct messaging_context *msg_ctx)
+void irpc_remove_name(struct imessaging_context *msg_ctx, const char *name)
+{
+        server_id_db_remove(msg_ctx->names, name);
+}
+
+struct server_id imessaging_get_server_id(struct imessaging_context *msg_ctx)
 {
         return msg_ctx->server_id;
@@ -1012 +776 @@
 
 struct irpc_bh_state {
-        struct messaging_context *msg_ctx;
+        struct imessaging_context *msg_ctx;
         struct server_id server_id;
         const struct ndr_interface_table *table;
@@ -1085 +849 @@
         ok = irpc_bh_is_connected(h);
         if (!ok) {
-                tevent_req_nterror(req, NT_STATUS_INVALID_CONNECTION);
+                tevent_req_nterror(req, NT_STATUS_CONNECTION_DISCONNECTED);
                 return tevent_req_post(req, ev);
         }
@@ -1138 +902 @@
         /* and send it */
         state->in_packet = ndr_push_blob(ndr);
-        status = messaging_send(hs->msg_ctx, hs->server_id,
+        status = imessaging_send(hs->msg_ctx, hs->server_id,
                                 MSG_IRPC, &state->in_packet);
         if (!NT_STATUS_IS_OK(status)) {
@@ -1177 +941 @@
                 m->ndr->data_size - m->ndr->offset);
         if ((m->ndr->data_size - m->ndr->offset) > 0 && !state->out_data.data) {
-                tevent_req_nomem(NULL, req);
+                tevent_req_oom(req);
                 return;
         }
@@ -1229 +993 @@
         ok = irpc_bh_is_connected(h);
         if (!ok) {
-                tevent_req_nterror(req, NT_STATUS_INVALID_CONNECTION);
+                tevent_req_nterror(req, NT_STATUS_CONNECTION_DISCONNECTED);
                 return tevent_req_post(req, ev);
         }
@@ -1271 +1035 @@
 /* initialise a irpc binding handle */
 struct dcerpc_binding_handle *irpc_binding_handle(TALLOC_CTX *mem_ctx,
-                                        struct messaging_context *msg_ctx,
-                                        struct server_id server_id,
-                                        const struct ndr_interface_table *table)
+                                                  struct imessaging_context *msg_ctx,
+                                                  struct server_id server_id,
+                                                  const struct ndr_interface_table *table)
 {
         struct dcerpc_binding_handle *h;
@@ -1293 +1057 @@
         hs->timeout = IRPC_CALL_TIMEOUT;
 
-        dcerpc_binding_handle_set_sync_ev(h, msg_ctx->event.ev);
-
         return h;
 }
 
 struct dcerpc_binding_handle *irpc_binding_handle_by_name(TALLOC_CTX *mem_ctx,
-                                        struct messaging_context *msg_ctx,
-                                        const char *dest_task,
-                                        const struct ndr_interface_table *table)
+                                                          struct imessaging_context *msg_ctx,
+                                                          const char *dest_task,
+                                                          const struct ndr_interface_table *table)
 {
         struct dcerpc_binding_handle *h;
+        unsigned num_sids;
         struct server_id *sids;
         struct server_id sid;
+        NTSTATUS status;
 
         /* find the server task */
-        sids = irpc_servers_byname(msg_ctx, mem_ctx, dest_task);
-        if (sids == NULL) {
-                errno = EADDRNOTAVAIL;
-                return NULL;
-        }
-        if (sids[0].id == 0) {
-                talloc_free(sids);
+
+        status = irpc_servers_byname(msg_ctx, mem_ctx, dest_task,
+                                     &num_sids, &sids);
+        if (!NT_STATUS_IS_OK(status)) {
                 errno = EADDRNOTAVAIL;
                 return NULL;

vendor/current/source4/lib/messaging/messaging.h

@@ -19 +19 @@
 */
 
-#ifndef _MESSAGES_H_
-#define _MESSAGES_H_
+#ifndef _SOURCE4_LIB_MESSAGING_MESSAGES_H_
+#define _SOURCE4_LIB_MESSAGING_MESSAGES_H_
 
-#include "librpc/gen_ndr/server_id4.h"
+#include "librpc/gen_ndr/server_id.h"
+#include "librpc/gen_ndr/messaging.h"
 
-struct messaging_context;
-
-/* general messages */
-#define MSG_DEBUG               1
-#define MSG_PING                2
-#define MSG_PONG                3
-#define MSG_BRL_RETRY           4
-#define MSG_PVFS_RETRY_OPEN     5
-#define MSG_IRPC                6
-#define MSG_PVFS_NOTIFY         7
-#define MSG_NTVFS_OPLOCK_BREAK  8
-#define MSG_DREPL_ALLOCATE_RID  9
-
-/* temporary messaging endpoints are allocated above this line */
-#define MSG_TMP_BASE            1000
+struct imessaging_context;
 
 /* taskid for messaging of parent process */
 #define SAMBA_PARENT_TASKID     0
 
-typedef void (*msg_callback_t)(struct messaging_context *msg, void *private_data,
+typedef void (*msg_callback_t)(struct imessaging_context *msg, void *private_data,
                                uint32_t msg_type,
                                struct server_id server_id, DATA_BLOB *data);
 
-NTSTATUS messaging_send(struct messaging_context *msg, struct server_id server,
+NTSTATUS imessaging_send(struct imessaging_context *msg, struct server_id server,
                         uint32_t msg_type, const DATA_BLOB *data);
-NTSTATUS messaging_register(struct messaging_context *msg, void *private_data,
+NTSTATUS imessaging_register(struct imessaging_context *msg, void *private_data,
                             uint32_t msg_type,
                             msg_callback_t fn);
-NTSTATUS messaging_register_tmp(struct messaging_context *msg, void *private_data,
+NTSTATUS imessaging_register_tmp(struct imessaging_context *msg, void *private_data,
                                 msg_callback_t fn, uint32_t *msg_type);
-struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx,
-                                         const char *dir,
-                                         struct server_id server_id,
+struct imessaging_context *imessaging_init(TALLOC_CTX *mem_ctx,
+                                           struct loadparm_context *lp_ctx,
+                                           struct server_id server_id,
+                                           struct tevent_context *ev,
+                                           bool auto_remove);
+int imessaging_cleanup(struct imessaging_context *msg);
+struct imessaging_context *imessaging_client_init(TALLOC_CTX *mem_ctx,
+                                           struct loadparm_context *lp_ctx,
                                          struct tevent_context *ev);
-struct messaging_context *messaging_client_init(TALLOC_CTX *mem_ctx,
-                                         const char *dir,
-                                         struct tevent_context *ev);
-NTSTATUS messaging_send_ptr(struct messaging_context *msg, struct server_id server,
+NTSTATUS imessaging_send_ptr(struct imessaging_context *msg, struct server_id server,
                             uint32_t msg_type, void *ptr);
-void messaging_deregister(struct messaging_context *msg, uint32_t msg_type, void *private_data);
-struct server_id messaging_get_server_id(struct messaging_context *msg_ctx);
+void imessaging_deregister(struct imessaging_context *msg, uint32_t msg_type, void *private_data);
+struct server_id imessaging_get_server_id(struct imessaging_context *msg_ctx);
 
 #endif

vendor/current/source4/lib/messaging/pymessaging.c

@@ -22 +22 @@
 #include <Python.h>
 #include "includes.h"
-#include "scripting/python/modules.h"
+#include "python/modules.h"
 #include "libcli/util/pyerrors.h"
 #include "librpc/rpc/pyrpc_util.h"
 #include "librpc/ndr/libndr.h"
 #include "lib/messaging/messaging.h"
+#include "lib/messaging/irpc.h"
 #include "lib/events/events.h"
 #include "cluster/cluster.h"
@@ -32 +33 @@
 #include "param/pyparam.h"
 #include "librpc/rpc/dcerpc.h"
-#include "librpc/gen_ndr/server_id4.h"
+#include "librpc/gen_ndr/server_id.h"
+#include <pytalloc.h>
 
 void initmessaging(void);
 
-extern PyTypeObject messaging_Type;
+extern PyTypeObject imessaging_Type;
 
 static bool server_id_from_py(PyObject *object, struct server_id *server_id)
 {
         if (!PyTuple_Check(object)) {
-                PyErr_SetString(PyExc_ValueError, "Expected tuple");
-                return false;
-        }
-
+                if (!py_check_dcerpc_type(object, "samba.dcerpc.server_id", "server_id")) {
+
+                        PyErr_SetString(PyExc_ValueError, "Expected tuple or server_id");
+                        return false;
+                }
+                *server_id = *pytalloc_get_type(object, struct server_id);
+                return true;
+        }
         if (PyTuple_Size(object) == 3) {
-                return PyArg_ParseTuple(object, "iii", &server_id->id, &server_id->id2, &server_id->node);
+                unsigned long long pid;
+                int task_id, vnn;
+
+                if (!PyArg_ParseTuple(object, "KII", &pid, &task_id, &vnn)) {
+                        return false;
+                }
+                server_id->pid = pid;
+                server_id->task_id = task_id;
+                server_id->vnn = vnn;
+                return true;
         } else {
-                int id, id2;
-                if (!PyArg_ParseTuple(object, "ii", &id, &id2))
+                unsigned long long pid;
+                int task_id;
+                if (!PyArg_ParseTuple(object, "KI", &pid, &task_id))
                         return false;
-                *server_id = cluster_id(id, id2);
+                *server_id = cluster_id(pid, task_id);
                 return true;
         }
@@ -59 +75 @@
         PyObject_HEAD
         TALLOC_CTX *mem_ctx;
-        struct messaging_context *msg_ctx;
-} messaging_Object;
-
-static PyObject *py_messaging_connect(PyTypeObject *self, PyObject *args, PyObject *kwargs)
+        struct imessaging_context *msg_ctx;
+} imessaging_Object;
+
+static PyObject *py_imessaging_connect(PyTypeObject *self, PyObject *args, PyObject *kwargs)
 {
         struct tevent_context *ev;
-        const char *kwnames[] = { "own_id", "messaging_path", NULL };
+        const char *kwnames[] = { "own_id", "lp_ctx", NULL };
         PyObject *own_id = Py_None;
-        const char *messaging_path = NULL;
-        messaging_Object *ret;
-
-        if (!PyArg_ParseTupleAndKeywords(args, kwargs, "|Oz:connect", 
-                discard_const_p(char *, kwnames), &own_id, &messaging_path)) {
-                return NULL;
-        }
-
-        ret = PyObject_New(messaging_Object, &messaging_Type);
+        PyObject *py_lp_ctx = Py_None;
+        imessaging_Object *ret;
+        struct loadparm_context *lp_ctx;
+
+        if (!PyArg_ParseTupleAndKeywords(args, kwargs, "|OO:connect", 
+                discard_const_p(char *, kwnames), &own_id, &py_lp_ctx)) {
+                return NULL;
+        }
+
+        ret = PyObject_New(imessaging_Object, &imessaging_Type);
         if (ret == NULL)
                 return NULL;
@@ -81 +98 @@
         ret->mem_ctx = talloc_new(NULL);
 
+        lp_ctx = lpcfg_from_py_object(ret->mem_ctx, py_lp_ctx);
+        if (lp_ctx == NULL) {
+                PyErr_SetString(PyExc_RuntimeError, "imessaging_connect unable to interpret loadparm_context");
+                talloc_free(ret->mem_ctx);
+                return NULL;
+        }
+
         ev = s4_event_context_init(ret->mem_ctx);
-
-        if (messaging_path == NULL) {
-                messaging_path = lpcfg_messaging_path(ret->mem_ctx,
-                                                                   py_default_loadparm_context(ret->mem_ctx));
-        } else {
-                messaging_path = talloc_strdup(ret->mem_ctx, messaging_path);
-        }
 
         if (own_id != Py_None) {
@@ -96 +113 @@
                         return NULL;
 
-                ret->msg_ctx = messaging_init(ret->mem_ctx, 
-                                            messaging_path,
-                                            server_id,
-                                            ev);
+                ret->msg_ctx = imessaging_init(ret->mem_ctx,
+                                               lp_ctx,
+                                               server_id,
+                                               ev, true);
         } else {
-                ret->msg_ctx = messaging_client_init(ret->mem_ctx, 
-                                            messaging_path,
-                                            ev);
+                ret->msg_ctx = imessaging_client_init(ret->mem_ctx,
+                                                      lp_ctx,
+                                                      ev);
         }
 
         if (ret->msg_ctx == NULL) {
-                PyErr_SetString(PyExc_RuntimeError, "messaging_connect unable to create a messaging context");
+                PyErr_SetString(PyExc_RuntimeError, "imessaging_connect unable to create a messaging context");
                 talloc_free(ret->mem_ctx);
                 return NULL;
@@ -115 +132 @@
 }
 
-static void py_messaging_dealloc(PyObject *self)
-{
-        messaging_Object *iface = (messaging_Object *)self;
+static void py_imessaging_dealloc(PyObject *self)
+{
+        imessaging_Object *iface = (imessaging_Object *)self;
         talloc_free(iface->msg_ctx);
         self->ob_type->tp_free(self);
 }
 
-static PyObject *py_messaging_send(PyObject *self, PyObject *args, PyObject *kwargs)
-{
-        messaging_Object *iface = (messaging_Object *)self;
+static PyObject *py_imessaging_send(PyObject *self, PyObject *args, PyObject *kwargs)
+{
+        imessaging_Object *iface = (imessaging_Object *)self;
         uint32_t msg_type;
         DATA_BLOB data;
@@ -131 +148 @@
         struct server_id server;
         const char *kwnames[] = { "target", "msg_type", "data", NULL };
-        int length;
+        Py_ssize_t length;
 
         if (!PyArg_ParseTupleAndKeywords(args, kwargs, "Ois#:send", 
@@ -144 +161 @@
                 return NULL;
 
-        status = messaging_send(iface->msg_ctx, server, msg_type, &data);
+        status = imessaging_send(iface->msg_ctx, server, msg_type, &data);
         if (NT_STATUS_IS_ERR(status)) {
                 PyErr_SetNTSTATUS(status);
@@ -153 +170 @@
 }
 
-static void py_msg_callback_wrapper(struct messaging_context *msg, void *private_data,
+static void py_msg_callback_wrapper(struct imessaging_context *msg, void *private_data,
                                uint32_t msg_type, 
                                struct server_id server_id, DATA_BLOB *data)
 {
-        PyObject *callback = (PyObject *)private_data;
-
-        PyObject_CallFunction(callback, discard_const_p(char, "i(iii)s#"), msg_type, 
-                              server_id.id, server_id.id2, server_id.node, 
+        PyObject *py_server_id, *callback = (PyObject *)private_data;
+
+        struct server_id *p_server_id = talloc(NULL, struct server_id);
+        if (!p_server_id) {
+                PyErr_NoMemory();
+                return;
+        }
+        *p_server_id = server_id;
+
+        py_server_id = py_return_ndr_struct("samba.dcerpc.server_id", "server_id", p_server_id, p_server_id);
+        talloc_unlink(NULL, p_server_id);
+
+        PyObject_CallFunction(callback, discard_const_p(char, "i(O)s#"), msg_type,
+                              py_server_id,
                               data->data, data->length);
 }
 
-static PyObject *py_messaging_register(PyObject *self, PyObject *args, PyObject *kwargs)
-{
-        messaging_Object *iface = (messaging_Object *)self;
+static PyObject *py_imessaging_register(PyObject *self, PyObject *args, PyObject *kwargs)
+{
+        imessaging_Object *iface = (imessaging_Object *)self;
         int msg_type = -1;
         PyObject *callback;
@@ -181 +208 @@
         if (msg_type == -1) {
                 uint32_t msg_type32 = msg_type;
-                status = messaging_register_tmp(iface->msg_ctx, callback,
+                status = imessaging_register_tmp(iface->msg_ctx, callback,
                                                 py_msg_callback_wrapper, &msg_type32);
                 msg_type = msg_type32;
         } else {
-                status = messaging_register(iface->msg_ctx, callback,
+                status = imessaging_register(iface->msg_ctx, callback,
                                     msg_type, py_msg_callback_wrapper);
         }
@@ -196 +223 @@
 }
 
-static PyObject *py_messaging_deregister(PyObject *self, PyObject *args, PyObject *kwargs)
-{
-        messaging_Object *iface = (messaging_Object *)self;
+static PyObject *py_imessaging_deregister(PyObject *self, PyObject *args, PyObject *kwargs)
+{
+        imessaging_Object *iface = (imessaging_Object *)self;
         int msg_type = -1;
         PyObject *callback;
@@ -208 +235 @@
         }
 
-        messaging_deregister(iface->msg_ctx, msg_type, callback);
+        imessaging_deregister(iface->msg_ctx, msg_type, callback);
 
         Py_DECREF(callback);
@@ -215 +242 @@
 }
 
-static PyMethodDef py_messaging_methods[] = {
-        { "send", (PyCFunction)py_messaging_send, METH_VARARGS|METH_KEYWORDS, 
+static PyObject *py_irpc_servers_byname(PyObject *self, PyObject *args, PyObject *kwargs)
+{
+        imessaging_Object *iface = (imessaging_Object *)self;
+        char *server_name;
+        unsigned i, num_ids;
+        struct server_id *ids;
+        PyObject *pylist;
+        TALLOC_CTX *mem_ctx = talloc_new(NULL);
+        NTSTATUS status;
+
+        if (!mem_ctx) {
+                PyErr_NoMemory();
+                return NULL;
+        }
+
+        if (!PyArg_ParseTuple(args, "s", &server_name)) {
+                TALLOC_FREE(mem_ctx);
+                return NULL;
+        }
+
+        status = irpc_servers_byname(iface->msg_ctx, mem_ctx, server_name,
+                                     &num_ids, &ids);
+        if (!NT_STATUS_IS_OK(status)) {
+                TALLOC_FREE(mem_ctx);
+                PyErr_SetString(PyExc_KeyError, "No such name");
+                return NULL;
+        }
+
+        pylist = PyList_New(num_ids);
+        if (pylist == NULL) {
+                TALLOC_FREE(mem_ctx);
+                PyErr_NoMemory();
+                return NULL;
+        }
+        for (i = 0; i < num_ids; i++) {
+                PyObject *py_server_id;
+                struct server_id *p_server_id = talloc(NULL, struct server_id);
+                if (!p_server_id) {
+                        PyErr_NoMemory();
+                        return NULL;
+                }
+                *p_server_id = ids[i];
+
+                py_server_id = py_return_ndr_struct("samba.dcerpc.server_id", "server_id", p_server_id, p_server_id);
+                if (!py_server_id) {
+                        return NULL;
+                }
+                PyList_SetItem(pylist, i, py_server_id);
+                talloc_unlink(NULL, p_server_id);
+        }
+        TALLOC_FREE(mem_ctx);
+        return pylist;
+}
+
+static PyObject *py_irpc_all_servers(PyObject *self, PyObject *args, PyObject *kwargs)
+{
+        imessaging_Object *iface = (imessaging_Object *)self;
+        PyObject *pylist;
+        int i;
+        struct irpc_name_records *records;
+        TALLOC_CTX *mem_ctx = talloc_new(NULL);
+        if (!mem_ctx) {
+                PyErr_NoMemory();
+                return NULL;
+        }
+
+        records = irpc_all_servers(iface->msg_ctx, mem_ctx);
+        if (records == NULL) {
+                return NULL;
+        }
+
+        pylist = PyList_New(records->num_records);
+        if (pylist == NULL) {
+                TALLOC_FREE(mem_ctx);
+                PyErr_NoMemory();
+                return NULL;
+        }
+        for (i = 0; i < records->num_records; i++) {
+                PyObject *py_name_record
+                        = py_return_ndr_struct("samba.dcerpc.irpc",
+                                               "name_record",
+                                               records->names[i],
+                                               records->names[i]);
+                if (!py_name_record) {
+                        return NULL;
+                }
+                PyList_SetItem(pylist, i,
+                               py_name_record);
+        }
+        TALLOC_FREE(mem_ctx);
+        return pylist;
+}
+
+static PyMethodDef py_imessaging_methods[] = {
+        { "send", (PyCFunction)py_imessaging_send, METH_VARARGS|METH_KEYWORDS,
                 "S.send(target, msg_type, data) -> None\nSend a message" },
-        { "register", (PyCFunction)py_messaging_register, METH_VARARGS|METH_KEYWORDS,
+        { "register", (PyCFunction)py_imessaging_register, METH_VARARGS|METH_KEYWORDS,
                 "S.register(callback, msg_type=None) -> msg_type\nRegister a message handler" },
-        { "deregister", (PyCFunction)py_messaging_deregister, METH_VARARGS|METH_KEYWORDS,
+        { "deregister", (PyCFunction)py_imessaging_deregister, METH_VARARGS|METH_KEYWORDS,
                 "S.deregister(callback, msg_type) -> None\nDeregister a message handler" },
+        { "irpc_servers_byname", (PyCFunction)py_irpc_servers_byname, METH_VARARGS,
+                "S.irpc_servers_byname(name) -> list\nGet list of server_id values that are registered for a particular name" },
+        { "irpc_all_servers", (PyCFunction)py_irpc_all_servers, METH_NOARGS,
+                "S.irpc_servers_byname() -> list\nGet list of all registered names and the associated server_id values" },
         { NULL, NULL, 0, NULL }
 };
 
-static PyObject *py_messaging_server_id(PyObject *obj, void *closure)
-{
-        messaging_Object *iface = (messaging_Object *)obj;
-        struct server_id server_id = messaging_get_server_id(iface->msg_ctx);
-
-        return Py_BuildValue("(iii)", server_id.id, server_id.id2, 
-                             server_id.node);
-}
-
-static PyGetSetDef py_messaging_getset[] = {
-        { discard_const_p(char, "server_id"), py_messaging_server_id, NULL, 
+static PyObject *py_imessaging_server_id(PyObject *obj, void *closure)
+{
+        imessaging_Object *iface = (imessaging_Object *)obj;
+        PyObject *py_server_id;
+        struct server_id server_id = imessaging_get_server_id(iface->msg_ctx);
+        struct server_id *p_server_id = talloc(NULL, struct server_id);
+        if (!p_server_id) {
+                PyErr_NoMemory();
+                return NULL;
+        }
+        *p_server_id = server_id;
+
+        py_server_id = py_return_ndr_struct("samba.dcerpc.server_id", "server_id", p_server_id, p_server_id);
+        talloc_unlink(NULL, p_server_id);
+
+        return py_server_id;
+}
+
+static PyGetSetDef py_imessaging_getset[] = {
+        { discard_const_p(char, "server_id"), py_imessaging_server_id, NULL,
           discard_const_p(char, "local server id") },
         { NULL },
@@ -241 +374 @@
 
 
-PyTypeObject messaging_Type = {
+PyTypeObject imessaging_Type = {
         PyObject_HEAD_INIT(NULL) 0,
         .tp_name = "messaging.Messaging",
-        .tp_basicsize = sizeof(messaging_Object),
+        .tp_basicsize = sizeof(imessaging_Object),
         .tp_flags = Py_TPFLAGS_DEFAULT|Py_TPFLAGS_BASETYPE,
-        .tp_new = py_messaging_connect,
-        .tp_dealloc = py_messaging_dealloc,
-        .tp_methods = py_messaging_methods,
-        .tp_getset = py_messaging_getset,
-        .tp_doc = "Messaging(own_id=None, messaging_path=None)\n" \
-                  "Create a new object that can be used to communicate with the peers in the specified messaging path.\n" \
-                  "If no path is specified, the default path from smb.conf will be used."
+        .tp_new = py_imessaging_connect,
+        .tp_dealloc = py_imessaging_dealloc,
+        .tp_methods = py_imessaging_methods,
+        .tp_getset = py_imessaging_getset,
+        .tp_doc = "Messaging(own_id=None)\n" \
+                  "Create a new object that can be used to communicate with the peers in the specified messaging path.\n"
 };
 
@@ -259 +391 @@
         PyObject *mod;
 
-        if (PyType_Ready(&messaging_Type) < 0)
+        if (PyType_Ready(&imessaging_Type) < 0)
                 return;
 
@@ -266 +398 @@
                 return;
 
-        Py_INCREF((PyObject *)&messaging_Type);
-        PyModule_AddObject(mod, "Messaging", (PyObject *)&messaging_Type);
-}
+        Py_INCREF((PyObject *)&imessaging_Type);
+        PyModule_AddObject(mod, "Messaging", (PyObject *)&imessaging_Type);
+}

vendor/current/source4/lib/messaging/tests/irpc.c

@@ -28 +28 @@
 #include "cluster/cluster.h"
 #include "param/param.h"
+#include "torture/local/proto.h"
 
 const uint32_t MSG_ID1 = 1, MSG_ID2 = 2;
@@ -35 +36 @@
 struct irpc_test_data
 {
-        struct messaging_context *msg_ctx1, *msg_ctx2;
+        struct imessaging_context *msg_ctx1, *msg_ctx2;
         struct tevent_context *ev;
 };
@@ -74 +75 @@
 static NTSTATUS irpc_EchoData(struct irpc_message *irpc, struct echo_EchoData *r)
 {
+        struct irpc_test_data *data = talloc_get_type_abort(irpc->private_data, struct irpc_test_data);
         irpc->defer_reply = true;
-        event_add_timed(irpc->ev, irpc, timeval_zero(), deferred_echodata, irpc);
+        tevent_add_timer(data->ev, irpc, timeval_zero(), deferred_echodata, irpc);
         return NT_STATUS_OK;
 }
@@ -101 +103 @@
 
         test_debug = true;
+        /*
+         * Note: this makes use of nested event loops
+         * as client and server use the same loop.
+         */
+        dcerpc_binding_handle_set_sync_ev(irpc_handle, data->ev);
         status = dcerpc_echo_AddOne_r(irpc_handle, test, &r);
         test_debug = false;
@@ -135 +142 @@
         r.in.len = strlen((char *)r.in.in_data);
 
+        /*
+         * Note: this makes use of nested event loops
+         * as client and server use the same loop.
+         */
+        dcerpc_binding_handle_set_sync_ev(irpc_handle, data->ev);
         status = dcerpc_echo_EchoData_r(irpc_handle, mem_ctx, &r);
         torture_assert_ntstatus_ok(tctx, status, "EchoData failed");
@@ -219 +231 @@
 
                 while (ping_count > pong_count + 20) {
-                        event_loop_once(data->ev);
+                        tevent_loop_once(data->ev);
                 }
         }
@@ -226 +238 @@
                ping_count - pong_count, pong_count);
         while (timeval_elapsed(&tv) < 30 && pong_count < ping_count) {
-                event_loop_once(data->ev);
+                tevent_loop_once(data->ev);
         }
 
@@ -247 +259 @@
         data->ev = tctx->ev;
         torture_assert(tctx, data->msg_ctx1 = 
-                       messaging_init(tctx, 
-                                      lpcfg_messaging_path(tctx, tctx->lp_ctx),
+                       imessaging_init(tctx,
+                                      tctx->lp_ctx,
                                       cluster_id(0, MSG_ID1),
-                                      data->ev),
+                                      data->ev, true),
                        "Failed to init first messaging context");
 
         torture_assert(tctx, data->msg_ctx2 = 
-                       messaging_init(tctx, 
-                                      lpcfg_messaging_path(tctx, tctx->lp_ctx),
+                       imessaging_init(tctx,
+                                      tctx->lp_ctx,
                                       cluster_id(0, MSG_ID2), 
-                                      data->ev),
+                                      data->ev, true),
                        "Failed to init second messaging context");
 
         /* register the server side function */
-        IRPC_REGISTER(data->msg_ctx1, rpcecho, ECHO_ADDONE, irpc_AddOne, NULL);
-        IRPC_REGISTER(data->msg_ctx2, rpcecho, ECHO_ADDONE, irpc_AddOne, NULL);
-
-        IRPC_REGISTER(data->msg_ctx1, rpcecho, ECHO_ECHODATA, irpc_EchoData, NULL);
-        IRPC_REGISTER(data->msg_ctx2, rpcecho, ECHO_ECHODATA, irpc_EchoData, NULL);
+        IRPC_REGISTER(data->msg_ctx1, rpcecho, ECHO_ADDONE, irpc_AddOne, data);
+        IRPC_REGISTER(data->msg_ctx2, rpcecho, ECHO_ADDONE, irpc_AddOne, data);
+
+        IRPC_REGISTER(data->msg_ctx1, rpcecho, ECHO_ECHODATA, irpc_EchoData, data);
+        IRPC_REGISTER(data->msg_ctx2, rpcecho, ECHO_ECHODATA, irpc_EchoData, data);
 
         return true;

vendor/current/source4/lib/messaging/tests/messaging.c

@@ -26 +26 @@
 #include "cluster/cluster.h"
 #include "param/param.h"
-
+#include "torture/local/proto.h"
 
 static uint32_t msg_pong;
 
-static void ping_message(struct messaging_context *msg, void *private_data,
+static void ping_message(struct imessaging_context *msg, void *private_data,
                          uint32_t msg_type, struct server_id src, DATA_BLOB *data)
 {
         NTSTATUS status;
-        status = messaging_send(msg, src, msg_pong, data);
+        status = imessaging_send(msg, src, msg_pong, data);
         if (!NT_STATUS_IS_OK(status)) {
                 printf("pong failed - %s\n", nt_errstr(status));
@@ -40 +40 @@
 }
 
-static void pong_message(struct messaging_context *msg, void *private_data,
+static void pong_message(struct imessaging_context *msg, void *private_data,
                          uint32_t msg_type, struct server_id src, DATA_BLOB *data)
 {
@@ -47 +47 @@
 }
 
-static void exit_message(struct messaging_context *msg, void *private_data,
+static void exit_message(struct imessaging_context *msg, void *private_data,
                          uint32_t msg_type, struct server_id src, DATA_BLOB *data)
 {
@@ -60 +60 @@
 {
         struct tevent_context *ev;
-        struct messaging_context *msg_client_ctx;
-        struct messaging_context *msg_server_ctx;
+        struct imessaging_context *msg_client_ctx;
+        struct imessaging_context *msg_server_ctx;
         int ping_count = 0;
         int pong_count = 0;
@@ -72 +72 @@
         ev = tctx->ev;
 
-        msg_server_ctx = messaging_init(tctx, 
-                                        lpcfg_messaging_path(tctx, tctx->lp_ctx), cluster_id(0, 1),
-                                        ev);
+        msg_server_ctx = imessaging_init(tctx,
+                                         tctx->lp_ctx, cluster_id(0, 1),
+                                         ev, true);
         
         torture_assert(tctx, msg_server_ctx != NULL, "Failed to init ping messaging context");
                 
-        messaging_register_tmp(msg_server_ctx, NULL, ping_message, &msg_ping);
-        messaging_register_tmp(msg_server_ctx, tctx, exit_message, &msg_exit);
+        imessaging_register_tmp(msg_server_ctx, NULL, ping_message, &msg_ping);
+        imessaging_register_tmp(msg_server_ctx, tctx, exit_message, &msg_exit);
 
-        msg_client_ctx = messaging_init(tctx, 
-                                        lpcfg_messaging_path(tctx, tctx->lp_ctx),
-                                        cluster_id(0, 2), 
-                                        ev);
+        msg_client_ctx = imessaging_init(tctx,
+                                         tctx->lp_ctx,
+                                         cluster_id(0, 2),
+                                         ev, true);
 
         torture_assert(tctx, msg_client_ctx != NULL, 
-                       "msg_client_ctx messaging_init() failed");
+                       "msg_client_ctx imessaging_init() failed");
 
-        messaging_register_tmp(msg_client_ctx, &pong_count, pong_message, &msg_pong);
+        imessaging_register_tmp(msg_client_ctx, &pong_count, pong_message, &msg_pong);
 
         tv = timeval_current();
@@ -101 +101 @@
                 data.length = strlen((const char *)data.data);
 
-                status1 = messaging_send(msg_client_ctx, cluster_id(0, 1), msg_ping, &data);
-                status2 = messaging_send(msg_client_ctx, cluster_id(0, 1), msg_ping, NULL);
+                status1 = imessaging_send(msg_client_ctx, cluster_id(0, 1), msg_ping, &data);
+                status2 = imessaging_send(msg_client_ctx, cluster_id(0, 1), msg_ping, NULL);
 
                 torture_assert_ntstatus_ok(tctx, status1, "msg1 failed");
@@ -111 +111 @@
 
                 while (ping_count > pong_count + 20) {
-                        event_loop_once(ev);
+                        tevent_loop_once(ev);
                 }
         }
@@ -118 +118 @@
                ping_count - pong_count, pong_count);
         while (timeval_elapsed(&tv) < 30 && pong_count < ping_count) {
-                event_loop_once(ev);
+                tevent_loop_once(ev);
         }
 
         torture_comment(tctx, "sending exit\n");
-        messaging_send(msg_client_ctx, cluster_id(0, 1), msg_exit, NULL);
+        imessaging_send(msg_client_ctx, cluster_id(0, 1), msg_exit, NULL);
 
         torture_assert_int_equal(tctx, ping_count, pong_count, "ping test failed");

vendor/current/source4/lib/messaging/wscript_build

@@ -2 +2 @@
 
 
-bld.SAMBA_SUBSYSTEM('MESSAGING',
+bld.SAMBA_LIBRARY('MESSAGING',
         source='messaging.c',
-        public_deps='samba-util tdb-wrap NDR_IRPC UNIX_PRIVS UTIL_TDB cluster ndr samba_socket dcerpc'
+        public_deps='samba-util NDR_IRPC UNIX_PRIVS cluster ndr dcerpc messages_util server_id_db talloc_report',
+        private_library=True
         )
-
 
 bld.SAMBA_PYTHON('python_messaging',
         source='pymessaging.c',
-        deps='MESSAGING events pyparam_util',
+        deps='MESSAGING events pyparam_util pytalloc-util',
         realname='samba/messaging.so'
         )