source/cluster/ctdb/common/ctdb.c

   1 /*
   2    ctdb main protocol code
   3
   4    Copyright (C) Andrew Tridgell  2006
   5
   6    This library is free software; you can redistribute it and/or
   7    modify it under the terms of the GNU Lesser General Public
   8    License as published by the Free Software Foundation; either
   9    version 3 of the License, or (at your option) any later version.
  10
  11    This library 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 GNU
  14    Lesser General Public License for more details.
  15
  16    You should have received a copy of the GNU Lesser General Public
  17    License along with this library; if not, write to the Free Software
  18    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  19 */
  20
  21 #include "includes.h"
  22 #include "lib/tdb/include/tdb.h"
  23 #include "lib/events/events.h"
  24 #include "lib/util/dlinklist.h"
  25 #include "system/network.h"
  26 #include "system/filesys.h"
  27 #include "../include/ctdb_private.h"
  28
  29 /*
  30   choose the transport we will use
  31 */
  32 int ctdb_set_transport(struct ctdb_context *ctdb, const char *transport)
  33 {
  34         int ctdb_tcp_init(struct ctdb_context *ctdb);
  35 #ifdef USE_INFINIBAND
  36         int ctdb_ibw_init(struct ctdb_context *ctdb);
  37 #endif /* USE_INFINIBAND */
  38
  39         if (strcmp(transport, "tcp") == 0) {
  40                 return ctdb_tcp_init(ctdb);
  41         }
  42 #ifdef USE_INFINIBAND
  43         if (strcmp(transport, "ib") == 0) {
  44                 return ctdb_ibw_init(ctdb);
  45         }
  46 #endif /* USE_INFINIBAND */
  47
  48         ctdb_set_error(ctdb, "Unknown transport '%s'\n", transport);
  49         return -1;
  50 }
  51
  52 /*
  53   set some ctdb flags
  54 */
  55 void ctdb_set_flags(struct ctdb_context *ctdb, unsigned flags)
  56 {
  57         ctdb->flags |= flags;
  58 }
  59
  60 /*
  61   clear some ctdb flags
  62 */
  63 void ctdb_clear_flags(struct ctdb_context *ctdb, unsigned flags)
  64 {
  65         ctdb->flags &= ~flags;
  66 }
  67
  68 /*
  69   set max acess count before a dmaster migration
  70 */
  71 void ctdb_set_max_lacount(struct ctdb_context *ctdb, unsigned count)
  72 {
  73         ctdb->max_lacount = count;
  74 }
  75
  76 /*
  77   set the directory for the local databases
  78 */
  79 int ctdb_set_tdb_dir(struct ctdb_context *ctdb, const char *dir)
  80 {
  81         if (dir == NULL) {
  82                 ctdb->db_directory = talloc_asprintf(ctdb, "ctdb-%u", ctdb_get_vnn(ctdb));
  83         } else {
  84                 ctdb->db_directory = talloc_strdup(ctdb, dir);
  85         }
  86         if (ctdb->db_directory == NULL) {
  87                 return -1;
  88         }
  89         return 0;
  90 }
  91
  92 /*
  93   add a node to the list of active nodes
  94 */
  95 static int ctdb_add_node(struct ctdb_context *ctdb, char *nstr)
  96 {
  97         struct ctdb_node *node, **nodep;
  98
  99         nodep = talloc_realloc(ctdb, ctdb->nodes, struct ctdb_node *, ctdb->num_nodes+1);
 100         CTDB_NO_MEMORY(ctdb, nodep);
 101
 102         ctdb->nodes = nodep;
 103         nodep = &ctdb->nodes[ctdb->num_nodes];
 104         (*nodep) = talloc_zero(ctdb->nodes, struct ctdb_node);
 105         CTDB_NO_MEMORY(ctdb, *nodep);
 106         node = *nodep;
 107
 108         if (ctdb_parse_address(ctdb, node, nstr, &node->address) != 0) {
 109                 return -1;
 110         }
 111         node->ctdb = ctdb;
 112         node->name = talloc_asprintf(node, "%s:%u",
 113                                      node->address.address,
 114                                      node->address.port);
 115         /* for now we just set the vnn to the line in the file - this
 116            will change! */
 117         node->vnn = ctdb->num_nodes;
 118
 119         if (ctdb->methods->add_node(node) != 0) {
 120                 talloc_free(node);
 121                 return -1;
 122         }
 123
 124         if (ctdb_same_address(&ctdb->address, &node->address)) {
 125                 ctdb->vnn = node->vnn;
 126         }
 127
 128         ctdb->num_nodes++;
 129
 130         return 0;
 131 }
 132
 133 /*
 134   setup the node list from a file
 135 */
 136 int ctdb_set_nlist(struct ctdb_context *ctdb, const char *nlist)
 137 {
 138         char **lines;
 139         int nlines;
 140         int i;
 141
 142         lines = file_lines_load(nlist, &nlines, ctdb);
 143         if (lines == NULL) {
 144                 ctdb_set_error(ctdb, "Failed to load nlist '%s'\n", nlist);
 145                 return -1;
 146         }
 147
 148         for (i=0;i<nlines;i++) {
 149                 if (ctdb_add_node(ctdb, lines[i]) != 0) {
 150                         talloc_free(lines);
 151                         return -1;
 152                 }
 153         }
 154
 155         talloc_free(lines);
 156         return 0;
 157 }
 158
 159 /*
 160   setup the local node address
 161 */
 162 int ctdb_set_address(struct ctdb_context *ctdb, const char *address)
 163 {
 164         if (ctdb_parse_address(ctdb, ctdb, address, &ctdb->address) != 0) {
 165                 return -1;
 166         }
 167
 168         ctdb->name = talloc_asprintf(ctdb, "%s:%u",
 169                                      ctdb->address.address,
 170                                      ctdb->address.port);
 171         return 0;
 172 }
 173
 174 /*
 175   add a node to the list of active nodes
 176 */
 177 int ctdb_set_call(struct ctdb_db_context *ctdb_db, ctdb_fn_t fn, int id)
 178 {
 179         struct ctdb_registered_call *call;
 180
 181         call = talloc(ctdb_db, struct ctdb_registered_call);
 182         call->fn = fn;
 183         call->id = id;
 184
 185         DLIST_ADD(ctdb_db->calls, call);
 186         return 0;
 187 }
 188
 189 /*
 190   return the vnn of this node
 191 */
 192 uint32_t ctdb_get_vnn(struct ctdb_context *ctdb)
 193 {
 194         return ctdb->vnn;
 195 }
 196
 197 /*
 198   return the number of nodes
 199 */
 200 uint32_t ctdb_get_num_nodes(struct ctdb_context *ctdb)
 201 {
 202         return ctdb->num_nodes;
 203 }
 204
 205
 206 /*
 207   called by the transport layer when a packet comes in
 208 */
 209 void ctdb_recv_pkt(struct ctdb_context *ctdb, uint8_t *data, uint32_t length)
 210 {
 211         struct ctdb_req_header *hdr = (struct ctdb_req_header *)data;
 212         TALLOC_CTX *tmp_ctx;
 213
 214         ctdb->status.node_packets_recv++;
 215
 216         /* place the packet as a child of the tmp_ctx. We then use
 217            talloc_free() below to free it. If any of the calls want
 218            to keep it, then they will steal it somewhere else, and the
 219            talloc_free() will only free the tmp_ctx */
 220         tmp_ctx = talloc_new(ctdb);
 221         talloc_steal(tmp_ctx, hdr);
 222
 223         if (length < sizeof(*hdr)) {
 224                 ctdb_set_error(ctdb, "Bad packet length %d\n", length);
 225                 goto done;
 226         }
 227         if (length != hdr->length) {
 228                 ctdb_set_error(ctdb, "Bad header length %d expected %d\n",
 229                                hdr->length, length);
 230                 goto done;
 231         }
 232
 233         if (hdr->ctdb_magic != CTDB_MAGIC) {
 234                 ctdb_set_error(ctdb, "Non CTDB packet rejected\n");
 235                 goto done;
 236         }
 237
 238         if (hdr->ctdb_version != CTDB_VERSION) {
 239                 ctdb_set_error(ctdb, "Bad CTDB version 0x%x rejected\n", hdr->ctdb_version);
 240                 goto done;
 241         }
 242
 243         DEBUG(3,(__location__ " ctdb request %d of type %d length %d from "
 244                  "node %d to %d\n", hdr->reqid, hdr->operation, hdr->length,
 245                  hdr->srcnode, hdr->destnode));
 246
 247         switch (hdr->operation) {
 248         case CTDB_REQ_CALL:
 249                 ctdb->status.count.req_call++;
 250                 ctdb_request_call(ctdb, hdr);
 251                 break;
 252
 253         case CTDB_REPLY_CALL:
 254                 ctdb->status.count.reply_call++;
 255                 ctdb_reply_call(ctdb, hdr);
 256                 break;
 257
 258         case CTDB_REPLY_ERROR:
 259                 ctdb->status.count.reply_error++;
 260                 ctdb_reply_error(ctdb, hdr);
 261                 break;
 262
 263         case CTDB_REPLY_REDIRECT:
 264                 ctdb->status.count.reply_redirect++;
 265                 ctdb_reply_redirect(ctdb, hdr);
 266                 break;
 267
 268         case CTDB_REQ_DMASTER:
 269                 ctdb->status.count.req_dmaster++;
 270                 ctdb_request_dmaster(ctdb, hdr);
 271                 break;
 272
 273         case CTDB_REPLY_DMASTER:
 274                 ctdb->status.count.reply_dmaster++;
 275                 ctdb_reply_dmaster(ctdb, hdr);
 276                 break;
 277
 278         case CTDB_REQ_MESSAGE:
 279                 ctdb->status.count.req_message++;
 280                 ctdb_request_message(ctdb, hdr);
 281                 break;
 282
 283         case CTDB_REQ_FINISHED:
 284                 ctdb->status.count.req_finished++;
 285                 ctdb_request_finished(ctdb, hdr);
 286                 break;
 287
 288         default:
 289                 DEBUG(0,("%s: Packet with unknown operation %d\n",
 290                          __location__, hdr->operation));
 291                 break;
 292         }
 293
 294 done:
 295         talloc_free(tmp_ctx);
 296 }
 297
 298 /*
 299   called by the transport layer when a packet comes in
 300 */
 301 void ctdb_recv_raw_pkt(void *p, uint8_t *data, uint32_t length)
 302 {
 303         struct ctdb_context *ctdb = talloc_get_type(p, struct ctdb_context);
 304         ctdb_recv_pkt(ctdb, data, length);
 305 }
 306
 307 /*
 308   called by the transport layer when a node is dead
 309 */
 310 static void ctdb_node_dead(struct ctdb_node *node)
 311 {
 312         node->ctdb->num_connected--;
 313         DEBUG(1,("%s: node %s is dead: %d connected\n",
 314                  node->ctdb->name, node->name, node->ctdb->num_connected));
 315 }
 316
 317 /*
 318   called by the transport layer when a node is connected
 319 */
 320 static void ctdb_node_connected(struct ctdb_node *node)
 321 {
 322         node->ctdb->num_connected++;
 323         DEBUG(1,("%s: connected to %s - %d connected\n",
 324                  node->ctdb->name, node->name, node->ctdb->num_connected));
 325 }
 326
 327 /*
 328   wait for all nodes to be connected
 329 */
 330 void ctdb_daemon_connect_wait(struct ctdb_context *ctdb)
 331 {
 332         int expected = ctdb->num_nodes - 1;
 333         if (ctdb->flags & CTDB_FLAG_SELF_CONNECT) {
 334                 expected++;
 335         }
 336         while (ctdb->num_connected != expected) {
 337                 DEBUG(3,("ctdb_connect_wait: waiting for %d nodes (have %d)\n",
 338                          expected, ctdb->num_connected));
 339                 event_loop_once(ctdb->ev);
 340         }
 341         DEBUG(3,("ctdb_connect_wait: got all %d nodes\n", expected));
 342 }
 343
 344 struct queue_next {
 345         struct ctdb_context *ctdb;
 346         struct ctdb_req_header *hdr;
 347 };
 348
 349
 350 /*
 351   trigered when a deferred packet is due
 352  */
 353 static void queue_next_trigger(struct event_context *ev, struct timed_event *te,
 354                                struct timeval t, void *private_data)
 355 {
 356         struct queue_next *q = talloc_get_type(private_data, struct queue_next);
 357         ctdb_recv_pkt(q->ctdb, (uint8_t *)q->hdr, q->hdr->length);
 358         talloc_free(q);
 359 }
 360
 361 /*
 362   defer a packet, so it is processed on the next event loop
 363   this is used for sending packets to ourselves
 364  */
 365 static void ctdb_defer_packet(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
 366 {
 367         struct queue_next *q;
 368         q = talloc(ctdb, struct queue_next);
 369         if (q == NULL) {
 370                 DEBUG(0,(__location__ " Failed to allocate deferred packet\n"));
 371                 return;
 372         }
 373         q->ctdb = ctdb;
 374         q->hdr = talloc_memdup(ctdb, hdr, hdr->length);
 375         if (q->hdr == NULL) {
 376                 DEBUG(0,("Error copying deferred packet to self\n"));
 377                 return;
 378         }
 379         event_add_timed(ctdb->ev, q, timeval_zero(), queue_next_trigger, q);
 380 }
 381
 382 /*
 383   queue a packet or die
 384 */
 385 void ctdb_queue_packet(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
 386 {
 387         struct ctdb_node *node;
 388         ctdb->status.node_packets_sent++;
 389         node = ctdb->nodes[hdr->destnode];
 390         if (hdr->destnode == ctdb->vnn && !(ctdb->flags & CTDB_FLAG_SELF_CONNECT)) {
 391                 ctdb_defer_packet(ctdb, hdr);
 392         } else if (ctdb->methods->queue_pkt(node, (uint8_t *)hdr, hdr->length) != 0) {
 393                 ctdb_fatal(ctdb, "Unable to queue packet\n");
 394         }
 395 }
 396
 397
 398 static const struct ctdb_upcalls ctdb_upcalls = {
 399         .recv_pkt       = ctdb_recv_pkt,
 400         .node_dead      = ctdb_node_dead,
 401         .node_connected = ctdb_node_connected
 402 };
 403
 404 /*
 405   initialise the ctdb daemon.
 406
 407   NOTE: In current code the daemon does not fork. This is for testing purposes only
 408   and to simplify the code.
 409 */
 410 struct ctdb_context *ctdb_init(struct event_context *ev)
 411 {
 412         struct ctdb_context *ctdb;
 413
 414         ctdb = talloc_zero(ev, struct ctdb_context);
 415         ctdb->ev = ev;
 416         ctdb->upcalls = &ctdb_upcalls;
 417         ctdb->idr = idr_init(ctdb);
 418         ctdb->max_lacount = CTDB_DEFAULT_MAX_LACOUNT;
 419
 420         return ctdb;
 421 }
 422