common/ctdb_traverse.c

   1 /*
   2    efficient async ctdb traverse
   3
   4    Copyright (C) Andrew Tridgell  2007
   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 2 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, write to the Free Software
  18    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  19 */
  20
  21 #include "includes.h"
  22 #include "lib/events/events.h"
  23 #include "system/filesys.h"
  24 #include "system/wait.h"
  25 #include "db_wrap.h"
  26 #include "lib/tdb/include/tdb.h"
  27 #include "../include/ctdb_private.h"
  28
  29 typedef void (*ctdb_traverse_fn_t)(void *private_data, TDB_DATA key, TDB_DATA data);
  30
  31 /*
  32   handle returned to caller - freeing this handler will kill the child and
  33   terminate the traverse
  34  */
  35 struct ctdb_traverse_local_handle {
  36         struct ctdb_db_context *ctdb_db;
  37         int fd[2];
  38         pid_t child;
  39         void *private_data;
  40         ctdb_traverse_fn_t callback;
  41         struct timeval start_time;
  42         struct ctdb_queue *queue;
  43 };
  44
  45 /*
  46   called when data is available from the child
  47  */
  48 static void ctdb_traverse_local_handler(uint8_t *rawdata, size_t length, void *private_data)
  49 {
  50         struct ctdb_traverse_local_handle *h = talloc_get_type(private_data,
  51                                                                struct ctdb_traverse_local_handle);
  52         TDB_DATA key, data;
  53         ctdb_traverse_fn_t callback = h->callback;
  54         void *p = h->private_data;
  55         struct ctdb_rec_data *tdata = (struct ctdb_rec_data *)rawdata;
  56
  57         if (rawdata == NULL || length < 4 || length != tdata->length) {
  58                 /* end of traverse */
  59                 talloc_free(h);
  60                 callback(p, tdb_null, tdb_null);
  61                 return;
  62         }
  63
  64         key.dsize = tdata->keylen;
  65         key.dptr  = &tdata->data[0];
  66         data.dsize = tdata->datalen;
  67         data.dptr = &tdata->data[tdata->keylen];
  68
  69         callback(p, key, data);
  70 }
  71
  72 /*
  73   destroy a in-flight traverse operation
  74  */
  75 static int traverse_local_destructor(struct ctdb_traverse_local_handle *h)
  76 {
  77         kill(h->child, SIGKILL);
  78         waitpid(h->child, NULL, 0);
  79         return 0;
  80 }
  81
  82 /*
  83   callback from tdb_traverse_read()
  84  */
  85 static int ctdb_traverse_local_fn(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *p)
  86 {
  87         struct ctdb_traverse_local_handle *h = talloc_get_type(p,
  88                                                                struct ctdb_traverse_local_handle);
  89         struct ctdb_rec_data *d;
  90         struct ctdb_ltdb_header *hdr;
  91
  92         /* filter out non-authoritative and zero-length records */
  93         hdr = (struct ctdb_ltdb_header *)data.dptr;
  94         if (data.dsize <= sizeof(struct ctdb_ltdb_header) ||
  95             hdr->dmaster != h->ctdb_db->ctdb->vnn) {
  96                 return 0;
  97         }
  98
  99         d = ctdb_marshall_record(h, 0, key, data);
 100         if (d == NULL) {
 101                 /* error handling is tricky in this child code .... */
 102                 return -1;
 103         }
 104
 105         if (write(h->fd[1], (uint8_t *)d, d->length) != d->length) {
 106                 return -1;
 107         }
 108         return 0;
 109 }
 110
 111
 112 /*
 113   setup a non-blocking traverse of a local ltdb. The callback function
 114   will be called on every record in the local ltdb. To stop the
 115   travserse, talloc_free() the travserse_handle.
 116
 117   The traverse is finished when the callback is called with tdb_null for key and data
 118  */
 119 static struct ctdb_traverse_local_handle *ctdb_traverse_local(struct ctdb_db_context *ctdb_db,
 120                                                               ctdb_traverse_fn_t callback,
 121                                                               void *private_data)
 122 {
 123         struct ctdb_traverse_local_handle *h;
 124         int ret;
 125
 126         h = talloc_zero(ctdb_db, struct ctdb_traverse_local_handle);
 127         if (h == NULL) {
 128                 return NULL;
 129         }
 130
 131         ret = pipe(h->fd);
 132
 133         if (ret != 0) {
 134                 talloc_free(h);
 135                 return NULL;
 136         }
 137
 138         h->child = fork();
 139
 140         if (h->child == (pid_t)-1) {
 141                 close(h->fd[0]);
 142                 close(h->fd[1]);
 143                 talloc_free(h);
 144                 return NULL;
 145         }
 146
 147         h->callback = callback;
 148         h->private_data = private_data;
 149         h->ctdb_db = ctdb_db;
 150
 151         if (h->child == 0) {
 152                 /* start the traverse in the child */
 153                 close(h->fd[0]);
 154                 tdb_traverse_read(ctdb_db->ltdb->tdb, ctdb_traverse_local_fn, h);
 155                 _exit(0);
 156         }
 157
 158         close(h->fd[1]);
 159         talloc_set_destructor(h, traverse_local_destructor);
 160
 161         /*
 162           setup a packet queue between the child and the parent. This
 163           copes with all the async and packet boundary issues
 164          */
 165         h->queue = ctdb_queue_setup(ctdb_db->ctdb, h, h->fd[0], 0, ctdb_traverse_local_handler, h);
 166         if (h->queue == NULL) {
 167                 talloc_free(h);
 168                 return NULL;
 169         }
 170
 171         h->start_time = timeval_current();
 172
 173         return h;
 174 }
 175
 176
 177 struct ctdb_traverse_all_handle {
 178         struct ctdb_context *ctdb;
 179         uint32_t reqid;
 180         ctdb_traverse_fn_t callback;
 181         void *private_data;
 182         uint32_t null_count;
 183 };
 184
 185 /*
 186   destroy a traverse_all op
 187  */
 188 static int ctdb_traverse_all_destructor(struct ctdb_traverse_all_handle *state)
 189 {
 190         ctdb_reqid_remove(state->ctdb, state->reqid);
 191         return 0;
 192 }
 193
 194 struct ctdb_traverse_all {
 195         uint32_t db_id;
 196         uint32_t reqid;
 197         uint32_t vnn;
 198 };
 199
 200 /* called when a traverse times out */
 201 static void ctdb_traverse_all_timeout(struct event_context *ev, struct timed_event *te,
 202                                       struct timeval t, void *private_data)
 203 {
 204         struct ctdb_traverse_all_handle *state = talloc_get_type(private_data, struct ctdb_traverse_all_handle);
 205
 206         state->ctdb->statistics.timeouts.traverse++;
 207
 208         state->callback(state->private_data, tdb_null, tdb_null);
 209         talloc_free(state);
 210 }
 211
 212 /*
 213   setup a cluster-wide non-blocking traverse of a ctdb. The
 214   callback function will be called on every record in the local
 215   ltdb. To stop the travserse, talloc_free() the traverse_handle.
 216
 217   The traverse is finished when the callback is called with tdb_null
 218   for key and data
 219  */
 220 static struct ctdb_traverse_all_handle *ctdb_daemon_traverse_all(struct ctdb_db_context *ctdb_db,
 221                                                                  ctdb_traverse_fn_t callback,
 222                                                                  void *private_data)
 223 {
 224         struct ctdb_traverse_all_handle *state;
 225         struct ctdb_context *ctdb = ctdb_db->ctdb;
 226         int ret;
 227         TDB_DATA data;
 228         struct ctdb_traverse_all r;
 229
 230         state = talloc(ctdb_db, struct ctdb_traverse_all_handle);
 231         if (state == NULL) {
 232                 return NULL;
 233         }
 234
 235         state->ctdb = ctdb;
 236         state->reqid = ctdb_reqid_new(ctdb_db->ctdb, state);
 237         state->callback = callback;
 238         state->private_data = private_data;
 239         state->null_count = 0;
 240
 241         talloc_set_destructor(state, ctdb_traverse_all_destructor);
 242
 243         r.db_id = ctdb_db->db_id;
 244         r.reqid = state->reqid;
 245         r.vnn   = ctdb->vnn;
 246
 247         data.dptr = (uint8_t *)&r;
 248         data.dsize = sizeof(r);
 249
 250         /* tell all the nodes in the cluster to start sending records to this node */
 251         ret = ctdb_daemon_send_control(ctdb, CTDB_BROADCAST_VNNMAP, 0, CTDB_CONTROL_TRAVERSE_ALL,
 252                                        0, CTDB_CTRL_FLAG_NOREPLY, data, NULL, NULL);
 253         if (ret != 0) {
 254                 talloc_free(state);
 255                 return NULL;
 256         }
 257
 258         /* timeout the traverse */
 259         event_add_timed(ctdb->ev, state,
 260                         timeval_current_ofs(ctdb->tunable.traverse_timeout, 0),
 261                         ctdb_traverse_all_timeout, state);
 262
 263         return state;
 264 }
 265
 266 struct traverse_all_state {
 267         struct ctdb_context *ctdb;
 268         struct ctdb_traverse_local_handle *h;
 269         uint32_t reqid;
 270         uint32_t srcnode;
 271 };
 272
 273 /*
 274   called for each record during a traverse all
 275  */
 276 static void traverse_all_callback(void *p, TDB_DATA key, TDB_DATA data)
 277 {
 278         struct traverse_all_state *state = talloc_get_type(p, struct traverse_all_state);
 279         int ret;
 280         struct ctdb_rec_data *d;
 281         TDB_DATA cdata;
 282
 283         d = ctdb_marshall_record(state, state->reqid, key, data);
 284         if (d == NULL) {
 285                 /* darn .... */
 286                 DEBUG(0,("Out of memory in traverse_all_callback\n"));
 287                 return;
 288         }
 289
 290         cdata.dptr = (uint8_t *)d;
 291         cdata.dsize = d->length;
 292
 293         ret = ctdb_daemon_send_control(state->ctdb, state->srcnode, 0, CTDB_CONTROL_TRAVERSE_DATA,
 294                                        0, CTDB_CTRL_FLAG_NOREPLY, cdata, NULL, NULL);
 295         if (ret != 0) {
 296                 DEBUG(0,("Failed to send traverse data\n"));
 297         }
 298
 299         if (key.dsize == 0 && data.dsize == 0) {
 300                 /* we're done */
 301                 talloc_free(state);
 302         }
 303 }
 304
 305 /*
 306   called when a CTDB_CONTROL_TRAVERSE_ALL control comes in. We then
 307   setup a traverse of our local ltdb, sending the records as
 308   CTDB_CONTROL_TRAVERSE_DATA records back to the originator
 309  */
 310 int32_t ctdb_control_traverse_all(struct ctdb_context *ctdb, TDB_DATA data, TDB_DATA *outdata)
 311 {
 312         struct ctdb_traverse_all *c = (struct ctdb_traverse_all *)data.dptr;
 313         struct traverse_all_state *state;
 314         struct ctdb_db_context *ctdb_db;
 315
 316         if (data.dsize != sizeof(struct ctdb_traverse_all)) {
 317                 DEBUG(0,("Invalid size in ctdb_control_traverse_all\n"));
 318                 return -1;
 319         }
 320
 321         ctdb_db = find_ctdb_db(ctdb, c->db_id);
 322         if (ctdb_db == NULL) {
 323                 return -1;
 324         }
 325
 326         state = talloc(ctdb_db, struct traverse_all_state);
 327         if (state == NULL) {
 328                 return -1;
 329         }
 330
 331         state->reqid = c->reqid;
 332         state->srcnode = c->vnn;
 333         state->ctdb = ctdb;
 334
 335         state->h = ctdb_traverse_local(ctdb_db, traverse_all_callback, state);
 336         if (state->h == NULL) {
 337                 talloc_free(state);
 338                 return -1;
 339         }
 340
 341         return 0;
 342 }
 343
 344
 345 /*
 346   called when a CTDB_CONTROL_TRAVERSE_DATA control comes in. We then
 347   call the traverse_all callback with the record
 348  */
 349 int32_t ctdb_control_traverse_data(struct ctdb_context *ctdb, TDB_DATA data, TDB_DATA *outdata)
 350 {
 351         struct ctdb_rec_data *d = (struct ctdb_rec_data *)data.dptr;
 352         struct ctdb_traverse_all_handle *state;
 353         TDB_DATA key;
 354         ctdb_traverse_fn_t callback;
 355         void *private_data;
 356
 357         if (data.dsize < sizeof(uint32_t) || data.dsize != d->length) {
 358                 DEBUG(0,("Bad record size in ctdb_control_traverse_data\n"));
 359                 return -1;
 360         }
 361
 362         state = ctdb_reqid_find(ctdb, d->reqid, struct ctdb_traverse_all_handle);
 363         if (state == NULL || d->reqid != state->reqid) {
 364                 /* traverse might have been terminated already */
 365                 return -1;
 366         }
 367
 368         key.dsize = d->keylen;
 369         key.dptr  = &d->data[0];
 370         data.dsize = d->datalen;
 371         data.dptr = &d->data[d->keylen];
 372
 373         if (key.dsize == 0 && data.dsize == 0) {
 374                 state->null_count++;
 375                 if (state->null_count != ctdb_get_num_enabled_nodes(ctdb)) {
 376                         return 0;
 377                 }
 378         }
 379
 380         callback = state->callback;
 381         private_data = state->private_data;
 382
 383         callback(private_data, key, data);
 384         if (key.dsize == 0 && data.dsize == 0) {
 385                 /* we've received all of the null replies, so all
 386                    nodes are finished */
 387                 talloc_free(state);
 388         }
 389         return 0;
 390 }
 391
 392 struct traverse_start_state {
 393         struct ctdb_context *ctdb;
 394         struct ctdb_traverse_all_handle *h;
 395         uint32_t srcnode;
 396         uint32_t reqid;
 397         uint64_t srvid;
 398 };
 399
 400 /*
 401   callback which sends records as messages to the client
 402  */
 403 static void traverse_start_callback(void *p, TDB_DATA key, TDB_DATA data)
 404 {
 405         struct traverse_start_state *state;
 406         struct ctdb_rec_data *d;
 407         TDB_DATA cdata;
 408
 409         state = talloc_get_type(p, struct traverse_start_state);
 410
 411         d = ctdb_marshall_record(state, state->reqid, key, data);
 412         if (d == NULL) {
 413                 return;
 414         }
 415
 416         cdata.dptr = (uint8_t *)d;
 417         cdata.dsize = d->length;
 418
 419         ctdb_dispatch_message(state->ctdb, state->srvid, cdata);
 420         if (key.dsize == 0 && data.dsize == 0) {
 421                 /* end of traverse */
 422                 talloc_free(state);
 423         }
 424 }
 425
 426 /*
 427   start a traverse_all - called as a control from a client
 428  */
 429 int32_t ctdb_control_traverse_start(struct ctdb_context *ctdb, TDB_DATA data,
 430                                     TDB_DATA *outdata, uint32_t srcnode)
 431 {
 432         struct ctdb_traverse_start *d = (struct ctdb_traverse_start *)data.dptr;
 433         struct traverse_start_state *state;
 434         struct ctdb_db_context *ctdb_db;
 435
 436         if (data.dsize != sizeof(*d)) {
 437                 DEBUG(0,("Bad record size in ctdb_control_traverse_start\n"));
 438                 return -1;
 439         }
 440
 441         ctdb_db = find_ctdb_db(ctdb, d->db_id);
 442         if (ctdb_db == NULL) {
 443                 return -1;
 444         }
 445
 446         state = talloc(ctdb_db, struct traverse_start_state);
 447         if (state == NULL) {
 448                 return -1;
 449         }
 450
 451         state->srcnode = srcnode;
 452         state->reqid = d->reqid;
 453         state->srvid = d->srvid;
 454         state->ctdb = ctdb;
 455
 456         state->h = ctdb_daemon_traverse_all(ctdb_db, traverse_start_callback, state);
 457         if (state->h == NULL) {
 458                 talloc_free(state);
 459                 return -1;
 460         }
 461
 462         return 0;
 463 }