ctdb/tests/src/fake_ctdbd.c

   1 /*
   2    Fake CTDB server for testing
   3
   4    Copyright (C) Amitay Isaacs  2016
   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 "replace.h"
  21 #include "system/network.h"
  22 #include "system/time.h"
  23
  24 #include <popt.h>
  25 #include <talloc.h>
  26 #include <tevent.h>
  27 #include <tdb.h>
  28
  29 #include "lib/util/dlinklist.h"
  30 #include "lib/util/tevent_unix.h"
  31 #include "lib/util/debug.h"
  32 #include "lib/util/samba_util.h"
  33 #include "lib/async_req/async_sock.h"
  34
  35 #include "protocol/protocol.h"
  36 #include "protocol/protocol_api.h"
  37
  38 #include "common/comm.h"
  39 #include "common/system.h"
  40 #include "common/logging.h"
  41
  42
  43 #define CTDB_PORT 4379
  44
  45 /* A fake flag that is only supported by some functions */
  46 #define NODE_FLAGS_FAKE_TIMEOUT 0x80000000
  47
  48 struct node {
  49         ctdb_sock_addr addr;
  50         uint32_t pnn;
  51         uint32_t flags;
  52         uint32_t capabilities;
  53         bool recovery_disabled;
  54         void *recovery_substate;
  55 };
  56
  57 struct node_map {
  58         uint32_t num_nodes;
  59         struct node *node;
  60         uint32_t pnn;
  61         uint32_t recmaster;
  62 };
  63
  64 struct interface {
  65         const char *name;
  66         bool link_up;
  67         uint32_t references;
  68 };
  69
  70 struct interface_map {
  71         int num;
  72         struct interface *iface;
  73 };
  74
  75 struct vnn_map {
  76         uint32_t recmode;
  77         uint32_t generation;
  78         uint32_t size;
  79         uint32_t *map;
  80 };
  81
  82 struct srvid_register_state {
  83         struct srvid_register_state *prev, *next;
  84         struct ctdbd_context *ctdb;
  85         uint64_t srvid;
  86 };
  87
  88 struct ctdbd_context {
  89         struct node_map *node_map;
  90         struct interface_map *iface_map;
  91         struct vnn_map *vnn_map;
  92         struct srvid_register_state *rstate;
  93         int num_clients;
  94         struct timeval start_time;
  95         struct timeval recovery_start_time;
  96         struct timeval recovery_end_time;
  97         bool takeover_disabled;
  98         enum debug_level log_level;
  99 };
 100
 101 /*
 102  * Parse routines
 103  */
 104
 105 static struct node_map *nodemap_init(TALLOC_CTX *mem_ctx)
 106 {
 107         struct node_map *node_map;
 108
 109         node_map = talloc_zero(mem_ctx, struct node_map);
 110         if (node_map == NULL) {
 111                 return NULL;
 112         }
 113
 114         node_map->pnn = CTDB_UNKNOWN_PNN;
 115         node_map->recmaster = CTDB_UNKNOWN_PNN;
 116
 117         return node_map;
 118 }
 119
 120 /* Read a nodemap from stdin.  Each line looks like:
 121  *  <PNN> <FLAGS> [RECMASTER] [CURRENT] [CAPABILITIES]
 122  * EOF or a blank line terminates input.
 123  *
 124  * By default, capablities for each node are
 125  * CTDB_CAP_RECMASTER|CTDB_CAP_LMASTER.  These 2
 126  * capabilities can be faked off by adding, for example,
 127  * -CTDB_CAP_RECMASTER.
 128  */
 129
 130 static bool nodemap_parse(struct node_map *node_map)
 131 {
 132         char line[1024];
 133
 134         while ((fgets(line, sizeof(line), stdin) != NULL)) {
 135                 uint32_t pnn, flags, capabilities;
 136                 char *tok, *t;
 137                 char *ip;
 138                 ctdb_sock_addr saddr;
 139                 struct node *node;
 140
 141                 if (line[0] == '\n') {
 142                         break;
 143                 }
 144
 145                 /* Get rid of pesky newline */
 146                 if ((t = strchr(line, '\n')) != NULL) {
 147                         *t = '\0';
 148                 }
 149
 150                 /* Get PNN */
 151                 tok = strtok(line, " \t");
 152                 if (tok == NULL) {
 153                         fprintf(stderr, "bad line (%s) - missing PNN\n", line);
 154                         continue;
 155                 }
 156                 pnn = (uint32_t)strtoul(tok, NULL, 0);
 157
 158                 /* Get IP */
 159                 tok = strtok(NULL, " \t");
 160                 if (tok == NULL) {
 161                         fprintf(stderr, "bad line (%s) - missing IP\n", line);
 162                         continue;
 163                 }
 164                 if (!parse_ip(tok, NULL, CTDB_PORT, &saddr)) {
 165                         fprintf(stderr, "bad line (%s) - invalid IP\n", line);
 166                         continue;
 167                 }
 168                 ip = talloc_strdup(node_map, tok);
 169                 if (ip == NULL) {
 170                         goto fail;
 171                 }
 172
 173                 /* Get flags */
 174                 tok = strtok(NULL, " \t");
 175                 if (tok == NULL) {
 176                         fprintf(stderr, "bad line (%s) - missing flags\n",
 177                                 line);
 178                         continue;
 179                 }
 180                 flags = (uint32_t)strtoul(tok, NULL, 0);
 181                 /* Handle deleted nodes */
 182                 if (flags & NODE_FLAGS_DELETED) {
 183                         talloc_free(ip);
 184                         ip = talloc_strdup(node_map, "0.0.0.0");
 185                         if (ip == NULL) {
 186                                 goto fail;
 187                         }
 188                 }
 189                 capabilities = CTDB_CAP_RECMASTER|CTDB_CAP_LMASTER;
 190
 191                 tok = strtok(NULL, " \t");
 192                 while (tok != NULL) {
 193                         if (strcmp(tok, "CURRENT") == 0) {
 194                                 node_map->pnn = pnn;
 195                         } else if (strcmp(tok, "RECMASTER") == 0) {
 196                                 node_map->recmaster = pnn;
 197                         } else if (strcmp(tok, "-CTDB_CAP_RECMASTER") == 0) {
 198                                 capabilities &= ~CTDB_CAP_RECMASTER;
 199                         } else if (strcmp(tok, "-CTDB_CAP_LMASTER") == 0) {
 200                                 capabilities &= ~CTDB_CAP_LMASTER;
 201                         } else if (strcmp(tok, "TIMEOUT") == 0) {
 202                                 /* This can be done with just a flag
 203                                  * value but it is probably clearer
 204                                  * and less error-prone to fake this
 205                                  * with an explicit token */
 206                                 flags |= NODE_FLAGS_FAKE_TIMEOUT;
 207                         }
 208                         tok = strtok(NULL, " \t");
 209                 }
 210
 211                 node_map->node = talloc_realloc(node_map, node_map->node,
 212                                                 struct node,
 213                                                 node_map->num_nodes + 1);
 214                 if (node_map->node == NULL) {
 215                         goto fail;
 216                 }
 217                 node = &node_map->node[node_map->num_nodes];
 218
 219                 parse_ip(ip, NULL, CTDB_PORT, &node->addr);
 220                 node->pnn = pnn;
 221                 node->flags = flags;
 222                 node->capabilities = capabilities;
 223                 node->recovery_disabled = false;
 224                 node->recovery_substate = NULL;
 225
 226                 node_map->num_nodes += 1;
 227         }
 228
 229         DEBUG(DEBUG_INFO, ("Parsing nodemap done\n"));
 230         return true;
 231
 232 fail:
 233         DEBUG(DEBUG_INFO, ("Parsing nodemap failed\n"));
 234         return false;
 235
 236 }
 237
 238 /* Append a node to a node map with given address and flags */
 239 static bool node_map_add(struct ctdb_node_map *nodemap,
 240                          const char *nstr, uint32_t flags)
 241 {
 242         ctdb_sock_addr addr;
 243         uint32_t num;
 244         struct ctdb_node_and_flags *n;
 245
 246         if (! parse_ip(nstr, NULL, CTDB_PORT, &addr)) {
 247                 fprintf(stderr, "Invalid IP address %s\n", nstr);
 248                 return false;
 249         }
 250
 251         num = nodemap->num;
 252         nodemap->node = talloc_realloc(nodemap, nodemap->node,
 253                                        struct ctdb_node_and_flags, num+1);
 254         if (nodemap->node == NULL) {
 255                 return false;
 256         }
 257
 258         n = &nodemap->node[num];
 259         n->addr = addr;
 260         n->pnn = num;
 261         n->flags = flags;
 262
 263         nodemap->num = num+1;
 264         return true;
 265 }
 266
 267 /* Read a nodes file into a node map */
 268 static struct ctdb_node_map *ctdb_read_nodes_file(TALLOC_CTX *mem_ctx,
 269                                                   const char *nlist)
 270 {
 271         char **lines;
 272         int nlines;
 273         int i;
 274         struct ctdb_node_map *nodemap;
 275
 276         nodemap = talloc_zero(mem_ctx, struct ctdb_node_map);
 277         if (nodemap == NULL) {
 278                 return NULL;
 279         }
 280
 281         lines = file_lines_load(nlist, &nlines, 0, mem_ctx);
 282         if (lines == NULL) {
 283                 return NULL;
 284         }
 285
 286         while (nlines > 0 && strcmp(lines[nlines-1], "") == 0) {
 287                 nlines--;
 288         }
 289
 290         for (i=0; i<nlines; i++) {
 291                 char *node;
 292                 uint32_t flags;
 293                 size_t len;
 294
 295                 node = lines[i];
 296                 /* strip leading spaces */
 297                 while((*node == ' ') || (*node == '\t')) {
 298                         node++;
 299                 }
 300
 301                 len = strlen(node);
 302
 303                 /* strip trailing spaces */
 304                 while ((len > 1) &&
 305                        ((node[len-1] == ' ') || (node[len-1] == '\t')))
 306                 {
 307                         node[len-1] = '\0';
 308                         len--;
 309                 }
 310
 311                 if (len == 0) {
 312                         continue;
 313                 }
 314                 if (*node == '#') {
 315                         /* A "deleted" node is a node that is
 316                            commented out in the nodes file.  This is
 317                            used instead of removing a line, which
 318                            would cause subsequent nodes to change
 319                            their PNN. */
 320                         flags = NODE_FLAGS_DELETED;
 321                         node = discard_const("0.0.0.0");
 322                 } else {
 323                         flags = 0;
 324                 }
 325                 if (! node_map_add(nodemap, node, flags)) {
 326                         talloc_free(lines);
 327                         TALLOC_FREE(nodemap);
 328                         return NULL;
 329                 }
 330         }
 331
 332         talloc_free(lines);
 333         return nodemap;
 334 }
 335
 336 static struct ctdb_node_map *read_nodes_file(TALLOC_CTX *mem_ctx,
 337                                              uint32_t pnn)
 338 {
 339         struct ctdb_node_map *nodemap;
 340         char nodepath[PATH_MAX];
 341         const char *nodes_list;
 342
 343         /* read the nodes file */
 344         sprintf(nodepath, "CTDB_NODES_%u", pnn);
 345         nodes_list = getenv(nodepath);
 346         if (nodes_list == NULL) {
 347                 nodes_list = getenv("CTDB_NODES");
 348                 if (nodes_list == NULL) {
 349                         DEBUG(DEBUG_INFO, ("Nodes file not defined\n"));
 350                         return NULL;
 351                 }
 352         }
 353
 354         nodemap = ctdb_read_nodes_file(mem_ctx, nodes_list);
 355         if (nodemap == NULL) {
 356                 DEBUG(DEBUG_INFO, ("Failed to read nodes file \"%s\"\n",
 357                                    nodes_list));
 358                 return NULL;
 359         }
 360
 361         return nodemap;
 362 }
 363
 364 static struct interface_map *interfaces_init(TALLOC_CTX *mem_ctx)
 365 {
 366         struct interface_map *iface_map;
 367
 368         iface_map = talloc_zero(mem_ctx, struct interface_map);
 369         if (iface_map == NULL) {
 370                 return NULL;
 371         }
 372
 373         return iface_map;
 374 }
 375
 376 /* Read interfaces information.  Same format as "ctdb ifaces -Y"
 377  * output:
 378  *   :Name:LinkStatus:References:
 379  *   :eth2:1:4294967294
 380  *   :eth1:1:4294967292
 381  */
 382
 383 static bool interfaces_parse(struct interface_map *iface_map)
 384 {
 385         char line[1024];
 386
 387         while ((fgets(line, sizeof(line), stdin) != NULL)) {
 388                 uint16_t link_state;
 389                 uint32_t references;
 390                 char *tok, *t, *name;
 391                 struct interface *iface;
 392
 393                 if (line[0] == '\n') {
 394                         break;
 395                 }
 396
 397                 /* Get rid of pesky newline */
 398                 if ((t = strchr(line, '\n')) != NULL) {
 399                         *t = '\0';
 400                 }
 401
 402                 if (strcmp(line, ":Name:LinkStatus:References:") == 0) {
 403                         continue;
 404                 }
 405
 406                 /* Leading colon... */
 407                 // tok = strtok(line, ":");
 408
 409                 /* name */
 410                 tok = strtok(line, ":");
 411                 if (tok == NULL) {
 412                         fprintf(stderr, "bad line (%s) - missing name\n", line);
 413                         continue;
 414                 }
 415                 name = tok;
 416
 417                 /* link_state */
 418                 tok = strtok(NULL, ":");
 419                 if (tok == NULL) {
 420                         fprintf(stderr, "bad line (%s) - missing link state\n",
 421                                 line);
 422                         continue;
 423                 }
 424                 link_state = (uint16_t)strtoul(tok, NULL, 0);
 425
 426                 /* references... */
 427                 tok = strtok(NULL, ":");
 428                 if (tok == NULL) {
 429                         fprintf(stderr, "bad line (%s) - missing references\n",
 430                                 line);
 431                         continue;
 432                 }
 433                 references = (uint32_t)strtoul(tok, NULL, 0);
 434
 435                 iface_map->iface = talloc_realloc(iface_map, iface_map->iface,
 436                                                   struct interface,
 437                                                   iface_map->num + 1);
 438                 if (iface_map->iface == NULL) {
 439                         goto fail;
 440                 }
 441
 442                 iface = &iface_map->iface[iface_map->num];
 443
 444                 iface->name = talloc_strdup(iface_map, name);
 445                 if (iface->name == NULL) {
 446                         goto fail;
 447                 }
 448                 iface->link_up = link_state;
 449                 iface->references = references;
 450
 451                 iface_map->num += 1;
 452         }
 453
 454         DEBUG(DEBUG_INFO, ("Parsing interfaces done\n"));
 455         return true;
 456
 457 fail:
 458         fprintf(stderr, "Parsing interfaces failed\n");
 459         return false;
 460 }
 461
 462 static struct vnn_map *vnnmap_init(TALLOC_CTX *mem_ctx)
 463 {
 464         struct vnn_map *vnn_map;
 465
 466         vnn_map = talloc_zero(mem_ctx, struct vnn_map);
 467         if (vnn_map == NULL) {
 468                 fprintf(stderr, "Memory error\n");
 469                 return NULL;
 470         }
 471         vnn_map->recmode = CTDB_RECOVERY_ACTIVE;
 472         vnn_map->generation = INVALID_GENERATION;
 473
 474         return vnn_map;
 475 }
 476
 477 /* Read vnn map.
 478  * output:
 479  *   <GENERATION>
 480  *   <LMASTER0>
 481  *   <LMASTER1>
 482  *   ...
 483  */
 484
 485 static bool vnnmap_parse(struct vnn_map *vnn_map)
 486 {
 487         char line[1024];
 488
 489         while (fgets(line, sizeof(line), stdin) != NULL) {
 490                 uint32_t n;
 491                 char *t;
 492
 493                 if (line[0] == '\n') {
 494                         break;
 495                 }
 496
 497                 /* Get rid of pesky newline */
 498                 if ((t = strchr(line, '\n')) != NULL) {
 499                         *t = '\0';
 500                 }
 501
 502                 n = (uint32_t) strtol(line, NULL, 0);
 503
 504                 /* generation */
 505                 if (vnn_map->generation == INVALID_GENERATION) {
 506                         vnn_map->generation = n;
 507                         continue;
 508                 }
 509
 510                 vnn_map->map = talloc_realloc(vnn_map, vnn_map->map, uint32_t,
 511                                               vnn_map->size + 1);
 512                 if (vnn_map->map == NULL) {
 513                         fprintf(stderr, "Memory error\n");
 514                         goto fail;
 515                 }
 516
 517                 vnn_map->map[vnn_map->size] = n;
 518                 vnn_map->size += 1;
 519         }
 520
 521         DEBUG(DEBUG_INFO, ("Parsing vnnmap done\n"));
 522         return true;
 523
 524 fail:
 525         fprintf(stderr, "Parsing vnnmap failed\n");
 526         return false;
 527 }
 528
 529 /*
 530  * CTDB context setup
 531  */
 532
 533 static uint32_t new_generation(uint32_t old_generation)
 534 {
 535         uint32_t generation;
 536
 537         while (1) {
 538                 generation = random();
 539                 if (generation != INVALID_GENERATION &&
 540                     generation != old_generation) {
 541                         break;
 542                 }
 543         }
 544
 545         return generation;
 546 }
 547
 548 static struct ctdbd_context *ctdbd_setup(TALLOC_CTX *mem_ctx)
 549 {
 550         struct ctdbd_context *ctdb;
 551         char line[1024];
 552         bool status;
 553
 554         ctdb = talloc_zero(mem_ctx, struct ctdbd_context);
 555         if (ctdb == NULL) {
 556                 return NULL;
 557         }
 558
 559         ctdb->node_map = nodemap_init(ctdb);
 560         if (ctdb->node_map == NULL) {
 561                 goto fail;
 562         }
 563
 564         ctdb->iface_map = interfaces_init(ctdb);
 565         if (ctdb->iface_map == NULL) {
 566                 goto fail;
 567         }
 568
 569         ctdb->vnn_map = vnnmap_init(ctdb);
 570         if (ctdb->vnn_map == NULL) {
 571                 goto fail;
 572         }
 573
 574         while (fgets(line, sizeof(line), stdin) != NULL) {
 575                 char *t;
 576
 577                 if ((t = strchr(line, '\n')) != NULL) {
 578                         *t = '\0';
 579                 }
 580
 581                 if (strcmp(line, "NODEMAP") == 0) {
 582                         status = nodemap_parse(ctdb->node_map);
 583                 } else if (strcmp(line, "IFACES") == 0) {
 584                         status = interfaces_parse(ctdb->iface_map);
 585                 } else if (strcmp(line, "VNNMAP") == 0) {
 586                         status = vnnmap_parse(ctdb->vnn_map);
 587                 } else {
 588                         fprintf(stderr, "Unknown line %s\n", line);
 589                         status = false;
 590                 }
 591
 592                 if (! status) {
 593                         goto fail;
 594                 }
 595         }
 596
 597         ctdb->start_time = tevent_timeval_current();
 598         ctdb->recovery_start_time = tevent_timeval_current();
 599         ctdb->vnn_map->recmode = CTDB_RECOVERY_NORMAL;
 600         if (ctdb->vnn_map->generation == INVALID_GENERATION) {
 601                 ctdb->vnn_map->generation =
 602                         new_generation(ctdb->vnn_map->generation);
 603         }
 604         ctdb->recovery_end_time = tevent_timeval_current();
 605
 606         ctdb->log_level = DEBUG_ERR;
 607
 608         return ctdb;
 609
 610 fail:
 611         TALLOC_FREE(ctdb);
 612         return NULL;
 613 }
 614
 615 static bool ctdbd_verify(struct ctdbd_context *ctdb)
 616 {
 617         struct node *node;
 618         int i;
 619
 620         if (ctdb->node_map->num_nodes == 0) {
 621                 return true;
 622         }
 623
 624         /* Make sure all the nodes are in order */
 625         for (i=0; i<ctdb->node_map->num_nodes; i++) {
 626                 node = &ctdb->node_map->node[i];
 627                 if (node->pnn != i) {
 628                         fprintf(stderr, "Expected node %u, found %u\n",
 629                                 i, node->pnn);
 630                         return false;
 631                 }
 632         }
 633
 634         node = &ctdb->node_map->node[ctdb->node_map->pnn];
 635         if (node->flags & NODE_FLAGS_DISCONNECTED) {
 636                 DEBUG(DEBUG_INFO, ("Node disconnected, exiting\n"));
 637                 exit(0);
 638         }
 639
 640         return true;
 641 }
 642
 643 /*
 644  * Doing a recovery
 645  */
 646
 647 struct recover_state {
 648         struct tevent_context *ev;
 649         struct ctdbd_context *ctdb;
 650 };
 651
 652 static int recover_check(struct tevent_req *req);
 653 static void recover_wait_done(struct tevent_req *subreq);
 654 static void recover_done(struct tevent_req *subreq);
 655
 656 static struct tevent_req *recover_send(TALLOC_CTX *mem_ctx,
 657                                        struct tevent_context *ev,
 658                                        struct ctdbd_context *ctdb)
 659 {
 660         struct tevent_req *req;
 661         struct recover_state *state;
 662         int ret;
 663
 664         req = tevent_req_create(mem_ctx, &state, struct recover_state);
 665         if (req == NULL) {
 666                 return NULL;
 667         }
 668
 669         state->ev = ev;
 670         state->ctdb = ctdb;
 671
 672         ret = recover_check(req);
 673         if (ret != 0) {
 674                 tevent_req_error(req, ret);
 675                 return tevent_req_post(req, ev);
 676         }
 677
 678         return req;
 679 }
 680
 681 static int recover_check(struct tevent_req *req)
 682 {
 683         struct recover_state *state = tevent_req_data(
 684                 req, struct recover_state);
 685         struct ctdbd_context *ctdb = state->ctdb;
 686         struct tevent_req *subreq;
 687         bool recovery_disabled;
 688         int i;
 689
 690         recovery_disabled = false;
 691         for (i=0; i<ctdb->node_map->num_nodes; i++) {
 692                 if (ctdb->node_map->node[i].recovery_disabled) {
 693                         recovery_disabled = true;
 694                         break;
 695                 }
 696         }
 697
 698         subreq = tevent_wakeup_send(state, state->ev,
 699                                     tevent_timeval_current_ofs(1, 0));
 700         if (subreq == NULL) {
 701                 return ENOMEM;
 702         }
 703
 704         if (recovery_disabled) {
 705                 tevent_req_set_callback(subreq, recover_wait_done, req);
 706         } else {
 707                 ctdb->recovery_start_time = tevent_timeval_current();
 708                 tevent_req_set_callback(subreq, recover_done, req);
 709         }
 710
 711         return 0;
 712 }
 713
 714 static void recover_wait_done(struct tevent_req *subreq)
 715 {
 716         struct tevent_req *req = tevent_req_callback_data(
 717                 subreq, struct tevent_req);
 718         int ret;
 719         bool status;
 720
 721         status = tevent_wakeup_recv(subreq);
 722         TALLOC_FREE(subreq);
 723         if (! status) {
 724                 tevent_req_error(req, EIO);
 725                 return;
 726         }
 727
 728         ret = recover_check(req);
 729         if (ret != 0) {
 730                 tevent_req_error(req, ret);
 731         }
 732 }
 733
 734 static void recover_done(struct tevent_req *subreq)
 735 {
 736         struct tevent_req *req = tevent_req_callback_data(
 737                 subreq, struct tevent_req);
 738         struct recover_state *state = tevent_req_data(
 739                 req, struct recover_state);
 740         struct ctdbd_context *ctdb = state->ctdb;
 741         bool status;
 742
 743         status = tevent_wakeup_recv(subreq);
 744         TALLOC_FREE(subreq);
 745         if (! status) {
 746                 tevent_req_error(req, EIO);
 747                 return;
 748         }
 749
 750         ctdb->vnn_map->recmode = CTDB_RECOVERY_NORMAL;
 751         ctdb->recovery_end_time = tevent_timeval_current();
 752         ctdb->vnn_map->generation = new_generation(ctdb->vnn_map->generation);
 753
 754         tevent_req_done(req);
 755 }
 756
 757 static bool recover_recv(struct tevent_req *req, int *perr)
 758 {
 759         int err;
 760
 761         if (tevent_req_is_unix_error(req, &err)) {
 762                 if (perr != NULL) {
 763                         *perr = err;
 764                 }
 765                 return false;
 766         }
 767
 768         return true;
 769 }
 770
 771 /*
 772  * Routines for ctdb_req_header
 773  */
 774
 775 static void header_fix_pnn(struct ctdb_req_header *header,
 776                            struct ctdbd_context *ctdb)
 777 {
 778         if (header->srcnode == CTDB_CURRENT_NODE) {
 779                 header->srcnode = ctdb->node_map->pnn;
 780         }
 781
 782         if (header->destnode == CTDB_CURRENT_NODE) {
 783                 header->destnode = ctdb->node_map->pnn;
 784         }
 785 }
 786
 787 static struct ctdb_req_header header_reply_control(
 788                                         struct ctdb_req_header *header,
 789                                         struct ctdbd_context *ctdb)
 790 {
 791         struct ctdb_req_header reply_header;
 792
 793         reply_header = (struct ctdb_req_header) {
 794                 .ctdb_magic = CTDB_MAGIC,
 795                 .ctdb_version = CTDB_PROTOCOL,
 796                 .generation = ctdb->vnn_map->generation,
 797                 .operation = CTDB_REPLY_CONTROL,
 798                 .destnode = header->srcnode,
 799                 .srcnode = header->destnode,
 800                 .reqid = header->reqid,
 801         };
 802
 803         return reply_header;
 804 }
 805
 806 static struct ctdb_req_header header_reply_message(
 807                                         struct ctdb_req_header *header,
 808                                         struct ctdbd_context *ctdb)
 809 {
 810         struct ctdb_req_header reply_header;
 811
 812         reply_header = (struct ctdb_req_header) {
 813                 .ctdb_magic = CTDB_MAGIC,
 814                 .ctdb_version = CTDB_PROTOCOL,
 815                 .generation = ctdb->vnn_map->generation,
 816                 .operation = CTDB_REQ_MESSAGE,
 817                 .destnode = header->srcnode,
 818                 .srcnode = header->destnode,
 819                 .reqid = 0,
 820         };
 821
 822         return reply_header;
 823 }
 824
 825 /*
 826  * Client state
 827  */
 828
 829 struct client_state {
 830         struct tevent_context *ev;
 831         int fd;
 832         struct ctdbd_context *ctdb;
 833         int pnn;
 834         struct comm_context *comm;
 835         struct srvid_register_state *rstate;
 836         int status;
 837 };
 838
 839 /*
 840  * Send replies to controls and messages
 841  */
 842
 843 static void client_reply_done(struct tevent_req *subreq);
 844
 845 static void client_send_message(struct tevent_req *req,
 846                                 struct ctdb_req_header *header,
 847                                 struct ctdb_req_message_data *message)
 848 {
 849         struct client_state *state = tevent_req_data(
 850                 req, struct client_state);
 851         struct ctdbd_context *ctdb = state->ctdb;
 852         struct tevent_req *subreq;
 853         struct ctdb_req_header reply_header;
 854         uint8_t *buf;
 855         size_t datalen, buflen;
 856         int ret;
 857
 858         reply_header = header_reply_message(header, ctdb);
 859
 860         datalen = ctdb_req_message_data_len(&reply_header, message);
 861         ret = ctdb_allocate_pkt(state, datalen, &buf, &buflen);
 862         if (ret != 0) {
 863                 tevent_req_error(req, ret);
 864                 return;
 865         }
 866
 867         ret = ctdb_req_message_data_push(&reply_header, message,
 868                                          buf, &buflen);
 869         if (ret != 0) {
 870                 tevent_req_error(req, ret);
 871                 return;
 872         }
 873
 874         DEBUG(DEBUG_INFO, ("message srvid = 0x%"PRIx64"\n", message->srvid));
 875
 876         subreq = comm_write_send(state, state->ev, state->comm, buf, buflen);
 877         if (tevent_req_nomem(subreq, req)) {
 878                 return;
 879         }
 880         tevent_req_set_callback(subreq, client_reply_done, req);
 881
 882         talloc_steal(subreq, buf);
 883 }
 884
 885 static void client_send_control(struct tevent_req *req,
 886                                 struct ctdb_req_header *header,
 887                                 struct ctdb_reply_control *reply)
 888 {
 889         struct client_state *state = tevent_req_data(
 890                 req, struct client_state);
 891         struct ctdbd_context *ctdb = state->ctdb;
 892         struct tevent_req *subreq;
 893         struct ctdb_req_header reply_header;
 894         uint8_t *buf;
 895         size_t datalen, buflen;
 896         int ret;
 897
 898         reply_header = header_reply_control(header, ctdb);
 899
 900         datalen = ctdb_reply_control_len(&reply_header, reply);
 901         ret = ctdb_allocate_pkt(state, datalen, &buf, &buflen);
 902         if (ret != 0) {
 903                 tevent_req_error(req, ret);
 904                 return;
 905         }
 906
 907         ret = ctdb_reply_control_push(&reply_header, reply, buf, &buflen);
 908         if (ret != 0) {
 909                 tevent_req_error(req, ret);
 910                 return;
 911         }
 912
 913         DEBUG(DEBUG_INFO, ("reply opcode = %u\n", reply->rdata.opcode));
 914
 915         subreq = comm_write_send(state, state->ev, state->comm, buf, buflen);
 916         if (tevent_req_nomem(subreq, req)) {
 917                 return;
 918         }
 919         tevent_req_set_callback(subreq, client_reply_done, req);
 920
 921         talloc_steal(subreq, buf);
 922 }
 923
 924 static void client_reply_done(struct tevent_req *subreq)
 925 {
 926         struct tevent_req *req = tevent_req_callback_data(
 927                 subreq, struct tevent_req);
 928         int ret;
 929         bool status;
 930
 931         status = comm_write_recv(subreq, &ret);
 932         TALLOC_FREE(subreq);
 933         if (! status) {
 934                 tevent_req_error(req, ret);
 935         }
 936 }
 937
 938 /*
 939  * Handling protocol - controls
 940  */
 941
 942 static void control_process_exists(TALLOC_CTX *mem_ctx,
 943                                    struct tevent_req *req,
 944                                    struct ctdb_req_header *header,
 945                                    struct ctdb_req_control *request)
 946 {
 947         struct ctdb_reply_control reply;
 948
 949         reply.rdata.opcode = request->opcode;
 950         reply.status = kill(request->rdata.data.pid, 0);
 951         reply.errmsg = NULL;
 952
 953         client_send_control(req, header, &reply);
 954 }
 955
 956 static void control_ping(TALLOC_CTX *mem_ctx,
 957                          struct tevent_req *req,
 958                          struct ctdb_req_header *header,
 959                          struct ctdb_req_control *request)
 960 {
 961         struct client_state *state = tevent_req_data(
 962                 req, struct client_state);
 963         struct ctdbd_context *ctdb = state->ctdb;
 964         struct ctdb_reply_control reply;
 965
 966         reply.rdata.opcode = request->opcode;
 967         reply.status = ctdb->num_clients;
 968         reply.errmsg = NULL;
 969
 970         client_send_control(req, header, &reply);
 971 }
 972
 973 static void control_getvnnmap(TALLOC_CTX *mem_ctx,
 974                               struct tevent_req *req,
 975                               struct ctdb_req_header *header,
 976                               struct ctdb_req_control *request)
 977 {
 978         struct client_state *state = tevent_req_data(
 979                 req, struct client_state);
 980         struct ctdbd_context *ctdb = state->ctdb;
 981         struct ctdb_reply_control reply;
 982         struct ctdb_vnn_map *vnnmap;
 983
 984         reply.rdata.opcode = request->opcode;
 985
 986         vnnmap = talloc_zero(mem_ctx, struct ctdb_vnn_map);
 987         if (vnnmap == NULL) {
 988                 reply.status = ENOMEM;
 989                 reply.errmsg = "Memory error";
 990         } else {
 991                 vnnmap->generation = ctdb->vnn_map->generation;
 992                 vnnmap->size = ctdb->vnn_map->size;
 993                 vnnmap->map = ctdb->vnn_map->map;
 994
 995                 reply.rdata.data.vnnmap = vnnmap;
 996                 reply.status = 0;
 997                 reply.errmsg = NULL;
 998         }
 999
1000         client_send_control(req, header, &reply);
1001 }
1002
1003 static void control_get_debug(TALLOC_CTX *mem_ctx,
1004                               struct tevent_req *req,
1005                               struct ctdb_req_header *header,
1006                               struct ctdb_req_control *request)
1007 {
1008         struct client_state *state = tevent_req_data(
1009                 req, struct client_state);
1010         struct ctdbd_context *ctdb = state->ctdb;
1011         struct ctdb_reply_control reply;
1012
1013         reply.rdata.opcode = request->opcode;
1014         reply.rdata.data.loglevel = debug_level_to_int(ctdb->log_level);
1015         reply.status = 0;
1016         reply.errmsg = NULL;
1017
1018         client_send_control(req, header, &reply);
1019 }
1020
1021 static void control_set_debug(TALLOC_CTX *mem_ctx,
1022                               struct tevent_req *req,
1023                               struct ctdb_req_header *header,
1024                               struct ctdb_req_control *request)
1025 {
1026         struct client_state *state = tevent_req_data(
1027                 req, struct client_state);
1028         struct ctdbd_context *ctdb = state->ctdb;
1029         struct ctdb_reply_control reply;
1030
1031         ctdb->log_level = debug_level_from_int(request->rdata.data.loglevel);
1032
1033         reply.rdata.opcode = request->opcode;
1034         reply.status = 0;
1035         reply.errmsg = NULL;
1036
1037         client_send_control(req, header, &reply);
1038 }
1039
1040 static void control_get_recmode(TALLOC_CTX *mem_ctx,
1041                                 struct tevent_req *req,
1042                                 struct ctdb_req_header *header,
1043                                 struct ctdb_req_control *request)
1044 {
1045         struct client_state *state = tevent_req_data(
1046                 req, struct client_state);
1047         struct ctdbd_context *ctdb = state->ctdb;
1048         struct ctdb_reply_control reply;
1049
1050         reply.rdata.opcode = request->opcode;
1051         reply.status = ctdb->vnn_map->recmode;
1052         reply.errmsg = NULL;
1053
1054         client_send_control(req, header, &reply);
1055 }
1056
1057 struct set_recmode_state {
1058         struct tevent_req *req;
1059         struct ctdbd_context *ctdb;
1060         struct ctdb_req_header header;
1061         struct ctdb_reply_control reply;
1062 };
1063
1064 static void set_recmode_callback(struct tevent_req *subreq)
1065 {
1066         struct set_recmode_state *substate = tevent_req_callback_data(
1067                 subreq, struct set_recmode_state);
1068         bool status;
1069         int ret;
1070
1071         status = recover_recv(subreq, &ret);
1072         TALLOC_FREE(subreq);
1073         if (! status) {
1074                 substate->reply.status = ret;
1075                 substate->reply.errmsg = "recovery failed";
1076         } else {
1077                 substate->reply.status = 0;
1078                 substate->reply.errmsg = NULL;
1079         }
1080
1081         client_send_control(substate->req, &substate->header, &substate->reply);
1082         talloc_free(substate);
1083 }
1084
1085 static void control_set_recmode(TALLOC_CTX *mem_ctx,
1086                                 struct tevent_req *req,
1087                                 struct ctdb_req_header *header,
1088                                 struct ctdb_req_control *request)
1089 {
1090         struct client_state *state = tevent_req_data(
1091                 req, struct client_state);
1092         struct tevent_req *subreq;
1093         struct ctdbd_context *ctdb = state->ctdb;
1094         struct set_recmode_state *substate;
1095         struct ctdb_reply_control reply;
1096
1097         reply.rdata.opcode = request->opcode;
1098
1099         if (request->rdata.data.recmode == CTDB_RECOVERY_NORMAL) {
1100                 reply.status = -1;
1101                 reply.errmsg = "Client cannot set recmode to NORMAL";
1102                 goto fail;
1103         }
1104
1105         substate = talloc_zero(ctdb, struct set_recmode_state);
1106         if (substate == NULL) {
1107                 reply.status = -1;
1108                 reply.errmsg = "Memory error";
1109                 goto fail;
1110         }
1111
1112         substate->req = req;
1113         substate->ctdb = ctdb;
1114         substate->header = *header;
1115         substate->reply.rdata.opcode = request->opcode;
1116
1117         subreq = recover_send(substate, state->ev, state->ctdb);
1118         if (subreq == NULL) {
1119                 talloc_free(substate);
1120                 goto fail;
1121         }
1122         tevent_req_set_callback(subreq, set_recmode_callback, substate);
1123
1124         ctdb->vnn_map->recmode = CTDB_RECOVERY_ACTIVE;
1125         return;
1126
1127 fail:
1128         client_send_control(req, header, &reply);
1129
1130 }
1131
1132 static int srvid_register_state_destructor(struct srvid_register_state *rstate)
1133 {
1134         DLIST_REMOVE(rstate->ctdb->rstate, rstate);
1135         return 0;
1136 }
1137
1138 static void control_register_srvid(TALLOC_CTX *mem_ctx,
1139                                    struct tevent_req *req,
1140                                    struct ctdb_req_header *header,
1141                                    struct ctdb_req_control *request)
1142 {
1143         struct client_state *state = tevent_req_data(
1144                 req, struct client_state);
1145         struct ctdbd_context *ctdb = state->ctdb;
1146         struct ctdb_reply_control reply;
1147         struct srvid_register_state *rstate;
1148
1149         reply.rdata.opcode = request->opcode;
1150
1151         rstate = talloc_zero(ctdb, struct srvid_register_state);
1152         if (rstate == NULL) {
1153                 reply.status = -1;
1154                 reply.errmsg = "Memory error";
1155                 goto fail;
1156         }
1157         rstate->ctdb = ctdb;
1158         rstate->srvid = request->srvid;
1159
1160         talloc_set_destructor(rstate, srvid_register_state_destructor);
1161
1162         DLIST_ADD_END(ctdb->rstate, rstate);
1163
1164         DEBUG(DEBUG_INFO, ("Register srvid 0x%"PRIx64"\n", rstate->srvid));
1165
1166         reply.status = 0;
1167         reply.errmsg = NULL;
1168
1169 fail:
1170         client_send_control(req, header, &reply);
1171 }
1172
1173 static void control_deregister_srvid(TALLOC_CTX *mem_ctx,
1174                                      struct tevent_req *req,
1175                                      struct ctdb_req_header *header,
1176                                      struct ctdb_req_control *request)
1177 {
1178         struct client_state *state = tevent_req_data(
1179                 req, struct client_state);
1180         struct ctdbd_context *ctdb = state->ctdb;
1181         struct ctdb_reply_control reply;
1182         struct srvid_register_state *rstate = NULL;
1183
1184         reply.rdata.opcode = request->opcode;
1185
1186         for (rstate = ctdb->rstate; rstate != NULL; rstate = rstate->next) {
1187                 if (rstate->srvid == request->srvid) {
1188                         break;
1189                 }
1190         }
1191
1192         if (rstate == NULL) {
1193                 reply.status = -1;
1194                 reply.errmsg = "srvid not registered";
1195                 goto fail;
1196         }
1197
1198         DEBUG(DEBUG_INFO, ("Deregister srvid 0x%"PRIx64"\n", rstate->srvid));
1199         talloc_free(rstate);
1200
1201         reply.status = 0;
1202         reply.errmsg = NULL;
1203
1204         client_send_control(req, header, &reply);
1205         return;
1206
1207 fail:
1208         TALLOC_FREE(rstate);
1209         client_send_control(req, header, &reply);
1210 }
1211
1212 static void control_get_pid(TALLOC_CTX *mem_ctx,
1213                             struct tevent_req *req,
1214                             struct ctdb_req_header *header,
1215                             struct ctdb_req_control *request)
1216 {
1217         struct ctdb_reply_control reply;
1218
1219         reply.rdata.opcode = request->opcode;
1220         reply.status = getpid();
1221         reply.errmsg = NULL;
1222
1223         client_send_control(req, header, &reply);
1224 }
1225
1226 static void control_get_recmaster(TALLOC_CTX *mem_ctx,
1227                                   struct tevent_req *req,
1228                                   struct ctdb_req_header *header,
1229                                   struct ctdb_req_control *request)
1230 {
1231         struct client_state *state = tevent_req_data(
1232                 req, struct client_state);
1233         struct ctdbd_context *ctdb = state->ctdb;
1234         struct ctdb_reply_control reply;
1235
1236         reply.rdata.opcode = request->opcode;
1237         reply.status = ctdb->node_map->recmaster;
1238         reply.errmsg = NULL;
1239
1240         client_send_control(req, header, &reply);
1241 }
1242
1243 static void control_get_pnn(TALLOC_CTX *mem_ctx,
1244                             struct tevent_req *req,
1245                             struct ctdb_req_header *header,
1246                             struct ctdb_req_control *request)
1247 {
1248         struct ctdb_reply_control reply;
1249
1250         reply.rdata.opcode = request->opcode;
1251         reply.status = header->destnode;
1252         reply.errmsg = NULL;
1253
1254         client_send_control(req, header, &reply);
1255 }
1256
1257 static void control_shutdown(TALLOC_CTX *mem_ctx,
1258                              struct tevent_req *req,
1259                              struct ctdb_req_header *hdr,
1260                              struct ctdb_req_control *request)
1261 {
1262         struct client_state *state = tevent_req_data(
1263                 req, struct client_state);
1264
1265         state->status = 99;
1266 }
1267
1268 static void control_uptime(TALLOC_CTX *mem_ctx,
1269                            struct tevent_req *req,
1270                            struct ctdb_req_header *header,
1271                            struct ctdb_req_control *request)
1272 {
1273         struct client_state *state = tevent_req_data(
1274                 req, struct client_state);
1275         struct ctdbd_context *ctdb = state->ctdb;
1276         struct ctdb_reply_control reply;
1277         struct ctdb_uptime *uptime;;
1278
1279         reply.rdata.opcode = request->opcode;
1280
1281         uptime = talloc_zero(mem_ctx, struct ctdb_uptime);
1282         if (uptime == NULL) {
1283                 goto fail;
1284         }
1285
1286         uptime->current_time = tevent_timeval_current();
1287         uptime->ctdbd_start_time = ctdb->start_time;
1288         uptime->last_recovery_started = ctdb->recovery_start_time;
1289         uptime->last_recovery_finished = ctdb->recovery_end_time;
1290
1291         reply.rdata.data.uptime = uptime;
1292         reply.status = 0;
1293         reply.errmsg = NULL;
1294         client_send_control(req, header, &reply);
1295         return;
1296
1297 fail:
1298         reply.status = -1;
1299         reply.errmsg = "Memory error";
1300         client_send_control(req, header, &reply);
1301 }
1302
1303 static void control_reload_nodes_file(TALLOC_CTX *mem_ctx,
1304                                       struct tevent_req *req,
1305                                       struct ctdb_req_header *header,
1306                                       struct ctdb_req_control *request)
1307 {
1308         struct client_state *state = tevent_req_data(
1309                 req, struct client_state);
1310         struct ctdbd_context *ctdb = state->ctdb;
1311         struct ctdb_reply_control reply;
1312         struct ctdb_node_map *nodemap;
1313         struct node_map *node_map = ctdb->node_map;
1314         int i;
1315
1316         reply.rdata.opcode = request->opcode;
1317
1318         nodemap = read_nodes_file(mem_ctx, header->destnode);
1319         if (nodemap == NULL) {
1320                 goto fail;
1321         }
1322
1323         for (i=0; i<nodemap->num; i++) {
1324                 struct node *node;
1325
1326                 if (i < node_map->num_nodes &&
1327                     ctdb_sock_addr_same(&nodemap->node[i].addr,
1328                                         &node_map->node[i].addr)) {
1329                         continue;
1330                 }
1331
1332                 if (nodemap->node[i].flags & NODE_FLAGS_DELETED) {
1333                         node = &node_map->node[i];
1334
1335                         node->flags |= NODE_FLAGS_DELETED;
1336                         parse_ip("0.0.0.0", NULL, 0, &node->addr);
1337
1338                         continue;
1339                 }
1340
1341                 if (i < node_map->num_nodes &&
1342                     node_map->node[i].flags & NODE_FLAGS_DELETED) {
1343                         node = &node_map->node[i];
1344
1345                         node->flags &= ~NODE_FLAGS_DELETED;
1346                         node->addr = nodemap->node[i].addr;
1347
1348                         continue;
1349                 }
1350
1351                 node_map->node = talloc_realloc(node_map, node_map->node,
1352                                                 struct node,
1353                                                 node_map->num_nodes+1);
1354                 if (node_map->node == NULL) {
1355                         goto fail;
1356                 }
1357                 node = &node_map->node[node_map->num_nodes];
1358
1359                 node->addr = nodemap->node[i].addr;
1360                 node->pnn = nodemap->node[i].pnn;
1361                 node->flags = 0;
1362                 node->capabilities = CTDB_CAP_DEFAULT;
1363                 node->recovery_disabled = false;
1364                 node->recovery_substate = NULL;
1365
1366                 node_map->num_nodes += 1;
1367         }
1368
1369         talloc_free(nodemap);
1370
1371         reply.status = 0;
1372         reply.errmsg = NULL;
1373         client_send_control(req, header, &reply);
1374         return;
1375
1376 fail:
1377         reply.status = -1;
1378         reply.errmsg = "Memory error";
1379         client_send_control(req, header, &reply);
1380 }
1381
1382 static void control_get_capabilities(TALLOC_CTX *mem_ctx,
1383                                      struct tevent_req *req,
1384                                      struct ctdb_req_header *header,
1385                                      struct ctdb_req_control *request)
1386 {
1387         struct client_state *state = tevent_req_data(
1388                 req, struct client_state);
1389         struct ctdbd_context *ctdb = state->ctdb;
1390         struct ctdb_reply_control reply;
1391         struct node *node;
1392         uint32_t caps = 0;
1393
1394         reply.rdata.opcode = request->opcode;
1395
1396         node = &ctdb->node_map->node[header->destnode];
1397         caps = node->capabilities;
1398
1399         if (node->flags & NODE_FLAGS_FAKE_TIMEOUT) {
1400                 /* Don't send reply */
1401                 return;
1402         }
1403
1404         reply.rdata.data.caps = caps;
1405         reply.status = 0;
1406         reply.errmsg = NULL;
1407
1408         client_send_control(req, header, &reply);
1409 }
1410
1411 static void control_get_nodemap(TALLOC_CTX *mem_ctx,
1412                                 struct tevent_req *req,
1413                                 struct ctdb_req_header *header,
1414                                 struct ctdb_req_control *request)
1415 {
1416         struct client_state *state = tevent_req_data(
1417                 req, struct client_state);
1418         struct ctdbd_context *ctdb = state->ctdb;
1419         struct ctdb_reply_control reply;
1420         struct ctdb_node_map *nodemap;
1421         struct node *node;
1422         int i;
1423
1424         reply.rdata.opcode = request->opcode;
1425
1426         nodemap = talloc_zero(mem_ctx, struct ctdb_node_map);
1427         if (nodemap == NULL) {
1428                 goto fail;
1429         }
1430
1431         nodemap->num = ctdb->node_map->num_nodes;
1432         nodemap->node = talloc_array(nodemap, struct ctdb_node_and_flags,
1433                                      nodemap->num);
1434         if (nodemap->node == NULL) {
1435                 goto fail;
1436         }
1437
1438         for (i=0; i<nodemap->num; i++) {
1439                 node = &ctdb->node_map->node[i];
1440                 nodemap->node[i] = (struct ctdb_node_and_flags) {
1441                         .pnn = node->pnn,
1442                         .flags = node->flags,
1443                         .addr = node->addr,
1444                 };
1445         }
1446
1447         reply.rdata.data.nodemap = nodemap;
1448         reply.status = 0;
1449         reply.errmsg = NULL;
1450         client_send_control(req, header, &reply);
1451         return;
1452
1453 fail:
1454         reply.status = -1;
1455         reply.errmsg = "Memory error";
1456         client_send_control(req, header, &reply);
1457 }
1458
1459 static void control_get_ifaces(TALLOC_CTX *mem_ctx,
1460                                struct tevent_req *req,
1461                                struct ctdb_req_header *header,
1462                                struct ctdb_req_control *request)
1463 {
1464         struct client_state *state = tevent_req_data(
1465                 req, struct client_state);
1466         struct ctdbd_context *ctdb = state->ctdb;
1467         struct ctdb_reply_control reply;
1468         struct ctdb_iface_list *iface_list;
1469         struct interface *iface;
1470         int i;
1471
1472         reply.rdata.opcode = request->opcode;
1473
1474         iface_list = talloc_zero(mem_ctx, struct ctdb_iface_list);
1475         if (iface_list == NULL) {
1476                 goto fail;
1477         }
1478
1479         iface_list->num = ctdb->iface_map->num;
1480         iface_list->iface = talloc_array(iface_list, struct ctdb_iface,
1481                                          iface_list->num);
1482         if (iface_list->iface == NULL) {
1483                 goto fail;
1484         }
1485
1486         for (i=0; i<iface_list->num; i++) {
1487                 iface = &ctdb->iface_map->iface[i];
1488                 iface_list->iface[i] = (struct ctdb_iface) {
1489                         .link_state = iface->link_up,
1490                         .references = iface->references,
1491                 };
1492                 strncpy(iface_list->iface[i].name, iface->name,
1493                         CTDB_IFACE_SIZE+2);
1494         }
1495
1496         reply.rdata.data.iface_list = iface_list;
1497         reply.status = 0;
1498         reply.errmsg = NULL;
1499         client_send_control(req, header, &reply);
1500         return;
1501
1502 fail:
1503         reply.status = -1;
1504         reply.errmsg = "Memory error";
1505         client_send_control(req, header, &reply);
1506 }
1507
1508 static void control_get_nodes_file(TALLOC_CTX *mem_ctx,
1509                                    struct tevent_req *req,
1510                                    struct ctdb_req_header *header,
1511                                    struct ctdb_req_control *request)
1512 {
1513         struct ctdb_reply_control reply;
1514         struct ctdb_node_map *nodemap;
1515
1516         reply.rdata.opcode = request->opcode;
1517
1518         nodemap = read_nodes_file(mem_ctx, header->destnode);
1519         if (nodemap == NULL) {
1520                 goto fail;
1521         }
1522
1523         reply.rdata.data.nodemap = nodemap;
1524         reply.status = 0;
1525         reply.errmsg = NULL;
1526         client_send_control(req, header, &reply);
1527         return;
1528
1529 fail:
1530         reply.status = -1;
1531         reply.errmsg = "Failed to read nodes file";
1532         client_send_control(req, header, &reply);
1533 }
1534
1535 static void control_error(TALLOC_CTX *mem_ctx,
1536                           struct tevent_req *req,
1537                           struct ctdb_req_header *header,
1538                           struct ctdb_req_control *request)
1539 {
1540         struct ctdb_reply_control reply;
1541
1542         reply.rdata.opcode = request->opcode;
1543         reply.status = -1;
1544         reply.errmsg = "Not implemented";
1545
1546         client_send_control(req, header, &reply);
1547 }
1548
1549 /*
1550  * Handling protocol - messages
1551  */
1552
1553 struct disable_recoveries_state {
1554         struct node *node;
1555 };
1556
1557 static void disable_recoveries_callback(struct tevent_req *subreq)
1558 {
1559         struct disable_recoveries_state *substate = tevent_req_callback_data(
1560                 subreq, struct disable_recoveries_state);
1561         bool status;
1562
1563         status = tevent_wakeup_recv(subreq);
1564         TALLOC_FREE(subreq);
1565         if (! status) {
1566                 DEBUG(DEBUG_INFO, ("tevent_wakeup_recv failed\n"));
1567         }
1568
1569         substate->node->recovery_disabled = false;
1570         TALLOC_FREE(substate->node->recovery_substate);
1571 }
1572
1573 static void message_disable_recoveries(TALLOC_CTX *mem_ctx,
1574                                        struct tevent_req *req,
1575                                        struct ctdb_req_header *header,
1576                                        struct ctdb_req_message *request)
1577 {
1578         struct client_state *state = tevent_req_data(
1579                 req, struct client_state);
1580         struct tevent_req *subreq;
1581         struct ctdbd_context *ctdb = state->ctdb;
1582         struct disable_recoveries_state *substate;
1583         struct ctdb_disable_message *disable = request->data.disable;
1584         struct ctdb_req_message_data reply;
1585         struct node *node;
1586         int ret = -1;
1587         TDB_DATA data;
1588
1589         node = &ctdb->node_map->node[header->destnode];
1590
1591         if (disable->timeout == 0) {
1592                 TALLOC_FREE(node->recovery_substate);
1593                 node->recovery_disabled = false;
1594                 DEBUG(DEBUG_INFO, ("Enabled recoveries on node %u\n",
1595                                    header->destnode));
1596                 goto done;
1597         }
1598
1599         substate = talloc_zero(ctdb->node_map,
1600                                struct disable_recoveries_state);
1601         if (substate == NULL) {
1602                 goto fail;
1603         }
1604
1605         substate->node = node;
1606
1607         subreq = tevent_wakeup_send(substate, state->ev,
1608                                     tevent_timeval_current_ofs(
1609                                             disable->timeout, 0));
1610         if (subreq == NULL) {
1611                 talloc_free(substate);
1612                 goto fail;
1613         }
1614         tevent_req_set_callback(subreq, disable_recoveries_callback, substate);
1615
1616         DEBUG(DEBUG_INFO, ("Disabled recoveries for %d seconds on node %u\n",
1617                            disable->timeout, header->destnode));
1618         node->recovery_substate = substate;
1619         node->recovery_disabled = true;
1620
1621 done:
1622         ret = header->destnode;
1623
1624 fail:
1625         reply.srvid = disable->srvid;
1626         data.dptr = (uint8_t *)&ret;
1627         data.dsize = sizeof(int);
1628         reply.data = data;
1629
1630         client_send_message(req, header, &reply);
1631 }
1632
1633 /*
1634  * Handle a single client
1635  */
1636
1637 static void client_read_handler(uint8_t *buf, size_t buflen,
1638                                 void *private_data);
1639 static void client_dead_handler(void *private_data);
1640 static void client_process_packet(struct tevent_req *req,
1641                                   uint8_t *buf, size_t buflen);
1642 static void client_process_message(struct tevent_req *req,
1643                                    uint8_t *buf, size_t buflen);
1644 static void client_process_control(struct tevent_req *req,
1645                                    uint8_t *buf, size_t buflen);
1646 static void client_reply_done(struct tevent_req *subreq);
1647
1648 static struct tevent_req *client_send(TALLOC_CTX *mem_ctx,
1649                                       struct tevent_context *ev,
1650                                       int fd, struct ctdbd_context *ctdb,
1651                                       int pnn)
1652 {
1653         struct tevent_req *req;
1654         struct client_state *state;
1655         int ret;
1656
1657         req = tevent_req_create(mem_ctx, &state, struct client_state);
1658         if (req == NULL) {
1659                 return NULL;
1660         }
1661
1662         state->ev = ev;
1663         state->fd = fd;
1664         state->ctdb = ctdb;
1665         state->pnn = pnn;
1666
1667         ret = comm_setup(state, ev, fd, client_read_handler, req,
1668                          client_dead_handler, req, &state->comm);
1669         if (ret != 0) {
1670                 tevent_req_error(req, ret);
1671                 return tevent_req_post(req, ev);
1672         }
1673
1674         DEBUG(DEBUG_INFO, ("New client fd=%d\n", fd));
1675
1676         return req;
1677 }
1678
1679 static void client_read_handler(uint8_t *buf, size_t buflen,
1680                                 void *private_data)
1681 {
1682         struct tevent_req *req = talloc_get_type_abort(
1683                 private_data, struct tevent_req);
1684         struct client_state *state = tevent_req_data(
1685                 req, struct client_state);
1686         struct ctdbd_context *ctdb = state->ctdb;
1687         struct ctdb_req_header header;
1688         int ret, i;
1689
1690         ret = ctdb_req_header_pull(buf, buflen, &header);
1691         if (ret != 0) {
1692                 return;
1693         }
1694
1695         if (buflen != header.length) {
1696                 return;
1697         }
1698
1699         ret = ctdb_req_header_verify(&header, 0);
1700         if (ret != 0) {
1701                 return;
1702         }
1703
1704         header_fix_pnn(&header, ctdb);
1705
1706         if (header.destnode == CTDB_BROADCAST_ALL) {
1707                 for (i=0; i<ctdb->node_map->num_nodes; i++) {
1708                         header.destnode = i;
1709
1710                         ctdb_req_header_push(&header, buf);
1711                         client_process_packet(req, buf, buflen);
1712                 }
1713                 return;
1714         }
1715
1716         if (header.destnode == CTDB_BROADCAST_CONNECTED) {
1717                 for (i=0; i<ctdb->node_map->num_nodes; i++) {
1718                         if (ctdb->node_map->node[i].flags &
1719                             NODE_FLAGS_DISCONNECTED) {
1720                                 continue;
1721                         }
1722
1723                         header.destnode = i;
1724
1725                         ctdb_req_header_push(&header, buf);
1726                         client_process_packet(req, buf, buflen);
1727                 }
1728                 return;
1729         }
1730
1731         if (header.destnode > ctdb->node_map->num_nodes) {
1732                 fprintf(stderr, "Invalid destination pnn 0x%x\n",
1733                         header.destnode);
1734                 return;
1735         }
1736
1737
1738         if (ctdb->node_map->node[header.destnode].flags & NODE_FLAGS_DISCONNECTED) {
1739                 fprintf(stderr, "Packet for disconnected node pnn %u\n",
1740                         header.destnode);
1741                 return;
1742         }
1743
1744         ctdb_req_header_push(&header, buf);
1745         client_process_packet(req, buf, buflen);
1746 }
1747
1748 static void client_dead_handler(void *private_data)
1749 {
1750         struct tevent_req *req = talloc_get_type_abort(
1751                 private_data, struct tevent_req);
1752
1753         tevent_req_done(req);
1754 }
1755
1756 static void client_process_packet(struct tevent_req *req,
1757                                   uint8_t *buf, size_t buflen)
1758 {
1759         struct ctdb_req_header header;
1760         int ret;
1761
1762         ret = ctdb_req_header_pull(buf, buflen, &header);
1763         if (ret != 0) {
1764                 return;
1765         }
1766
1767         switch (header.operation) {
1768         case CTDB_REQ_MESSAGE:
1769                 client_process_message(req, buf, buflen);
1770                 break;
1771
1772         case CTDB_REQ_CONTROL:
1773                 client_process_control(req, buf, buflen);
1774                 break;
1775
1776         default:
1777                 break;
1778         }
1779 }
1780
1781 static void client_process_message(struct tevent_req *req,
1782                                    uint8_t *buf, size_t buflen)
1783 {
1784         struct client_state *state = tevent_req_data(
1785                 req, struct client_state);
1786         struct ctdbd_context *ctdb = state->ctdb;
1787         TALLOC_CTX *mem_ctx;
1788         struct ctdb_req_header header;
1789         struct ctdb_req_message request;
1790         uint64_t srvid;
1791         int ret;
1792
1793         mem_ctx = talloc_new(state);
1794         if (tevent_req_nomem(mem_ctx, req)) {
1795                 return;
1796         }
1797
1798         ret = ctdb_req_message_pull(buf, buflen, &header, mem_ctx, &request);
1799         if (ret != 0) {
1800                 talloc_free(mem_ctx);
1801                 tevent_req_error(req, ret);
1802                 return;
1803         }
1804
1805         header_fix_pnn(&header, ctdb);
1806
1807         if (header.destnode >= ctdb->node_map->num_nodes) {
1808                 /* Many messages are not replied to, so just behave as
1809                  * though this message was not received */
1810                 fprintf(stderr, "Invalid node %d\n", header.destnode);
1811                 talloc_free(mem_ctx);
1812                 return;
1813         }
1814
1815         srvid = request.srvid;
1816         DEBUG(DEBUG_INFO, ("request srvid = 0x%"PRIx64"\n", srvid));
1817
1818         if (srvid == CTDB_SRVID_DISABLE_RECOVERIES) {
1819                 message_disable_recoveries(mem_ctx, req, &header, &request);
1820         }
1821
1822         /* check srvid */
1823         talloc_free(mem_ctx);
1824 }
1825
1826 static void client_process_control(struct tevent_req *req,
1827                                    uint8_t *buf, size_t buflen)
1828 {
1829         struct client_state *state = tevent_req_data(
1830                 req, struct client_state);
1831         struct ctdbd_context *ctdb = state->ctdb;
1832         TALLOC_CTX *mem_ctx;
1833         struct ctdb_req_header header;
1834         struct ctdb_req_control request;
1835         int ret;
1836
1837         mem_ctx = talloc_new(state);
1838         if (tevent_req_nomem(mem_ctx, req)) {
1839                 return;
1840         }
1841
1842         ret = ctdb_req_control_pull(buf, buflen, &header, mem_ctx, &request);
1843         if (ret != 0) {
1844                 talloc_free(mem_ctx);
1845                 tevent_req_error(req, ret);
1846                 return;
1847         }
1848
1849         header_fix_pnn(&header, ctdb);
1850
1851         if (header.destnode >= ctdb->node_map->num_nodes) {
1852                 struct ctdb_reply_control reply;
1853
1854                 reply.rdata.opcode = request.opcode;
1855                 reply.errmsg = "Invalid node";
1856                 reply.status = -1;
1857                 client_send_control(req, &header, &reply);
1858                 return;
1859         }
1860
1861         DEBUG(DEBUG_INFO, ("request opcode = %u, reqid = %u\n",
1862                            request.opcode, header.reqid));
1863
1864         switch (request.opcode) {
1865         case CTDB_CONTROL_PROCESS_EXISTS:
1866                 control_process_exists(mem_ctx, req, &header, &request);
1867                 break;
1868
1869         case CTDB_CONTROL_PING:
1870                 control_ping(mem_ctx, req, &header, &request);
1871                 break;
1872
1873         case CTDB_CONTROL_GETVNNMAP:
1874                 control_getvnnmap(mem_ctx, req, &header, &request);
1875                 break;
1876
1877         case CTDB_CONTROL_GET_DEBUG:
1878                 control_get_debug(mem_ctx, req, &header, &request);
1879                 break;
1880
1881         case CTDB_CONTROL_SET_DEBUG:
1882                 control_set_debug(mem_ctx, req, &header, &request);
1883                 break;
1884
1885         case CTDB_CONTROL_GET_RECMODE:
1886                 control_get_recmode(mem_ctx, req, &header, &request);
1887                 break;
1888
1889         case CTDB_CONTROL_SET_RECMODE:
1890                 control_set_recmode(mem_ctx, req, &header, &request);
1891                 break;
1892
1893         case CTDB_CONTROL_REGISTER_SRVID:
1894                 control_register_srvid(mem_ctx, req, &header, &request);
1895                 break;
1896
1897         case CTDB_CONTROL_DEREGISTER_SRVID:
1898                 control_deregister_srvid(mem_ctx, req, &header, &request);
1899                 break;
1900
1901         case CTDB_CONTROL_GET_PID:
1902                 control_get_pid(mem_ctx, req, &header, &request);
1903                 break;
1904
1905         case CTDB_CONTROL_GET_RECMASTER:
1906                 control_get_recmaster(mem_ctx, req, &header, &request);
1907                 break;
1908
1909         case CTDB_CONTROL_GET_PNN:
1910                 control_get_pnn(mem_ctx, req, &header, &request);
1911                 break;
1912
1913         case CTDB_CONTROL_SHUTDOWN:
1914                 control_shutdown(mem_ctx, req, &header, &request);
1915                 break;
1916
1917         case CTDB_CONTROL_UPTIME:
1918                 control_uptime(mem_ctx, req, &header, &request);
1919                 break;
1920
1921         case CTDB_CONTROL_RELOAD_NODES_FILE:
1922                 control_reload_nodes_file(mem_ctx, req, &header, &request);
1923                 break;
1924
1925         case CTDB_CONTROL_GET_CAPABILITIES:
1926                 control_get_capabilities(mem_ctx, req, &header, &request);
1927                 break;
1928
1929         case CTDB_CONTROL_GET_NODEMAP:
1930                 control_get_nodemap(mem_ctx, req, &header, &request);
1931                 break;
1932
1933         case CTDB_CONTROL_GET_IFACES:
1934                 control_get_ifaces(mem_ctx, req, &header, &request);
1935                 break;
1936
1937         case CTDB_CONTROL_GET_NODES_FILE:
1938                 control_get_nodes_file(mem_ctx, req, &header, &request);
1939                 break;
1940
1941         default:
1942                 if (! (request.flags & CTDB_CTRL_FLAG_NOREPLY)) {
1943                         control_error(mem_ctx, req, &header, &request);
1944                 }
1945                 break;
1946         }
1947
1948         talloc_free(mem_ctx);
1949 }
1950
1951 static int client_recv(struct tevent_req *req, int *perr)
1952 {
1953         struct client_state *state = tevent_req_data(
1954                 req, struct client_state);
1955         int err;
1956
1957         DEBUG(DEBUG_INFO, ("Client done fd=%d\n", state->fd));
1958         close(state->fd);
1959
1960         if (tevent_req_is_unix_error(req, &err)) {
1961                 if (perr != NULL) {
1962                         *perr = err;
1963                 }
1964                 return -1;
1965         }
1966
1967         return state->status;
1968 }
1969
1970 /*
1971  * Fake CTDB server
1972  */
1973
1974 struct server_state {
1975         struct tevent_context *ev;
1976         struct ctdbd_context *ctdb;
1977         int fd;
1978 };
1979
1980 static void server_new_client(struct tevent_req *subreq);
1981 static void server_client_done(struct tevent_req *subreq);
1982
1983 static struct tevent_req *server_send(TALLOC_CTX *mem_ctx,
1984                                       struct tevent_context *ev,
1985                                       struct ctdbd_context *ctdb,
1986                                       int fd)
1987 {
1988         struct tevent_req *req, *subreq;
1989         struct server_state *state;
1990
1991         req = tevent_req_create(mem_ctx, &state, struct server_state);
1992         if (req == NULL) {
1993                 return NULL;
1994         }
1995
1996         state->ev = ev;
1997         state->ctdb = ctdb;
1998         state->fd = fd;
1999
2000         subreq = accept_send(state, ev, fd);
2001         if (tevent_req_nomem(subreq, req)) {
2002                 return tevent_req_post(req, ev);
2003         }
2004         tevent_req_set_callback(subreq, server_new_client, req);
2005
2006         return req;
2007 }
2008
2009 static void server_new_client(struct tevent_req *subreq)
2010 {
2011         struct tevent_req *req = tevent_req_callback_data(
2012                 subreq, struct tevent_req);
2013         struct server_state *state = tevent_req_data(
2014                 req, struct server_state);
2015         struct ctdbd_context *ctdb = state->ctdb;
2016         int client_fd;
2017         int ret = 0;
2018
2019         client_fd = accept_recv(subreq, NULL, NULL, &ret);
2020         TALLOC_FREE(subreq);
2021         if (client_fd == -1) {
2022                 tevent_req_error(req, ret);
2023                 return;
2024         }
2025
2026         subreq = client_send(state, state->ev, client_fd,
2027                              ctdb, ctdb->node_map->pnn);
2028         if (tevent_req_nomem(subreq, req)) {
2029                 return;
2030         }
2031         tevent_req_set_callback(subreq, server_client_done, req);
2032
2033         ctdb->num_clients += 1;
2034
2035         subreq = accept_send(state, state->ev, state->fd);
2036         if (tevent_req_nomem(subreq, req)) {
2037                 return;
2038         }
2039         tevent_req_set_callback(subreq, server_new_client, req);
2040 }
2041
2042 static void server_client_done(struct tevent_req *subreq)
2043 {
2044         struct tevent_req *req = tevent_req_callback_data(
2045                 subreq, struct tevent_req);
2046         struct server_state *state = tevent_req_data(
2047                 req, struct server_state);
2048         struct ctdbd_context *ctdb = state->ctdb;
2049         int ret = 0;
2050         int status;
2051
2052         status = client_recv(subreq, &ret);
2053         TALLOC_FREE(subreq);
2054         if (status < 0) {
2055                 tevent_req_error(req, ret);
2056                 return;
2057         }
2058
2059         ctdb->num_clients -= 1;
2060
2061         if (status == 99) {
2062                 /* Special status, to shutdown server */
2063                 DEBUG(DEBUG_INFO, ("Shutting down server\n"));
2064                 tevent_req_done(req);
2065         }
2066 }
2067
2068 static bool server_recv(struct tevent_req *req, int *perr)
2069 {
2070         int err;
2071
2072         if (tevent_req_is_unix_error(req, &err)) {
2073                 if (perr != NULL) {
2074                         *perr = err;
2075                 }
2076                 return false;
2077         }
2078         return true;
2079 }
2080
2081 /*
2082  * Main functions
2083  */
2084
2085 static int socket_init(const char *sockpath)
2086 {
2087         struct sockaddr_un addr;
2088         size_t len;
2089         int ret, fd;
2090
2091         memset(&addr, 0, sizeof(addr));
2092         addr.sun_family = AF_UNIX;
2093
2094         len = strlcpy(addr.sun_path, sockpath, sizeof(addr.sun_path));
2095         if (len >= sizeof(addr.sun_path)) {
2096                 fprintf(stderr, "path too long: %s\n", sockpath);
2097                 return -1;
2098         }
2099
2100         fd = socket(AF_UNIX, SOCK_STREAM, 0);
2101         if (fd == -1) {
2102                 fprintf(stderr, "socket failed - %s\n", sockpath);
2103                 return -1;
2104         }
2105
2106         ret = bind(fd, (struct sockaddr *)&addr, sizeof(addr));
2107         if (ret != 0) {
2108                 fprintf(stderr, "bind failed - %s\n", sockpath);
2109                 goto fail;
2110         }
2111
2112         ret = listen(fd, 10);
2113         if (ret != 0) {
2114                 fprintf(stderr, "listen failed\n");
2115                 goto fail;
2116         }
2117
2118         DEBUG(DEBUG_INFO, ("Socket init done\n"));
2119
2120         return fd;
2121
2122 fail:
2123         if (fd != -1) {
2124                 close(fd);
2125         }
2126         return -1;
2127 }
2128
2129 static struct options {
2130         const char *sockpath;
2131         const char *pidfile;
2132         const char *debuglevel;
2133 } options;
2134
2135 static struct poptOption cmdline_options[] = {
2136         { "socket", 's', POPT_ARG_STRING, &options.sockpath, 0,
2137                 "Unix domain socket path", "filename" },
2138         { "pidfile", 'p', POPT_ARG_STRING, &options.pidfile, 0,
2139                 "pid file", "filename" } ,
2140         { "debug", 'd', POPT_ARG_STRING, &options.debuglevel, 0,
2141                 "debug level", "ERR|WARNING|NOTICE|INFO|DEBUG" } ,
2142 };
2143
2144 static void cleanup(void)
2145 {
2146         unlink(options.sockpath);
2147         unlink(options.pidfile);
2148 }
2149
2150 static void signal_handler(int sig)
2151 {
2152         cleanup();
2153         exit(0);
2154 }
2155
2156 static void start_server(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
2157                          struct ctdbd_context *ctdb, int fd, int pfd)
2158 {
2159         struct tevent_req *req;
2160         int ret = 0;
2161         ssize_t len;
2162
2163         atexit(cleanup);
2164         signal(SIGTERM, signal_handler);
2165
2166         req = server_send(mem_ctx, ev, ctdb, fd);
2167         if (req == NULL) {
2168                 fprintf(stderr, "Memory error\n");
2169                 exit(1);
2170         }
2171
2172         len = write(pfd, &ret, sizeof(ret));
2173         if (len != sizeof(ret)) {
2174                 fprintf(stderr, "Failed to send message to parent\n");
2175                 exit(1);
2176         }
2177         close(pfd);
2178
2179         tevent_req_poll(req, ev);
2180
2181         server_recv(req, &ret);
2182         if (ret != 0) {
2183                 exit(1);
2184         }
2185 }
2186
2187 int main(int argc, const char *argv[])
2188 {
2189         TALLOC_CTX *mem_ctx;
2190         struct ctdbd_context *ctdb;
2191         struct tevent_context *ev;
2192         enum debug_level debug_level;
2193         poptContext pc;
2194         int opt, fd, ret, pfd[2];
2195         ssize_t len;
2196         pid_t pid;
2197         FILE *fp;
2198
2199         pc = poptGetContext(argv[0], argc, argv, cmdline_options,
2200                             POPT_CONTEXT_KEEP_FIRST);
2201         while ((opt = poptGetNextOpt(pc)) != -1) {
2202                 fprintf(stderr, "Invalid option %s\n", poptBadOption(pc, 0));
2203                 exit(1);
2204         }
2205
2206         if (options.sockpath == NULL) {
2207                 fprintf(stderr, "Please specify socket path\n");
2208                 poptPrintHelp(pc, stdout, 0);
2209                 exit(1);
2210         }
2211
2212         if (options.pidfile == NULL) {
2213                 fprintf(stderr, "Please specify pid file\n");
2214                 poptPrintHelp(pc, stdout, 0);
2215                 exit(1);
2216         }
2217
2218         if (options.debuglevel == NULL) {
2219                 DEBUGLEVEL = debug_level_to_int(DEBUG_ERR);
2220         } else {
2221                 if (debug_level_parse(options.debuglevel, &debug_level)) {
2222                         DEBUGLEVEL = debug_level_to_int(debug_level);
2223                 } else {
2224                         fprintf(stderr, "Invalid debug level\n");
2225                         poptPrintHelp(pc, stdout, 0);
2226                         exit(1);
2227                 }
2228         }
2229
2230         mem_ctx = talloc_new(NULL);
2231         if (mem_ctx == NULL) {
2232                 fprintf(stderr, "Memory error\n");
2233                 exit(1);
2234         }
2235
2236         ctdb = ctdbd_setup(mem_ctx);
2237         if (ctdb == NULL) {
2238                 exit(1);
2239         }
2240
2241         if (! ctdbd_verify(ctdb)) {
2242                 exit(1);
2243         }
2244
2245         ev = tevent_context_init(mem_ctx);
2246         if (ev == NULL) {
2247                 fprintf(stderr, "Memory error\n");
2248                 exit(1);
2249         }
2250
2251         fd = socket_init(options.sockpath);
2252         if (fd == -1) {
2253                 exit(1);
2254         }
2255
2256         ret = pipe(pfd);
2257         if (ret != 0) {
2258                 fprintf(stderr, "Failed to create pipe\n");
2259                 cleanup();
2260                 exit(1);
2261         }
2262
2263         pid = fork();
2264         if (pid == -1) {
2265                 fprintf(stderr, "Failed to fork\n");
2266                 cleanup();
2267                 exit(1);
2268         }
2269
2270         if (pid == 0) {
2271                 /* Child */
2272                 close(pfd[0]);
2273                 start_server(mem_ctx, ev, ctdb, fd, pfd[1]);
2274                 exit(1);
2275         }
2276
2277         /* Parent */
2278         close(pfd[1]);
2279
2280         len = read(pfd[0], &ret, sizeof(ret));
2281         close(pfd[0]);
2282         if (len != sizeof(ret)) {
2283                 fprintf(stderr, "len = %zi\n", len);
2284                 fprintf(stderr, "Failed to get message from child\n");
2285                 kill(pid, SIGTERM);
2286                 exit(1);
2287         }
2288
2289         fp = fopen(options.pidfile, "w");
2290         if (fp == NULL) {
2291                 fprintf(stderr, "Failed to open pid file %s\n",
2292                         options.pidfile);
2293                 kill(pid, SIGTERM);
2294                 exit(1);
2295         }
2296         fprintf(fp, "%d\n", pid);
2297         fclose(fp);
2298
2299         return 0;
2300 }