summaryrefslogtreecommitdiff
path: root/source4
diff options
context:
space:
mode:
Diffstat (limited to 'source4')
-rw-r--r--source4/lib/messaging/messaging.c213
1 files changed, 94 insertions, 119 deletions
diff --git a/source4/lib/messaging/messaging.c b/source4/lib/messaging/messaging.c
index 2130958b36..4d2cd9c910 100644
--- a/source4/lib/messaging/messaging.c
+++ b/source4/lib/messaging/messaging.c
@@ -31,9 +31,6 @@
/* change the message version with any incompatible changes in the protocol */
#define MESSAGING_VERSION 1
-/* the number of microseconds to backoff in retrying to send a message */
-#define MESSAGING_BACKOFF 250000
-
/* maximum message size */
#define MESSAGING_MAX_SIZE 512
@@ -42,6 +39,7 @@ struct messaging_context {
struct socket_context *sock;
const char *path;
struct dispatch_fn *dispatch;
+ struct messaging_rec *pending;
struct {
struct event_context *ev;
@@ -61,8 +59,8 @@ struct dispatch_fn {
/* an individual message */
struct messaging_rec {
+ struct messaging_rec *next, *prev;
struct messaging_context *msg;
- struct socket_context *sock;
const char *path;
struct {
@@ -76,6 +74,7 @@ struct messaging_rec {
DATA_BLOB data;
};
+
/*
A useful function for testing the message system.
*/
@@ -111,19 +110,67 @@ static void messaging_dispatch(struct messaging_context *msg, struct messaging_r
d->fn(msg, d->private, d->msg_type, rec->header.from, &rec->data);
}
}
-
rec->header.length = 0;
}
/*
- handle a new incoming connection
+ try to send the message
*/
-static void messaging_recv_handler(struct event_context *ev, struct fd_event *fde,
- uint16_t flags, void *private)
+static NTSTATUS try_send(struct messaging_rec *rec)
+{
+ struct messaging_context *msg = rec->msg;
+ DATA_BLOB blob;
+ size_t nsent;
+ void *priv;
+ NTSTATUS status;
+
+ blob = data_blob_talloc(rec, NULL, sizeof(rec->header) + rec->data.length);
+ NT_STATUS_HAVE_NO_MEMORY(blob.data);
+
+ memcpy(blob.data, &rec->header, sizeof(rec->header));
+ memcpy(blob.data + sizeof(rec->header), rec->data.data, rec->data.length);
+
+ /* we send with privileges so messages work from any context */
+ priv = root_privileges();
+ status = socket_sendto(msg->sock, &blob, &nsent, 0, rec->path, 0);
+ talloc_free(priv);
+
+ data_blob_free(&blob);
+
+ return status;
+}
+
+/*
+ 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)) {
+ break;
+ }
+ if (!NT_STATUS_IS_OK(status)) {
+ DEBUG(1,("messaging: Lost message from %u to %u of type %u - %s\n",
+ rec->header.from, 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_context *msg = talloc_get_type(private,
- struct messaging_context);
struct messaging_rec *rec;
NTSTATUS status;
uint8_t data[MESSAGING_MAX_SIZE];
@@ -146,7 +193,6 @@ static void messaging_recv_handler(struct event_context *ev, struct fd_event *fd
rec->msg = msg;
rec->path = msg->path;
- rec->sock = NULL;
memcpy(&rec->header, data, sizeof(rec->header));
if (msize != sizeof(rec->header) + rec->header.length) {
@@ -166,6 +212,24 @@ static void messaging_recv_handler(struct event_context *ev, struct fd_event *fd
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.
*/
@@ -200,89 +264,11 @@ void messaging_deregister(struct messaging_context *msg, uint32_t msg_type, void
}
-
-/*
- handle IO for sending a message
-*/
-static void messaging_send_handler(struct event_context *ev, struct fd_event *fde,
- uint16_t flags, void *private)
-{
- struct messaging_rec *rec = talloc_get_type(private, struct messaging_rec);
- uint8_t data[MESSAGING_MAX_SIZE];
- DATA_BLOB blob;
- size_t nsent;
- NTSTATUS status;
-
- memcpy(data, &rec->header, sizeof(rec->header));
- memcpy(data + sizeof(rec->header), rec->data.data, rec->data.length);
-
- blob.data = data;
- blob.length = sizeof(rec->header) + rec->header.length;
-
- status = socket_send(rec->sock, &blob, &nsent, 0);
- if (NT_STATUS_IS_ERR(status)) {
- DEBUG(3,("Unable to send message of type %d length %d - %s\n",
- rec->header.msg_type,
- rec->header.length,
- nt_errstr(status)));
- talloc_free(rec);
- return;
- }
-
- if (NT_STATUS_IS_OK(status)) {
- talloc_free(rec);
- }
-}
-
-
-/*
- wrapper around socket_connect with raised privileges
-*/
-static NTSTATUS try_connect(struct messaging_rec *rec)
-{
- NTSTATUS status;
- void *priv = root_privileges();
- status = socket_connect(rec->sock, NULL, 0, rec->path, 0, 0);
- talloc_free(priv);
- return status;
-}
-
-
-/*
- when the servers listen queue is full we use this to backoff the message
-*/
-static void messaging_backoff_handler(struct event_context *ev, struct timed_event *te,
- struct timeval t, void *private)
-{
- struct messaging_rec *rec = talloc_get_type(private, struct messaging_rec);
- struct messaging_context *msg = rec->msg;
- NTSTATUS status;
-
- status = try_connect(rec);
- if (NT_STATUS_EQUAL(status, STATUS_MORE_ENTRIES)) {
- /* backoff again */
- event_add_timed(msg->event.ev, rec,
- timeval_add(&t, 0, MESSAGING_BACKOFF),
- messaging_backoff_handler, rec);
- return;
- }
-
- if (!NT_STATUS_IS_OK(status)) {
- DEBUG(1,("messaging: Lost message from %u to %u of type %u after backoff - %s\n",
- rec->header.from, rec->header.to, rec->header.msg_type, nt_errstr(status)));
- talloc_free(rec);
- return;
- }
-
- event_add_fd(msg->event.ev, rec, socket_get_fd(rec->sock),
- EVENT_FD_WRITE, messaging_send_handler, rec);
-}
-
-
/*
Send a message to a particular server
*/
-NTSTATUS messaging_send(struct messaging_context *msg, uint32_t server, uint32_t msg_type, DATA_BLOB *data)
+NTSTATUS messaging_send(struct messaging_context *msg, uint32_t server,
+ uint32_t msg_type, DATA_BLOB *data)
{
struct messaging_rec *rec;
NTSTATUS status;
@@ -292,45 +278,32 @@ NTSTATUS messaging_send(struct messaging_context *msg, uint32_t server, uint32_t
return NT_STATUS_NO_MEMORY;
}
- rec->msg = msg;
- rec->header.version = MESSAGING_VERSION;
+ rec->msg = msg;
+ rec->header.version = MESSAGING_VERSION;
rec->header.msg_type = msg_type;
- rec->header.from = msg->server_id;
- rec->header.to = server;
- rec->header.length = data?data->length:0;
+ rec->header.from = msg->server_id;
+ rec->header.to = server;
+ rec->header.length = data?data->length:0;
if (rec->header.length != 0) {
rec->data = data_blob_talloc(rec, data->data, data->length);
} else {
rec->data = data_blob(NULL, 0);
}
- status = socket_create("unix", SOCKET_TYPE_DGRAM, &rec->sock, 0);
- if (!NT_STATUS_IS_OK(status)) {
- talloc_free(rec);
- return status;
- }
- talloc_steal(rec, rec->sock);
-
rec->path = messaging_path(rec, server);
- status = try_connect(rec);
+ status = try_send(rec);
if (NT_STATUS_EQUAL(status, STATUS_MORE_ENTRIES)) {
- /* backoff on this message - the servers listen queue is full */
- event_add_timed(msg->event.ev, rec,
- timeval_current_ofs(0, MESSAGING_BACKOFF),
- messaging_backoff_handler, rec);
+ if (msg->pending == NULL) {
+ EVENT_FD_WRITEABLE(msg->event.fde);
+ }
+ DLIST_ADD(msg->pending, rec);
return NT_STATUS_OK;
}
- if (!NT_STATUS_IS_OK(status)) {
- talloc_free(rec);
- return status;
- }
-
- event_add_fd(msg->event.ev, rec, socket_get_fd(rec->sock),
- EVENT_FD_WRITE, messaging_send_handler, rec);
+ talloc_free(rec);
- return NT_STATUS_OK;
+ return status;
}
/*
@@ -381,6 +354,7 @@ struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx, uint32_t server_id
msg->path = messaging_path(msg, server_id);
msg->server_id = server_id;
msg->dispatch = NULL;
+ msg->pending = NULL;
status = socket_create("unix", SOCKET_TYPE_DGRAM, &msg->sock, 0);
if (!NT_STATUS_IS_OK(status)) {
@@ -399,9 +373,12 @@ struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx, uint32_t server_id
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_recv_handler, msg);
+ EVENT_FD_READ, messaging_handler, msg);
talloc_set_destructor(msg, messaging_destructor);
@@ -409,5 +386,3 @@ struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx, uint32_t server_id
return msg;
}
-
-
generated by cgit (git 2.34.1) at 2024-11-25 00:20:55 +0000