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