#include "dbwrap/dbwrap.h"
#include "serverid.h"
#include "messages.h"
+#include "lib/util/tevent_unix.h"
+#include "lib/background.h"
struct messaging_callback {
struct messaging_callback *prev, *next;
* the msg has already been deleted from the messages.tdb.*/
status = messaging_send_buf(msg_all->msg_ctx, *id, msg_all->msg_type,
- (const uint8 *)msg_all->buf, msg_all->len);
+ (const uint8_t *)msg_all->buf, msg_all->len);
if (NT_STATUS_EQUAL(status, NT_STATUS_INVALID_HANDLE)) {
}
struct messaging_context *messaging_init(TALLOC_CTX *mem_ctx,
- struct event_context *ev)
+ struct tevent_context *ev)
{
struct messaging_context *ctx;
NTSTATUS status;
ctx->id = procid_self();
ctx->event_ctx = ev;
- status = messaging_tdb_init(ctx, ctx, &ctx->local);
+ status = messaging_dgm_init(ctx, ctx, &ctx->local);
if (!NT_STATUS_IS_OK(status)) {
- DEBUG(2, ("messaging_tdb_init failed: %s\n",
+ DEBUG(2, ("messaging_dgm_init failed: %s\n",
nt_errstr(status)));
TALLOC_FREE(ctx);
return NULL;
}
-#ifdef CLUSTER_SUPPORT
if (lp_clustering()) {
status = messaging_ctdbd_init(ctx, ctx, &ctx->remote);
}
}
ctx->id.vnn = get_my_vnn();
-#endif
messaging_register(ctx, NULL, MSG_PING, ping_message);
msg_ctx->id = procid_self();
- status = messaging_tdb_init(msg_ctx, msg_ctx, &msg_ctx->local);
+ status = messaging_dgm_init(msg_ctx, msg_ctx, &msg_ctx->local);
if (!NT_STATUS_IS_OK(status)) {
- DEBUG(0, ("messaging_tdb_init failed: %s\n",
+ DEBUG(0, ("messaging_dgm_init failed: %s\n",
nt_errstr(status)));
return status;
}
-#ifdef CLUSTER_SUPPORT
TALLOC_FREE(msg_ctx->remote);
if (lp_clustering()) {
}
}
-#endif
-
return NT_STATUS_OK;
}
}
}
+struct messaging_selfsend_state {
+ struct messaging_context *msg;
+ struct messaging_rec rec;
+};
+
+static void messaging_trigger_self(struct tevent_context *ev,
+ struct tevent_immediate *im,
+ void *private_data);
+
/*
Send a message to a particular server
*/
struct server_id server, uint32_t msg_type,
const DATA_BLOB *data)
{
-#ifdef CLUSTER_SUPPORT
+ if (server_id_is_disconnected(&server)) {
+ return NT_STATUS_INVALID_PARAMETER_MIX;
+ }
+
if (!procid_is_local(&server)) {
return msg_ctx->remote->send_fn(msg_ctx, server,
msg_type, data,
msg_ctx->remote);
}
-#endif
+
+ if (server_id_equal(&msg_ctx->id, &server)) {
+ struct messaging_selfsend_state *state;
+ struct tevent_immediate *im;
+
+ state = talloc_pooled_object(
+ msg_ctx, struct messaging_selfsend_state,
+ 1, data->length);
+ if (state == NULL) {
+ return NT_STATUS_NO_MEMORY;
+ }
+ state->msg = msg_ctx;
+ state->rec.msg_version = MESSAGE_VERSION;
+ state->rec.msg_type = msg_type & MSG_TYPE_MASK;
+ state->rec.dest = server;
+ state->rec.src = msg_ctx->id;
+
+ /* Can't fail, it's a pooled_object */
+ state->rec.buf = data_blob_talloc(
+ state, data->data, data->length);
+
+ im = tevent_create_immediate(state);
+ if (im == NULL) {
+ TALLOC_FREE(state);
+ return NT_STATUS_NO_MEMORY;
+ }
+
+ tevent_schedule_immediate(im, msg_ctx->event_ctx,
+ messaging_trigger_self, state);
+ return NT_STATUS_OK;
+ }
+
return msg_ctx->local->send_fn(msg_ctx, server, msg_type, data,
msg_ctx->local);
}
+static void messaging_trigger_self(struct tevent_context *ev,
+ struct tevent_immediate *im,
+ void *private_data)
+{
+ struct messaging_selfsend_state *state = talloc_get_type_abort(
+ private_data, struct messaging_selfsend_state);
+ messaging_dispatch_rec(state->msg, &state->rec);
+ TALLOC_FREE(state);
+}
+
NTSTATUS messaging_send_buf(struct messaging_context *msg_ctx,
struct server_id server, uint32_t msg_type,
- const uint8 *buf, size_t len)
+ const uint8_t *buf, size_t len)
{
DATA_BLOB blob = data_blob_const(buf, len);
return messaging_send(msg_ctx, server, msg_type, &blob);
}
+NTSTATUS messaging_send_iov(struct messaging_context *msg_ctx,
+ struct server_id server, uint32_t msg_type,
+ const struct iovec *iov, int iovlen)
+{
+ uint8_t *buf;
+ NTSTATUS status;
+
+ buf = iov_buf(talloc_tos(), iov, iovlen);
+ if (buf == NULL) {
+ return NT_STATUS_NO_MEMORY;
+ }
+
+ status = messaging_send_buf(msg_ctx, server, msg_type,
+ buf, talloc_get_size(buf));
+
+ TALLOC_FREE(buf);
+ return status;
+}
+
+static struct messaging_rec *messaging_rec_dup(TALLOC_CTX *mem_ctx,
+ struct messaging_rec *rec)
+{
+ struct messaging_rec *result;
+
+ result = talloc_pooled_object(mem_ctx, struct messaging_rec,
+ 1, rec->buf.length);
+ if (result == NULL) {
+ return NULL;
+ }
+ *result = *rec;
+
+ /* Doesn't fail, see talloc_pooled_object */
+
+ result->buf.data = talloc_memdup(result, rec->buf.data,
+ rec->buf.length);
+ return result;
+}
+
+struct messaging_filtered_read_state {
+ struct tevent_context *ev;
+ struct messaging_context *msg_ctx;
+
+ bool (*filter)(struct messaging_rec *rec, void *private_data);
+ void *private_data;
+
+ struct messaging_rec *rec;
+};
+
+static void messaging_filtered_read_cleanup(struct tevent_req *req,
+ enum tevent_req_state req_state);
+
+struct tevent_req *messaging_filtered_read_send(
+ TALLOC_CTX *mem_ctx, struct tevent_context *ev,
+ struct messaging_context *msg_ctx,
+ bool (*filter)(struct messaging_rec *rec, void *private_data),
+ void *private_data)
+{
+ struct tevent_req *req;
+ struct messaging_filtered_read_state *state;
+ size_t new_waiters_len;
+
+ req = tevent_req_create(mem_ctx, &state,
+ struct messaging_filtered_read_state);
+ if (req == NULL) {
+ return NULL;
+ }
+ state->ev = ev;
+ state->msg_ctx = msg_ctx;
+ state->filter = filter;
+ state->private_data = private_data;
+
+ /*
+ * We add ourselves to the "new_waiters" array, not the "waiters"
+ * array. If we are called from within messaging_read_done,
+ * messaging_dispatch_rec will be in an active for-loop on
+ * "waiters". We must be careful not to mess with this array, because
+ * it could mean that a single event is being delivered twice.
+ */
+
+ new_waiters_len = talloc_array_length(msg_ctx->new_waiters);
+
+ if (new_waiters_len == msg_ctx->num_new_waiters) {
+ struct tevent_req **tmp;
+
+ tmp = talloc_realloc(msg_ctx, msg_ctx->new_waiters,
+ struct tevent_req *, new_waiters_len+1);
+ if (tevent_req_nomem(tmp, req)) {
+ return tevent_req_post(req, ev);
+ }
+ msg_ctx->new_waiters = tmp;
+ }
+
+ msg_ctx->new_waiters[msg_ctx->num_new_waiters] = req;
+ msg_ctx->num_new_waiters += 1;
+ tevent_req_set_cleanup_fn(req, messaging_filtered_read_cleanup);
+
+ return req;
+}
+
+static void messaging_filtered_read_cleanup(struct tevent_req *req,
+ enum tevent_req_state req_state)
+{
+ struct messaging_filtered_read_state *state = tevent_req_data(
+ req, struct messaging_filtered_read_state);
+ struct messaging_context *msg_ctx = state->msg_ctx;
+ unsigned i;
+
+ tevent_req_set_cleanup_fn(req, NULL);
+
+ /*
+ * Just set the [new_]waiters entry to NULL, be careful not to mess
+ * with the other "waiters" array contents. We are often called from
+ * within "messaging_dispatch_rec", which loops over
+ * "waiters". Messing with the "waiters" array will mess up that
+ * for-loop.
+ */
+
+ for (i=0; i<msg_ctx->num_waiters; i++) {
+ if (msg_ctx->waiters[i] == req) {
+ msg_ctx->waiters[i] = NULL;
+ return;
+ }
+ }
+
+ for (i=0; i<msg_ctx->num_new_waiters; i++) {
+ if (msg_ctx->new_waiters[i] == req) {
+ msg_ctx->new_waiters[i] = NULL;
+ return;
+ }
+ }
+}
+
+static void messaging_filtered_read_done(struct tevent_req *req,
+ struct messaging_rec *rec)
+{
+ struct messaging_filtered_read_state *state = tevent_req_data(
+ req, struct messaging_filtered_read_state);
+
+ state->rec = messaging_rec_dup(state, rec);
+ if (tevent_req_nomem(state->rec, req)) {
+ return;
+ }
+ tevent_req_done(req);
+}
+
+int messaging_filtered_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
+ struct messaging_rec **presult)
+{
+ struct messaging_filtered_read_state *state = tevent_req_data(
+ req, struct messaging_filtered_read_state);
+ int err;
+
+ if (tevent_req_is_unix_error(req, &err)) {
+ tevent_req_received(req);
+ return err;
+ }
+ *presult = talloc_move(mem_ctx, &state->rec);
+ return 0;
+}
+
+struct messaging_read_state {
+ uint32_t msg_type;
+ struct messaging_rec *rec;
+};
+
+static bool messaging_read_filter(struct messaging_rec *rec,
+ void *private_data);
+static void messaging_read_done(struct tevent_req *subreq);
+
+struct tevent_req *messaging_read_send(TALLOC_CTX *mem_ctx,
+ struct tevent_context *ev,
+ struct messaging_context *msg,
+ uint32_t msg_type)
+{
+ struct tevent_req *req, *subreq;
+ struct messaging_read_state *state;
+
+ req = tevent_req_create(mem_ctx, &state,
+ struct messaging_read_state);
+ if (req == NULL) {
+ return NULL;
+ }
+ state->msg_type = msg_type;
+
+ subreq = messaging_filtered_read_send(state, ev, msg,
+ messaging_read_filter, state);
+ if (tevent_req_nomem(subreq, req)) {
+ return tevent_req_post(req, ev);
+ }
+ tevent_req_set_callback(subreq, messaging_read_done, req);
+ return req;
+}
+
+static bool messaging_read_filter(struct messaging_rec *rec,
+ void *private_data)
+{
+ struct messaging_read_state *state = talloc_get_type_abort(
+ private_data, struct messaging_read_state);
+
+ return rec->msg_type == state->msg_type;
+}
+
+static void messaging_read_done(struct tevent_req *subreq)
+{
+ struct tevent_req *req = tevent_req_callback_data(
+ subreq, struct tevent_req);
+ struct messaging_read_state *state = tevent_req_data(
+ req, struct messaging_read_state);
+ int ret;
+
+ ret = messaging_filtered_read_recv(subreq, state, &state->rec);
+ TALLOC_FREE(subreq);
+ if (tevent_req_error(req, ret)) {
+ return;
+ }
+ tevent_req_done(req);
+}
+
+int messaging_read_recv(struct tevent_req *req, TALLOC_CTX *mem_ctx,
+ struct messaging_rec **presult)
+{
+ struct messaging_read_state *state = tevent_req_data(
+ req, struct messaging_read_state);
+ int err;
+
+ if (tevent_req_is_unix_error(req, &err)) {
+ return err;
+ }
+ if (presult != NULL) {
+ *presult = talloc_move(mem_ctx, &state->rec);
+ }
+ return 0;
+}
+
+static bool messaging_append_new_waiters(struct messaging_context *msg_ctx)
+{
+ if (msg_ctx->num_new_waiters == 0) {
+ return true;
+ }
+
+ if (talloc_array_length(msg_ctx->waiters) <
+ (msg_ctx->num_waiters + msg_ctx->num_new_waiters)) {
+ struct tevent_req **tmp;
+ tmp = talloc_realloc(
+ msg_ctx, msg_ctx->waiters, struct tevent_req *,
+ msg_ctx->num_waiters + msg_ctx->num_new_waiters);
+ if (tmp == NULL) {
+ DEBUG(1, ("%s: talloc failed\n", __func__));
+ return false;
+ }
+ msg_ctx->waiters = tmp;
+ }
+
+ memcpy(&msg_ctx->waiters[msg_ctx->num_waiters], msg_ctx->new_waiters,
+ sizeof(struct tevent_req *) * msg_ctx->num_new_waiters);
+
+ msg_ctx->num_waiters += msg_ctx->num_new_waiters;
+ msg_ctx->num_new_waiters = 0;
+
+ return true;
+}
+
/*
Dispatch one messaging_rec
*/
struct messaging_rec *rec)
{
struct messaging_callback *cb, *next;
+ unsigned i;
for (cb = msg_ctx->callbacks; cb != NULL; cb = next) {
next = cb->next;
the same message type */
}
}
+
+ if (!messaging_append_new_waiters(msg_ctx)) {
+ return;
+ }
+
+ i = 0;
+ while (i < msg_ctx->num_waiters) {
+ struct tevent_req *req;
+ struct messaging_filtered_read_state *state;
+
+ req = msg_ctx->waiters[i];
+ if (req == NULL) {
+ /*
+ * This got cleaned up. In the meantime,
+ * move everything down one. We need
+ * to keep the order of waiters, as
+ * other code may depend on this.
+ */
+ if (i < msg_ctx->num_waiters - 1) {
+ memmove(&msg_ctx->waiters[i],
+ &msg_ctx->waiters[i+1],
+ sizeof(struct tevent_req *) *
+ (msg_ctx->num_waiters - i - 1));
+ }
+ msg_ctx->num_waiters -= 1;
+ continue;
+ }
+
+ state = tevent_req_data(
+ req, struct messaging_filtered_read_state);
+ if (state->filter(rec, state->private_data)) {
+ messaging_filtered_read_done(req, rec);
+ }
+
+ i += 1;
+ }
return;
}
+static int mess_parent_dgm_cleanup(void *private_data);
+static void mess_parent_dgm_cleanup_done(struct tevent_req *req);
+
+bool messaging_parent_dgm_cleanup_init(struct messaging_context *msg)
+{
+ struct tevent_req *req;
+
+ req = background_job_send(
+ msg, msg->event_ctx, msg, NULL, 0,
+ lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
+ 60*15),
+ mess_parent_dgm_cleanup, msg);
+ if (req == NULL) {
+ return false;
+ }
+ tevent_req_set_callback(req, mess_parent_dgm_cleanup_done, msg);
+ return true;
+}
+
+static int mess_parent_dgm_cleanup(void *private_data)
+{
+ struct messaging_context *msg_ctx = talloc_get_type_abort(
+ private_data, struct messaging_context);
+ NTSTATUS status;
+
+ status = messaging_dgm_wipe(msg_ctx);
+ DEBUG(10, ("messaging_dgm_wipe returned %s\n", nt_errstr(status)));
+ return lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
+ 60*15);
+}
+
+static void mess_parent_dgm_cleanup_done(struct tevent_req *req)
+{
+ struct messaging_context *msg = tevent_req_callback_data(
+ req, struct messaging_context);
+ NTSTATUS status;
+
+ status = background_job_recv(req);
+ TALLOC_FREE(req);
+ DEBUG(1, ("messaging dgm cleanup job ended with %s\n",
+ nt_errstr(status)));
+
+ req = background_job_send(
+ msg, msg->event_ctx, msg, NULL, 0,
+ lp_parm_int(-1, "messaging", "messaging dgm cleanup interval",
+ 60*15),
+ mess_parent_dgm_cleanup, msg);
+ if (req == NULL) {
+ DEBUG(1, ("background_job_send failed\n"));
+ }
+ tevent_req_set_callback(req, mess_parent_dgm_cleanup_done, msg);
+}
+
/** @} **/