ctdb/common/ctdb_daemon.c

   1 /*
   2    ctdb daemon 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 "db_wrap.h"
  23 #include "lib/tdb/include/tdb.h"
  24 #include "lib/events/events.h"
  25 #include "lib/util/dlinklist.h"
  26 #include "system/network.h"
  27 #include "system/filesys.h"
  28 #include "../include/ctdb.h"
  29 #include "../include/ctdb_private.h"
  30
  31 #define CTDB_PATH       "/tmp/ctdb.socket"
  32
  33
  34 static void ctdb_main_loop(struct ctdb_context *ctdb)
  35 {
  36         ctdb->methods->start(ctdb);
  37
  38         /* go into a wait loop to allow other nodes to complete */
  39         event_loop_wait(ctdb->ev);
  40
  41         printf("event_loop_wait() returned. this should not happen\n");
  42         exit(1);
  43 }
  44
  45
  46 static void set_non_blocking(int fd)
  47 {
  48         unsigned v;
  49         v = fcntl(fd, F_GETFL, 0);
  50         fcntl(fd, F_SETFL, v | O_NONBLOCK);
  51 }
  52
  53
  54
  55 struct ctdb_client {
  56         struct ctdb_context *ctdb;
  57         struct fd_event *fde;
  58         int fd;
  59         struct ctdb_partial partial;
  60 };
  61
  62
  63 /*
  64   destroy a ctdb_client
  65 */
  66 static int ctdb_client_destructor(struct ctdb_client *client)
  67 {
  68         close(client->fd);
  69         client->fd = -1;
  70         return 0;
  71 }
  72
  73
  74
  75 static void client_request_call(struct ctdb_client *client, struct ctdb_req_call *c)
  76 {
  77         struct ctdb_call_state *state;
  78         struct ctdb_db_context *ctdb_db;
  79         struct ctdb_call call;
  80         struct ctdb_reply_call r;
  81         int res;
  82
  83         for (ctdb_db=client->ctdb->db_list; ctdb_db; ctdb_db=ctdb_db->next) {
  84                 if (ctdb_db->db_id == c->db_id) {
  85                         break;
  86                 }
  87         }
  88         if (!ctdb_db) {
  89                 printf("Unknown database in request. db_id==0x%08x",c->db_id);
  90                 return;
  91         }
  92
  93
  94
  95         ZERO_STRUCT(call);
  96         call.call_id = c->callid;
  97         call.key.dptr = c->data;
  98         call.key.dsize = c->keylen;
  99         call.call_data.dptr = c->data + c->keylen;
 100         call.call_data.dsize = c->calldatalen;
 101
 102         state = ctdb_call_send(ctdb_db, &call);
 103
 104 /* XXX this must be converted to fully async */
 105         res = ctdb_call_recv(state, &call);
 106         if (res != 0) {
 107                 printf("ctdbd_call_recv() returned error\n");
 108                 exit(1);
 109         }
 110
 111         ZERO_STRUCT(r);
 112 #if 0
 113         r.status =
 114 #endif
 115         r.datalen          = call.reply_data.dsize;
 116
 117         r.hdr.length       = offsetof(struct ctdb_reply_call, data) + r.datalen;
 118         r.hdr.ctdb_magic   = c->hdr.ctdb_magic;
 119         r.hdr.ctdb_version = c->hdr.ctdb_version;
 120         r.hdr.operation    = CTDB_REPLY_CALL;
 121 #if 0
 122         r.hdr.destnode     =
 123         r.hdr.srcnode      =
 124 #endif
 125         r.hdr.reqid        = c->hdr.reqid;
 126
 127
 128 /*XXX need to handle the case of partial writes    logic for partial writes in tcp/ctdb_tcp_node_write */
 129         res = write(client->fd, &r, offsetof(struct ctdb_reply_call, data));
 130         if (r.datalen) {
 131                 res = write(client->fd, call.reply_data.dptr, r.datalen);
 132         }
 133 }
 134
 135
 136 /* data contains a packet from the client */
 137 static void client_incoming_packet(struct ctdb_client *client, void *data, size_t nread)
 138 {
 139         struct ctdb_req_header *hdr = data;
 140
 141         if (hdr->ctdb_magic != CTDB_MAGIC) {
 142                 ctdb_set_error(client->ctdb, "Non CTDB packet rejected\n");
 143                 return;
 144         }
 145
 146         if (hdr->ctdb_version != CTDB_VERSION) {
 147                 ctdb_set_error(client->ctdb, "Bad CTDB version 0x%x rejected\n", hdr->ctdb_version);
 148                 return;
 149         }
 150
 151         switch (hdr->operation) {
 152         case CTDB_REQ_CALL:
 153                 client_request_call(client, (struct ctdb_req_call *)hdr);
 154                 break;
 155
 156         }
 157
 158         talloc_free(data);
 159 }
 160
 161
 162 static void ctdb_client_read_cb(uint8_t *data, int cnt, void *args)
 163 {
 164         struct ctdb_client *client = talloc_get_type(args, struct ctdb_client);
 165         struct ctdb_req_header *hdr;
 166
 167         if (cnt < sizeof(*hdr)) {
 168                 ctdb_set_error(client->ctdb, "Bad packet length %d\n", cnt);
 169                 return;
 170         }
 171         hdr = (struct ctdb_req_header *)data;
 172         if (cnt != hdr->length) {
 173                 ctdb_set_error(client->ctdb, "Bad header length %d expected %d\n",
 174                                hdr->length, cnt);
 175                 return;
 176         }
 177
 178         if (hdr->ctdb_magic != CTDB_MAGIC) {
 179                 ctdb_set_error(client->ctdb, "Non CTDB packet rejected\n");
 180                 return;
 181         }
 182
 183         if (hdr->ctdb_version != CTDB_VERSION) {
 184                 ctdb_set_error(client->ctdb, "Bad CTDB version 0x%x rejected\n", hdr->ctdb_version);
 185                 return;
 186         }
 187
 188         /* it is the responsibility of the incoming packet function to free 'data' */
 189         client_incoming_packet(client, data, cnt);
 190 }
 191
 192 static void ctdb_client_read(struct event_context *ev, struct fd_event *fde,
 193                          uint16_t flags, void *private)
 194 {
 195         struct ctdb_client *client = talloc_get_type(private, struct ctdb_client);
 196
 197         ctdb_read_pdu(client->fd, client, &client->partial, ctdb_client_read_cb, client);
 198 }
 199
 200
 201 static void ctdb_accept_client(struct event_context *ev, struct fd_event *fde,
 202                          uint16_t flags, void *private)
 203 {
 204         struct sockaddr_in addr;
 205         socklen_t len;
 206         int fd;
 207         struct ctdb_context *ctdb = talloc_get_type(private, struct ctdb_context);
 208         struct ctdb_client *client;
 209
 210         memset(&addr, 0, sizeof(addr));
 211         len = sizeof(addr);
 212         fd = accept(ctdb->daemon.sd, (struct sockaddr *)&addr, &len);
 213         if (fd == -1) {
 214                 return;
 215         }
 216         set_non_blocking(fd);
 217
 218         client = talloc_zero(ctdb, struct ctdb_client);
 219         client->ctdb = ctdb;
 220         client->fd = fd;
 221
 222         event_add_fd(ctdb->ev, client, client->fd, EVENT_FD_READ,
 223                      ctdb_client_read, client);
 224
 225         talloc_set_destructor(client, ctdb_client_destructor);
 226 }
 227
 228
 229
 230 static void ctdb_read_from_parent(struct event_context *ev, struct fd_event *fde,
 231                          uint16_t flags, void *private)
 232 {
 233         int *fd = private;
 234         int cnt;
 235         char buf;
 236
 237         /* XXX this is a good place to try doing some cleaning up before exiting */
 238         cnt = read(*fd, &buf, 1);
 239         if (cnt==0) {
 240                 printf("parent process exited. filedescriptor dissappeared\n");
 241                 exit(1);
 242         } else {
 243                 printf("ctdb: did not expect data from parent process\n");
 244                 exit(1);
 245         }
 246 }
 247
 248
 249
 250 /*
 251   create a unix domain socket and bind it
 252   return a file descriptor open on the socket
 253 */
 254 static int ux_socket_bind(struct ctdb_context *ctdb)
 255 {
 256         struct sockaddr_un addr;
 257
 258         ctdb->daemon.sd = socket(AF_UNIX, SOCK_STREAM, 0);
 259         if (ctdb->daemon.sd == -1) {
 260                 ctdb->daemon.sd = -1;
 261                 return -1;
 262         }
 263
 264         set_non_blocking(ctdb->daemon.sd);
 265
 266         memset(&addr, 0, sizeof(addr));
 267         addr.sun_family = AF_UNIX;
 268         strncpy(addr.sun_path, ctdb->daemon.name, sizeof(addr.sun_path));
 269
 270         if (bind(ctdb->daemon.sd, (struct sockaddr *)&addr, sizeof(addr)) == -1) {
 271                 close(ctdb->daemon.sd);
 272                 ctdb->daemon.sd = -1;
 273                 return -1;
 274         }
 275         listen(ctdb->daemon.sd, 1);
 276
 277         return 0;
 278 }
 279
 280 static char *domain_socket_name=NULL;
 281 static void unlink_domain_socket(void)
 282 {
 283         if (domain_socket_name) {
 284                 unlink(domain_socket_name);
 285         }
 286 }
 287
 288 /*
 289   start the protocol going
 290 */
 291 int ctdbd_start(struct ctdb_context *ctdb)
 292 {
 293         pid_t pid;
 294         static int fd[2];
 295         int res;
 296         struct fd_event *fde;
 297
 298         /* generate a name to use for our local socket */
 299         ctdb->daemon.name = talloc_asprintf(ctdb, "%s.%s", CTDB_PATH, ctdb->address.address);
 300         /* get rid of any old sockets */
 301         unlink(ctdb->daemon.name);
 302
 303         domain_socket_name = ctdb->daemon.name;
 304         atexit(unlink_domain_socket);
 305
 306         /* create a unix domain stream socket to listen to */
 307         res = ux_socket_bind(ctdb);
 308         if (res!=0) {
 309                 printf("Failed to open CTDB unix domain socket\n");
 310                 exit(10);
 311         }
 312
 313         res = pipe(&fd[0]);
 314         if (res) {
 315                 printf("Failed to open pipe for CTDB\n");
 316                 exit(1);
 317         }
 318         pid = fork();
 319         if (pid==-1) {
 320                 printf("Failed to fork CTDB daemon\n");
 321                 exit(1);
 322         }
 323
 324         if (pid) {
 325                 close(fd[0]);
 326                 close(ctdb->daemon.sd);
 327                 ctdb->daemon.sd = -1;
 328                 return 0;
 329         }
 330
 331
 332         close(fd[1]);
 333         ctdb_clear_flags(ctdb, CTDB_FLAG_DAEMON_MODE);
 334         ctdb->ev = event_context_init(NULL);
 335         fde = event_add_fd(ctdb->ev, ctdb, fd[0], EVENT_FD_READ, ctdb_read_from_parent, &fd[0]);
 336         fde = event_add_fd(ctdb->ev, ctdb, ctdb->daemon.sd, EVENT_FD_READ, ctdb_accept_client, ctdb);
 337         ctdb_main_loop(ctdb);
 338
 339         return 0;
 340 }
 341
 342
 343 static void ctdb_daemon_read_cb(uint8_t *data, int cnt, void *args)
 344 {
 345         struct ctdb_context *ctdb = talloc_get_type(args, struct ctdb_context);
 346         struct ctdb_req_header *hdr;
 347
 348         if (cnt < sizeof(*hdr)) {
 349                 ctdb_set_error(ctdb, "Bad packet length %d\n", cnt);
 350                 return;
 351         }
 352         hdr = (struct ctdb_req_header *)data;
 353         if (cnt != hdr->length) {
 354                 ctdb_set_error(ctdb, "Bad header length %d expected %d\n",
 355                                hdr->length, cnt);
 356                 return;
 357         }
 358
 359         if (hdr->ctdb_magic != CTDB_MAGIC) {
 360                 ctdb_set_error(ctdb, "Non CTDB packet rejected\n");
 361                 return;
 362         }
 363
 364         if (hdr->ctdb_version != CTDB_VERSION) {
 365                 ctdb_set_error(ctdb, "Bad CTDB version 0x%x rejected\n", hdr->ctdb_version);
 366                 return;
 367         }
 368
 369         ctdb_reply_call(ctdb, hdr);
 370 }
 371
 372
 373
 374 static void ctdb_daemon_io(struct event_context *ev, struct fd_event *fde,
 375                          uint16_t flags, void *private)
 376 {
 377         struct ctdb_context *ctdb = talloc_get_type(private, struct ctdb_context);
 378
 379
 380         if (flags&EVENT_FD_READ) {
 381                 ctdb_read_pdu(ctdb->daemon.sd, ctdb, &ctdb->daemon.partial, ctdb_daemon_read_cb, ctdb);
 382         }
 383         if (flags&EVENT_FD_WRITE) {
 384                 printf("socket is filled.   fix this   see tcp_io/ctdb_tcp_node_write how to do this\n");
 385 /*              ctdb_daemon_write(ctdb);*/
 386         }
 387 }
 388
 389 /*
 390   connect to a unix domain socket
 391 */
 392 static int ux_socket_connect(struct ctdb_context *ctdb)
 393 {
 394         struct sockaddr_un addr;
 395
 396         memset(&addr, 0, sizeof(addr));
 397         addr.sun_family = AF_UNIX;
 398         strncpy(addr.sun_path, ctdb->daemon.name, sizeof(addr.sun_path));
 399
 400         ctdb->daemon.sd = socket(AF_UNIX, SOCK_STREAM, 0);
 401         if (ctdb->daemon.sd == -1) {
 402                 return -1;
 403         }
 404
 405         if (connect(ctdb->daemon.sd, (struct sockaddr *)&addr, sizeof(addr)) == -1) {
 406                 close(ctdb->daemon.sd);
 407                 ctdb->daemon.sd = -1;
 408                 return -1;
 409         }
 410
 411         ctdb->daemon.fde = event_add_fd(ctdb->ev, ctdb, ctdb->daemon.sd, EVENT_FD_READ,
 412                      ctdb_daemon_io, ctdb);
 413         return 0;
 414 }
 415
 416
 417
 418
 419 static int ctdb_ltdb_lock(struct ctdb_db_context *ctdb_db, TDB_DATA key)
 420 {
 421         return tdb_chainlock(ctdb_db->ltdb->tdb, key);
 422 }
 423
 424 static int ctdb_ltdb_unlock(struct ctdb_db_context *ctdb_db, TDB_DATA key)
 425 {
 426         return tdb_chainunlock(ctdb_db->ltdb->tdb, key);
 427 }
 428
 429
 430 #define CTDB_DS_ALIGNMENT 8
 431 static void *ctdbd_allocate_pkt(struct ctdb_context *ctdb, size_t len)
 432 {
 433         int size;
 434
 435         size = (len+(CTDB_DS_ALIGNMENT-1)) & ~(CTDB_DS_ALIGNMENT-1);
 436         return talloc_size(ctdb, size);
 437 }
 438
 439
 440 struct ctdbd_queue_packet {
 441         struct ctdbd_queue_packet *next, *prev;
 442         uint8_t *data;
 443         uint32_t length;
 444 };
 445
 446 /*
 447   queue a packet for sending
 448 */
 449 int ctdbd_queue_pkt(struct ctdb_context *ctdb, struct ctdb_req_header *hdr)
 450 {
 451         uint8_t *data = (uint8_t *)hdr;
 452         uint32_t length = hdr->length;
 453         struct ctdbd_queue_packet *pkt;
 454         uint32_t length2;
 455
 456         /* enforce the length and alignment rules from the tcp packet allocator */
 457         length2 = (length+(CTDB_DS_ALIGNMENT-1)) & ~(CTDB_DS_ALIGNMENT-1);
 458         *(uint32_t *)data = length2;
 459
 460         if (length2 != length) {
 461                 memset(data+length, 0, length2-length);
 462         }
 463
 464         /* if the queue is empty then try an immediate write, avoiding
 465            queue overhead. This relies on non-blocking sockets */
 466         if (ctdb->daemon.queue == NULL) {
 467                 ssize_t n = write(ctdb->daemon.sd, data, length2);
 468                 if (n == -1 && errno != EAGAIN && errno != EWOULDBLOCK) {
 469                         printf("socket to ctdb daemon has died\n");
 470                         return -1;
 471                 }
 472                 if (n > 0) {
 473                         data += n;
 474                         length2 -= n;
 475                 }
 476                 if (length2 == 0) return 0;
 477         }
 478
 479         pkt = talloc(ctdb, struct ctdbd_queue_packet);
 480         CTDB_NO_MEMORY(ctdb, pkt);
 481
 482         pkt->data = talloc_memdup(pkt, data, length2);
 483         CTDB_NO_MEMORY(ctdb, pkt->data);
 484
 485         pkt->length = length2;
 486
 487         if (ctdb->daemon.queue == NULL) {
 488                 EVENT_FD_WRITEABLE(ctdb->daemon.fde);
 489         }
 490
 491         DLIST_ADD_END(ctdb->daemon.queue, pkt, struct ctdbd_queue_packet *);
 492
 493         return 0;
 494 }
 495
 496
 497 /*
 498   destroy a ctdb_call
 499 */
 500 static int ctdbd_call_destructor(struct ctdb_call_state *state)
 501 {
 502         idr_remove(state->node->ctdb->idr, state->c->hdr.reqid);
 503         return 0;
 504 }
 505
 506 /*
 507   make a recv call to the local ctdb daemon
 508
 509   This is called when the program wants to wait for a ctdb_call to complete and get the
 510   results. This call will block unless the call has already completed.
 511 */
 512 int ctdbd_call_recv(struct ctdb_call_state *state, struct ctdb_call *call)
 513 {
 514         struct ctdb_record_handle *rec;
 515
 516         while (state->state < CTDB_CALL_DONE) {
 517                 event_loop_once(state->node->ctdb->ev);
 518         }
 519         if (state->state != CTDB_CALL_DONE) {
 520                 ctdb_set_error(state->node->ctdb, "%s", state->errmsg);
 521                 talloc_free(state);
 522                 return -1;
 523         }
 524
 525         rec = state->fetch_private;
 526
 527         /* ugly hack to manage forced migration */
 528         if (rec != NULL) {
 529                 rec->data->dptr = talloc_steal(rec, state->call.reply_data.dptr);
 530                 rec->data->dsize = state->call.reply_data.dsize;
 531                 talloc_free(state);
 532                 return 0;
 533         }
 534
 535         if (state->call.reply_data.dsize) {
 536                 call->reply_data.dptr = talloc_memdup(state->node->ctdb,
 537                                                       state->call.reply_data.dptr,
 538                                                       state->call.reply_data.dsize);
 539                 call->reply_data.dsize = state->call.reply_data.dsize;
 540         } else {
 541                 call->reply_data.dptr = NULL;
 542                 call->reply_data.dsize = 0;
 543         }
 544         call->status = state->call.status;
 545         talloc_free(state);
 546
 547         return 0;
 548 }
 549
 550 /*
 551   make a ctdb call to the local daemon - async send
 552
 553   This constructs a ctdb_call request and queues it for processing.
 554   This call never blocks.
 555 */
 556 struct ctdb_call_state *ctdbd_call_send(struct ctdb_db_context *ctdb_db, struct ctdb_call *call)
 557 {
 558         struct ctdb_call_state *state;
 559         struct ctdb_context *ctdb = ctdb_db->ctdb;
 560         struct ctdb_ltdb_header header;
 561         TDB_DATA data;
 562         int ret;
 563         size_t len;
 564
 565         /* if the domain socket is not yet open, open it */
 566         if (ctdb->daemon.sd==-1) {
 567                 ux_socket_connect(ctdb);
 568         }
 569
 570         ret = ctdb_ltdb_lock(ctdb_db, call->key);
 571         if (ret != 0) {
 572                 printf("failed to lock ltdb record\n");
 573                 return NULL;
 574         }
 575
 576         ret = ctdb_ltdb_fetch(ctdb_db, call->key, &header, ctdb_db, &data);
 577         if (ret != 0) {
 578                 ctdb_ltdb_unlock(ctdb_db, call->key);
 579                 return NULL;
 580         }
 581
 582 #if 0
 583         if (header.dmaster == ctdb->vnn && !(ctdb->flags & CTDB_FLAG_SELF_CONNECT)) {
 584                 state = ctdb_call_local_send(ctdb_db, call, &header, &data);
 585                 ctdb_ltdb_unlock(ctdb_db, call->key);
 586                 return state;
 587         }
 588 #endif
 589
 590         state = talloc_zero(ctdb_db, struct ctdb_call_state);
 591         if (state == NULL) {
 592                 printf("failed to allocate state\n");
 593                 ctdb_ltdb_unlock(ctdb_db, call->key);
 594                 return NULL;
 595         }
 596
 597         talloc_steal(state, data.dptr);
 598
 599         len = offsetof(struct ctdb_req_call, data) + call->key.dsize + call->call_data.dsize;
 600         state->c = ctdbd_allocate_pkt(ctdb, len);
 601         if (state->c == NULL) {
 602                 printf("failed to allocate packet\n");
 603                 ctdb_ltdb_unlock(ctdb_db, call->key);
 604                 return NULL;
 605         }
 606         talloc_set_name_const(state->c, "ctdbd req_call packet");
 607         talloc_steal(state, state->c);
 608
 609         state->c->hdr.length    = len;
 610         state->c->hdr.ctdb_magic = CTDB_MAGIC;
 611         state->c->hdr.ctdb_version = CTDB_VERSION;
 612         state->c->hdr.operation = CTDB_REQ_CALL;
 613         state->c->hdr.destnode  = header.dmaster;
 614         state->c->hdr.srcnode   = ctdb->vnn;
 615         /* this limits us to 16k outstanding messages - not unreasonable */
 616         state->c->hdr.reqid     = idr_get_new(ctdb->idr, state, 0xFFFF);
 617         state->c->flags         = call->flags;
 618         state->c->db_id         = ctdb_db->db_id;
 619         state->c->callid        = call->call_id;
 620         state->c->keylen        = call->key.dsize;
 621         state->c->calldatalen   = call->call_data.dsize;
 622         memcpy(&state->c->data[0], call->key.dptr, call->key.dsize);
 623         memcpy(&state->c->data[call->key.dsize],
 624                call->call_data.dptr, call->call_data.dsize);
 625         state->call                = *call;
 626         state->call.call_data.dptr = &state->c->data[call->key.dsize];
 627         state->call.key.dptr       = &state->c->data[0];
 628
 629         state->node   = ctdb->nodes[header.dmaster];
 630         state->state  = CTDB_CALL_WAIT;
 631         state->header = header;
 632         state->ctdb_db = ctdb_db;
 633
 634         talloc_set_destructor(state, ctdbd_call_destructor);
 635
 636         ctdbd_queue_pkt(ctdb, &state->c->hdr);
 637
 638 /*XXX set up timeout to cleanup if server doesnt respond
 639         event_add_timed(ctdb->ev, state, timeval_current_ofs(CTDB_REQ_TIMEOUT, 0),
 640                         ctdb_call_timeout, state);
 641 */
 642
 643         ctdb_ltdb_unlock(ctdb_db, call->key);
 644         return state;
 645 }
 646
 647
 648