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