source3/lib/messages_ctdb.c

   1 /*
   2  * Unix SMB/CIFS implementation.
   3  * Samba internal messaging functions
   4  * Copyright (C) 2017 by Volker Lendecke
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License as published by
   8  * the Free Software Foundation; either version 3 of the License, or
   9  * (at your option) any later version.
  10  *
  11  * This program is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14  * GNU General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  18  */
  19
  20 #include "includes.h"
  21 #include "lib/messages_ctdb.h"
  22 #include "lib/util/server_id.h"
  23 #include "messages.h"
  24 #include "util_tdb.h"
  25 #include "lib/util/iov_buf.h"
  26 #include "lib/messages_util.h"
  27 #include "ctdbd_conn.h"
  28 #include "lib/cluster_support.h"
  29
  30 struct messaging_ctdb_context;
  31
  32 /*
  33  * We can only have one tevent_fd per ctdb_context and per
  34  * tevent_context. Maintain a list of registered tevent_contexts per
  35  * ctdb_context.
  36  */
  37 struct messaging_ctdb_fde_ev {
  38         struct messaging_ctdb_fde_ev *prev, *next;
  39
  40         /*
  41          * Backreference to enable DLIST_REMOVE from our
  42          * destructor. Also, set to NULL when the ctdb_context dies
  43          * before the messaging_ctdb_fde_ev.
  44          */
  45         struct messaging_ctdb_context *ctx;
  46
  47         struct tevent_context *ev;
  48         struct tevent_fd *fde;
  49 };
  50
  51 struct messaging_ctdb_context {
  52         struct ctdbd_connection *conn;
  53
  54         void (*recv_cb)(struct tevent_context *ev,
  55                         const uint8_t *msg, size_t msg_len,
  56                         int *fds, size_t num_fds,
  57                         void *private_data);
  58         void *recv_cb_private_data;
  59
  60         struct messaging_ctdb_fde_ev *fde_evs;
  61 };
  62
  63 static int messaging_ctdb_recv(
  64         struct tevent_context *ev,
  65         uint32_t src_vnn, uint32_t dst_vnn, uint64_t dst_srvid,
  66         const uint8_t *msg, size_t msg_len, void *private_data)
  67 {
  68         struct messaging_ctdb_context *state = talloc_get_type_abort(
  69                 private_data, struct messaging_ctdb_context);
  70
  71         state->recv_cb(ev, msg, msg_len, NULL, 0, state->recv_cb_private_data);
  72
  73         return 0;
  74 }
  75
  76 struct messaging_ctdb_context *global_ctdb_context;
  77
  78 int messaging_ctdb_init(const char *sockname, int timeout, uint64_t unique_id,
  79                         void (*recv_cb)(struct tevent_context *ev,
  80                                         const uint8_t *msg, size_t msg_len,
  81                                         int *fds, size_t num_fds,
  82                                         void *private_data),
  83                         void *private_data)
  84 {
  85         struct messaging_ctdb_context *ctx;
  86         int ret;
  87
  88         if (global_ctdb_context != NULL) {
  89                 return EBUSY;
  90         }
  91
  92         ctx = talloc_zero(NULL, struct messaging_ctdb_context);
  93         if (ctx == NULL) {
  94                 return ENOMEM;
  95         }
  96         ctx->recv_cb = recv_cb;
  97         ctx->recv_cb_private_data = private_data;
  98
  99         ret = ctdbd_init_connection(ctx, sockname, timeout, &ctx->conn);
 100         if (ret != 0) {
 101                 DBG_DEBUG("ctdbd_init_connection returned %s\n",
 102                           strerror(ret));
 103                 goto fail;
 104         }
 105
 106         ret = register_with_ctdbd(ctx->conn, getpid(), messaging_ctdb_recv,
 107                                   ctx);
 108         if (ret != 0) {
 109                 DBG_DEBUG("register_with_ctdbd returned %s (%d)\n",
 110                           strerror(ret), ret);
 111                 goto fail;
 112         }
 113
 114         ret = register_with_ctdbd(ctx->conn, unique_id, NULL, NULL);
 115         if (ret != 0) {
 116                 DBG_DEBUG("register_with_ctdbd returned %s (%d)\n",
 117                           strerror(ret), ret);
 118                 goto fail;
 119         }
 120
 121         set_my_vnn(ctdbd_vnn(ctx->conn));
 122
 123         global_ctdb_context = ctx;
 124         return 0;
 125 fail:
 126         TALLOC_FREE(ctx);
 127         return ret;
 128 }
 129
 130 void messaging_ctdb_destroy(void)
 131 {
 132         TALLOC_FREE(global_ctdb_context);
 133 }
 134
 135 int messaging_ctdb_send(uint32_t dst_vnn, uint64_t dst_srvid,
 136                         const struct iovec *iov, int iovlen)
 137 {
 138         struct messaging_ctdb_context *ctx = global_ctdb_context;
 139         int ret;
 140
 141         if (ctx == NULL) {
 142                 return ENOTCONN;
 143         }
 144
 145         ret = ctdbd_messaging_send_iov(ctx->conn, dst_vnn, dst_srvid,
 146                                        iov, iovlen);
 147         return ret;
 148 }
 149
 150 static void messaging_ctdb_read_handler(struct tevent_context *ev,
 151                                         struct tevent_fd *fde,
 152                                         uint16_t flags,
 153                                         void *private_data)
 154 {
 155         struct messaging_ctdb_context *ctx = talloc_get_type_abort(
 156                 private_data, struct messaging_ctdb_context);
 157
 158         if ((flags & TEVENT_FD_READ) == 0) {
 159                 return;
 160         }
 161         ctdbd_socket_readable(ev, ctx->conn);
 162 }
 163
 164 struct messaging_ctdb_fde {
 165         struct tevent_fd *fde;
 166 };
 167
 168 static int messaging_ctdb_fde_ev_destructor(
 169         struct messaging_ctdb_fde_ev *fde_ev)
 170 {
 171         if (fde_ev->ctx != NULL) {
 172                 DLIST_REMOVE(fde_ev->ctx->fde_evs, fde_ev);
 173                 fde_ev->ctx = NULL;
 174         }
 175         return 0;
 176 }
 177
 178 /*
 179  * Reference counter for a struct tevent_fd messaging read event
 180  * (with callback function) on a struct tevent_context registered
 181  * on a messaging context.
 182  *
 183  * If we've already registered this struct tevent_context before
 184  * (so already have a read event), just increase the reference count.
 185  *
 186  * Otherwise create a new struct tevent_fd messaging read event on the
 187  * previously unseen struct tevent_context - this is what drives
 188  * the message receive processing.
 189  *
 190  */
 191
 192 struct messaging_ctdb_fde *messaging_ctdb_register_tevent_context(
 193         TALLOC_CTX *mem_ctx, struct tevent_context *ev)
 194 {
 195         struct messaging_ctdb_context *ctx = global_ctdb_context;
 196         struct messaging_ctdb_fde_ev *fde_ev;
 197         struct messaging_ctdb_fde *fde;
 198
 199         if (ctx == NULL) {
 200                 return NULL;
 201         }
 202
 203         fde = talloc(mem_ctx, struct messaging_ctdb_fde);
 204         if (fde == NULL) {
 205                 return NULL;
 206         }
 207
 208         for (fde_ev = ctx->fde_evs; fde_ev != NULL; fde_ev = fde_ev->next) {
 209                 if ((fde_ev->ev == ev) &&
 210                     (tevent_fd_get_flags(fde_ev->fde) != 0)) {
 211                         break;
 212                 }
 213         }
 214
 215         if (fde_ev == NULL) {
 216                 int sock = ctdbd_conn_get_fd(ctx->conn);
 217
 218                 fde_ev = talloc(fde, struct messaging_ctdb_fde_ev);
 219                 if (fde_ev == NULL) {
 220                         return NULL;
 221                 }
 222                 fde_ev->fde = tevent_add_fd(
 223                         ev, fde_ev, sock, TEVENT_FD_READ,
 224                         messaging_ctdb_read_handler, ctx);
 225                 if (fde_ev->fde == NULL) {
 226                         TALLOC_FREE(fde);
 227                         return NULL;
 228                 }
 229                 fde_ev->ev = ev;
 230                 fde_ev->ctx = ctx;
 231                 DLIST_ADD(ctx->fde_evs, fde_ev);
 232                 talloc_set_destructor(
 233                         fde_ev, messaging_ctdb_fde_ev_destructor);
 234         } else {
 235                 /*
 236                  * Same trick as with tdb_wrap: The caller will never
 237                  * see the talloc_referenced object, the
 238                  * messaging_ctdb_fde_ev, so problems with
 239                  * talloc_unlink will not happen.
 240                  */
 241                 if (talloc_reference(fde, fde_ev) == NULL) {
 242                         TALLOC_FREE(fde);
 243                         return NULL;
 244                 }
 245         }
 246
 247         fde->fde = fde_ev->fde;
 248         return fde;
 249 }
 250
 251 bool messaging_ctdb_fde_active(struct messaging_ctdb_fde *fde)
 252 {
 253         uint16_t flags;
 254
 255         if (fde == NULL) {
 256                 return false;
 257         }
 258         flags = tevent_fd_get_flags(fde->fde);
 259         return (flags != 0);
 260 }
 261
 262 struct ctdbd_connection *messaging_ctdb_connection(void)
 263 {
 264         if (global_ctdb_context == NULL) {
 265                 smb_panic("messaging not initialized\n");
 266         }
 267         return global_ctdb_context->conn;
 268 }