server/ctdb_vacuum.c

   1 /*
   2    ctdb vacuuming events
   3
   4    Copyright (C) Ronnie Sahlberg  2009
   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 "includes.h"
  21 #include "lib/tevent/tevent.h"
  22 #include "lib/tdb/include/tdb.h"
  23 #include "system/network.h"
  24 #include "system/filesys.h"
  25 #include "system/dir.h"
  26 #include "../include/ctdb_private.h"
  27 #include "db_wrap.h"
  28 #include "lib/util/dlinklist.h"
  29 #include "lib/tevent/tevent.h"
  30 #include "../include/ctdb_private.h"
  31 #include "../common/rb_tree.h"
  32
  33 #define TIMELIMIT() timeval_current_ofs(10, 0)
  34 #define TUNINGDBNAME "vactune.tdb"
  35
  36 enum vacuum_child_status { VACUUM_RUNNING, VACUUM_OK, VACUUM_ERROR, VACUUM_TIMEOUT};
  37
  38 struct ctdb_vacuum_child_context {
  39         struct ctdb_vacuum_child_context *next, *prev;
  40         struct ctdb_vacuum_handle *vacuum_handle;
  41         /* fd child writes status to */
  42         int fd[2];
  43         pid_t child_pid;
  44         enum vacuum_child_status status;
  45         struct timeval start_time;
  46 };
  47
  48 struct ctdb_vacuum_handle {
  49         struct ctdb_db_context *ctdb_db;
  50         struct ctdb_vacuum_child_context *child_ctx;
  51         uint32_t fast_path_count;
  52 };
  53
  54
  55 /*  a list of records to possibly delete */
  56 struct vacuum_data {
  57         uint32_t vacuum_limit;
  58         uint32_t repack_limit;
  59         struct ctdb_context *ctdb;
  60         struct ctdb_db_context *ctdb_db;
  61         struct tdb_context *dest_db;
  62         trbt_tree_t *delete_tree;
  63         uint32_t delete_count;
  64         struct ctdb_marshall_buffer **list;
  65         struct timeval start;
  66         bool traverse_error;
  67         bool vacuum;
  68         uint32_t total;
  69         uint32_t vacuumed;
  70         uint32_t copied;
  71         uint32_t fast_added_to_vacuum_fetch_list;
  72         uint32_t fast_added_to_delete_tree;
  73         uint32_t fast_deleted;
  74         uint32_t fast_skipped;
  75         uint32_t fast_error;
  76         uint32_t fast_total;
  77         uint32_t full_added_to_vacuum_fetch_list;
  78         uint32_t full_added_to_delete_tree;
  79         uint32_t full_skipped;
  80         uint32_t full_error;
  81         uint32_t full_total;
  82 };
  83
  84 /* tuning information stored for every db */
  85 struct vacuum_tuning_data {
  86         uint32_t last_num_repack;
  87         uint32_t last_num_empty;
  88         uint32_t last_interval;
  89         uint32_t new_interval;
  90         struct timeval last_start;
  91         double   last_duration;
  92 };
  93
  94 /* this structure contains the information for one record to be deleted */
  95 struct delete_record_data {
  96         struct ctdb_context *ctdb;
  97         struct ctdb_db_context *ctdb_db;
  98         struct ctdb_ltdb_header hdr;
  99         TDB_DATA key;
 100 };
 101
 102 struct delete_records_list {
 103         struct ctdb_marshall_buffer *records;
 104 };
 105
 106 /**
 107  * Store key and header in a tree, indexed by the key hash.
 108  */
 109 static int insert_delete_record_data_into_tree(struct ctdb_context *ctdb,
 110                                                struct ctdb_db_context *ctdb_db,
 111                                                trbt_tree_t *tree,
 112                                                const struct ctdb_ltdb_header *hdr,
 113                                                TDB_DATA key)
 114 {
 115         struct delete_record_data *dd;
 116         uint32_t hash;
 117
 118         dd = talloc_zero(tree, struct delete_record_data);
 119         if (dd == NULL) {
 120                 DEBUG(DEBUG_ERR,(__location__ " Out of memory\n"));
 121                 return -1;
 122         }
 123
 124         dd->ctdb      = ctdb;
 125         dd->ctdb_db   = ctdb_db;
 126         dd->key.dsize = key.dsize;
 127         dd->key.dptr  = talloc_memdup(dd, key.dptr, key.dsize);
 128         if (dd->key.dptr == NULL) {
 129                 DEBUG(DEBUG_ERR,(__location__ " Out of memory\n"));
 130                 return -1;
 131         }
 132
 133         dd->hdr = *hdr;
 134
 135         hash = ctdb_hash(&key);
 136
 137         trbt_insert32(tree, hash, dd);
 138
 139         return 0;
 140 }
 141
 142 static int add_record_to_delete_tree(struct vacuum_data *vdata, TDB_DATA key,
 143                                      struct ctdb_ltdb_header *hdr)
 144 {
 145         struct ctdb_context *ctdb = vdata->ctdb;
 146         struct ctdb_db_context *ctdb_db = vdata->ctdb_db;
 147         uint32_t hash;
 148         int ret;
 149
 150         hash = ctdb_hash(&key);
 151
 152         if (trbt_lookup32(vdata->delete_tree, hash)) {
 153                 DEBUG(DEBUG_INFO, (__location__ " Hash collission when vacuuming, skipping this record.\n"));
 154                 return 0;
 155         }
 156
 157         ret = insert_delete_record_data_into_tree(ctdb, ctdb_db,
 158                                                   vdata->delete_tree,
 159                                                   hdr, key);
 160         if (ret != 0) {
 161                 return -1;
 162         }
 163
 164         vdata->delete_count++;
 165
 166         return 0;
 167 }
 168
 169 /**
 170  * Add a record to the list of records to be sent
 171  * to their lmaster with VACUUM_FETCH.
 172  */
 173 static int add_record_to_vacuum_fetch_list(struct vacuum_data *vdata,
 174                                            TDB_DATA key)
 175 {
 176         struct ctdb_context *ctdb = vdata->ctdb;
 177         struct ctdb_rec_data *rec;
 178         uint32_t lmaster;
 179         size_t old_size;
 180
 181         lmaster = ctdb_lmaster(ctdb, &key);
 182
 183         rec = ctdb_marshall_record(vdata->list[lmaster], ctdb->pnn, key, NULL, tdb_null);
 184         if (rec == NULL) {
 185                 DEBUG(DEBUG_ERR,(__location__ " Out of memory\n"));
 186                 vdata->traverse_error = true;
 187                 return -1;
 188         }
 189
 190         old_size = talloc_get_size(vdata->list[lmaster]);
 191         vdata->list[lmaster] = talloc_realloc_size(NULL, vdata->list[lmaster],
 192                                                    old_size + rec->length);
 193         if (vdata->list[lmaster] == NULL) {
 194                 DEBUG(DEBUG_ERR,(__location__ " Failed to expand\n"));
 195                 vdata->traverse_error = true;
 196                 return -1;
 197         }
 198
 199         vdata->list[lmaster]->count++;
 200         memcpy(old_size+(uint8_t *)vdata->list[lmaster], rec, rec->length);
 201         talloc_free(rec);
 202
 203         vdata->total++;
 204
 205         return 0;
 206 }
 207
 208
 209 static void ctdb_vacuum_event(struct event_context *ev, struct timed_event *te,
 210                                                           struct timeval t, void *private_data);
 211
 212
 213 /*
 214  * traverse function for gathering the records that can be deleted
 215  */
 216 static int vacuum_traverse(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *private)
 217 {
 218         struct vacuum_data *vdata = talloc_get_type(private, struct vacuum_data);
 219         struct ctdb_context *ctdb = vdata->ctdb;
 220         uint32_t lmaster;
 221         struct ctdb_ltdb_header *hdr;
 222         int res = 0;
 223
 224         vdata->full_total++;
 225
 226         lmaster = ctdb_lmaster(ctdb, &key);
 227         if (lmaster >= ctdb->num_nodes) {
 228                 vdata->full_error++;
 229                 DEBUG(DEBUG_CRIT, (__location__
 230                                    " lmaster[%u] >= ctdb->num_nodes[%u] for key"
 231                                    " with hash[%u]!\n",
 232                                    (unsigned)lmaster,
 233                                    (unsigned)ctdb->num_nodes,
 234                                    (unsigned)ctdb_hash(&key)));
 235                 return -1;
 236         }
 237
 238         if (data.dsize != sizeof(struct ctdb_ltdb_header)) {
 239                 /* its not a deleted record */
 240                 vdata->full_skipped++;
 241                 return 0;
 242         }
 243
 244         hdr = (struct ctdb_ltdb_header *)data.dptr;
 245
 246         if (hdr->dmaster != ctdb->pnn) {
 247                 vdata->full_skipped++;
 248                 return 0;
 249         }
 250
 251         if (lmaster == ctdb->pnn) {
 252                 /*
 253                  * We are both lmaster and dmaster, and the record * is empty.
 254                  * So we should be able to delete it.
 255                  */
 256                 res = add_record_to_delete_tree(vdata, key, hdr);
 257                 if (res != 0) {
 258                         vdata->full_error++;
 259                 } else {
 260                         vdata->full_added_to_delete_tree++;
 261                 }
 262         } else {
 263                 /*
 264                  * We are not lmaster.
 265                  * Add the record to the blob ready to send to the nodes.
 266                  */
 267                 res = add_record_to_vacuum_fetch_list(vdata, key);
 268                 if (res != 0) {
 269                         vdata->full_error++;
 270                 } else {
 271                         vdata->full_added_to_vacuum_fetch_list++;
 272                 }
 273         }
 274
 275         return res;
 276 }
 277
 278 /*
 279  * traverse the tree of records to delete and marshall them into
 280  * a blob
 281  */
 282 static void delete_traverse(void *param, void *data)
 283 {
 284         struct delete_record_data *dd = talloc_get_type(data, struct delete_record_data);
 285         struct delete_records_list *recs = talloc_get_type(param, struct delete_records_list);
 286         struct ctdb_rec_data *rec;
 287         size_t old_size;
 288
 289         rec = ctdb_marshall_record(dd, recs->records->db_id, dd->key, &dd->hdr, tdb_null);
 290         if (rec == NULL) {
 291                 DEBUG(DEBUG_ERR, (__location__ " failed to marshall record\n"));
 292                 return;
 293         }
 294
 295         old_size = talloc_get_size(recs->records);
 296         recs->records = talloc_realloc_size(NULL, recs->records, old_size + rec->length);
 297         if (recs->records == NULL) {
 298                 DEBUG(DEBUG_ERR,(__location__ " Failed to expand\n"));
 299                 return;
 300         }
 301         recs->records->count++;
 302         memcpy(old_size+(uint8_t *)(recs->records), rec, rec->length);
 303 }
 304
 305 /**
 306  * traverse function for the traversal of the delete_queue,
 307  * the fast-path vacuuming list.
 308  *
 309  *  - If the record has been migrated off the node
 310  *    or has been revived (filled with data) on the node,
 311  *    then skip the record.
 312  *
 313  *  - If the current node is the record's lmaster and it is
 314  *    a record that has never been migrated with data, then
 315  *    delete the record from the local tdb.
 316  *
 317  *  - If the current node is the record's lmaster and it has
 318  *    been migrated with data, then schedule it for the normal
 319  *    vacuuming procedure (i.e. add it to the delete_list).
 320  *
 321  *  - If the current node is NOT the record's lmaster then
 322  *    add it to the list of records that are to be sent to
 323  *    the lmaster with the VACUUM_FETCH message.
 324  */
 325 static void delete_queue_traverse(void *param, void *data)
 326 {
 327         struct delete_record_data *dd =
 328                 talloc_get_type(data, struct delete_record_data);
 329         struct vacuum_data *vdata = talloc_get_type(param, struct vacuum_data);
 330         struct ctdb_db_context *ctdb_db = dd->ctdb_db;
 331         struct ctdb_context *ctdb = ctdb_db->ctdb; /* or dd->ctdb ??? */
 332         int res;
 333         struct ctdb_ltdb_header *header;
 334         TDB_DATA tdb_data;
 335         uint32_t lmaster;
 336
 337         vdata->fast_total++;
 338
 339         res = tdb_chainlock(ctdb_db->ltdb->tdb, dd->key);
 340         if (res != 0) {
 341                 DEBUG(DEBUG_ERR, (__location__ " Error getting chainlock.\n"));
 342                 vdata->fast_error++;
 343                 return;
 344         }
 345
 346         tdb_data = tdb_fetch(ctdb_db->ltdb->tdb, dd->key);
 347         if (tdb_data.dsize < sizeof(struct ctdb_ltdb_header)) {
 348                 /* Does not exist or not a ctdb record. Skip. */
 349                 goto skipped;
 350         }
 351
 352         if (tdb_data.dsize > sizeof(struct ctdb_ltdb_header)) {
 353                 /* The record has been recycled (filled with data). Skip. */
 354                 goto skipped;
 355         }
 356
 357         header = (struct ctdb_ltdb_header *)tdb_data.dptr;
 358
 359         if (header->dmaster != ctdb->pnn) {
 360                 /* The record has been migrated off the node. Skip. */
 361                 goto skipped;
 362         }
 363
 364
 365         if (header->rsn != dd->hdr.rsn) {
 366                 /*
 367                  * The record has been migrated off the node and back again.
 368                  * But not requeued for deletion. Skip it.
 369                  */
 370                 goto skipped;
 371         }
 372
 373         /*
 374          * We are dmaster, and the record has no data, and it has
 375          * not been migrated after it has been queued for deletion.
 376          *
 377          * At this stage, the record could still have been revived locally
 378          * and last been written with empty data. This can only be
 379          * fixed with the addition of an active or delete flag. (TODO)
 380          */
 381
 382         lmaster = ctdb_lmaster(ctdb_db->ctdb, &dd->key);
 383
 384         if (lmaster != ctdb->pnn) {
 385                 res = add_record_to_vacuum_fetch_list(vdata, dd->key);
 386
 387                 if (res != 0) {
 388                         DEBUG(DEBUG_ERR,
 389                               (__location__ " Error adding record to list "
 390                                "of records to send to lmaster.\n"));
 391                         vdata->fast_error++;
 392                 } else {
 393                         vdata->fast_added_to_vacuum_fetch_list++;
 394                 }
 395                 goto done;
 396         }
 397
 398         /* use header->flags or dd->hdr.flags ?? */
 399         if (dd->hdr.flags & CTDB_REC_FLAG_MIGRATED_WITH_DATA) {
 400                 res = add_record_to_delete_tree(vdata, dd->key, &dd->hdr);
 401
 402                 if (res != 0) {
 403                         DEBUG(DEBUG_ERR,
 404                               (__location__ " Error adding record to list "
 405                                "of records for deletion on lmaster.\n"));
 406                         vdata->fast_error++;
 407                 } else {
 408                         vdata->fast_added_to_delete_tree++;
 409                 }
 410         } else {
 411                 res = tdb_delete(ctdb_db->ltdb->tdb, dd->key);
 412
 413                 if (res != 0) {
 414                         DEBUG(DEBUG_ERR,
 415                               (__location__ " Error deleting record from local "
 416                                "data base.\n"));
 417                         vdata->fast_error++;
 418                 } else {
 419                         vdata->fast_deleted++;
 420                 }
 421         }
 422
 423         goto done;
 424
 425 skipped:
 426         vdata->fast_skipped++;
 427
 428 done:
 429         if (tdb_data.dptr != NULL) {
 430                 free(tdb_data.dptr);
 431         }
 432         tdb_chainunlock(ctdb_db->ltdb->tdb, dd->key);
 433
 434         return;
 435 }
 436
 437 /*
 438  * read-only traverse the database in order to find
 439  * records that can be deleted and try to delete these
 440  * records on the other nodes
 441  * this executes in the child context
 442  */
 443 static int ctdb_vacuum_db(struct ctdb_db_context *ctdb_db,
 444                           struct vacuum_data *vdata,
 445                           bool full_vacuum_run)
 446 {
 447         struct ctdb_context *ctdb = ctdb_db->ctdb;
 448         const char *name = ctdb_db->db_name;
 449         int ret, i, pnn;
 450
 451         DEBUG(DEBUG_INFO, (__location__ " Entering %s vacuum run for db "
 452                            "%s db_id[0x%08x]\n",
 453                            full_vacuum_run ? "full" : "fast",
 454                            ctdb_db->db_name, ctdb_db->db_id));
 455
 456         ret = ctdb_ctrl_getvnnmap(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE, ctdb, &ctdb->vnn_map);
 457         if (ret != 0) {
 458                 DEBUG(DEBUG_ERR, ("Unable to get vnnmap from local node\n"));
 459                 return ret;
 460         }
 461
 462         pnn = ctdb_ctrl_getpnn(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE);
 463         if (pnn == -1) {
 464                 DEBUG(DEBUG_ERR, ("Unable to get pnn from local node\n"));
 465                 return -1;
 466         }
 467
 468         ctdb->pnn = pnn;
 469
 470         vdata->fast_added_to_delete_tree = 0;
 471         vdata->fast_added_to_vacuum_fetch_list = 0;
 472         vdata->fast_deleted = 0;
 473         vdata->fast_skipped = 0;
 474         vdata->fast_error = 0;
 475         vdata->fast_total = 0;
 476         vdata->full_added_to_delete_tree = 0;
 477         vdata->full_added_to_vacuum_fetch_list = 0;
 478         vdata->full_skipped = 0;
 479         vdata->full_error = 0;
 480         vdata->full_total = 0;
 481
 482         /* the list needs to be of length num_nodes */
 483         vdata->list = talloc_array(vdata, struct ctdb_marshall_buffer *, ctdb->num_nodes);
 484         if (vdata->list == NULL) {
 485                 DEBUG(DEBUG_ERR,(__location__ " Out of memory\n"));
 486                 return -1;
 487         }
 488         for (i = 0; i < ctdb->num_nodes; i++) {
 489                 vdata->list[i] = (struct ctdb_marshall_buffer *)
 490                         talloc_zero_size(vdata->list,
 491                                                          offsetof(struct ctdb_marshall_buffer, data));
 492                 if (vdata->list[i] == NULL) {
 493                         DEBUG(DEBUG_ERR,(__location__ " Out of memory\n"));
 494                         return -1;
 495                 }
 496                 vdata->list[i]->db_id = ctdb_db->db_id;
 497         }
 498
 499         /*
 500          * Traverse the delete_queue.
 501          * This builds the same lists as the db traverse.
 502          */
 503         trbt_traversearray32(ctdb_db->delete_queue, 1, delete_queue_traverse, vdata);
 504
 505         if (vdata->fast_total > 0) {
 506                 DEBUG(DEBUG_INFO,
 507                       (__location__
 508                        " fast vacuuming delete_queue traverse statistics: "
 509                        "db[%s] "
 510                        "total[%u] "
 511                        "del[%u] "
 512                        "skp[%u] "
 513                        "err[%u] "
 514                        "adt[%u] "
 515                        "avf[%u]\n",
 516                        ctdb_db->db_name,
 517                        (unsigned)vdata->fast_total,
 518                        (unsigned)vdata->fast_deleted,
 519                        (unsigned)vdata->fast_skipped,
 520                        (unsigned)vdata->fast_error,
 521                        (unsigned)vdata->fast_added_to_delete_tree,
 522                        (unsigned)vdata->fast_added_to_vacuum_fetch_list));
 523         }
 524
 525         /*
 526          * read-only traverse of the database, looking for records that
 527          * might be able to be vacuumed.
 528          *
 529          * This is not done each time but only every tunable
 530          * VacuumFastPathCount times.
 531          */
 532         if (full_vacuum_run) {
 533                 ret = tdb_traverse_read(ctdb_db->ltdb->tdb, vacuum_traverse, vdata);
 534                 if (ret == -1 || vdata->traverse_error) {
 535                         DEBUG(DEBUG_ERR,(__location__ " Traverse error in vacuuming '%s'\n", name));
 536                         return -1;
 537                 }
 538                 if (vdata->full_total > 0) {
 539                         DEBUG(DEBUG_INFO,
 540                               (__location__
 541                                " full vacuuming db traverse statistics: "
 542                                "db[%s] "
 543                                "total[%u] "
 544                                "skp[%u] "
 545                                "err[%u] "
 546                                "adt[%u] "
 547                                "avf[%u]\n",
 548                                ctdb_db->db_name,
 549                                (unsigned)vdata->full_total,
 550                                (unsigned)vdata->full_skipped,
 551                                (unsigned)vdata->full_error,
 552                                (unsigned)vdata->full_added_to_delete_tree,
 553                                (unsigned)vdata->full_added_to_vacuum_fetch_list));
 554                 }
 555         }
 556
 557         /*
 558          * For records where we are not the lmaster,
 559          * tell the lmaster to fetch the record.
 560          */
 561         for (i = 0; i < ctdb->num_nodes; i++) {
 562                 TDB_DATA data;
 563
 564                 if (ctdb->nodes[i]->pnn == ctdb->pnn) {
 565                         continue;
 566                 }
 567
 568                 if (vdata->list[i]->count == 0) {
 569                         continue;
 570                 }
 571
 572                 DEBUG(DEBUG_INFO, ("Found %u records for lmaster %u in '%s'\n",
 573                                    vdata->list[i]->count, ctdb->nodes[i]->pnn,
 574                                    name));
 575
 576                 data.dsize = talloc_get_size(vdata->list[i]);
 577                 data.dptr  = (void *)vdata->list[i];
 578                 if (ctdb_client_send_message(ctdb, ctdb->nodes[i]->pnn, CTDB_SRVID_VACUUM_FETCH, data) != 0) {
 579                         DEBUG(DEBUG_ERR, (__location__ " Failed to send vacuum "
 580                                           "fetch message to %u\n",
 581                                           ctdb->nodes[i]->pnn));
 582                         return -1;
 583                 }
 584         }
 585
 586         /* Process all records we can delete (if any) */
 587         if (vdata->delete_count > 0) {
 588                 struct delete_records_list *recs;
 589                 TDB_DATA indata, outdata;
 590                 int32_t res;
 591                 struct ctdb_node_map *nodemap;
 592                 uint32_t *active_nodes;
 593                 int num_active_nodes;
 594
 595                 recs = talloc_zero(vdata, struct delete_records_list);
 596                 if (recs == NULL) {
 597                         DEBUG(DEBUG_ERR,(__location__ " Out of memory\n"));
 598                         return -1;
 599                 }
 600                 recs->records = (struct ctdb_marshall_buffer *)
 601                         talloc_zero_size(vdata,
 602                                     offsetof(struct ctdb_marshall_buffer, data));
 603                 if (recs->records == NULL) {
 604                         DEBUG(DEBUG_ERR,(__location__ " Out of memory\n"));
 605                         return -1;
 606                 }
 607                 recs->records->db_id = ctdb_db->db_id;
 608
 609                 /*
 610                  * traverse the tree of all records we want to delete and
 611                  * create a blob we can send to the other nodes.
 612                  */
 613                 trbt_traversearray32(vdata->delete_tree, 1, delete_traverse, recs);
 614
 615                 indata.dsize = talloc_get_size(recs->records);
 616                 indata.dptr  = (void *)recs->records;
 617
 618                 /*
 619                  * now tell all the active nodes to delete all these records
 620                  * (if possible)
 621                  */
 622
 623                 ret = ctdb_ctrl_getnodemap(ctdb, TIMELIMIT(),
 624                                            CTDB_CURRENT_NODE,
 625                                            recs, /* talloc context */
 626                                            &nodemap);
 627                 if (ret != 0) {
 628                         DEBUG(DEBUG_ERR,(__location__ " unable to get node map\n"));
 629                         return -1;
 630                 }
 631
 632                 active_nodes = list_of_active_nodes(ctdb, nodemap,
 633                                                     nodemap, /* talloc context */
 634                                                     false /* include self */);
 635                 /* yuck! ;-) */
 636                 num_active_nodes = talloc_get_size(active_nodes)/sizeof(*active_nodes);
 637
 638                 for (i = 0; i < num_active_nodes; i++) {
 639                         struct ctdb_marshall_buffer *records;
 640                         struct ctdb_rec_data *rec;
 641
 642                         ret = ctdb_control(ctdb, active_nodes[i], 0,
 643                                         CTDB_CONTROL_TRY_DELETE_RECORDS, 0,
 644                                         indata, recs, &outdata, &res,
 645                                         NULL, NULL);
 646                         if (ret != 0 || res != 0) {
 647                                 DEBUG(DEBUG_ERR, ("Failed to delete records on "
 648                                                   "node %u: ret[%d] res[%d]\n",
 649                                                   active_nodes[i], ret, res));
 650                                 return -1;
 651                         }
 652
 653                         /*
 654                          * outdata countains the list of records coming back
 655                          * from the node which the node could not delete
 656                          */
 657                         records = (struct ctdb_marshall_buffer *)outdata.dptr;
 658                         rec = (struct ctdb_rec_data *)&records->data[0];
 659                         while (records->count-- > 1) {
 660                                 TDB_DATA reckey, recdata;
 661                                 struct ctdb_ltdb_header *rechdr;
 662
 663                                 reckey.dptr = &rec->data[0];
 664                                 reckey.dsize = rec->keylen;
 665                                 recdata.dptr = &rec->data[reckey.dsize];
 666                                 recdata.dsize = rec->datalen;
 667
 668                                 if (recdata.dsize < sizeof(struct ctdb_ltdb_header)) {
 669                                         DEBUG(DEBUG_CRIT,(__location__ " bad ltdb record\n"));
 670                                         return -1;
 671                                 }
 672                                 rechdr = (struct ctdb_ltdb_header *)recdata.dptr;
 673                                 recdata.dptr += sizeof(*rechdr);
 674                                 recdata.dsize -= sizeof(*rechdr);
 675
 676                                 /*
 677                                  * that other node couldnt delete the record
 678                                  * so we should delete it and thereby remove it from the tree
 679                                  */
 680                                 talloc_free(trbt_lookup32(vdata->delete_tree, ctdb_hash(&reckey)));
 681
 682                                 rec = (struct ctdb_rec_data *)(rec->length + (uint8_t *)rec);
 683                         }
 684                 }
 685
 686                 /* free nodemap and active_nodes */
 687                 talloc_free(nodemap);
 688
 689                 /*
 690                  * The only records remaining in the tree would be those
 691                  * records where all other nodes could successfully
 692                  * delete them, so we can safely delete them on the
 693                  * lmaster as well. Deletion implictely happens while
 694                  * we repack the database. The repack algorithm revisits
 695                  * the tree in order to find the records that don't need
 696                  * to be copied / repacked.
 697                  */
 698         }
 699
 700         /* this ensures we run our event queue */
 701         ctdb_ctrl_getpnn(ctdb, TIMELIMIT(), CTDB_CURRENT_NODE);
 702
 703         return 0;
 704 }
 705
 706
 707 /*
 708  * traverse function for repacking
 709  */
 710 static int repack_traverse(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *private)
 711 {
 712         struct vacuum_data *vdata = (struct vacuum_data *)private;
 713
 714         if (vdata->vacuum) {
 715                 uint32_t hash = ctdb_hash(&key);
 716                 struct delete_record_data *kd;
 717                 /*
 718                  * check if we can ignore this record because it's in the delete_tree
 719                  */
 720                 kd = (struct delete_record_data *)trbt_lookup32(vdata->delete_tree, hash);
 721                 /*
 722                  * there might be hash collisions so we have to compare the keys here to be sure
 723                  */
 724                 if (kd && kd->key.dsize == key.dsize && memcmp(kd->key.dptr, key.dptr, key.dsize) == 0) {
 725                         struct ctdb_ltdb_header *hdr = (struct ctdb_ltdb_header *)data.dptr;
 726                         /*
 727                          * we have to check if the record hasn't changed in the meantime in order to
 728                          * savely remove it from the database
 729                          */
 730                         if (data.dsize == sizeof(struct ctdb_ltdb_header) &&
 731                                 hdr->dmaster == kd->ctdb->pnn &&
 732                                 ctdb_lmaster(kd->ctdb, &(kd->key)) == kd->ctdb->pnn &&
 733                                 kd->hdr.rsn == hdr->rsn) {
 734                                 vdata->vacuumed++;
 735                                 return 0;
 736                         }
 737                 }
 738         }
 739         if (tdb_store(vdata->dest_db, key, data, TDB_INSERT) != 0) {
 740                 vdata->traverse_error = true;
 741                 return -1;
 742         }
 743         vdata->copied++;
 744         return 0;
 745 }
 746
 747 /*
 748  * repack a tdb
 749  */
 750 static int ctdb_repack_tdb(struct tdb_context *tdb, TALLOC_CTX *mem_ctx, struct vacuum_data *vdata)
 751 {
 752         struct tdb_context *tmp_db;
 753
 754         if (tdb_transaction_start(tdb) != 0) {
 755                 DEBUG(DEBUG_ERR,(__location__ " Failed to start transaction\n"));
 756                 return -1;
 757         }
 758
 759         tmp_db = tdb_open("tmpdb", tdb_hash_size(tdb),
 760                           TDB_INTERNAL|TDB_DISALLOW_NESTING,
 761                           O_RDWR|O_CREAT, 0);
 762         if (tmp_db == NULL) {
 763                 DEBUG(DEBUG_ERR,(__location__ " Failed to create tmp_db\n"));
 764                 tdb_transaction_cancel(tdb);
 765                 return -1;
 766         }
 767
 768         vdata->traverse_error = false;
 769         vdata->dest_db = tmp_db;
 770         vdata->vacuum = true;
 771         vdata->vacuumed = 0;
 772         vdata->copied = 0;
 773
 774         /*
 775          * repack and vacuum on-the-fly by not writing the records that are
 776          * no longer needed
 777          */
 778         if (tdb_traverse_read(tdb, repack_traverse, vdata) == -1) {
 779                 DEBUG(DEBUG_ERR,(__location__ " Failed to traverse copying out\n"));
 780                 tdb_transaction_cancel(tdb);
 781                 tdb_close(tmp_db);
 782                 return -1;
 783         }
 784
 785         DEBUG(DEBUG_INFO,(__location__ " %u records vacuumed\n", vdata->vacuumed));
 786
 787         if (vdata->traverse_error) {
 788                 DEBUG(DEBUG_ERR,(__location__ " Error during traversal\n"));
 789                 tdb_transaction_cancel(tdb);
 790                 tdb_close(tmp_db);
 791                 return -1;
 792         }
 793
 794         if (tdb_wipe_all(tdb) != 0) {
 795                 DEBUG(DEBUG_ERR,(__location__ " Failed to wipe database\n"));
 796                 tdb_transaction_cancel(tdb);
 797                 tdb_close(tmp_db);
 798                 return -1;
 799         }
 800
 801         vdata->traverse_error = false;
 802         vdata->dest_db = tdb;
 803         vdata->vacuum = false;
 804         vdata->copied = 0;
 805
 806         if (tdb_traverse_read(tmp_db, repack_traverse, vdata) == -1) {
 807                 DEBUG(DEBUG_ERR,(__location__ " Failed to traverse copying back\n"));
 808                 tdb_transaction_cancel(tdb);
 809                 tdb_close(tmp_db);
 810                 return -1;
 811         }
 812
 813         if (vdata->traverse_error) {
 814                 DEBUG(DEBUG_ERR,(__location__ " Error during second traversal\n"));
 815                 tdb_transaction_cancel(tdb);
 816                 tdb_close(tmp_db);
 817                 return -1;
 818         }
 819
 820         tdb_close(tmp_db);
 821
 822
 823         if (tdb_transaction_commit(tdb) != 0) {
 824                 DEBUG(DEBUG_ERR,(__location__ " Failed to commit\n"));
 825                 return -1;
 826         }
 827         DEBUG(DEBUG_INFO,(__location__ " %u records copied\n", vdata->copied));
 828
 829         return 0;
 830 }
 831
 832 static int update_tuning_db(struct ctdb_db_context *ctdb_db, struct vacuum_data *vdata, uint32_t freelist)
 833 {
 834         TALLOC_CTX *tmp_ctx = talloc_new(NULL);
 835         TDB_CONTEXT *tune_tdb;
 836         TDB_DATA key, value;
 837         struct vacuum_tuning_data tdata;
 838         struct vacuum_tuning_data *tptr;
 839         char *vac_dbname;
 840         int flags;
 841
 842         vac_dbname = talloc_asprintf(tmp_ctx, "%s/%s.%u",
 843                                      ctdb_db->ctdb->db_directory_state,
 844                                      TUNINGDBNAME, ctdb_db->ctdb->pnn);
 845         if (vac_dbname == NULL) {
 846                 DEBUG(DEBUG_CRIT,(__location__ " Out of memory error while allocating '%s'\n", vac_dbname));
 847                 talloc_free(tmp_ctx);
 848                 return -1;
 849         }
 850
 851         flags  = ctdb_db->ctdb->valgrinding ? TDB_NOMMAP : 0;
 852         flags |= TDB_DISALLOW_NESTING;
 853         tune_tdb = tdb_open(vac_dbname, 0,
 854                             flags,
 855                             O_RDWR|O_CREAT, 0600);
 856         if (tune_tdb == NULL) {
 857                 DEBUG(DEBUG_ERR,(__location__ " Failed to create/open %s\n", TUNINGDBNAME));
 858                 talloc_free(tmp_ctx);
 859                 return -1;
 860         }
 861
 862         if (tdb_transaction_start(tune_tdb) != 0) {
 863                 DEBUG(DEBUG_ERR,(__location__ " Failed to start transaction\n"));
 864                 tdb_close(tune_tdb);
 865                 return -1;
 866         }
 867         key.dptr = discard_const(ctdb_db->db_name);
 868         key.dsize = strlen(ctdb_db->db_name);
 869         value = tdb_fetch(tune_tdb, key);
 870
 871         if (value.dptr != NULL && value.dsize == sizeof(struct vacuum_tuning_data)) {
 872                 tptr = (struct vacuum_tuning_data *)value.dptr;
 873                 tdata = *tptr;
 874
 875                 /*
 876                  * re-calc new vacuum interval:
 877                  * in case no limit was reached we continuously increase the interval
 878                  * until vacuum_max_interval is reached
 879                  * in case a limit was reached we divide the current interval by 2
 880                  * unless vacuum_min_interval is reached
 881                  */
 882                 if (freelist < vdata->repack_limit &&
 883                     vdata->delete_count < vdata->vacuum_limit) {
 884                         if (tdata.last_interval < ctdb_db->ctdb->tunable.vacuum_max_interval) {
 885                                 tdata.new_interval = tdata.last_interval * 110 / 100;
 886                                 DEBUG(DEBUG_INFO,("Increasing vacuum interval %u -> %u for %s\n",
 887                                         tdata.last_interval, tdata.new_interval, ctdb_db->db_name));
 888                         }
 889                 } else {
 890                         tdata.new_interval = tdata.last_interval / 2;
 891                         if (tdata.new_interval < ctdb_db->ctdb->tunable.vacuum_min_interval ||
 892                                 tdata.new_interval > ctdb_db->ctdb->tunable.vacuum_max_interval) {
 893                                 tdata.new_interval = ctdb_db->ctdb->tunable.vacuum_min_interval;
 894                         }
 895                         DEBUG(DEBUG_INFO,("Decreasing vacuum interval %u -> %u for %s\n",
 896                                          tdata.last_interval, tdata.new_interval, ctdb_db->db_name));
 897                 }
 898                 tdata.last_interval = tdata.new_interval;
 899         } else {
 900                 DEBUG(DEBUG_DEBUG,(__location__ " Cannot find tunedb record for %s. Using default interval\n", ctdb_db->db_name));
 901                 tdata.last_num_repack = freelist;
 902                 tdata.last_num_empty = vdata->delete_count;
 903                 tdata.last_interval = ctdb_db->ctdb->tunable.vacuum_default_interval;
 904         }
 905
 906         if (value.dptr != NULL) {
 907                 free(value.dptr);
 908         }
 909
 910         tdata.last_start = vdata->start;
 911         tdata.last_duration = timeval_elapsed(&vdata->start);
 912
 913         value.dptr = (unsigned char *)&tdata;
 914         value.dsize = sizeof(tdata);
 915
 916         if (tdb_store(tune_tdb, key, value, 0) != 0) {
 917                 DEBUG(DEBUG_ERR,(__location__ " Unable to store tundb record for %s\n", ctdb_db->db_name));
 918                 tdb_transaction_cancel(tune_tdb);
 919                 tdb_close(tune_tdb);
 920                 talloc_free(tmp_ctx);
 921                 return -1;
 922         }
 923         tdb_transaction_commit(tune_tdb);
 924         tdb_close(tune_tdb);
 925         talloc_free(tmp_ctx);
 926
 927         return 0;
 928 }
 929
 930 /*
 931  * repack and vaccum a db
 932  * called from the child context
 933  */
 934 static int ctdb_vacuum_and_repack_db(struct ctdb_db_context *ctdb_db,
 935                                      TALLOC_CTX *mem_ctx,
 936                                      bool full_vacuum_run)
 937 {
 938         uint32_t repack_limit = ctdb_db->ctdb->tunable.repack_limit;
 939         uint32_t vacuum_limit = ctdb_db->ctdb->tunable.vacuum_limit;
 940         const char *name = ctdb_db->db_name;
 941         int size;
 942         struct vacuum_data *vdata;
 943
 944         size = tdb_freelist_size(ctdb_db->ltdb->tdb);
 945         if (size == -1) {
 946                 DEBUG(DEBUG_ERR,(__location__ " Failed to get freelist size for '%s'\n", name));
 947                 return -1;
 948         }
 949
 950         vdata = talloc_zero(mem_ctx, struct vacuum_data);
 951         if (vdata == NULL) {
 952                 DEBUG(DEBUG_ERR,(__location__ " Out of memory\n"));
 953                 return -1;
 954         }
 955
 956         vdata->ctdb = ctdb_db->ctdb;
 957         vdata->vacuum_limit = vacuum_limit;
 958         vdata->repack_limit = repack_limit;
 959         vdata->delete_tree = trbt_create(vdata, 0);
 960         vdata->ctdb_db = ctdb_db;
 961         if (vdata->delete_tree == NULL) {
 962                 DEBUG(DEBUG_ERR,(__location__ " Out of memory\n"));
 963                 talloc_free(vdata);
 964                 return -1;
 965         }
 966
 967         vdata->start = timeval_current();
 968
 969         /*
 970          * gather all records that can be deleted in vdata
 971          */
 972         if (ctdb_vacuum_db(ctdb_db, vdata, full_vacuum_run) != 0) {
 973                 DEBUG(DEBUG_ERR,(__location__ " Failed to vacuum '%s'\n", name));
 974         }
 975
 976         /*
 977          * decide if a repack is necessary
 978          */
 979         if (size < repack_limit && vdata->delete_count < vacuum_limit) {
 980                 update_tuning_db(ctdb_db, vdata, size);
 981                 talloc_free(vdata);
 982                 return 0;
 983         }
 984
 985         DEBUG(DEBUG_INFO,("Repacking %s with %u freelist entries and %u records to delete\n",
 986                         name, size, vdata->delete_count));
 987
 988         /*
 989          * repack and implicitely get rid of the records we can delete
 990          */
 991         if (ctdb_repack_tdb(ctdb_db->ltdb->tdb, mem_ctx, vdata) != 0) {
 992                 DEBUG(DEBUG_ERR,(__location__ " Failed to repack '%s'\n", name));
 993                 update_tuning_db(ctdb_db, vdata, size);
 994                 talloc_free(vdata);
 995                 return -1;
 996         }
 997         update_tuning_db(ctdb_db, vdata, size);
 998         talloc_free(vdata);
 999
1000         return 0;
1001 }
1002
1003 static int get_vacuum_interval(struct ctdb_db_context *ctdb_db)
1004 {
1005         TALLOC_CTX *tmp_ctx = talloc_new(NULL);
1006         TDB_CONTEXT *tdb;
1007         TDB_DATA key, value;
1008         char *vac_dbname;
1009         uint interval = ctdb_db->ctdb->tunable.vacuum_default_interval;
1010         struct ctdb_context *ctdb = ctdb_db->ctdb;
1011         int flags;
1012
1013         vac_dbname = talloc_asprintf(tmp_ctx, "%s/%s.%u", ctdb->db_directory, TUNINGDBNAME, ctdb->pnn);
1014         if (vac_dbname == NULL) {
1015                 DEBUG(DEBUG_CRIT,(__location__ " Out of memory error while allocating '%s'\n", vac_dbname));
1016                 talloc_free(tmp_ctx);
1017                 return interval;
1018         }
1019
1020         flags  = ctdb_db->ctdb->valgrinding ? TDB_NOMMAP : 0;
1021         flags |= TDB_DISALLOW_NESTING;
1022         tdb = tdb_open(vac_dbname, 0,
1023                        flags,
1024                        O_RDWR|O_CREAT, 0600);
1025         if (!tdb) {
1026                 DEBUG(DEBUG_ERR,("Unable to open/create database %s using default interval. Errno : %s (%d)\n", vac_dbname, strerror(errno), errno));
1027                 talloc_free(tmp_ctx);
1028                 return interval;
1029         }
1030
1031         key.dptr = discard_const(ctdb_db->db_name);
1032         key.dsize = strlen(ctdb_db->db_name);
1033
1034         value = tdb_fetch(tdb, key);
1035
1036         if (value.dptr != NULL) {
1037                 if (value.dsize == sizeof(struct vacuum_tuning_data)) {
1038                         struct vacuum_tuning_data *tptr = (struct vacuum_tuning_data *)value.dptr;
1039
1040                         interval = tptr->new_interval;
1041
1042                         if (interval < ctdb->tunable.vacuum_min_interval) {
1043                                 interval = ctdb->tunable.vacuum_min_interval;
1044                         }
1045                         if (interval > ctdb->tunable.vacuum_max_interval) {
1046                                 interval = ctdb->tunable.vacuum_max_interval;
1047                         }
1048                 }
1049                 free(value.dptr);
1050         }
1051         tdb_close(tdb);
1052
1053         talloc_free(tmp_ctx);
1054
1055         return interval;
1056 }
1057
1058 static int vacuum_child_destructor(struct ctdb_vacuum_child_context *child_ctx)
1059 {
1060         double l = timeval_elapsed(&child_ctx->start_time);
1061         struct ctdb_db_context *ctdb_db = child_ctx->vacuum_handle->ctdb_db;
1062         struct ctdb_context *ctdb = ctdb_db->ctdb;
1063
1064         DEBUG(DEBUG_INFO,("Vacuuming took %.3f seconds for database %s\n", l, ctdb_db->db_name));
1065
1066         if (child_ctx->child_pid != -1) {
1067                 kill(child_ctx->child_pid, SIGKILL);
1068         } else {
1069                 /* Bump the number of successful fast-path runs. */
1070                 child_ctx->vacuum_handle->fast_path_count++;
1071         }
1072
1073         DLIST_REMOVE(ctdb->vacuumers, child_ctx);
1074
1075         event_add_timed(ctdb->ev, child_ctx->vacuum_handle,
1076                         timeval_current_ofs(get_vacuum_interval(ctdb_db), 0),
1077                         ctdb_vacuum_event, child_ctx->vacuum_handle);
1078
1079         return 0;
1080 }
1081
1082 /*
1083  * this event is generated when a vacuum child process times out
1084  */
1085 static void vacuum_child_timeout(struct event_context *ev, struct timed_event *te,
1086                                          struct timeval t, void *private_data)
1087 {
1088         struct ctdb_vacuum_child_context *child_ctx = talloc_get_type(private_data, struct ctdb_vacuum_child_context);
1089
1090         DEBUG(DEBUG_ERR,("Vacuuming child process timed out for db %s\n", child_ctx->vacuum_handle->ctdb_db->db_name));
1091
1092         child_ctx->status = VACUUM_TIMEOUT;
1093
1094         talloc_free(child_ctx);
1095 }
1096
1097
1098 /*
1099  * this event is generated when a vacuum child process has completed
1100  */
1101 static void vacuum_child_handler(struct event_context *ev, struct fd_event *fde,
1102                              uint16_t flags, void *private_data)
1103 {
1104         struct ctdb_vacuum_child_context *child_ctx = talloc_get_type(private_data, struct ctdb_vacuum_child_context);
1105         char c = 0;
1106         int ret;
1107
1108         DEBUG(DEBUG_INFO,("Vacuuming child process %d finished for db %s\n", child_ctx->child_pid, child_ctx->vacuum_handle->ctdb_db->db_name));
1109         child_ctx->child_pid = -1;
1110
1111         ret = read(child_ctx->fd[0], &c, 1);
1112         if (ret != 1 || c != 0) {
1113                 child_ctx->status = VACUUM_ERROR;
1114                 DEBUG(DEBUG_ERR, ("A vacuum child process failed with an error for database %s. ret=%d c=%d\n", child_ctx->vacuum_handle->ctdb_db->db_name, ret, c));
1115         } else {
1116                 child_ctx->status = VACUUM_OK;
1117         }
1118
1119         talloc_free(child_ctx);
1120 }
1121
1122 /*
1123  * this event is called every time we need to start a new vacuum process
1124  */
1125 static void
1126 ctdb_vacuum_event(struct event_context *ev, struct timed_event *te,
1127                                struct timeval t, void *private_data)
1128 {
1129         struct ctdb_vacuum_handle *vacuum_handle = talloc_get_type(private_data, struct ctdb_vacuum_handle);
1130         struct ctdb_db_context *ctdb_db = vacuum_handle->ctdb_db;
1131         struct ctdb_context *ctdb = ctdb_db->ctdb;
1132         struct ctdb_vacuum_child_context *child_ctx;
1133         struct tevent_fd *fde;
1134         int ret;
1135
1136         /* we dont vacuum if we are in recovery mode, or db frozen */
1137         if (ctdb->recovery_mode == CTDB_RECOVERY_ACTIVE ||
1138             ctdb->freeze_mode[ctdb_db->priority] != CTDB_FREEZE_NONE) {
1139                 DEBUG(DEBUG_INFO, ("Not vacuuming %s (%s)\n", ctdb_db->db_name,
1140                                    ctdb->recovery_mode == CTDB_RECOVERY_ACTIVE ? "in recovery"
1141                                    : ctdb->freeze_mode[ctdb_db->priority] == CTDB_FREEZE_PENDING
1142                                    ? "freeze pending"
1143                                    : "frozen"));
1144                 event_add_timed(ctdb->ev, vacuum_handle, timeval_current_ofs(ctdb->tunable.vacuum_default_interval, 0), ctdb_vacuum_event, vacuum_handle);
1145                 return;
1146         }
1147
1148         child_ctx = talloc(vacuum_handle, struct ctdb_vacuum_child_context);
1149         if (child_ctx == NULL) {
1150                 DEBUG(DEBUG_CRIT, (__location__ " Failed to allocate child context for vacuuming of %s\n", ctdb_db->db_name));
1151                 ctdb_fatal(ctdb, "Out of memory when crating vacuum child context. Shutting down\n");
1152         }
1153
1154
1155         ret = pipe(child_ctx->fd);
1156         if (ret != 0) {
1157                 talloc_free(child_ctx);
1158                 DEBUG(DEBUG_ERR, ("Failed to create pipe for vacuum child process.\n"));
1159                 event_add_timed(ctdb->ev, vacuum_handle, timeval_current_ofs(ctdb->tunable.vacuum_default_interval, 0), ctdb_vacuum_event, vacuum_handle);
1160                 return;
1161         }
1162
1163         if (vacuum_handle->fast_path_count > ctdb->tunable.vacuum_fast_path_count) {
1164                 vacuum_handle->fast_path_count = 0;
1165         }
1166
1167         child_ctx->child_pid = ctdb_fork(ctdb);
1168         if (child_ctx->child_pid == (pid_t)-1) {
1169                 close(child_ctx->fd[0]);
1170                 close(child_ctx->fd[1]);
1171                 talloc_free(child_ctx);
1172                 DEBUG(DEBUG_ERR, ("Failed to fork vacuum child process.\n"));
1173                 event_add_timed(ctdb->ev, vacuum_handle, timeval_current_ofs(ctdb->tunable.vacuum_default_interval, 0), ctdb_vacuum_event, vacuum_handle);
1174                 return;
1175         }
1176
1177
1178         if (child_ctx->child_pid == 0) {
1179                 char cc = 0;
1180                 bool full_vacuum_run = false;
1181                 close(child_ctx->fd[0]);
1182
1183                 DEBUG(DEBUG_INFO,("Vacuuming child process %d for db %s started\n", getpid(), ctdb_db->db_name));
1184
1185                 if (switch_from_server_to_client(ctdb, "vacuum-%s", ctdb_db->db_name) != 0) {
1186                         DEBUG(DEBUG_CRIT, (__location__ "ERROR: failed to switch vacuum daemon into client mode. Shutting down.\n"));
1187                         _exit(1);
1188                 }
1189
1190                 /*
1191                  * repack the db
1192                  */
1193                 if ((ctdb->tunable.vacuum_fast_path_count > 0) &&
1194                     (vacuum_handle->fast_path_count == 0))
1195                 {
1196                         full_vacuum_run = true;
1197                 }
1198                 cc = ctdb_vacuum_and_repack_db(ctdb_db, child_ctx,
1199                                                full_vacuum_run);
1200
1201                 write(child_ctx->fd[1], &cc, 1);
1202                 _exit(0);
1203         }
1204
1205         set_close_on_exec(child_ctx->fd[0]);
1206         close(child_ctx->fd[1]);
1207
1208         child_ctx->status = VACUUM_RUNNING;
1209         child_ctx->start_time = timeval_current();
1210
1211         DLIST_ADD(ctdb->vacuumers, child_ctx);
1212         talloc_set_destructor(child_ctx, vacuum_child_destructor);
1213
1214         /*
1215          * Clear the fastpath vacuuming list in the parent.
1216          */
1217         talloc_free(ctdb_db->delete_queue);
1218         ctdb_db->delete_queue = trbt_create(ctdb_db, 0);
1219         if (ctdb_db->delete_queue == NULL) {
1220                 /* fatal here? ... */
1221                 ctdb_fatal(ctdb, "Out of memory when re-creating vacuum tree "
1222                                  "in parent context. Shutting down\n");
1223         }
1224
1225         event_add_timed(ctdb->ev, child_ctx,
1226                 timeval_current_ofs(ctdb->tunable.vacuum_max_run_time, 0),
1227                 vacuum_child_timeout, child_ctx);
1228
1229         DEBUG(DEBUG_DEBUG, (__location__ " Created PIPE FD:%d to child vacuum process\n", child_ctx->fd[0]));
1230
1231         fde = event_add_fd(ctdb->ev, child_ctx, child_ctx->fd[0],
1232                            EVENT_FD_READ, vacuum_child_handler, child_ctx);
1233         tevent_fd_set_auto_close(fde);
1234
1235         vacuum_handle->child_ctx = child_ctx;
1236         child_ctx->vacuum_handle = vacuum_handle;
1237 }
1238
1239 void ctdb_stop_vacuuming(struct ctdb_context *ctdb)
1240 {
1241         /* Simply free them all. */
1242         while (ctdb->vacuumers) {
1243                 DEBUG(DEBUG_INFO, ("Aborting vacuuming for %s (%i)\n",
1244                            ctdb->vacuumers->vacuum_handle->ctdb_db->db_name,
1245                            (int)ctdb->vacuumers->child_pid));
1246                 /* vacuum_child_destructor kills it, removes from list */
1247                 talloc_free(ctdb->vacuumers);
1248         }
1249 }
1250
1251 /* this function initializes the vacuuming context for a database
1252  * starts the vacuuming events
1253  */
1254 int ctdb_vacuum_init(struct ctdb_db_context *ctdb_db)
1255 {
1256         if (ctdb_db->persistent != 0) {
1257                 DEBUG(DEBUG_ERR,("Vacuuming is disabled for persistent database %s\n", ctdb_db->db_name));
1258                 return 0;
1259         }
1260
1261         ctdb_db->vacuum_handle = talloc(ctdb_db, struct ctdb_vacuum_handle);
1262         CTDB_NO_MEMORY(ctdb_db->ctdb, ctdb_db->vacuum_handle);
1263
1264         ctdb_db->vacuum_handle->ctdb_db         = ctdb_db;
1265         ctdb_db->vacuum_handle->fast_path_count = 0;
1266
1267         event_add_timed(ctdb_db->ctdb->ev, ctdb_db->vacuum_handle,
1268                         timeval_current_ofs(get_vacuum_interval(ctdb_db), 0),
1269                         ctdb_vacuum_event, ctdb_db->vacuum_handle);
1270
1271         return 0;
1272 }
1273
1274 /**
1275  * Insert a record into the ctdb_db context's delete queue,
1276  * handling hash collisions.
1277  */
1278 static int insert_record_into_delete_queue(struct ctdb_db_context *ctdb_db,
1279                                            const struct ctdb_ltdb_header *hdr,
1280                                            TDB_DATA key)
1281 {
1282         struct delete_record_data *kd;
1283         uint32_t hash;
1284         int ret;
1285
1286         hash = (uint32_t)ctdb_hash(&key);
1287
1288         DEBUG(DEBUG_INFO, (__location__ " Schedule for deletion: db[%s] "
1289                            "db_id[0x%08x] "
1290                            "key_hash[0x%08x] "
1291                            "lmaster[%u] "
1292                            "migrated_with_data[%s]\n",
1293                             ctdb_db->db_name, ctdb_db->db_id,
1294                             hash,
1295                             ctdb_lmaster(ctdb_db->ctdb, &key),
1296                             hdr->flags & CTDB_REC_FLAG_MIGRATED_WITH_DATA ? "yes" : "no"));
1297
1298         kd = (struct delete_record_data *)trbt_lookup32(ctdb_db->delete_queue, hash);
1299         if (kd != NULL) {
1300                 if ((kd->key.dsize != key.dsize) ||
1301                     (memcmp(kd->key.dptr, key.dptr, key.dsize) != 0))
1302                 {
1303                         DEBUG(DEBUG_INFO,
1304                               ("schedule for deletion: Hash collision (0x%08x)."
1305                                " Skipping the record.\n", hash));
1306                         return 0;
1307                 } else {
1308                         DEBUG(DEBUG_DEBUG,
1309                               ("schedule for deletion: Overwriting entry for "
1310                                "key with hash 0x%08x.\n", hash));
1311                 }
1312         }
1313
1314         ret = insert_delete_record_data_into_tree(ctdb_db->ctdb, ctdb_db,
1315                                                   ctdb_db->delete_queue,
1316                                                   hdr, key);
1317         if (ret != 0) {
1318                 return -1;
1319         }
1320
1321         return 0;
1322 }
1323
1324 /**
1325  * Schedule a record for deletetion.
1326  * Called from the parent context.
1327  */
1328 int32_t ctdb_control_schedule_for_deletion(struct ctdb_context *ctdb,
1329                                            TDB_DATA indata)
1330 {
1331         struct ctdb_control_schedule_for_deletion *dd;
1332         struct ctdb_db_context *ctdb_db;
1333         int ret;
1334         TDB_DATA key;
1335
1336         dd = (struct ctdb_control_schedule_for_deletion *)indata.dptr;
1337
1338         ctdb_db = find_ctdb_db(ctdb, dd->db_id);
1339         if (ctdb_db == NULL) {
1340                 DEBUG(DEBUG_ERR, (__location__ " Unknown db id 0x%08x\n",
1341                                   dd->db_id));
1342                 return -1;
1343         }
1344
1345         key.dsize = dd->keylen;
1346         key.dptr = dd->key;
1347
1348         ret = insert_record_into_delete_queue(ctdb_db, &dd->hdr, key);
1349
1350         return ret;
1351 }
1352
1353 int32_t ctdb_local_schedule_for_deletion(struct ctdb_db_context *ctdb_db,
1354                                          const struct ctdb_ltdb_header *hdr,
1355                                          TDB_DATA key)
1356 {
1357         int ret;
1358         struct ctdb_control_schedule_for_deletion *dd;
1359         TDB_DATA indata;
1360         int32_t status;
1361
1362         if (ctdb_db->ctdb->ctdbd_pid == getpid()) {
1363                 /* main daemon - directly queue */
1364                 ret = insert_record_into_delete_queue(ctdb_db, hdr, key);
1365
1366                 return ret;
1367         }
1368
1369         /* child process: send the main daemon a control */
1370
1371         indata.dsize = offsetof(struct ctdb_control_schedule_for_deletion, key) + key.dsize;
1372         indata.dptr = talloc_zero_array(ctdb_db, uint8_t, indata.dsize);
1373         if (indata.dptr == NULL) {
1374                 DEBUG(DEBUG_ERR, (__location__ " out of memory\n"));
1375                 return -1;
1376         }
1377         dd = (struct ctdb_control_schedule_for_deletion *)(void *)indata.dptr;
1378         dd->db_id = ctdb_db->db_id;
1379         dd->hdr = *hdr;
1380         dd->keylen = key.dsize;
1381         memcpy(dd->key, key.dptr, key.dsize);
1382
1383         ret = ctdb_control(ctdb_db->ctdb,
1384                            CTDB_CURRENT_NODE,
1385                            ctdb_db->db_id,
1386                            CTDB_CONTROL_SCHEDULE_FOR_DELETION,
1387                            CTDB_CTRL_FLAG_NOREPLY, /* flags */
1388                            indata,
1389                            NULL, /* mem_ctx */
1390                            NULL, /* outdata */
1391                            &status,
1392                            NULL, /* timeout : NULL == wait forever */
1393                            NULL); /* error message */
1394
1395         talloc_free(indata.dptr);
1396
1397         if (ret != 0 || status != 0) {
1398                 DEBUG(DEBUG_ERR, (__location__ " Error sending "
1399                                   "SCHEDULE_FOR_DELETION "
1400                                   "control.\n"));
1401                 if (status != 0) {
1402                         ret = -1;
1403                 }
1404         }
1405
1406         return ret;
1407 }