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