messaging: Add messaging_ctdb_init/destroy

This models connecting to ctdb after the dgm code. The main point
is that we should never open more than more ctdb socket for messaging.

With more than one socket, we might end up with our pid registered with
ctdb on more than one socket. This could lead to memory overconsumption
in ctdb. ctdbd will eventually throw away messages, but they will take
up space unnecessarily.

Signed-off-by: Volker Lendecke <vl@samba.org>
Reviewed-by: Ralph Boehme <slow@samba.org>

diff --git a/source3/lib/ctdb_dummy.c b/source3/lib/ctdb_dummy.c
index b6ec2285ec7b02948241123920ec1bc2ce669eff..855f56d114863ecfefcda5a02d945eac7f5813a0 100644 (file)
--- a/source3/lib/ctdb_dummy.c
+++ b/source3/lib/ctdb_dummy.c
@@ -20,6 +20,7 @@
 #include "includes.h"
 #include "messages.h"
 #include "lib/messages_ctdbd.h"
+#include "lib/messages_ctdb.h"
 #include "ctdbd_conn.h"
 #include "lib/dbwrap/dbwrap.h"
 #include "lib/dbwrap/dbwrap_ctdb.h"
@@ -91,6 +92,18 @@ int messaging_ctdbd_init(struct messaging_context *msg_ctx,
        return ENOSYS;
 }
 
+int messaging_ctdb_send(uint32_t dst_vnn, uint64_t dst_srvid,
+                       const struct iovec *iov, int iovlen)
+{
+       return ENOSYS;
+}
+
+struct messaging_ctdb_fde *messaging_ctdb_register_tevent_context(
+       TALLOC_CTX *mem_ctx, struct tevent_context *ev)
+{
+       return NULL;
+}
+
 int messaging_ctdbd_reinit(struct messaging_context *msg_ctx,
                           TALLOC_CTX *mem_ctx,
                           void (*recv_cb)(struct tevent_context *ev,

diff --git a/source3/lib/messages_ctdb.c b/source3/lib/messages_ctdb.c
new file mode 100644 (file)
index 0000000..5bc494d
--- /dev/null
+++ b/source3/lib/messages_ctdb.c
@@ -0,0 +1,260 @@
+/*
+ * Unix SMB/CIFS implementation.
+ * Samba internal messaging functions
+ * Copyright (C) 2017 by Volker Lendecke
+ *
+ * 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/messages_ctdb.h"
+#include "lib/util/server_id.h"
+#include "messages.h"
+#include "util_tdb.h"
+#include "lib/util/iov_buf.h"
+#include "lib/messages_util.h"
+#include "ctdbd_conn.h"
+#include "lib/cluster_support.h"
+
+struct messaging_ctdb_context;
+
+/*
+ * We can only have one tevent_fd per ctdb_context and per
+ * tevent_context. Maintain a list of registered tevent_contexts per
+ * ctdb_context.
+ */
+struct messaging_ctdb_fde_ev {
+       struct messaging_ctdb_fde_ev *prev, *next;
+
+       /*
+        * Backreference to enable DLIST_REMOVE from our
+        * destructor. Also, set to NULL when the ctdb_context dies
+        * before the messaging_ctdb_fde_ev.
+        */
+       struct messaging_ctdb_context *ctx;
+
+       struct tevent_context *ev;
+       struct tevent_fd *fde;
+};
+
+struct messaging_ctdb_context {
+       struct ctdbd_connection *conn;
+
+       void (*recv_cb)(struct tevent_context *ev,
+                       const uint8_t *msg, size_t msg_len,
+                       int *fds, size_t num_fds,
+                       void *private_data);
+       void *recv_cb_private_data;
+
+       struct messaging_ctdb_fde_ev *fde_evs;
+};
+
+static int messaging_ctdb_recv(
+       struct tevent_context *ev,
+       uint32_t src_vnn, uint32_t dst_vnn, uint64_t dst_srvid,
+       const uint8_t *msg, size_t msg_len, void *private_data)
+{
+       struct messaging_ctdb_context *state = talloc_get_type_abort(
+               private_data, struct messaging_ctdb_context);
+
+       state->recv_cb(ev, msg, msg_len, NULL, 0, state->recv_cb_private_data);
+
+       return 0;
+}
+
+struct messaging_ctdb_context *global_ctdb_context;
+
+int messaging_ctdb_init(const char *sockname, int timeout, uint64_t unique_id,
+                       void (*recv_cb)(struct tevent_context *ev,
+                                       const uint8_t *msg, size_t msg_len,
+                                       int *fds, size_t num_fds,
+                                       void *private_data),
+                       void *private_data)
+{
+       struct messaging_ctdb_context *ctx;
+       int ret;
+
+       if (global_ctdb_context != NULL) {
+               return EBUSY;
+       }
+
+       ctx = talloc_zero(NULL, struct messaging_ctdb_context);
+       if (ctx == NULL) {
+               return ENOMEM;
+       }
+       ctx->recv_cb = recv_cb;
+       ctx->recv_cb_private_data = private_data;
+
+       ret = ctdbd_init_connection(ctx, sockname, timeout, &ctx->conn);
+       if (ret != 0) {
+               DBG_DEBUG("ctdbd_init_connection returned %s\n",
+                         strerror(ret));
+               goto fail;
+       }
+
+       ret = register_with_ctdbd(ctx->conn, getpid(), messaging_ctdb_recv,
+                                 ctx);
+       if (ret != 0) {
+               DBG_DEBUG("register_with_ctdbd returned %s (%d)\n",
+                         strerror(ret), ret);
+               goto fail;
+       }
+
+       ret = register_with_ctdbd(ctx->conn, unique_id, NULL, NULL);
+       if (ret != 0) {
+               DBG_DEBUG("register_with_ctdbd returned %s (%d)\n",
+                         strerror(ret), ret);
+               goto fail;
+       }
+
+       set_my_vnn(ctdbd_vnn(ctx->conn));
+
+       global_ctdb_context = ctx;
+       return 0;
+fail:
+       TALLOC_FREE(ctx);
+       return ret;
+}
+
+void messaging_ctdb_destroy(void)
+{
+       TALLOC_FREE(global_ctdb_context);
+}
+
+int messaging_ctdb_send(uint32_t dst_vnn, uint64_t dst_srvid,
+                       const struct iovec *iov, int iovlen)
+{
+       struct messaging_ctdb_context *ctx = global_ctdb_context;
+       int ret;
+
+       if (ctx == NULL) {
+               return ENOTCONN;
+       }
+
+       ret = ctdbd_messaging_send_iov(ctx->conn, dst_vnn, dst_srvid,
+                                      iov, iovlen);
+       return ret;
+}
+
+static void messaging_ctdb_read_handler(struct tevent_context *ev,
+                                       struct tevent_fd *fde,
+                                       uint16_t flags,
+                                       void *private_data)
+{
+       struct messaging_ctdb_context *ctx = talloc_get_type_abort(
+               private_data, struct messaging_ctdb_context);
+
+       if ((flags & TEVENT_FD_READ) == 0) {
+               return;
+       }
+       ctdbd_socket_readable(ev, ctx->conn);
+}
+
+struct messaging_ctdb_fde {
+       struct tevent_fd *fde;
+};
+
+static int messaging_ctdb_fde_ev_destructor(
+       struct messaging_ctdb_fde_ev *fde_ev)
+{
+       if (fde_ev->ctx != NULL) {
+               DLIST_REMOVE(fde_ev->ctx->fde_evs, fde_ev);
+               fde_ev->ctx = NULL;
+       }
+       return 0;
+}
+
+/*
+ * Reference counter for a struct tevent_fd messaging read event
+ * (with callback function) on a struct tevent_context registered
+ * on a messaging context.
+ *
+ * If we've already registered this struct tevent_context before
+ * (so already have a read event), just increase the reference count.
+ *
+ * Otherwise create a new struct tevent_fd messaging read event on the
+ * previously unseen struct tevent_context - this is what drives
+ * the message receive processing.
+ *
+ */
+
+struct messaging_ctdb_fde *messaging_ctdb_register_tevent_context(
+       TALLOC_CTX *mem_ctx, struct tevent_context *ev)
+{
+       struct messaging_ctdb_context *ctx = global_ctdb_context;
+       struct messaging_ctdb_fde_ev *fde_ev;
+       struct messaging_ctdb_fde *fde;
+
+       if (ctx == NULL) {
+               return NULL;
+       }
+
+       fde = talloc(mem_ctx, struct messaging_ctdb_fde);
+       if (fde == NULL) {
+               return NULL;
+       }
+
+       for (fde_ev = ctx->fde_evs; fde_ev != NULL; fde_ev = fde_ev->next) {
+               if ((fde_ev->ev == ev) &&
+                   (tevent_fd_get_flags(fde_ev->fde) != 0)) {
+                       break;
+               }
+       }
+
+       if (fde_ev == NULL) {
+               int sock = ctdbd_conn_get_fd(ctx->conn);
+
+               fde_ev = talloc(fde, struct messaging_ctdb_fde_ev);
+               if (fde_ev == NULL) {
+                       return NULL;
+               }
+               fde_ev->fde = tevent_add_fd(
+                       ev, fde_ev, sock, TEVENT_FD_READ,
+                       messaging_ctdb_read_handler, ctx);
+               if (fde_ev->fde == NULL) {
+                       TALLOC_FREE(fde);
+                       return NULL;
+               }
+               fde_ev->ev = ev;
+               fde_ev->ctx = ctx;
+               DLIST_ADD(ctx->fde_evs, fde_ev);
+               talloc_set_destructor(
+                       fde_ev, messaging_ctdb_fde_ev_destructor);
+       } else {
+               /*
+                * Same trick as with tdb_wrap: The caller will never
+                * see the talloc_referenced object, the
+                * messaging_ctdb_fde_ev, so problems with
+                * talloc_unlink will not happen.
+                */
+               if (talloc_reference(fde, fde_ev) == NULL) {
+                       TALLOC_FREE(fde);
+                       return NULL;
+               }
+       }
+
+       fde->fde = fde_ev->fde;
+       return fde;
+}
+
+bool messaging_ctdb_fde_active(struct messaging_ctdb_fde *fde)
+{
+       uint16_t flags;
+
+       if (fde == NULL) {
+               return false;
+       }
+       flags = tevent_fd_get_flags(fde->fde);
+       return (flags != 0);
+}

diff --git a/source3/lib/messages_ctdb.h b/source3/lib/messages_ctdb.h
new file mode 100644 (file)
index 0000000..006821b
--- /dev/null
+++ b/source3/lib/messages_ctdb.h
@@ -0,0 +1,42 @@
+/*
+ * Unix SMB/CIFS implementation.
+ * Samba internal messaging functions
+ * Copyright (C) 2017 by Volker Lendecke
+ *
+ * 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_CTDB_H__
+#define __MESSAGES_CTDB_H__
+
+#include "replace.h"
+#include "system/filesys.h"
+#include <tevent.h>
+
+int messaging_ctdb_init(const char *sockname, int timeout, uint64_t unique_id,
+                       void (*recv_cb)(struct tevent_context *ev,
+                                       const uint8_t *msg, size_t msg_len,
+                                       int *fds, size_t num_fds,
+                                       void *private_data),
+                       void *private_data);
+void messaging_ctdb_destroy(void);
+int messaging_ctdb_send(uint32_t dst_vnn, uint64_t dst_srvid,
+                       const struct iovec *iov, int iovlen);
+
+struct messaging_ctdb_fde;
+struct messaging_ctdb_fde *messaging_ctdb_register_tevent_context(
+       TALLOC_CTX *mem_ctx, struct tevent_context *ev);
+bool messaging_ctdb_fde_active(struct messaging_ctdb_fde *fde);
+
+#endif

diff --git a/source3/wscript_build b/source3/wscript_build
index 7b3d383ab5d0c0c01b56b12fbe96212065817b57..3f3629e93fc337067cfb54c9878a1fdf74eb0358 100644 (file)
--- a/source3/wscript_build
+++ b/source3/wscript_build
@@ -323,6 +323,7 @@ if bld.env.with_ctdb:
                      lib/cluster_support.c
                      lib/dbwrap/dbwrap_ctdb.c
                      lib/messages_ctdbd.c
+                     lib/messages_ctdb.c
                      lib/ctdbd_conn.c
                    '''
     SAMBA_CLUSTER_SUPPORT_DEPS='''