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