ctdb/server/ctdb_freeze.c

   1 /*
   2    ctdb freeze handling
   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 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 #include "includes.h"
  20 #include "tdb.h"
  21 #include "system/network.h"
  22 #include "system/filesys.h"
  23 #include "system/wait.h"
  24 #include "../include/ctdb_private.h"
  25 #include "lib/util/dlinklist.h"
  26 #include "lib/tdb_wrap/tdb_wrap.h"
  27 #include "../common/rb_tree.h"
  28
  29
  30 /**
  31  * Cancel a transaction on database
  32  */
  33 static int db_transaction_cancel_handler(struct ctdb_db_context *ctdb_db,
  34                                          void *private_data)
  35 {
  36         int ret;
  37
  38         tdb_add_flags(ctdb_db->ltdb->tdb, TDB_NOLOCK);
  39         ret = tdb_transaction_cancel(ctdb_db->ltdb->tdb);
  40         if (ret != 0) {
  41                 DEBUG(DEBUG_ERR, ("Failed to cancel transaction for db %s\n",
  42                                   ctdb_db->db_name));
  43         }
  44         tdb_remove_flags(ctdb_db->ltdb->tdb, TDB_NOLOCK);
  45         return 0;
  46 }
  47
  48 /**
  49  * Start a transaction on database
  50  */
  51 static int db_transaction_start_handler(struct ctdb_db_context *ctdb_db,
  52                                         void *private_data)
  53 {
  54         bool freeze_transaction_started = *(bool *)private_data;
  55         int ret;
  56
  57         tdb_add_flags(ctdb_db->ltdb->tdb, TDB_NOLOCK);
  58         if (freeze_transaction_started) {
  59                 ret = tdb_transaction_cancel(ctdb_db->ltdb->tdb);
  60                 if (ret != 0) {
  61                         DEBUG(DEBUG_ERR,
  62                               ("Failed to cancel transaction for db %s\n",
  63                                ctdb_db->db_name));
  64                 }
  65         }
  66         ret = tdb_transaction_start(ctdb_db->ltdb->tdb);
  67         tdb_remove_flags(ctdb_db->ltdb->tdb, TDB_NOLOCK);
  68         if (ret != 0) {
  69                 DEBUG(DEBUG_ERR, ("Failed to start transaction for db %s\n",
  70                                   ctdb_db->db_name));
  71                 return -1;
  72         }
  73         return 0;
  74 }
  75
  76 /**
  77  * Commit a transaction on database
  78  */
  79 static int db_transaction_commit_handler(struct ctdb_db_context *ctdb_db,
  80                                          void *private_data)
  81 {
  82         int healthy_nodes = *(int *)private_data;
  83         int ret;
  84
  85         tdb_add_flags(ctdb_db->ltdb->tdb, TDB_NOLOCK);
  86         ret = tdb_transaction_commit(ctdb_db->ltdb->tdb);
  87         tdb_remove_flags(ctdb_db->ltdb->tdb, TDB_NOLOCK);
  88         if (ret != 0) {
  89                 DEBUG(DEBUG_ERR, ("Failed to commit transaction for db %s\n",
  90                                   ctdb_db->db_name));
  91                 return -1;
  92         }
  93
  94         ret = ctdb_update_persistent_health(ctdb_db->ctdb, ctdb_db, NULL,
  95                                             healthy_nodes);
  96         if (ret != 0) {
  97                 DEBUG(DEBUG_ERR, ("Failed to update persistent health for db %s\n",
  98                                   ctdb_db->db_name));
  99         }
 100         return ret;
 101 }
 102
 103
 104 /* a list of control requests waiting for db freeze */
 105 struct ctdb_db_freeze_waiter {
 106         struct ctdb_db_freeze_waiter *next, *prev;
 107         struct ctdb_context *ctdb;
 108         void *private_data;
 109         int32_t status;
 110 };
 111
 112 /* a handle to a db freeze lock child process */
 113 struct ctdb_db_freeze_handle {
 114         struct ctdb_db_context *ctdb_db;
 115         struct lock_request *lreq;
 116         struct ctdb_db_freeze_waiter *waiters;
 117 };
 118
 119 /**
 120  * Called when freeing database freeze handle
 121  */
 122 static int ctdb_db_freeze_handle_destructor(struct ctdb_db_freeze_handle *h)
 123 {
 124         struct ctdb_db_context *ctdb_db = h->ctdb_db;
 125
 126         DEBUG(DEBUG_ERR, ("Release freeze handle for db %s\n",
 127                           ctdb_db->db_name));
 128
 129         /* Cancel any pending transactions */
 130         if (ctdb_db->freeze_transaction_started) {
 131                 db_transaction_cancel_handler(ctdb_db, NULL);
 132                 ctdb_db->freeze_transaction_started = false;
 133         }
 134         ctdb_db->freeze_mode = CTDB_FREEZE_NONE;
 135         ctdb_db->freeze_handle = NULL;
 136
 137         talloc_free(h->lreq);
 138         return 0;
 139 }
 140
 141 /**
 142  * Called when a database is frozen
 143  */
 144 static void ctdb_db_freeze_handler(void *private_data, bool locked)
 145 {
 146         struct ctdb_db_freeze_handle *h = talloc_get_type_abort(
 147                 private_data, struct ctdb_db_freeze_handle);
 148         struct ctdb_db_freeze_waiter *w;
 149
 150         if (h->ctdb_db->freeze_mode == CTDB_FREEZE_FROZEN) {
 151                 DEBUG(DEBUG_ERR, ("Freeze db child died - unfreezing\n"));
 152                 h->ctdb_db->freeze_mode = CTDB_FREEZE_NONE;
 153                 talloc_free(h);
 154                 return;
 155         }
 156
 157         if (!locked) {
 158                 DEBUG(DEBUG_ERR, ("Failed to get db lock for %s\n",
 159                                   h->ctdb_db->db_name));
 160                 h->ctdb_db->freeze_mode = CTDB_FREEZE_NONE;
 161                 talloc_free(h);
 162                 return;
 163         }
 164
 165         h->ctdb_db->freeze_mode = CTDB_FREEZE_FROZEN;
 166
 167         /* notify the waiters */
 168         while ((w = h->waiters) != NULL) {
 169                 w->status = 0;
 170                 DLIST_REMOVE(h->waiters, w);
 171                 talloc_free(w);
 172         }
 173 }
 174
 175 /**
 176  * Start freeze process for a database
 177  */
 178 static void ctdb_start_db_freeze(struct ctdb_db_context *ctdb_db)
 179 {
 180         struct ctdb_db_freeze_handle *h;
 181
 182         if (ctdb_db->freeze_mode == CTDB_FREEZE_FROZEN) {
 183                 return;
 184         }
 185
 186         if (ctdb_db->freeze_handle != NULL) {
 187                 return;
 188         }
 189
 190         DEBUG(DEBUG_ERR, ("Freeze db: %s\n", ctdb_db->db_name));
 191
 192         ctdb_stop_vacuuming(ctdb_db->ctdb);
 193
 194         h = talloc_zero(ctdb_db, struct ctdb_db_freeze_handle);
 195         CTDB_NO_MEMORY_FATAL(ctdb_db->ctdb, h);
 196
 197         h->ctdb_db = ctdb_db;
 198         h->lreq = ctdb_lock_db(h, ctdb_db, false, ctdb_db_freeze_handler, h);
 199         CTDB_NO_MEMORY_FATAL(ctdb_db->ctdb, h->lreq);
 200         talloc_set_destructor(h, ctdb_db_freeze_handle_destructor);
 201
 202         ctdb_db->freeze_handle = h;
 203         ctdb_db->freeze_mode = CTDB_FREEZE_PENDING;
 204 }
 205
 206 /**
 207  * Reply to a waiter for db freeze
 208  */
 209 static int ctdb_db_freeze_waiter_destructor(struct ctdb_db_freeze_waiter *w)
 210 {
 211         /* 'c' pointer is talloc_memdup(), so cannot use talloc_get_type */
 212         struct ctdb_req_control *c =
 213                 (struct ctdb_req_control *)w->private_data;
 214
 215         ctdb_request_control_reply(w->ctdb, c, NULL, w->status, NULL);
 216         return 0;
 217 }
 218
 219 /**
 220  * freeze a database
 221  */
 222 int32_t ctdb_control_db_freeze(struct ctdb_context *ctdb,
 223                                struct ctdb_req_control *c,
 224                                uint32_t db_id,
 225                                bool *async_reply)
 226 {
 227         struct ctdb_db_context *ctdb_db;
 228         struct ctdb_db_freeze_waiter *w;
 229
 230         ctdb_db = find_ctdb_db(ctdb, db_id);
 231         if (ctdb_db == NULL) {
 232                 DEBUG(DEBUG_ERR, ("Freeze db for unknown dbid 0x%08x\n", db_id));
 233                 return -1;
 234         }
 235
 236         if (ctdb_db->freeze_mode == CTDB_FREEZE_FROZEN) {
 237                 DEBUG(DEBUG_ERR, ("Freeze db: %s frozen\n", ctdb_db->db_name));
 238                 return 0;
 239         }
 240
 241         ctdb_start_db_freeze(ctdb_db);
 242
 243         /* add ourselves to the list of waiters */
 244         w = talloc(ctdb_db->freeze_handle, struct ctdb_db_freeze_waiter);
 245         CTDB_NO_MEMORY(ctdb, w);
 246         w->ctdb = ctdb;
 247         w->private_data = talloc_steal(w, c);
 248         w->status = -1;
 249         talloc_set_destructor(w, ctdb_db_freeze_waiter_destructor);
 250         DLIST_ADD(ctdb_db->freeze_handle->waiters, w);
 251
 252         *async_reply = true;
 253         return 0;
 254 }
 255
 256 /**
 257  * Thaw a database
 258  */
 259 int32_t ctdb_control_db_thaw(struct ctdb_context *ctdb, uint32_t db_id)
 260 {
 261         struct ctdb_db_context *ctdb_db;
 262
 263         ctdb_db = find_ctdb_db(ctdb, db_id);
 264         if (ctdb_db == NULL) {
 265                 DEBUG(DEBUG_ERR, ("Thaw db for unknown dbid 0x%08x\n", db_id));
 266                 return -1;
 267         }
 268
 269         DEBUG(DEBUG_ERR, ("Thaw db: %s\n", ctdb_db->db_name));
 270
 271         TALLOC_FREE(ctdb_db->freeze_handle);
 272         ctdb_call_resend_db(ctdb_db);
 273         return 0;
 274 }
 275
 276
 277 /*
 278   a list of control requests waiting for a freeze lock child to get
 279   the database locks
 280  */
 281 struct ctdb_freeze_waiter {
 282         struct ctdb_freeze_waiter *next, *prev;
 283         struct ctdb_context *ctdb;
 284         struct ctdb_req_control *c;
 285         uint32_t priority;
 286         int32_t status;
 287 };
 288
 289 /* a handle to a freeze lock child process */
 290 struct ctdb_freeze_handle {
 291         struct ctdb_context *ctdb;
 292         uint32_t priority;
 293         unsigned int num_total, num_locked, num_failed;
 294         struct ctdb_freeze_waiter *waiters;
 295 };
 296
 297 static int db_thaw(struct ctdb_db_context *ctdb_db, void *private_data)
 298 {
 299         talloc_free(ctdb_db->freeze_handle);
 300         return 0;
 301 }
 302
 303 /*
 304   destroy a freeze handle
 305  */
 306 static int ctdb_freeze_handle_destructor(struct ctdb_freeze_handle *h)
 307 {
 308         struct ctdb_context *ctdb = h->ctdb;
 309
 310         DEBUG(DEBUG_ERR,("Release freeze handler for prio %u\n", h->priority));
 311
 312         /* cancel any pending transactions */
 313         if (ctdb->freeze_transaction_started) {
 314                 ctdb_db_prio_iterator(ctdb, h->priority,
 315                                       db_transaction_cancel_handler, NULL);
 316                 ctdb->freeze_transaction_started = false;
 317         }
 318
 319         ctdb_db_prio_iterator(ctdb, h->priority, db_thaw, NULL);
 320
 321         ctdb->freeze_mode[h->priority]    = CTDB_FREEZE_NONE;
 322         ctdb->freeze_handles[h->priority] = NULL;
 323
 324         return 0;
 325 }
 326
 327 /*
 328   called when the child writes its status to us
 329  */
 330 static void ctdb_freeze_lock_handler(void *private_data, bool locked)
 331 {
 332         struct ctdb_freeze_handle *h = talloc_get_type_abort(private_data,
 333                                                              struct ctdb_freeze_handle);
 334         struct ctdb_freeze_waiter *w;
 335
 336         if (h->ctdb->freeze_mode[h->priority] == CTDB_FREEZE_FROZEN) {
 337                 DEBUG(DEBUG_INFO,("freeze child died - unfreezing\n"));
 338                 talloc_free(h);
 339                 return;
 340         }
 341
 342         if (!locked) {
 343                 DEBUG(DEBUG_ERR,("Failed to get locks in ctdb_freeze_child\n"));
 344                 /* we didn't get the locks - destroy the handle */
 345                 talloc_free(h);
 346                 return;
 347         }
 348
 349         h->ctdb->freeze_mode[h->priority] = CTDB_FREEZE_FROZEN;
 350
 351         /* notify the waiters */
 352         if (h != h->ctdb->freeze_handles[h->priority]) {
 353                 DEBUG(DEBUG_ERR,("lockwait finished but h is not linked\n"));
 354         }
 355         while ((w = h->waiters)) {
 356                 w->status = 0;
 357                 DLIST_REMOVE(h->waiters, w);
 358                 talloc_free(w);
 359         }
 360 }
 361
 362 /**
 363  * When single database is frozen
 364  */
 365 static int db_freeze_waiter_destructor(struct ctdb_db_freeze_waiter *w)
 366 {
 367         struct ctdb_freeze_handle *h = talloc_get_type_abort(
 368                 w->private_data, struct ctdb_freeze_handle);
 369
 370         if (w->status == 0) {
 371                 h->num_locked += 1;
 372         } else {
 373                 h->num_failed += 1;
 374         }
 375
 376         /* Call ctdb_freeze_lock_handler() only when the status of all
 377          * databases is known.
 378          */
 379         if (h->num_locked + h->num_failed == h->num_total) {
 380                 bool locked;
 381
 382                 if (h->num_locked == h->num_total) {
 383                         locked = true;
 384                 } else {
 385                         locked = false;
 386                 }
 387                 ctdb_freeze_lock_handler(h, locked);
 388         }
 389         return 0;
 390 }
 391
 392 /**
 393  * Count the number of databases
 394  */
 395 static int db_count(struct ctdb_db_context *ctdb_db, void *private_data)
 396 {
 397         int *count = (int *)private_data;
 398
 399         *count += 1;
 400
 401         return 0;
 402 }
 403
 404 /**
 405  * Freeze a single database
 406  */
 407 static int db_freeze(struct ctdb_db_context *ctdb_db, void *private_data)
 408 {
 409         struct ctdb_freeze_handle *h = talloc_get_type_abort(
 410                 private_data, struct ctdb_freeze_handle);
 411         struct ctdb_db_freeze_waiter *w;
 412
 413         ctdb_start_db_freeze(ctdb_db);
 414
 415         w = talloc(ctdb_db->freeze_handle, struct ctdb_db_freeze_waiter);
 416         CTDB_NO_MEMORY(h->ctdb, w);
 417         w->ctdb = h->ctdb;
 418         w->private_data = h;
 419         w->status = -1;
 420         talloc_set_destructor(w, db_freeze_waiter_destructor);
 421
 422         if (ctdb_db->freeze_mode == CTDB_FREEZE_FROZEN) {
 423                 /* Early return if already frozen */
 424                 w->status = 0;
 425                 talloc_free(w);
 426                 return 0;
 427         }
 428
 429         DLIST_ADD(ctdb_db->freeze_handle->waiters, w);
 430
 431         return 0;
 432 }
 433
 434 /*
 435   start the freeze process for a certain priority
 436  */
 437 void ctdb_start_freeze(struct ctdb_context *ctdb, uint32_t priority)
 438 {
 439         struct ctdb_freeze_handle *h;
 440         int ret;
 441
 442         if ((priority < 1) || (priority > NUM_DB_PRIORITIES)) {
 443                 DEBUG(DEBUG_ERR,(__location__ " Invalid db priority : %u\n", priority));
 444                 ctdb_fatal(ctdb, "Internal error");
 445         }
 446
 447         if (ctdb->freeze_mode[priority] == CTDB_FREEZE_FROZEN) {
 448                 /* we're already frozen */
 449                 return;
 450         }
 451
 452         if (ctdb->freeze_handles[priority] != NULL) {
 453                 /* already trying to freeze */
 454                 return;
 455         }
 456
 457         DEBUG(DEBUG_ERR, ("Freeze priority %u\n", priority));
 458
 459         /* Stop any vacuuming going on: we don't want to wait. */
 460         ctdb_stop_vacuuming(ctdb);
 461
 462         /* create freeze lock children for each database */
 463         h = talloc_zero(ctdb, struct ctdb_freeze_handle);
 464         CTDB_NO_MEMORY_FATAL(ctdb, h);
 465         h->ctdb = ctdb;
 466         h->priority = priority;
 467         talloc_set_destructor(h, ctdb_freeze_handle_destructor);
 468
 469         ret = ctdb_db_prio_iterator(ctdb, priority, db_count, &h->num_total);
 470         if (ret != 0) {
 471                 talloc_free(h);
 472                 return;
 473         }
 474
 475         ctdb->freeze_mode[priority] = CTDB_FREEZE_PENDING;
 476
 477         ret = ctdb_db_prio_iterator(ctdb, priority, db_freeze, h);
 478         if (ret != 0) {
 479                 talloc_free(h);
 480                 return;
 481         }
 482
 483         ctdb->freeze_handles[priority] = h;
 484         if (h->num_total == 0) {
 485                 ctdb->freeze_mode[priority] = CTDB_FREEZE_FROZEN;
 486         }
 487 }
 488
 489 /*
 490   destroy a waiter for a freeze mode change
 491  */
 492 static int ctdb_freeze_waiter_destructor(struct ctdb_freeze_waiter *w)
 493 {
 494         ctdb_request_control_reply(w->ctdb, w->c, NULL, w->status, NULL);
 495         return 0;
 496 }
 497
 498 /*
 499   freeze the databases
 500  */
 501 int32_t ctdb_control_freeze(struct ctdb_context *ctdb, struct ctdb_req_control *c, bool *async_reply)
 502 {
 503         struct ctdb_freeze_waiter *w;
 504         uint32_t priority;
 505
 506         priority = (uint32_t)c->srvid;
 507
 508         if (priority == 0) {
 509                 DEBUG(DEBUG_ERR,("Freeze priority 0 requested, remapping to priority 1\n"));
 510                 priority = 1;
 511         }
 512
 513         if ((priority < 1) || (priority > NUM_DB_PRIORITIES)) {
 514                 DEBUG(DEBUG_ERR,(__location__ " Invalid db priority : %u\n", priority));
 515                 return -1;
 516         }
 517
 518         if (ctdb->freeze_mode[priority] == CTDB_FREEZE_FROZEN) {
 519                 DEBUG(DEBUG_ERR, ("Freeze priority %u: frozen\n", priority));
 520                 /* we're already frozen */
 521                 return 0;
 522         }
 523
 524         ctdb_start_freeze(ctdb, priority);
 525
 526         if (ctdb->freeze_handles[priority] == NULL) {
 527                 DEBUG(DEBUG_ERR,("No freeze lock handle when adding a waiter\n"));
 528                 return -1;
 529         }
 530
 531         /* If there are no databases, we are done. */
 532         if (ctdb->freeze_handles[priority]->num_total == 0) {
 533                 return 0;
 534         }
 535
 536         /* add ourselves to list of waiters */
 537         w = talloc(ctdb->freeze_handles[priority], struct ctdb_freeze_waiter);
 538         CTDB_NO_MEMORY(ctdb, w);
 539         w->ctdb     = ctdb;
 540         w->c        = talloc_steal(w, c);
 541         w->priority = priority;
 542         w->status   = -1;
 543         talloc_set_destructor(w, ctdb_freeze_waiter_destructor);
 544         DLIST_ADD(ctdb->freeze_handles[priority]->waiters, w);
 545
 546         /* we won't reply till later */
 547         *async_reply = true;
 548         return 0;
 549 }
 550
 551
 552 /*
 553   block until we are frozen, used during daemon startup
 554  */
 555 bool ctdb_blocking_freeze(struct ctdb_context *ctdb)
 556 {
 557         int i;
 558
 559         for (i=1; i<=NUM_DB_PRIORITIES; i++) {
 560                 ctdb_start_freeze(ctdb, i);
 561
 562                 /* block until frozen */
 563                 while (ctdb->freeze_mode[i] == CTDB_FREEZE_PENDING) {
 564                         event_loop_once(ctdb->ev);
 565                 }
 566         }
 567
 568         return true;
 569 }
 570
 571
 572 static void thaw_priority(struct ctdb_context *ctdb, uint32_t priority)
 573 {
 574         DEBUG(DEBUG_ERR,("Thawing priority %u\n", priority));
 575
 576         /* cancel any pending transactions */
 577         if (ctdb->freeze_transaction_started) {
 578                 ctdb_db_prio_iterator(ctdb, priority,
 579                                       db_transaction_cancel_handler, NULL);
 580                 ctdb->freeze_transaction_started = false;
 581         }
 582
 583         ctdb_db_prio_iterator(ctdb, priority, db_thaw, NULL);
 584         if (ctdb->freeze_handles[priority] != NULL) {
 585                 talloc_free(ctdb->freeze_handles[priority]);
 586                 ctdb->freeze_handles[priority] = NULL;
 587         }
 588 }
 589
 590 /*
 591   thaw the databases
 592  */
 593 int32_t ctdb_control_thaw(struct ctdb_context *ctdb, uint32_t priority,
 594                           bool check_recmode)
 595 {
 596         if (priority > NUM_DB_PRIORITIES) {
 597                 DEBUG(DEBUG_ERR,(__location__ " Invalid db priority : %u\n",
 598                                  priority));
 599                 return -1;
 600         }
 601
 602         if (check_recmode && ctdb->recovery_mode == CTDB_RECOVERY_ACTIVE) {
 603                 DEBUG(DEBUG_ERR, ("Failing to thaw databases while "
 604                                   "recovery is active\n"));
 605                 return -1;
 606         }
 607
 608         if (priority == 0) {
 609                 int i;
 610                 for (i=1;i<=NUM_DB_PRIORITIES; i++) {
 611                         thaw_priority(ctdb, i);
 612                 }
 613         } else {
 614                 thaw_priority(ctdb, priority);
 615         }
 616
 617         ctdb_call_resend_all(ctdb);
 618         return 0;
 619 }
 620
 621
 622 /*
 623   start a transaction on all databases - used for recovery
 624  */
 625 int32_t ctdb_control_transaction_start(struct ctdb_context *ctdb, uint32_t id)
 626 {
 627         int ret;
 628
 629         if (!ctdb_db_all_frozen(ctdb)) {
 630                 DEBUG(DEBUG_ERR, (__location__
 631                       " failing transaction start while not frozen\n"));
 632                 return -1;
 633         }
 634
 635         ret = ctdb_db_iterator(ctdb, db_transaction_start_handler,
 636                                &ctdb->freeze_transaction_started);
 637         if (ret != 0) {
 638                 return -1;
 639         }
 640
 641         ctdb->freeze_transaction_started = true;
 642         ctdb->freeze_transaction_id = id;
 643
 644         return 0;
 645 }
 646
 647 /*
 648   cancel a transaction for all databases - used for recovery
 649  */
 650 int32_t ctdb_control_transaction_cancel(struct ctdb_context *ctdb)
 651 {
 652         DEBUG(DEBUG_ERR,(__location__ " recovery transaction cancelled called\n"));
 653
 654         ctdb_db_iterator(ctdb, db_transaction_cancel_handler, NULL);
 655         ctdb->freeze_transaction_started = false;
 656
 657         return 0;
 658 }
 659
 660 /*
 661   commit transactions on all databases
 662  */
 663 int32_t ctdb_control_transaction_commit(struct ctdb_context *ctdb, uint32_t id)
 664 {
 665         int i;
 666         int healthy_nodes = 0;
 667         int ret;
 668
 669         if (!ctdb_db_all_frozen(ctdb)) {
 670                 DEBUG(DEBUG_ERR, (__location__
 671                       " failing transaction commit while not frozen\n"));
 672                 return -1;
 673         }
 674
 675         if (!ctdb->freeze_transaction_started) {
 676                 DEBUG(DEBUG_ERR,(__location__ " transaction not started\n"));
 677                 return -1;
 678         }
 679
 680         if (id != ctdb->freeze_transaction_id) {
 681                 DEBUG(DEBUG_ERR,(__location__ " incorrect transaction id 0x%x in commit\n", id));
 682                 return -1;
 683         }
 684
 685         DEBUG(DEBUG_DEBUG,(__location__ " num_nodes[%d]\n", ctdb->num_nodes));
 686         for (i=0; i < ctdb->num_nodes; i++) {
 687                 DEBUG(DEBUG_DEBUG,(__location__ " node[%d].flags[0x%X]\n",
 688                                    i, ctdb->nodes[i]->flags));
 689                 if (ctdb->nodes[i]->flags == 0) {
 690                         healthy_nodes++;
 691                 }
 692         }
 693         DEBUG(DEBUG_INFO,(__location__ " healthy_nodes[%d]\n", healthy_nodes));
 694
 695         ret = ctdb_db_iterator(ctdb, db_transaction_commit_handler,
 696                                &healthy_nodes);
 697         if (ret != 0) {
 698                 DEBUG(DEBUG_ERR, ("Cancel all transactions\n"));
 699                 goto fail;
 700         }
 701
 702         ctdb->freeze_transaction_started = false;
 703         ctdb->freeze_transaction_id = 0;
 704
 705         return 0;
 706
 707 fail:
 708         /* cancel any pending transactions */
 709         ctdb_db_iterator(ctdb, db_transaction_cancel_handler, NULL);
 710         ctdb->freeze_transaction_started = false;
 711
 712         return -1;
 713 }
 714
 715 /*
 716   wipe a database - only possible when in a frozen transaction
 717  */
 718 int32_t ctdb_control_wipe_database(struct ctdb_context *ctdb, TDB_DATA indata)
 719 {
 720         struct ctdb_control_wipe_database w = *(struct ctdb_control_wipe_database *)indata.dptr;
 721         struct ctdb_db_context *ctdb_db;
 722
 723         ctdb_db = find_ctdb_db(ctdb, w.db_id);
 724         if (!ctdb_db) {
 725                 DEBUG(DEBUG_ERR,(__location__ " Unknown db 0x%x\n", w.db_id));
 726                 return -1;
 727         }
 728
 729         if (ctdb->freeze_mode[ctdb_db->priority] != CTDB_FREEZE_FROZEN) {
 730                 DEBUG(DEBUG_ERR,(__location__ " Failed transaction_start while not frozen\n"));
 731                 return -1;
 732         }
 733
 734         if (!ctdb->freeze_transaction_started) {
 735                 DEBUG(DEBUG_ERR,(__location__ " transaction not started\n"));
 736                 return -1;
 737         }
 738
 739         if (w.transaction_id != ctdb->freeze_transaction_id) {
 740                 DEBUG(DEBUG_ERR,(__location__ " incorrect transaction id 0x%x in commit\n", w.transaction_id));
 741                 return -1;
 742         }
 743
 744         if (tdb_wipe_all(ctdb_db->ltdb->tdb) != 0) {
 745                 DEBUG(DEBUG_ERR,(__location__ " Failed to wipe database for db '%s'\n",
 746                          ctdb_db->db_name));
 747                 return -1;
 748         }
 749
 750         if (!ctdb_db->persistent) {
 751                 talloc_free(ctdb_db->delete_queue);
 752                 ctdb_db->delete_queue = trbt_create(ctdb_db, 0);
 753                 if (ctdb_db->delete_queue == NULL) {
 754                         DEBUG(DEBUG_ERR, (__location__ " Failed to re-create "
 755                                           "the vacuum tree.\n"));
 756                         return -1;
 757                 }
 758         }
 759
 760         return 0;
 761 }
 762
 763 bool ctdb_db_frozen(struct ctdb_db_context *ctdb_db)
 764 {
 765         if (ctdb_db->freeze_mode != CTDB_FREEZE_FROZEN) {
 766                 return false;
 767         }
 768
 769         return true;
 770 }
 771
 772 bool ctdb_db_prio_frozen(struct ctdb_context *ctdb, uint32_t priority)
 773 {
 774         if (priority == 0) {
 775                 priority = 1;
 776         }
 777         if (priority > NUM_DB_PRIORITIES) {
 778                 DEBUG(DEBUG_ERR, ("Invalid DB priority specified\n"));
 779                 return false;
 780         }
 781
 782         if (ctdb->freeze_mode[priority] != CTDB_FREEZE_FROZEN) {
 783                 return false;
 784         }
 785
 786         return true;
 787 }
 788
 789 bool ctdb_db_all_frozen(struct ctdb_context *ctdb)
 790 {
 791         int i;
 792
 793         for (i=1; i<=NUM_DB_PRIORITIES; i++) {
 794                 if (ctdb->freeze_mode[i] != CTDB_FREEZE_FROZEN) {
 795                         return false;
 796                 }
 797         }
 798         return true;
 799 }