diff options
Diffstat (limited to 'source4/lib/messaging')
| -rw-r--r-- | source4/lib/messaging/config.mk | 18 | ||||
| -rw-r--r-- | source4/lib/messaging/irpc.h | 131 | ||||
| -rw-r--r-- | source4/lib/messaging/messaging.c | 1120 | ||||
| -rw-r--r-- | source4/lib/messaging/messaging.h | 39 | ||||
| -rw-r--r-- | source4/lib/messaging/pymessaging.c | 564 | ||||
| -rw-r--r-- | source4/lib/messaging/tests/bindings.py | 57 | ||||
| -rw-r--r-- | source4/lib/messaging/tests/irpc.c | 271 | ||||
| -rw-r--r-- | source4/lib/messaging/tests/messaging.c | 145 |
8 files changed, 2345 insertions, 0 deletions
diff --git a/source4/lib/messaging/config.mk b/source4/lib/messaging/config.mk new file mode 100644 index 0000000000..e92f78c8e3 --- /dev/null +++ b/source4/lib/messaging/config.mk @@ -0,0 +1,18 @@ +[SUBSYSTEM::MESSAGING] +PUBLIC_DEPENDENCIES = \ + LIBSAMBA-UTIL \ + TDB_WRAP \ + NDR_IRPC \ + UNIX_PRIVS \ + UTIL_TDB \ + CLUSTER \ + LIBNDR \ + samba-socket + +MESSAGING_OBJ_FILES = $(libmessagingsrcdir)/messaging.o + +[PYTHON::python_messaging] +LIBRARY_REALNAME = samba/messaging.$(SHLIBEXT) +PRIVATE_DEPENDENCIES = MESSAGING LIBEVENTS python_irpc + +python_messaging_OBJ_FILES = $(libmessagingsrcdir)/pymessaging.o diff --git a/source4/lib/messaging/irpc.h b/source4/lib/messaging/irpc.h new file mode 100644 index 0000000000..65e98dce2c --- /dev/null +++ b/source4/lib/messaging/irpc.h @@ -0,0 +1,131 @@ +/* + Unix SMB/CIFS implementation. + + Samba internal rpc code - header + + Copyright (C) Andrew Tridgell 2005 + + 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/>. +*/ + +#ifndef IRPC_H +#define IRPC_H + +#include "librpc/gen_ndr/irpc.h" + +/* + an incoming irpc message +*/ +struct irpc_message { + struct server_id from; + void *private; + struct irpc_header header; + struct ndr_pull *ndr; + bool defer_reply; + struct messaging_context *msg_ctx; + struct irpc_list *irpc; + void *data; + struct event_context *ev; +}; + +/* don't allow calls to take too long */ +#define IRPC_CALL_TIMEOUT 10 + + +/* the server function type */ +typedef NTSTATUS (*irpc_function_t)(struct irpc_message *, void *r); + +/* register a server function with the irpc messaging system */ +#define IRPC_REGISTER(msg_ctx, pipename, funcname, function, private) \ + irpc_register(msg_ctx, &ndr_table_ ## pipename, \ + NDR_ ## funcname, \ + (irpc_function_t)function, private) + +/* make a irpc call */ +#define IRPC_CALL(msg_ctx, server_id, pipename, funcname, ptr, ctx) \ + irpc_call(msg_ctx, server_id, &ndr_table_ ## pipename, NDR_ ## funcname, ptr, ctx) + +#define IRPC_CALL_SEND(msg_ctx, server_id, pipename, funcname, ptr, ctx) \ + irpc_call_send(msg_ctx, server_id, &ndr_table_ ## pipename, NDR_ ## funcname, ptr, ctx) + + +/* + a pending irpc call +*/ +struct irpc_request { + struct messaging_context *msg_ctx; + const struct ndr_interface_table *table; + int callnum; + int callid; + void *r; + NTSTATUS status; + bool done; + bool reject_free; + TALLOC_CTX *mem_ctx; + struct { + void (*fn)(struct irpc_request *); + void *private; + } async; +}; + +struct loadparm_context; + +typedef void (*msg_callback_t)(struct messaging_context *msg, void *private, + uint32_t msg_type, + struct server_id server_id, DATA_BLOB *data); + +NTSTATUS messaging_send(struct messaging_context *msg, struct server_id server, + uint32_t msg_type, DATA_BLOB *data); +NTSTATUS messaging_register(struct messaging_context *msg, void *private, + uint32_t msg_type, + msg_callback_t fn); +NTSTATUS messaging_register_tmp(struct messaging_context *msg, void *private, + 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 smb_iconv_convenience *iconv_convenience, + struct event_context *ev); +struct messaging_context *messaging_client_init(TALLOC_CTX *mem_ctx, + const char *dir, + struct smb_iconv_convenience *iconv_convenience, + struct event_context *ev); +NTSTATUS messaging_send_ptr(struct messaging_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); + + + + +NTSTATUS irpc_register(struct messaging_context *msg_ctx, + const struct ndr_interface_table *table, + int call, irpc_function_t fn, void *private); +struct irpc_request *irpc_call_send(struct messaging_context *msg_ctx, + struct server_id server_id, + const struct ndr_interface_table *table, + int callnum, void *r, TALLOC_CTX *ctx); +NTSTATUS irpc_call_recv(struct irpc_request *irpc); +NTSTATUS irpc_call(struct messaging_context *msg_ctx, + struct server_id server_id, + const struct ndr_interface_table *table, + int callnum, void *r, TALLOC_CTX *ctx); + +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_send_reply(struct irpc_message *m, NTSTATUS status); +struct server_id messaging_get_server_id(struct messaging_context *msg_ctx); + +#endif + diff --git a/source4/lib/messaging/messaging.c b/source4/lib/messaging/messaging.c new file mode 100644 index 0000000000..4b90e8c4fd --- /dev/null +++ b/source4/lib/messaging/messaging.c @@ -0,0 +1,1120 @@ +/* + Unix SMB/CIFS implementation. + + Samba internal messaging functions + + 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/>. +*/ + +#include "includes.h" +#include "lib/events/events.h" +#include "system/filesys.h" +#include "messaging/messaging.h" +#include "lib/util/dlinklist.h" +#include "lib/socket/socket.h" +#include "librpc/gen_ndr/ndr_irpc.h" +#include "lib/messaging/irpc.h" +#include "tdb_wrap.h" +#include "lib/util/unix_privs.h" +#include "librpc/rpc/dcerpc.h" +#include "lib/tdb/include/tdb.h" +#include "lib/util/util_tdb.h" +#include "lib/util/util_tdb.h" +#include "cluster/cluster.h" +#include "param/param.h" + +/* change the message version with any incompatible changes in the protocol */ +#define MESSAGING_VERSION 1 + +struct messaging_context { + struct server_id server_id; + struct socket_context *sock; + const char *base_path; + const char *path; + struct dispatch_fn **dispatch; + uint32_t num_types; + struct idr_context *dispatch_tree; + struct messaging_rec *pending; + struct messaging_rec *retry_queue; + struct smb_iconv_convenience *iconv_convenience; + struct irpc_list *irpc; + struct idr_context *idr; + const char **names; + struct timeval start_time; + struct timed_event *retry_te; + struct { + struct event_context *ev; + struct fd_event *fde; + } event; +}; + +/* we have a linked list of dispatch handlers for each msg_type that + this messaging server can deal with */ +struct dispatch_fn { + struct dispatch_fn *next, *prev; + uint32_t msg_type; + void *private; + msg_callback_t fn; +}; + +/* 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 *, + uint32_t, struct server_id, DATA_BLOB *); + + +/* + A useful function for testing the message system. +*/ +static void ping_message(struct messaging_context *msg, void *private, + uint32_t msg_type, struct server_id src, DATA_BLOB *data) +{ + DEBUG(1,("INFO: Received PING message from server %u.%u [%.*s]\n", + (uint_t)src.node, (uint_t)src.id, (int)data->length, + data->data?(const char *)data->data:"")); + messaging_send(msg, src, MSG_PONG, data); +} + +/* + return uptime of messaging server via irpc +*/ +static NTSTATUS irpc_uptime(struct irpc_message *msg, + struct irpc_uptime *r) +{ + struct messaging_context *ctx = talloc_get_type(msg->private, struct messaging_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) +{ + return talloc_asprintf(msg, "%s/msg.%s", msg->base_path, + cluster_id_string(msg, server_id)); +} + +/* + 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; + + /* 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, 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 event_context *ev, struct timed_event *te, + struct timeval t, void *private) +{ + struct messaging_context *msg = talloc_get_type(private, + 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)) { + DEBUG(1,("messaging: Lost message from %s to %s of type %u - %s\n", + cluster_id_string(debug_ctx(), rec->header->from), + cluster_id_string(debug_ctx(), rec->header->to), + rec->header->msg_type, + nt_errstr(status))); + } + 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 event_context *ev, struct fd_event *fde, + uint16_t flags, void *private) +{ + struct messaging_context *msg = talloc_get_type(private, + struct messaging_context); + if (flags & EVENT_FD_WRITE) { + messaging_send_handler(msg); + } + if (flags & EVENT_FD_READ) { + messaging_recv_handler(msg); + } +} + + +/* + Register a dispatch function for a particular message type. +*/ +NTSTATUS messaging_register(struct messaging_context *msg, void *private, + uint32_t msg_type, msg_callback_t fn) +{ + struct dispatch_fn *d; + + /* possibly expand dispatch array */ + if (msg_type >= msg->num_types) { + struct dispatch_fn **dp; + int i; + dp = talloc_realloc(msg, msg->dispatch, struct dispatch_fn *, msg_type+1); + NT_STATUS_HAVE_NO_MEMORY(dp); + msg->dispatch = dp; + for (i=msg->num_types;i<=msg_type;i++) { + msg->dispatch[i] = NULL; + } + msg->num_types = msg_type+1; + } + + d = talloc_zero(msg->dispatch, struct dispatch_fn); + NT_STATUS_HAVE_NO_MEMORY(d); + d->msg_type = msg_type; + d->private = private; + d->fn = fn; + + DLIST_ADD(msg->dispatch[msg_type], d); + + return NT_STATUS_OK; +} + +/* + register a temporary message handler. The msg_type is allocated + above MSG_TMP_BASE +*/ +NTSTATUS messaging_register_tmp(struct messaging_context *msg, void *private, + msg_callback_t fn, uint32_t *msg_type) +{ + struct dispatch_fn *d; + int id; + + d = talloc_zero(msg->dispatch, struct dispatch_fn); + NT_STATUS_HAVE_NO_MEMORY(d); + d->private = private; + d->fn = fn; + + id = idr_get_new_above(msg->dispatch_tree, d, MSG_TMP_BASE, UINT16_MAX); + if (id == -1) { + talloc_free(d); + return NT_STATUS_TOO_MANY_CONTEXT_IDS; + } + + d->msg_type = (uint32_t)id; + (*msg_type) = d->msg_type; + + return NT_STATUS_OK; +} + +/* + De-register the function for a particular message type. +*/ +void messaging_deregister(struct messaging_context *msg, uint32_t msg_type, void *private) +{ + struct dispatch_fn *d, *next; + + if (msg_type >= msg->num_types) { + d = (struct dispatch_fn *)idr_find(msg->dispatch_tree, + msg_type); + if (!d) return; + idr_remove(msg->dispatch_tree, msg_type); + talloc_free(d); + return; + } + + for (d = msg->dispatch[msg_type]; d; d = next) { + next = d->next; + if (d->private == private) { + DLIST_REMOVE(msg->dispatch[msg_type], d); + talloc_free(d); + } + } +} + +/* + Send a message to a particular server +*/ +NTSTATUS messaging_send(struct messaging_context *msg, struct server_id server, + uint32_t msg_type, 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); + } + + 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 *); + return NT_STATUS_OK; + } + + talloc_free(rec); + + return status; +} + +/* + 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, + uint32_t msg_type, void *ptr) +{ + DATA_BLOB blob; + + blob.data = (uint8_t *)&ptr; + 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 0; +} + +/* + 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 smb_iconv_convenience *iconv_convenience, + struct event_context *ev) +{ + struct messaging_context *msg; + NTSTATUS status; + struct socket_address *path; + + if (ev == NULL) { + return NULL; + } + + msg = talloc_zero(mem_ctx, struct messaging_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); + msg->server_id = server_id; + msg->iconv_convenience = iconv_convenience; + msg->idr = idr_init(msg); + msg->dispatch_tree = idr_init(msg); + 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 = talloc_reference(msg, ev); + msg->event.fde = event_add_fd(ev, msg, socket_get_fd(msg->sock), + EVENT_FD_READ, messaging_handler, msg); + + talloc_set_destructor(msg, messaging_destructor); + + messaging_register(msg, NULL, MSG_PING, ping_message); + messaging_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, + struct smb_iconv_convenience *iconv_convenience, + struct event_context *ev) +{ + struct server_id id; + ZERO_STRUCT(id); + id.id = random() % 0x10000000; + return messaging_init(mem_ctx, dir, id, iconv_convenience, ev); +} +/* + a list of registered irpc server functions +*/ +struct irpc_list { + struct irpc_list *next, *prev; + struct GUID uuid; + const struct ndr_interface_table *table; + int callnum; + irpc_function_t fn; + void *private; +}; + + +/* + register a irpc server function +*/ +NTSTATUS irpc_register(struct messaging_context *msg_ctx, + const struct ndr_interface_table *table, + int callnum, irpc_function_t fn, void *private) +{ + struct irpc_list *irpc; + + /* override an existing handler, if any */ + for (irpc=msg_ctx->irpc; irpc; irpc=irpc->next) { + if (irpc->table == table && irpc->callnum == callnum) { + break; + } + } + if (irpc == NULL) { + irpc = talloc(msg_ctx, struct irpc_list); + NT_STATUS_HAVE_NO_MEMORY(irpc); + DLIST_ADD(msg_ctx->irpc, irpc); + } + + irpc->table = table; + irpc->callnum = callnum; + irpc->fn = fn; + irpc->private = private; + irpc->uuid = irpc->table->syntax_id.uuid; + + return NT_STATUS_OK; +} + + +/* + handle an incoming irpc reply message +*/ +static void irpc_handler_reply(struct messaging_context *msg_ctx, struct irpc_message *m) +{ + struct irpc_request *irpc; + enum ndr_err_code ndr_err; + + irpc = (struct irpc_request *)idr_find(msg_ctx->idr, m->header.callid); + if (irpc == NULL) return; + + /* parse the reply data */ + ndr_err = irpc->table->calls[irpc->callnum].ndr_pull(m->ndr, NDR_OUT, irpc->r); + if (NDR_ERR_CODE_IS_SUCCESS(ndr_err)) { + irpc->status = m->header.status; + talloc_steal(irpc->mem_ctx, m); + } else { + irpc->status = ndr_map_error2ntstatus(ndr_err); + talloc_steal(irpc, m); + } + irpc->done = true; + if (irpc->async.fn) { + irpc->async.fn(irpc); + } +} + +/* + send a irpc reply +*/ +NTSTATUS irpc_send_reply(struct irpc_message *m, NTSTATUS status) +{ + struct ndr_push *push; + DATA_BLOB packet; + enum ndr_err_code ndr_err; + + m->header.status = status; + + /* setup the reply */ + push = ndr_push_init_ctx(m->ndr, m->msg_ctx->iconv_convenience); + if (push == NULL) { + status = NT_STATUS_NO_MEMORY; + goto failed; + } + + m->header.flags |= IRPC_FLAG_REPLY; + + /* construct the packet */ + ndr_err = ndr_push_irpc_header(push, NDR_SCALARS|NDR_BUFFERS, &m->header); + if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) { + status = ndr_map_error2ntstatus(ndr_err); + goto failed; + } + + ndr_err = m->irpc->table->calls[m->irpc->callnum].ndr_push(push, NDR_OUT, m->data); + if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) { + status = ndr_map_error2ntstatus(ndr_err); + goto failed; + } + + /* send the reply message */ + packet = ndr_push_blob(push); + status = messaging_send(m->msg_ctx, m->from, MSG_IRPC, &packet); + if (!NT_STATUS_IS_OK(status)) goto failed; + +failed: + talloc_free(m); + return status; +} + +/* + handle an incoming irpc request message +*/ +static void irpc_handler_request(struct messaging_context *msg_ctx, + struct irpc_message *m) +{ + struct irpc_list *i; + void *r; + enum ndr_err_code ndr_err; + + for (i=msg_ctx->irpc; i; i=i->next) { + if (GUID_equal(&i->uuid, &m->header.uuid) && + i->table->syntax_id.if_version == m->header.if_version && + i->callnum == m->header.callnum) { + break; + } + } + + if (i == NULL) { + /* no registered handler for this message */ + talloc_free(m); + return; + } + + /* allocate space for the structure */ + r = talloc_zero_size(m->ndr, i->table->calls[m->header.callnum].struct_size); + if (r == NULL) goto failed; + + /* parse the request data */ + ndr_err = i->table->calls[i->callnum].ndr_pull(m->ndr, NDR_IN, r); + if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) goto failed; + + /* make the call */ + m->private = i->private; + m->defer_reply = false; + m->msg_ctx = msg_ctx; + m->irpc = i; + m->data = r; + m->ev = msg_ctx->event.ev; + + m->header.status = i->fn(m, r); + + if (m->defer_reply) { + /* the server function has asked to defer the reply to later */ + talloc_steal(msg_ctx, m); + return; + } + + irpc_send_reply(m, m->header.status); + return; + +failed: + talloc_free(m); +} + +/* + handle an incoming irpc message +*/ +static void irpc_handler(struct messaging_context *msg_ctx, void *private, + uint32_t msg_type, struct server_id src, DATA_BLOB *packet) +{ + struct irpc_message *m; + enum ndr_err_code ndr_err; + + m = talloc(msg_ctx, struct irpc_message); + if (m == NULL) goto failed; + + m->from = src; + + m->ndr = ndr_pull_init_blob(packet, m, msg_ctx->iconv_convenience); + if (m->ndr == NULL) goto failed; + + m->ndr->flags |= LIBNDR_FLAG_REF_ALLOC; + + ndr_err = ndr_pull_irpc_header(m->ndr, NDR_BUFFERS|NDR_SCALARS, &m->header); + if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) goto failed; + + if (m->header.flags & IRPC_FLAG_REPLY) { + irpc_handler_reply(msg_ctx, m); + } else { + irpc_handler_request(msg_ctx, m); + } + return; + +failed: + talloc_free(m); +} + + +/* + destroy a irpc request +*/ +static int irpc_destructor(struct irpc_request *irpc) +{ + if (irpc->callid != -1) { + idr_remove(irpc->msg_ctx->idr, irpc->callid); + irpc->callid = -1; + } + + if (irpc->reject_free) { + return -1; + } + return 0; +} + +/* + timeout a irpc request +*/ +static void irpc_timeout(struct event_context *ev, struct timed_event *te, + struct timeval t, void *private) +{ + struct irpc_request *irpc = talloc_get_type(private, struct irpc_request); + irpc->status = NT_STATUS_IO_TIMEOUT; + irpc->done = true; + if (irpc->async.fn) { + irpc->async.fn(irpc); + } +} + + +/* + make a irpc call - async send +*/ +struct irpc_request *irpc_call_send(struct messaging_context *msg_ctx, + struct server_id server_id, + const struct ndr_interface_table *table, + int callnum, void *r, TALLOC_CTX *ctx) +{ + struct irpc_header header; + struct ndr_push *ndr; + NTSTATUS status; + DATA_BLOB packet; + struct irpc_request *irpc; + enum ndr_err_code ndr_err; + + irpc = talloc(msg_ctx, struct irpc_request); + if (irpc == NULL) goto failed; + + irpc->msg_ctx = msg_ctx; + irpc->table = table; + irpc->callnum = callnum; + irpc->callid = idr_get_new(msg_ctx->idr, irpc, UINT16_MAX); + if (irpc->callid == -1) goto failed; + irpc->r = r; + irpc->done = false; + irpc->async.fn = NULL; + irpc->mem_ctx = ctx; + irpc->reject_free = false; + + talloc_set_destructor(irpc, irpc_destructor); + + /* setup the header */ + header.uuid = table->syntax_id.uuid; + + header.if_version = table->syntax_id.if_version; + header.callid = irpc->callid; + header.callnum = callnum; + header.flags = 0; + header.status = NT_STATUS_OK; + + /* construct the irpc packet */ + ndr = ndr_push_init_ctx(irpc, msg_ctx->iconv_convenience); + if (ndr == NULL) goto failed; + + ndr_err = ndr_push_irpc_header(ndr, NDR_SCALARS|NDR_BUFFERS, &header); + if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) goto failed; + + ndr_err = table->calls[callnum].ndr_push(ndr, NDR_IN, r); + if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) goto failed; + + /* and send it */ + packet = ndr_push_blob(ndr); + status = messaging_send(msg_ctx, server_id, MSG_IRPC, &packet); + if (!NT_STATUS_IS_OK(status)) goto failed; + + event_add_timed(msg_ctx->event.ev, irpc, + timeval_current_ofs(IRPC_CALL_TIMEOUT, 0), + irpc_timeout, irpc); + + talloc_free(ndr); + return irpc; + +failed: + talloc_free(irpc); + return NULL; +} + +/* + wait for a irpc reply +*/ +NTSTATUS irpc_call_recv(struct irpc_request *irpc) +{ + NTSTATUS status; + + NT_STATUS_HAVE_NO_MEMORY(irpc); + + irpc->reject_free = true; + + while (!irpc->done) { + if (event_loop_once(irpc->msg_ctx->event.ev) != 0) { + return NT_STATUS_CONNECTION_DISCONNECTED; + } + } + + irpc->reject_free = false; + + status = irpc->status; + talloc_free(irpc); + return status; +} + +/* + perform a synchronous irpc request +*/ +NTSTATUS irpc_call(struct messaging_context *msg_ctx, + struct server_id server_id, + const struct ndr_interface_table *table, + int callnum, void *r, + TALLOC_CTX *mem_ctx) +{ + struct irpc_request *irpc = irpc_call_send(msg_ctx, server_id, + table, callnum, r, mem_ctx); + return irpc_call_recv(irpc); +} + +/* + 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) { + 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) { + 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; +} + +/* + 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) +{ + return msg_ctx->server_id; +} diff --git a/source4/lib/messaging/messaging.h b/source4/lib/messaging/messaging.h new file mode 100644 index 0000000000..c91a31d285 --- /dev/null +++ b/source4/lib/messaging/messaging.h @@ -0,0 +1,39 @@ +/* + Unix SMB/CIFS implementation. + messages.c header + Copyright (C) Andrew Tridgell 2000 + Copyright (C) 2001, 2002 by Martin Pool + + 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/>. +*/ + +#ifndef _MESSAGES_H_ +#define _MESSAGES_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 + +/* temporary messaging endpoints are allocated above this line */ +#define MSG_TMP_BASE 1000 + +#endif diff --git a/source4/lib/messaging/pymessaging.c b/source4/lib/messaging/pymessaging.c new file mode 100644 index 0000000000..621a2fe349 --- /dev/null +++ b/source4/lib/messaging/pymessaging.c @@ -0,0 +1,564 @@ +/* + Unix SMB/CIFS implementation. + Copyright © Jelmer Vernooij <jelmer@samba.org> 2008 + + Based on the equivalent for EJS: + Copyright © Andrew Tridgell <tridge@samba.org> 2005 + + 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/>. +*/ + +#include "includes.h" +#include <Python.h> +#include "libcli/util/pyerrors.h" +#include "librpc/rpc/pyrpc.h" +#include "lib/messaging/irpc.h" +#include "lib/messaging/messaging.h" +#include "lib/events/events.h" +#include "cluster/cluster.h" +#include "param/param.h" +#include "librpc/gen_ndr/py_irpc.h" + +PyAPI_DATA(PyTypeObject) messaging_Type; +PyAPI_DATA(PyTypeObject) irpc_ClientConnectionType; + +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 (PyTuple_Size(object) == 3) { + return PyArg_ParseTuple(object, "iii", &server_id->id, &server_id->id2, &server_id->node); + } else { + int id, id2; + if (!PyArg_ParseTuple(object, "ii", &id, &id2)) + return false; + *server_id = cluster_id(id, id2); + return true; + } +} + +typedef struct { + PyObject_HEAD + TALLOC_CTX *mem_ctx; + struct messaging_context *msg_ctx; +} messaging_Object; + +PyObject *py_messaging_connect(PyTypeObject *self, PyObject *args, PyObject *kwargs) +{ + struct event_context *ev; + const char *kwnames[] = { "own_id", "messaging_path", 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); + if (ret == NULL) + return NULL; + + ret->mem_ctx = talloc_new(NULL); + + ev = s4_event_context_init(ret->mem_ctx); + + if (messaging_path == NULL) { + messaging_path = lp_messaging_path(ret, global_loadparm); + } else { + messaging_path = talloc_strdup(ret->mem_ctx, messaging_path); + } + + if (own_id != Py_None) { + struct server_id server_id; + + if (!server_id_from_py(own_id, &server_id)) + return NULL; + + ret->msg_ctx = messaging_init(ret->mem_ctx, + messaging_path, + server_id, + lp_iconv_convenience(global_loadparm), + ev); + } else { + ret->msg_ctx = messaging_client_init(ret->mem_ctx, + messaging_path, + lp_iconv_convenience(global_loadparm), + ev); + } + + if (ret->msg_ctx == NULL) { + PyErr_SetString(PyExc_RuntimeError, "messaging_connect unable to create a messaging context"); + talloc_free(ret->mem_ctx); + return NULL; + } + + return (PyObject *)ret; +} + +static void py_messaging_dealloc(PyObject *self) +{ + messaging_Object *iface = (messaging_Object *)self; + talloc_free(iface->msg_ctx); + PyObject_Del(self); +} + +static PyObject *py_messaging_send(PyObject *self, PyObject *args, PyObject *kwargs) +{ + messaging_Object *iface = (messaging_Object *)self; + uint32_t msg_type; + DATA_BLOB data; + PyObject *target; + NTSTATUS status; + struct server_id server; + const char *kwnames[] = { "target", "msg_type", "data", NULL }; + int length; + + if (!PyArg_ParseTupleAndKeywords(args, kwargs, "Ois#|:send", + discard_const_p(char *, kwnames), &target, &msg_type, &data.data, &length)) { + return NULL; + } + + data.length = length; + + if (!server_id_from_py(target, &server)) + return NULL; + + status = messaging_send(iface->msg_ctx, server, msg_type, &data); + if (NT_STATUS_IS_ERR(status)) { + PyErr_SetNTSTATUS(status); + return NULL; + } + + return Py_None; +} + +static void py_msg_callback_wrapper(struct messaging_context *msg, void *private, + uint32_t msg_type, + struct server_id server_id, DATA_BLOB *data) +{ + PyObject *callback = (PyObject *)private; + + PyObject_CallFunction(callback, discard_const_p(char, "i(iii)s#"), msg_type, + server_id.id, server_id.id2, server_id.node, + data->data, data->length); +} + +static PyObject *py_messaging_register(PyObject *self, PyObject *args, PyObject *kwargs) +{ + messaging_Object *iface = (messaging_Object *)self; + int msg_type = -1; + PyObject *callback; + NTSTATUS status; + const char *kwnames[] = { "callback", "msg_type", NULL }; + + if (!PyArg_ParseTupleAndKeywords(args, kwargs, "O|i:send", + discard_const_p(char *, kwnames), &callback, &msg_type)) { + return NULL; + } + + Py_INCREF(callback); + + if (msg_type == -1) { + uint32_t msg_type32 = msg_type; + status = messaging_register_tmp(iface->msg_ctx, callback, + py_msg_callback_wrapper, &msg_type32); + msg_type = msg_type32; + } else { + status = messaging_register(iface->msg_ctx, callback, + msg_type, py_msg_callback_wrapper); + } + if (NT_STATUS_IS_ERR(status)) { + PyErr_SetNTSTATUS(status); + return NULL; + } + + return PyLong_FromLong(msg_type); +} + +static PyObject *py_messaging_deregister(PyObject *self, PyObject *args, PyObject *kwargs) +{ + messaging_Object *iface = (messaging_Object *)self; + int msg_type = -1; + PyObject *callback; + const char *kwnames[] = { "callback", "msg_type", NULL }; + + if (!PyArg_ParseTupleAndKeywords(args, kwargs, "O|i:send", + discard_const_p(char *, kwnames), &callback, &msg_type)) { + return NULL; + } + + messaging_deregister(iface->msg_ctx, msg_type, callback); + + Py_DECREF(callback); + + return Py_None; +} + +static PyObject *py_messaging_add_name(PyObject *self, PyObject *args, PyObject *kwargs) +{ + messaging_Object *iface = (messaging_Object *)self; + NTSTATUS status; + char *name; + const char *kwnames[] = { "name", NULL }; + + if (!PyArg_ParseTupleAndKeywords(args, kwargs, "s|:send", + discard_const_p(char *, kwnames), &name)) { + return NULL; + } + + status = irpc_add_name(iface->msg_ctx, name); + if (NT_STATUS_IS_ERR(status)) { + PyErr_SetNTSTATUS(status); + return NULL; + } + + return Py_None; +} + + +static PyObject *py_messaging_remove_name(PyObject *self, PyObject *args, PyObject *kwargs) +{ + messaging_Object *iface = (messaging_Object *)self; + char *name; + const char *kwnames[] = { "name", NULL }; + + if (!PyArg_ParseTupleAndKeywords(args, kwargs, "s|:send", + discard_const_p(char *, kwnames), &name)) { + return NULL; + } + + irpc_remove_name(iface->msg_ctx, name); + + return Py_None; +} + +static PyMethodDef py_messaging_methods[] = { + { "send", (PyCFunction)py_messaging_send, METH_VARARGS|METH_KEYWORDS, + "S.send(target, msg_type, data) -> None\nSend a message" }, + { "register", (PyCFunction)py_messaging_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, + "S.deregister(callback, msg_type) -> None\nDeregister a message handler" }, + { "add_name", (PyCFunction)py_messaging_add_name, METH_VARARGS|METH_KEYWORDS, "S.add_name(name) -> None\nListen on another name" }, + { "remove_name", (PyCFunction)py_messaging_remove_name, METH_VARARGS|METH_KEYWORDS, "S.remove_name(name) -> None\nStop listening on a name" }, + { 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, + discard_const_p(char, "local server id") }, + { NULL }, +}; + + +PyTypeObject messaging_Type = { + PyObject_HEAD_INIT(NULL) 0, + .tp_name = "irpc.Messaging", + .tp_basicsize = sizeof(messaging_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." +}; + + +/* + state of a irpc 'connection' +*/ +typedef struct { + PyObject_HEAD + const char *server_name; + struct server_id *dest_ids; + struct messaging_context *msg_ctx; + TALLOC_CTX *mem_ctx; +} irpc_ClientConnectionObject; + +/* + setup a context for talking to a irpc server + example: + status = irpc.connect("smb_server"); +*/ + +PyObject *py_irpc_connect(PyTypeObject *self, PyObject *args, PyObject *kwargs) +{ + struct event_context *ev; + const char *kwnames[] = { "server", "own_id", "messaging_path", NULL }; + char *server; + const char *messaging_path = NULL; + PyObject *own_id = Py_None; + irpc_ClientConnectionObject *ret; + + if (!PyArg_ParseTupleAndKeywords(args, kwargs, "s|Oz:connect", + discard_const_p(char *, kwnames), &server, &own_id, &messaging_path)) { + return NULL; + } + + ret = PyObject_New(irpc_ClientConnectionObject, &irpc_ClientConnectionType); + if (ret == NULL) + return NULL; + + ret->mem_ctx = talloc_new(NULL); + + ret->server_name = server; + + ev = s4_event_context_init(ret->mem_ctx); + + if (messaging_path == NULL) { + messaging_path = lp_messaging_path(ret, global_loadparm); + } else { + messaging_path = talloc_strdup(ret->mem_ctx, messaging_path); + } + + if (own_id != Py_None) { + struct server_id server_id; + + if (!server_id_from_py(own_id, &server_id)) + return NULL; + + ret->msg_ctx = messaging_init(ret->mem_ctx, + messaging_path, + server_id, + lp_iconv_convenience(global_loadparm), + ev); + } else { + ret->msg_ctx = messaging_client_init(ret->mem_ctx, + messaging_path, + lp_iconv_convenience(global_loadparm), + ev); + } + + if (ret->msg_ctx == NULL) { + PyErr_SetString(PyExc_RuntimeError, "irpc_connect unable to create a messaging context"); + talloc_free(ret->mem_ctx); + return NULL; + } + + ret->dest_ids = irpc_servers_byname(ret->msg_ctx, ret->mem_ctx, ret->server_name); + if (ret->dest_ids == NULL || ret->dest_ids[0].id == 0) { + talloc_free(ret->mem_ctx); + PyErr_SetNTSTATUS(NT_STATUS_OBJECT_NAME_NOT_FOUND); + return NULL; + } else { + return (PyObject *)ret; + } +} + +typedef struct { + PyObject_HEAD + struct irpc_request **reqs; + int count; + int current; + TALLOC_CTX *mem_ctx; + py_data_unpack_fn unpack_fn; +} irpc_ResultObject; + + +static PyObject *irpc_result_next(irpc_ResultObject *iterator) +{ + NTSTATUS status; + + if (iterator->current >= iterator->count) { + PyErr_SetString(PyExc_StopIteration, "No more results"); + return NULL; + } + + status = irpc_call_recv(iterator->reqs[iterator->current]); + iterator->current++; + if (!NT_STATUS_IS_OK(status)) { + PyErr_SetNTSTATUS(status); + return NULL; + } + + return iterator->unpack_fn(iterator->reqs[iterator->current-1]->r); +} + +static PyObject *irpc_result_len(irpc_ResultObject *self) +{ + return PyLong_FromLong(self->count); +} + +static PyMethodDef irpc_result_methods[] = { + { "__len__", (PyCFunction)irpc_result_len, METH_NOARGS, + "Number of elements returned"}, + { NULL } +}; + +static void irpc_result_dealloc(PyObject *self) +{ + talloc_free(((irpc_ResultObject *)self)->mem_ctx); + PyObject_Del(self); +} + +PyTypeObject irpc_ResultIteratorType = { + PyObject_HEAD_INIT(NULL) 0, + .tp_name = "irpc.ResultIterator", + .tp_basicsize = sizeof(irpc_ResultObject), + .tp_flags = Py_TPFLAGS_DEFAULT|Py_TPFLAGS_BASETYPE, + .tp_iternext = (iternextfunc)irpc_result_next, + .tp_iter = PyObject_SelfIter, + .tp_methods = irpc_result_methods, + .tp_dealloc = irpc_result_dealloc, +}; + +static PyObject *py_irpc_call(irpc_ClientConnectionObject *p, struct PyNdrRpcMethodDef *method_def, PyObject *args, PyObject *kwargs) +{ + void *ptr; + struct irpc_request **reqs; + int i, count; + NTSTATUS status; + TALLOC_CTX *mem_ctx = talloc_new(NULL); + irpc_ResultObject *ret; + + /* allocate the C structure */ + ptr = talloc_zero_size(mem_ctx, method_def->table->calls[method_def->opnum].struct_size); + if (ptr == NULL) { + status = NT_STATUS_NO_MEMORY; + goto done; + } + + /* convert the mpr object into a C structure */ + if (!method_def->pack_in_data(args, kwargs, ptr)) { + talloc_free(mem_ctx); + return NULL; + } + + for (count=0;p->dest_ids[count].id;count++) /* noop */ ; + + /* we need to make a call per server */ + reqs = talloc_array(mem_ctx, struct irpc_request *, count); + if (reqs == NULL) { + status = NT_STATUS_NO_MEMORY; + goto done; + } + + /* make the actual calls */ + for (i=0;i<count;i++) { + reqs[i] = irpc_call_send(p->msg_ctx, p->dest_ids[i], + method_def->table, method_def->opnum, ptr, ptr); + if (reqs[i] == NULL) { + status = NT_STATUS_NO_MEMORY; + goto done; + } + talloc_steal(reqs, reqs[i]); + } + + ret = PyObject_New(irpc_ResultObject, &irpc_ResultIteratorType); + ret->mem_ctx = mem_ctx; + ret->reqs = reqs; + ret->count = count; + ret->current = 0; + ret->unpack_fn = method_def->unpack_out_data; + + return (PyObject *)ret; +done: + talloc_free(mem_ctx); + PyErr_SetNTSTATUS(status); + return NULL; +} + +static PyObject *py_irpc_call_wrapper(PyObject *self, PyObject *args, void *wrapped, PyObject *kwargs) +{ + irpc_ClientConnectionObject *iface = (irpc_ClientConnectionObject *)self; + struct PyNdrRpcMethodDef *md = wrapped; + + return py_irpc_call(iface, md, args, kwargs); +} + +static void py_irpc_dealloc(PyObject *self) +{ + irpc_ClientConnectionObject *iface = (irpc_ClientConnectionObject *)self; + talloc_free(iface->mem_ctx); + PyObject_Del(self); +} + +PyTypeObject irpc_ClientConnectionType = { + PyObject_HEAD_INIT(NULL) 0, + .tp_name = "irpc.ClientConnection", + .tp_basicsize = sizeof(irpc_ClientConnectionObject), + .tp_flags = Py_TPFLAGS_DEFAULT|Py_TPFLAGS_BASETYPE, + .tp_new = py_irpc_connect, + .tp_dealloc = py_irpc_dealloc, + .tp_doc = "ClientConnection(server, own_id=None, messaging_path=None)\n" \ + "Create a new IRPC client connection to communicate with the servers in the specified path.\n" \ + "If no path is specified, the default path from smb.conf will be used." +}; + +static bool irpc_AddNdrRpcMethods(PyTypeObject *ifacetype, const struct PyNdrRpcMethodDef *mds) +{ + int i; + for (i = 0; mds[i].name; i++) { + PyObject *ret; + struct wrapperbase *wb = calloc(sizeof(struct wrapperbase), 1); + + wb->name = discard_const_p(char, mds[i].name); + wb->flags = PyWrapperFlag_KEYWORDS; + wb->wrapper = (wrapperfunc)py_irpc_call_wrapper; + wb->doc = discard_const_p(char, mds[i].doc); + + ret = PyDescr_NewWrapper(ifacetype, wb, discard_const_p(void, &mds[i])); + + PyDict_SetItemString(ifacetype->tp_dict, mds[i].name, + (PyObject *)ret); + } + + return true; +} + +void initmessaging(void) +{ + extern void initirpc(void); + PyObject *mod; + + if (PyType_Ready(&irpc_ClientConnectionType) < 0) + return; + + if (PyType_Ready(&messaging_Type) < 0) + return; + + if (PyType_Ready(&irpc_ResultIteratorType) < 0) + return; + + if (!irpc_AddNdrRpcMethods(&irpc_ClientConnectionType, py_ndr_irpc_methods)) + return; + + mod = Py_InitModule3("messaging", NULL, "Internal RPC"); + if (mod == NULL) + return; + + initirpc(); + + Py_INCREF((PyObject *)&irpc_ClientConnectionType); + PyModule_AddObject(mod, "ClientConnection", (PyObject *)&irpc_ClientConnectionType); + + Py_INCREF((PyObject *)&messaging_Type); + PyModule_AddObject(mod, "Messaging", (PyObject *)&messaging_Type); +} diff --git a/source4/lib/messaging/tests/bindings.py b/source4/lib/messaging/tests/bindings.py new file mode 100644 index 0000000000..c89538ddfa --- /dev/null +++ b/source4/lib/messaging/tests/bindings.py @@ -0,0 +1,57 @@ +#!/usr/bin/python +# -*- coding: utf-8 -*- + +# Unix SMB/CIFS implementation. +# Copyright © Jelmer Vernooij <jelmer@samba.org> 2008 +# +# 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/>. +# + +from samba.messaging import Messaging +from unittest import TestCase + +class MessagingTests(TestCase): + def get_context(self, *args, **kwargs): + kwargs["messaging_path"] = "." + return Messaging(*args, **kwargs) + + def test_register(self): + x = self.get_context() + def callback(): + pass + msg_type = x.register(callback) + x.deregister(callback, msg_type) + + def test_assign_server_id(self): + x = self.get_context() + self.assertTrue(isinstance(x.server_id, tuple)) + self.assertEquals(3, len(x.server_id)) + + def test_ping_speed(self): + server_ctx = self.get_context((0, 1)) + def ping_callback(src, data): + server_ctx.send(src, data) + def exit_callback(): + print "received exit" + msg_ping = server_ctx.register(ping_callback) + msg_exit = server_ctx.register(exit_callback) + + def pong_callback(): + print "received pong" + client_ctx = self.get_context((0, 2)) + msg_pong = client_ctx.register(pong_callback) + + client_ctx.send((0,1), msg_ping, "testing") + client_ctx.send((0,1), msg_ping, "") + diff --git a/source4/lib/messaging/tests/irpc.c b/source4/lib/messaging/tests/irpc.c new file mode 100644 index 0000000000..d9b0548643 --- /dev/null +++ b/source4/lib/messaging/tests/irpc.c @@ -0,0 +1,271 @@ +/* + Unix SMB/CIFS implementation. + + local test for irpc code + + 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/>. +*/ + +#include "includes.h" +#include "lib/events/events.h" +#include "lib/messaging/irpc.h" +#include "librpc/gen_ndr/ndr_echo.h" +#include "torture/torture.h" +#include "cluster/cluster.h" +#include "param/param.h" + +const uint32_t MSG_ID1 = 1, MSG_ID2 = 2; + +static bool test_debug; + +struct irpc_test_data +{ + struct messaging_context *msg_ctx1, *msg_ctx2; + struct event_context *ev; +}; + +/* + serve up AddOne over the irpc system +*/ +static NTSTATUS irpc_AddOne(struct irpc_message *irpc, struct echo_AddOne *r) +{ + *r->out.out_data = r->in.in_data + 1; + if (test_debug) { + printf("irpc_AddOne: in=%u in+1=%u out=%u\n", + r->in.in_data, r->in.in_data+1, *r->out.out_data); + } + return NT_STATUS_OK; +} + +/* + a deferred reply to echodata +*/ +static void deferred_echodata(struct event_context *ev, struct timed_event *te, + struct timeval t, void *private) +{ + struct irpc_message *irpc = talloc_get_type(private, struct irpc_message); + struct echo_EchoData *r = irpc->data; + r->out.out_data = talloc_memdup(r, r->in.in_data, r->in.len); + if (r->out.out_data == NULL) { + irpc_send_reply(irpc, NT_STATUS_NO_MEMORY); + } + printf("sending deferred reply\n"); + irpc_send_reply(irpc, NT_STATUS_OK); +} + + +/* + serve up EchoData over the irpc system +*/ +static NTSTATUS irpc_EchoData(struct irpc_message *irpc, struct echo_EchoData *r) +{ + irpc->defer_reply = true; + event_add_timed(irpc->ev, irpc, timeval_zero(), deferred_echodata, irpc); + return NT_STATUS_OK; +} + + +/* + test a addone call over the internal messaging system +*/ +static bool test_addone(struct torture_context *test, const void *_data, + const void *_value) +{ + struct echo_AddOne r; + NTSTATUS status; + const struct irpc_test_data *data = (const struct irpc_test_data *)_data; + uint32_t value = (uint32_t)_value; + + /* make the call */ + r.in.in_data = value; + + test_debug = true; + status = IRPC_CALL(data->msg_ctx1, cluster_id(0, MSG_ID2), + rpcecho, ECHO_ADDONE, &r, test); + test_debug = false; + torture_assert_ntstatus_ok(test, status, "AddOne failed"); + + /* check the answer */ + torture_assert(test, *r.out.out_data == r.in.in_data + 1, + "AddOne wrong answer"); + + torture_comment(test, "%u + 1 = %u\n", r.in.in_data, *r.out.out_data); + return true; +} + +/* + test a echodata call over the internal messaging system +*/ +static bool test_echodata(struct torture_context *tctx, + const void *tcase_data, + const void *test_data) +{ + struct echo_EchoData r; + NTSTATUS status; + const struct irpc_test_data *data = (const struct irpc_test_data *)tcase_data; + TALLOC_CTX *mem_ctx = tctx; + + /* make the call */ + r.in.in_data = (unsigned char *)talloc_strdup(mem_ctx, "0123456789"); + r.in.len = strlen((char *)r.in.in_data); + + status = IRPC_CALL(data->msg_ctx1, cluster_id(0, MSG_ID2), + rpcecho, ECHO_ECHODATA, &r, + mem_ctx); + torture_assert_ntstatus_ok(tctx, status, "EchoData failed"); + + /* check the answer */ + if (memcmp(r.out.out_data, r.in.in_data, r.in.len) != 0) { + NDR_PRINT_OUT_DEBUG(echo_EchoData, &r); + torture_fail(tctx, "EchoData wrong answer"); + } + + torture_comment(tctx, "Echo '%*.*s' -> '%*.*s'\n", + r.in.len, r.in.len, + r.in.in_data, + r.in.len, r.in.len, + r.out.out_data); + return true; +} + + +static void irpc_callback(struct irpc_request *irpc) +{ + struct echo_AddOne *r = (struct echo_AddOne *)irpc->r; + int *pong_count = (int *)irpc->async.private; + NTSTATUS status = irpc_call_recv(irpc); + if (!NT_STATUS_IS_OK(status)) { + printf("irpc call failed - %s\n", nt_errstr(status)); + } + if (*r->out.out_data != r->in.in_data + 1) { + printf("AddOne wrong answer - %u + 1 = %u should be %u\n", + r->in.in_data, *r->out.out_data, r->in.in_data+1); + } + (*pong_count)++; +} + +/* + test echo speed +*/ +static bool test_speed(struct torture_context *tctx, + const void *tcase_data, + const void *test_data) +{ + int ping_count = 0; + int pong_count = 0; + const struct irpc_test_data *data = (const struct irpc_test_data *)tcase_data; + struct timeval tv; + struct echo_AddOne r; + TALLOC_CTX *mem_ctx = tctx; + int timelimit = torture_setting_int(tctx, "timelimit", 10); + + tv = timeval_current(); + + r.in.in_data = 0; + + torture_comment(tctx, "Sending echo for %d seconds\n", timelimit); + while (timeval_elapsed(&tv) < timelimit) { + struct irpc_request *irpc; + + irpc = IRPC_CALL_SEND(data->msg_ctx1, cluster_id(0, MSG_ID2), + rpcecho, ECHO_ADDONE, + &r, mem_ctx); + torture_assert(tctx, irpc != NULL, "AddOne send failed"); + + irpc->async.fn = irpc_callback; + irpc->async.private = &pong_count; + + ping_count++; + + while (ping_count > pong_count + 20) { + event_loop_once(data->ev); + } + } + + torture_comment(tctx, "waiting for %d remaining replies (done %d)\n", + ping_count - pong_count, pong_count); + while (timeval_elapsed(&tv) < 30 && pong_count < ping_count) { + event_loop_once(data->ev); + } + + torture_assert_int_equal(tctx, ping_count, pong_count, "ping test failed"); + + torture_comment(tctx, "echo rate of %.0f messages/sec\n", + (ping_count+pong_count)/timeval_elapsed(&tv)); + return true; +} + + +static bool irpc_setup(struct torture_context *tctx, void **_data) +{ + struct irpc_test_data *data; + + *_data = data = talloc(tctx, struct irpc_test_data); + + lp_set_cmdline(tctx->lp_ctx, "pid directory", "piddir.tmp"); + + data->ev = tctx->ev; + torture_assert(tctx, data->msg_ctx1 = + messaging_init(tctx, + lp_messaging_path(tctx, tctx->lp_ctx), + cluster_id(0, MSG_ID1), + lp_iconv_convenience(tctx->lp_ctx), + data->ev), + "Failed to init first messaging context"); + + torture_assert(tctx, data->msg_ctx2 = + messaging_init(tctx, + lp_messaging_path(tctx, tctx->lp_ctx), + cluster_id(0, MSG_ID2), + lp_iconv_convenience(tctx->lp_ctx), + data->ev), + "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); + + return true; +} + +struct torture_suite *torture_local_irpc(TALLOC_CTX *mem_ctx) +{ + struct torture_suite *suite = torture_suite_create(mem_ctx, "IRPC"); + struct torture_tcase *tcase = torture_suite_add_tcase(suite, "irpc"); + int i; + uint32_t *values = talloc_array(tcase, uint32_t, 5); + + values[0] = 0; + values[1] = 0x7FFFFFFE; + values[2] = 0xFFFFFFFE; + values[3] = 0xFFFFFFFF; + values[4] = random() & 0xFFFFFFFF; + + tcase->setup = irpc_setup; + + for (i = 0; i < 5; i++) { + torture_tcase_add_test_const(tcase, "addone", test_addone, + (void *)values[i]); + } + + torture_tcase_add_test_const(tcase, "echodata", test_echodata, NULL); + torture_tcase_add_test_const(tcase, "speed", test_speed, NULL); + + return suite; +} diff --git a/source4/lib/messaging/tests/messaging.c b/source4/lib/messaging/tests/messaging.c new file mode 100644 index 0000000000..f66b3a5b43 --- /dev/null +++ b/source4/lib/messaging/tests/messaging.c @@ -0,0 +1,145 @@ +/* + Unix SMB/CIFS implementation. + + local test for messaging code + + 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/>. +*/ + +#include "includes.h" +#include "lib/events/events.h" +#include "lib/messaging/irpc.h" +#include "torture/torture.h" +#include "cluster/cluster.h" +#include "param/param.h" + + +static uint32_t msg_pong; + +static void ping_message(struct messaging_context *msg, void *private, + uint32_t msg_type, struct server_id src, DATA_BLOB *data) +{ + NTSTATUS status; + status = messaging_send(msg, src, msg_pong, data); + if (!NT_STATUS_IS_OK(status)) { + printf("pong failed - %s\n", nt_errstr(status)); + } +} + +static void pong_message(struct messaging_context *msg, void *private, + uint32_t msg_type, struct server_id src, DATA_BLOB *data) +{ + int *count = private; + (*count)++; +} + +static void exit_message(struct messaging_context *msg, void *private, + uint32_t msg_type, struct server_id src, DATA_BLOB *data) +{ + talloc_free(private); + exit(0); +} + +/* + test ping speed +*/ +static bool test_ping_speed(struct torture_context *tctx) +{ + struct event_context *ev; + struct messaging_context *msg_client_ctx; + struct messaging_context *msg_server_ctx; + int ping_count = 0; + int pong_count = 0; + struct timeval tv; + int timelimit = torture_setting_int(tctx, "timelimit", 10); + uint32_t msg_ping, msg_exit; + + lp_set_cmdline(tctx->lp_ctx, "pid directory", "piddir.tmp"); + + ev = tctx->ev; + + msg_server_ctx = messaging_init(tctx, + lp_messaging_path(tctx, tctx->lp_ctx), + cluster_id(0, 1), + lp_iconv_convenience(tctx->lp_ctx), + ev); + + 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); + + msg_client_ctx = messaging_init(tctx, + lp_messaging_path(tctx, tctx->lp_ctx), + cluster_id(0, 2), + lp_iconv_convenience(tctx->lp_ctx), + ev); + + torture_assert(tctx, msg_client_ctx != NULL, + "msg_client_ctx messaging_init() failed"); + + messaging_register_tmp(msg_client_ctx, &pong_count, pong_message, &msg_pong); + + tv = timeval_current(); + + torture_comment(tctx, "Sending pings for %d seconds\n", timelimit); + while (timeval_elapsed(&tv) < timelimit) { + DATA_BLOB data; + NTSTATUS status1, status2; + + data.data = discard_const_p(uint8_t, "testing"); + 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); + + torture_assert_ntstatus_ok(tctx, status1, "msg1 failed"); + ping_count++; + + torture_assert_ntstatus_ok(tctx, status2, "msg2 failed"); + ping_count++; + + while (ping_count > pong_count + 20) { + event_loop_once(ev); + } + } + + torture_comment(tctx, "waiting for %d remaining replies (done %d)\n", + ping_count - pong_count, pong_count); + while (timeval_elapsed(&tv) < 30 && pong_count < ping_count) { + event_loop_once(ev); + } + + torture_comment(tctx, "sending exit\n"); + messaging_send(msg_client_ctx, cluster_id(0, 1), msg_exit, NULL); + + torture_assert_int_equal(tctx, ping_count, pong_count, "ping test failed"); + + torture_comment(tctx, "ping rate of %.0f messages/sec\n", + (ping_count+pong_count)/timeval_elapsed(&tv)); + + talloc_free(msg_client_ctx); + talloc_free(msg_server_ctx); + + return true; +} + +struct torture_suite *torture_local_messaging(TALLOC_CTX *mem_ctx) +{ + struct torture_suite *s = torture_suite_create(mem_ctx, "MESSAGING"); + torture_suite_add_simple_test(s, "ping_speed", test_ping_speed); + return s; +} |