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