Adding upstream version 2:4.17.12+dfsg.upstream/2%4.17.12+dfsgupstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 923 insertions, 0 deletions

diff --git a/ctdb/server/ctdb_freeze.c b/ctdb/server/ctdb_freeze.c
new file mode 100644
index 0000000..06aeacf
--- /dev/null
+++ b/ctdb/server/ctdb_freeze.c
@@ -0,0 +1,923 @@
+/* 
+   ctdb freeze handling
+
+   Copyright (C) Andrew Tridgell  2007
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 3 of the License, or
+   (at your option) any later version.
+   
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+   
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, see <http://www.gnu.org/licenses/>.
+*/
+#include "replace.h"
+#include "system/network.h"
+#include "system/filesys.h"
+#include "system/wait.h"
+
+#include <talloc.h>
+#include <tevent.h>
+
+#include "lib/tdb_wrap/tdb_wrap.h"
+#include "lib/util/dlinklist.h"
+#include "lib/util/debug.h"
+
+#include "ctdb_private.h"
+
+#include "common/rb_tree.h"
+#include "common/common.h"
+#include "common/logging.h"
+
+/**
+ * Cancel a transaction on database
+ */
+static int db_transaction_cancel_handler(struct ctdb_db_context *ctdb_db,
+					 void *private_data)
+{
+	int ret;
+
+	tdb_add_flags(ctdb_db->ltdb->tdb, TDB_NOLOCK);
+	ret = tdb_transaction_cancel(ctdb_db->ltdb->tdb);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR, ("Failed to cancel transaction for db %s\n",
+				  ctdb_db->db_name));
+	}
+	tdb_remove_flags(ctdb_db->ltdb->tdb, TDB_NOLOCK);
+	return 0;
+}
+
+/**
+ * Start a transaction on database
+ */
+static int db_transaction_start_handler(struct ctdb_db_context *ctdb_db,
+					void *private_data)
+{
+	bool freeze_transaction_started = *(bool *)private_data;
+	int ret;
+
+	tdb_add_flags(ctdb_db->ltdb->tdb, TDB_NOLOCK);
+	if (freeze_transaction_started) {
+		ret = tdb_transaction_cancel(ctdb_db->ltdb->tdb);
+		if (ret != 0) {
+			DEBUG(DEBUG_ERR,
+			      ("Failed to cancel transaction for db %s\n",
+			       ctdb_db->db_name));
+		}
+	}
+	ret = tdb_transaction_start(ctdb_db->ltdb->tdb);
+	tdb_remove_flags(ctdb_db->ltdb->tdb, TDB_NOLOCK);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR, ("Failed to start transaction for db %s\n",
+				  ctdb_db->db_name));
+		return -1;
+	}
+	return 0;
+}
+
+/**
+ * Commit a transaction on database
+ */
+static int db_transaction_commit_handler(struct ctdb_db_context *ctdb_db,
+					 void *private_data)
+{
+	unsigned int healthy_nodes = *(unsigned int *)private_data;
+	int ret;
+
+	tdb_add_flags(ctdb_db->ltdb->tdb, TDB_NOLOCK);
+	ret = tdb_transaction_commit(ctdb_db->ltdb->tdb);
+	tdb_remove_flags(ctdb_db->ltdb->tdb, TDB_NOLOCK);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR, ("Failed to commit transaction for db %s\n",
+				  ctdb_db->db_name));
+		return -1;
+	}
+
+	ret = ctdb_update_persistent_health(ctdb_db->ctdb, ctdb_db, NULL,
+					    healthy_nodes);
+	if (ret != 0) {
+		DEBUG(DEBUG_ERR, ("Failed to update persistent health for db %s\n",
+				  ctdb_db->db_name));
+	}
+	return ret;
+}
+
+/* a list of control requests waiting for db freeze */
+struct ctdb_db_freeze_waiter {
+	struct ctdb_db_freeze_waiter *next, *prev;
+	struct ctdb_context *ctdb;
+	void *private_data;
+	int32_t status;
+};
+
+/* a handle to a db freeze lock child process */
+struct ctdb_db_freeze_handle {
+	struct ctdb_db_context *ctdb_db;
+	struct lock_request *lreq;
+	struct ctdb_db_freeze_waiter *waiters;
+};
+
+/**
+ * Called when freeing database freeze handle
+ */
+static int ctdb_db_freeze_handle_destructor(struct ctdb_db_freeze_handle *h)
+{
+	struct ctdb_db_context *ctdb_db = h->ctdb_db;
+
+	DEBUG(DEBUG_ERR, ("Release freeze handle for db %s\n",
+			  ctdb_db->db_name));
+
+	/* Cancel any pending transactions */
+	if (ctdb_db->freeze_transaction_started) {
+		db_transaction_cancel_handler(ctdb_db, NULL);
+		ctdb_db->freeze_transaction_started = false;
+	}
+	ctdb_db->freeze_mode = CTDB_FREEZE_NONE;
+	ctdb_db->freeze_handle = NULL;
+
+	/* Clear invalid records flag */
+	ctdb_db->invalid_records = false;
+
+	talloc_free(h->lreq);
+	return 0;
+}
+
+/**
+ * Called when a database is frozen
+ */
+static void ctdb_db_freeze_handler(void *private_data, bool locked)
+{
+	struct ctdb_db_freeze_handle *h = talloc_get_type_abort(
+		private_data, struct ctdb_db_freeze_handle);
+	struct ctdb_db_freeze_waiter *w;
+
+	if (h->ctdb_db->freeze_mode == CTDB_FREEZE_FROZEN) {
+		DEBUG(DEBUG_ERR, ("Freeze db child died - unfreezing\n"));
+		h->ctdb_db->freeze_mode = CTDB_FREEZE_NONE;
+		talloc_free(h);
+		return;
+	}
+
+	if (!locked) {
+		DEBUG(DEBUG_ERR, ("Failed to get db lock for %s\n",
+				  h->ctdb_db->db_name));
+		h->ctdb_db->freeze_mode = CTDB_FREEZE_NONE;
+		talloc_free(h);
+		return;
+	}
+
+	h->ctdb_db->freeze_mode = CTDB_FREEZE_FROZEN;
+
+	/* notify the waiters */
+	while ((w = h->waiters) != NULL) {
+		w->status = 0;
+		DLIST_REMOVE(h->waiters, w);
+		talloc_free(w);
+	}
+}
+
+/**
+ * Start freeze process for a database
+ */
+static void ctdb_start_db_freeze(struct ctdb_db_context *ctdb_db)
+{
+	struct ctdb_db_freeze_handle *h;
+
+	if (ctdb_db->freeze_mode == CTDB_FREEZE_FROZEN) {
+		return;
+	}
+
+	if (ctdb_db->freeze_handle != NULL) {
+		return;
+	}
+
+	DEBUG(DEBUG_ERR, ("Freeze db: %s\n", ctdb_db->db_name));
+
+	ctdb_stop_vacuuming(ctdb_db->ctdb);
+
+	h = talloc_zero(ctdb_db, struct ctdb_db_freeze_handle);
+	CTDB_NO_MEMORY_FATAL(ctdb_db->ctdb, h);
+
+	h->ctdb_db = ctdb_db;
+	h->lreq = ctdb_lock_db(h, ctdb_db, false, ctdb_db_freeze_handler, h);
+	CTDB_NO_MEMORY_FATAL(ctdb_db->ctdb, h->lreq);
+	talloc_set_destructor(h, ctdb_db_freeze_handle_destructor);
+
+	ctdb_db->freeze_handle = h;
+	ctdb_db->freeze_mode = CTDB_FREEZE_PENDING;
+}
+
+/**
+ * Reply to a waiter for db freeze
+ */
+static int ctdb_db_freeze_waiter_destructor(struct ctdb_db_freeze_waiter *w)
+{
+	/* 'c' pointer is talloc_memdup(), so cannot use talloc_get_type */
+	struct ctdb_req_control_old *c =
+		(struct ctdb_req_control_old *)w->private_data;
+
+	ctdb_request_control_reply(w->ctdb, c, NULL, w->status, NULL);
+	return 0;
+}
+
+/**
+ * freeze a database
+ */
+int32_t ctdb_control_db_freeze(struct ctdb_context *ctdb,
+			       struct ctdb_req_control_old *c,
+			       uint32_t db_id,
+			       bool *async_reply)
+{
+	struct ctdb_db_context *ctdb_db;
+	struct ctdb_db_freeze_waiter *w;
+
+	ctdb_db = find_ctdb_db(ctdb, db_id);
+	if (ctdb_db == NULL) {
+		DEBUG(DEBUG_ERR, ("Freeze db for unknown dbid 0x%08x\n", db_id));
+		return -1;
+	}
+
+	if (ctdb_db->freeze_mode == CTDB_FREEZE_FROZEN) {
+		DEBUG(DEBUG_ERR, ("Freeze db: %s frozen\n", ctdb_db->db_name));
+		return 0;
+	}
+
+	ctdb_start_db_freeze(ctdb_db);
+
+	/* add ourselves to the list of waiters */
+	w = talloc(ctdb_db->freeze_handle, struct ctdb_db_freeze_waiter);
+	CTDB_NO_MEMORY(ctdb, w);
+	w->ctdb = ctdb;
+	w->private_data = talloc_steal(w, c);
+	w->status = -1;
+	talloc_set_destructor(w, ctdb_db_freeze_waiter_destructor);
+	DLIST_ADD(ctdb_db->freeze_handle->waiters, w);
+
+	*async_reply = true;
+	return 0;
+}
+
+/**
+ * Thaw a database
+ */
+int32_t ctdb_control_db_thaw(struct ctdb_context *ctdb, uint32_t db_id)
+{
+	struct ctdb_db_context *ctdb_db;
+
+	ctdb_db = find_ctdb_db(ctdb, db_id);
+	if (ctdb_db == NULL) {
+		DEBUG(DEBUG_ERR, ("Thaw db for unknown dbid 0x%08x\n", db_id));
+		return -1;
+	}
+
+	DEBUG(DEBUG_ERR, ("Thaw db: %s generation %u\n", ctdb_db->db_name,
+			  ctdb_db->generation));
+
+	TALLOC_FREE(ctdb_db->freeze_handle);
+	ctdb_call_resend_db(ctdb_db);
+	return 0;
+}
+
+
+/*
+  a list of control requests waiting for a freeze lock child to get
+  the database locks
+ */
+struct ctdb_freeze_waiter {
+	struct ctdb_freeze_waiter *next, *prev;
+	struct ctdb_context *ctdb;
+	struct ctdb_req_control_old *c;
+	int32_t status;
+};
+
+/* a handle to a freeze lock child process */
+struct ctdb_freeze_handle {
+	struct ctdb_context *ctdb;
+	unsigned int num_total, num_locked, num_failed;
+	struct ctdb_freeze_waiter *waiters;
+};
+
+static int db_thaw(struct ctdb_db_context *ctdb_db, void *private_data)
+{
+	talloc_free(ctdb_db->freeze_handle);
+	return 0;
+}
+
+/*
+  destroy a freeze handle
+ */
+static int ctdb_freeze_handle_destructor(struct ctdb_freeze_handle *h)
+{
+	struct ctdb_context *ctdb = h->ctdb;
+
+	DEBUG(DEBUG_ERR,("Release freeze handle\n"));
+
+	/* cancel any pending transactions */
+	if (ctdb->freeze_transaction_started) {
+		ctdb_db_iterator(ctdb, db_transaction_cancel_handler, NULL);
+		ctdb->freeze_transaction_started = false;
+	}
+
+	ctdb_db_iterator(ctdb, db_thaw, NULL);
+
+	ctdb->freeze_mode   = CTDB_FREEZE_NONE;
+	ctdb->freeze_handle = NULL;
+
+	return 0;
+}
+
+/*
+  called when the child writes its status to us
+ */
+static void ctdb_freeze_lock_handler(void *private_data, bool locked)
+{
+	struct ctdb_freeze_handle *h = talloc_get_type_abort(private_data,
+							     struct ctdb_freeze_handle);
+	struct ctdb_freeze_waiter *w;
+
+	if (h->ctdb->freeze_mode == CTDB_FREEZE_FROZEN) {
+		DEBUG(DEBUG_INFO,("freeze child died - unfreezing\n"));
+		talloc_free(h);
+		return;
+	}
+
+	if (!locked) {
+		DEBUG(DEBUG_ERR,("Failed to get locks in ctdb_freeze_child\n"));
+		/* we didn't get the locks - destroy the handle */
+		talloc_free(h);
+		return;
+	}
+
+	h->ctdb->freeze_mode = CTDB_FREEZE_FROZEN;
+
+	/* notify the waiters */
+	if (h != h->ctdb->freeze_handle) {
+		DEBUG(DEBUG_ERR,("lockwait finished but h is not linked\n"));
+	}
+	while ((w = h->waiters)) {
+		w->status = 0;
+		DLIST_REMOVE(h->waiters, w);
+		talloc_free(w);
+	}
+}
+
+/**
+ * When single database is frozen
+ */
+static int db_freeze_waiter_destructor(struct ctdb_db_freeze_waiter *w)
+{
+	struct ctdb_freeze_handle *h = talloc_get_type_abort(
+		w->private_data, struct ctdb_freeze_handle);
+
+	if (w->status == 0) {
+		h->num_locked += 1;
+	} else {
+		h->num_failed += 1;
+	}
+
+	/* Call ctdb_freeze_lock_handler() only when the status of all
+	 * databases is known.
+	 */
+	if (h->num_locked + h->num_failed == h->num_total) {
+		bool locked;
+
+		if (h->num_locked == h->num_total) {
+			locked = true;
+		} else {
+			locked = false;
+		}
+		ctdb_freeze_lock_handler(h, locked);
+	}
+	return 0;
+}
+
+/**
+ * Invalidate the records in the database.
+ * This only applies to volatile databases.
+ */
+static int db_invalidate(struct ctdb_db_context *ctdb_db, void *private_data)
+{
+	if (ctdb_db_volatile(ctdb_db)) {
+		ctdb_db->invalid_records = true;
+	}
+
+	return 0;
+}
+
+/**
+ * Count the number of databases
+ */
+static int db_count(struct ctdb_db_context *ctdb_db, void *private_data)
+{
+	unsigned int *count = (unsigned int *)private_data;
+
+	*count += 1;
+
+	return 0;
+}
+
+/**
+ * Freeze a single database
+ */
+static int db_freeze(struct ctdb_db_context *ctdb_db, void *private_data)
+{
+	struct ctdb_freeze_handle *h = talloc_get_type_abort(
+		private_data, struct ctdb_freeze_handle);
+	struct ctdb_db_freeze_waiter *w;
+
+	ctdb_start_db_freeze(ctdb_db);
+
+	w = talloc(ctdb_db->freeze_handle, struct ctdb_db_freeze_waiter);
+	CTDB_NO_MEMORY(h->ctdb, w);
+	w->ctdb = h->ctdb;
+	w->private_data = h;
+	w->status = -1;
+	talloc_set_destructor(w, db_freeze_waiter_destructor);
+
+	if (ctdb_db->freeze_mode == CTDB_FREEZE_FROZEN) {
+		/* Early return if already frozen */
+		w->status = 0;
+		talloc_free(w);
+		return 0;
+	}
+
+	DLIST_ADD(ctdb_db->freeze_handle->waiters, w);
+
+	return 0;
+}
+
+/*
+  start the freeze process for all databases
+  This is only called from ctdb_control_freeze(), which is called
+  only on node becoming INACTIVE.  So mark the records invalid.
+ */
+static void ctdb_start_freeze(struct ctdb_context *ctdb)
+{
+	struct ctdb_freeze_handle *h;
+	int ret;
+
+	ctdb_db_iterator(ctdb, db_invalidate, NULL);
+
+	if (ctdb->freeze_mode == CTDB_FREEZE_FROZEN) {
+		unsigned int count = 0;
+
+		/*
+		 * Check if all the databases are frozen
+		 *
+		 * It's possible that the databases can get attached after
+		 * initial freeze. This typically happens during startup as
+		 * CTDB will only attach persistent databases and go in to
+		 * startup freeze.  The recovery master during recovery will
+		 * attach all the missing databases.
+		 */
+
+		h = ctdb->freeze_handle;
+		if (h == NULL) {
+			ctdb->freeze_mode = CTDB_FREEZE_NONE;
+			return;
+		}
+
+		ret = ctdb_db_iterator(ctdb, db_count, &count);
+		if (ret != 0) {
+			TALLOC_FREE(ctdb->freeze_handle);
+			ctdb->freeze_mode = CTDB_FREEZE_NONE;
+			return;
+		}
+
+		if (count != h->num_total) {
+			DEBUG(DEBUG_ERR, ("Freeze all: incremental\n"));
+
+			h->num_total = count;
+			h->num_locked = 0;
+			h->num_failed = 0;
+
+			ctdb->freeze_mode = CTDB_FREEZE_PENDING;
+
+			ret = ctdb_db_iterator(ctdb, db_freeze, h);
+			if (ret != 0) {
+				TALLOC_FREE(ctdb->freeze_handle);
+				ctdb->freeze_mode = CTDB_FREEZE_NONE;
+			}
+		}
+		return;
+	}
+
+	if (ctdb->freeze_handle != NULL) {
+		/* already trying to freeze */
+		return;
+	}
+
+	DEBUG(DEBUG_ERR, ("Freeze all\n"));
+
+	/* Stop any vacuuming going on: we don't want to wait. */
+	ctdb_stop_vacuuming(ctdb);
+
+	/* create freeze lock children for each database */
+	h = talloc_zero(ctdb, struct ctdb_freeze_handle);
+	CTDB_NO_MEMORY_FATAL(ctdb, h);
+	h->ctdb = ctdb;
+	talloc_set_destructor(h, ctdb_freeze_handle_destructor);
+	ctdb->freeze_handle = h;
+
+	ret = ctdb_db_iterator(ctdb, db_count, &h->num_total);
+	if (ret != 0) {
+		talloc_free(h);
+		return;
+	}
+
+	ctdb->freeze_mode = CTDB_FREEZE_PENDING;
+
+	ret = ctdb_db_iterator(ctdb, db_freeze, h);
+	if (ret != 0) {
+		talloc_free(h);
+		return;
+	}
+
+	if (h->num_total == 0) {
+		ctdb->freeze_mode = CTDB_FREEZE_FROZEN;
+	}
+}
+
+/*
+  destroy a waiter for a freeze mode change
+ */
+static int ctdb_freeze_waiter_destructor(struct ctdb_freeze_waiter *w)
+{
+	ctdb_request_control_reply(w->ctdb, w->c, NULL, w->status, NULL);
+	return 0;
+}
+
+/*
+  freeze all the databases
+  This control is only used when freezing database on node becoming INACTIVE.
+  So mark the records invalid in ctdb_start_freeze().
+ */
+int32_t ctdb_control_freeze(struct ctdb_context *ctdb,
+			    struct ctdb_req_control_old *c, bool *async_reply)
+{
+	struct ctdb_freeze_waiter *w;
+
+	ctdb_start_freeze(ctdb);
+
+	if (ctdb->freeze_mode == CTDB_FREEZE_FROZEN) {
+		DEBUG(DEBUG_ERR, ("Freeze all: frozen\n"));
+		/* we're already frozen */
+		return 0;
+	}
+
+	if (ctdb->freeze_handle == NULL) {
+		DEBUG(DEBUG_ERR,("No freeze lock handle when adding a waiter\n"));
+		return -1;
+	}
+
+	/* If there are no databases, we are done. */
+	if (ctdb->freeze_handle->num_total == 0) {
+		return 0;
+	}
+
+	/* add ourselves to list of waiters */
+	w = talloc(ctdb->freeze_handle, struct ctdb_freeze_waiter);
+	CTDB_NO_MEMORY(ctdb, w);
+	w->ctdb     = ctdb;
+	w->c        = talloc_steal(w, c);
+	w->status   = -1;
+	talloc_set_destructor(w, ctdb_freeze_waiter_destructor);
+	DLIST_ADD(ctdb->freeze_handle->waiters, w);
+
+	/* we won't reply till later */
+	*async_reply = true;
+	return 0;
+}
+
+
+static int db_freeze_block(struct ctdb_db_context *ctdb_db, void *private_data)
+{
+	struct tevent_context *ev = (struct tevent_context *)private_data;
+
+	ctdb_start_db_freeze(ctdb_db);
+
+	while (ctdb_db->freeze_mode == CTDB_FREEZE_PENDING) {
+		tevent_loop_once(ev);
+	}
+
+	if (ctdb_db->freeze_mode != CTDB_FREEZE_FROZEN) {
+		return -1;
+	}
+
+	return 0;
+}
+
+/*
+  block until we are frozen, used during daemon startup
+ */
+bool ctdb_blocking_freeze(struct ctdb_context *ctdb)
+{
+	int ret;
+
+	ret = ctdb_db_iterator(ctdb, db_freeze_block, ctdb->ev);
+	if (ret != 0) {
+		return false;
+	}
+
+	return true;
+}
+
+/*
+  thaw the databases
+ */
+int32_t ctdb_control_thaw(struct ctdb_context *ctdb, bool check_recmode)
+{
+	if (check_recmode && ctdb->recovery_mode == CTDB_RECOVERY_ACTIVE) {
+		DEBUG(DEBUG_ERR, ("Failing to thaw databases while "
+				  "recovery is active\n"));
+		return -1;
+	}
+
+	DEBUG(DEBUG_ERR,("Thawing all\n"));
+
+	/* cancel any pending transactions */
+	if (ctdb->freeze_transaction_started) {
+		ctdb_db_iterator(ctdb, db_transaction_cancel_handler, NULL);
+		ctdb->freeze_transaction_started = false;
+	}
+
+	ctdb_db_iterator(ctdb, db_thaw, NULL);
+	TALLOC_FREE(ctdb->freeze_handle);
+
+	ctdb_call_resend_all(ctdb);
+	return 0;
+}
+
+/**
+ * Database transaction wrappers
+ *
+ * These functions are wrappers around transaction start/cancel/commit handlers.
+ */
+
+struct db_start_transaction_state {
+	uint32_t transaction_id;
+	bool transaction_started;
+};
+
+static int db_start_transaction(struct ctdb_db_context *ctdb_db,
+				void *private_data)
+{
+	struct db_start_transaction_state *state =
+		(struct db_start_transaction_state *)private_data;
+	int ret;
+	bool transaction_started;
+
+	if (ctdb_db->freeze_mode != CTDB_FREEZE_FROZEN) {
+		DEBUG(DEBUG_ERR,
+		      ("Database %s not frozen, cannot start transaction\n",
+		       ctdb_db->db_name));
+		return -1;
+	}
+
+	transaction_started = state->transaction_started &
+			      ctdb_db->freeze_transaction_started;
+
+	ret = db_transaction_start_handler(ctdb_db,
+					   &transaction_started);
+	if (ret != 0) {
+		return -1;
+	}
+
+	ctdb_db->freeze_transaction_started = true;
+	ctdb_db->freeze_transaction_id = state->transaction_id;
+
+	return 0;
+}
+
+static int db_cancel_transaction(struct ctdb_db_context *ctdb_db,
+				 void *private_data)
+{
+	int ret;
+
+	ret = db_transaction_cancel_handler(ctdb_db, private_data);
+	if (ret != 0) {
+		return ret;
+	}
+
+	ctdb_db->freeze_transaction_started = false;
+
+	return 0;
+}
+
+struct db_commit_transaction_state {
+	uint32_t transaction_id;
+	unsigned int healthy_nodes;
+};
+
+static int db_commit_transaction(struct ctdb_db_context *ctdb_db,
+				 void *private_data)
+{
+	struct db_commit_transaction_state *state =
+		(struct db_commit_transaction_state *)private_data;
+	int ret;
+
+	if (ctdb_db->freeze_mode != CTDB_FREEZE_FROZEN) {
+		DEBUG(DEBUG_ERR,
+		      ("Database %s not frozen, cannot commit transaction\n",
+		       ctdb_db->db_name));
+		return -1;
+	}
+
+	if (!ctdb_db->freeze_transaction_started) {
+		DEBUG(DEBUG_ERR, ("Transaction not started on %s\n",
+				  ctdb_db->db_name));
+		return -1;
+	}
+
+	if (ctdb_db->freeze_transaction_id != state->transaction_id) {
+		DEBUG(DEBUG_ERR,
+		      ("Incorrect transaction commit id 0x%08x for %s\n",
+		       state->transaction_id, ctdb_db->db_name));
+		return -1;
+	}
+
+	ret = db_transaction_commit_handler(ctdb_db, &state->healthy_nodes);
+	if (ret != 0) {
+		return -1;
+	}
+
+	ctdb_db->freeze_transaction_started = false;
+	ctdb_db->freeze_transaction_id = 0;
+	ctdb_db->generation = state->transaction_id;
+	return 0;
+}
+
+/**
+ * Start a transaction on a database - used for db recovery
+ */
+int32_t ctdb_control_db_transaction_start(struct ctdb_context *ctdb,
+					  TDB_DATA indata)
+{
+	struct ctdb_transdb *w =
+		(struct ctdb_transdb *)indata.dptr;
+	struct ctdb_db_context *ctdb_db;
+	struct db_start_transaction_state state;
+
+	ctdb_db = find_ctdb_db(ctdb, w->db_id);
+	if (ctdb_db == NULL) {
+		DEBUG(DEBUG_ERR,
+		      ("Transaction start for unknown dbid 0x%08x\n",
+		       w->db_id));
+		return -1;
+	}
+
+	state.transaction_id = w->tid;
+	state.transaction_started = true;
+
+	return db_start_transaction(ctdb_db, &state);
+}
+
+/**
+ * Cancel a transaction on a database - used for db recovery
+ */
+int32_t ctdb_control_db_transaction_cancel(struct ctdb_context *ctdb,
+					   TDB_DATA indata)
+{
+	uint32_t db_id = *(uint32_t *)indata.dptr;
+	struct ctdb_db_context *ctdb_db;
+
+	ctdb_db = find_ctdb_db(ctdb, db_id);
+	if (ctdb_db == NULL) {
+		DEBUG(DEBUG_ERR,
+		      ("Transaction cancel for unknown dbid 0x%08x\n", db_id));
+		return -1;
+	}
+
+	DEBUG(DEBUG_ERR, ("Recovery db transaction cancelled for %s\n",
+			  ctdb_db->db_name));
+
+	return db_cancel_transaction(ctdb_db, NULL);
+}
+
+/**
+ * Commit a transaction on a database - used for db recovery
+ */
+int32_t ctdb_control_db_transaction_commit(struct ctdb_context *ctdb,
+					   TDB_DATA indata)
+{
+	struct ctdb_transdb *w =
+		(struct ctdb_transdb *)indata.dptr;
+	struct ctdb_db_context *ctdb_db;
+	struct db_commit_transaction_state state;
+	unsigned int healthy_nodes, i;
+
+	ctdb_db = find_ctdb_db(ctdb, w->db_id);
+	if (ctdb_db == NULL) {
+		DEBUG(DEBUG_ERR,
+		      ("Transaction commit for unknown dbid 0x%08x\n",
+		       w->db_id));
+		return -1;
+	}
+
+	healthy_nodes = 0;
+	for (i=0; i < ctdb->num_nodes; i++) {
+		if (ctdb->nodes[i]->flags == 0) {
+			healthy_nodes += 1;
+		}
+	}
+
+	state.transaction_id = w->tid;
+	state.healthy_nodes = healthy_nodes;
+
+	return db_commit_transaction(ctdb_db, &state);
+}
+
+/*
+  wipe a database - only possible when in a frozen transaction
+ */
+int32_t ctdb_control_wipe_database(struct ctdb_context *ctdb, TDB_DATA indata)
+{
+	struct ctdb_transdb w = *(struct ctdb_transdb *)indata.dptr;
+	struct ctdb_db_context *ctdb_db;
+
+	ctdb_db = find_ctdb_db(ctdb, w.db_id);
+	if (!ctdb_db) {
+		DEBUG(DEBUG_ERR,(__location__ " Unknown db 0x%x\n", w.db_id));
+		return -1;
+	}
+
+	if (ctdb_db->freeze_mode != CTDB_FREEZE_FROZEN) {
+		DEBUG(DEBUG_ERR,(__location__ " Failed transaction_start while not frozen\n"));
+		return -1;
+	}
+
+	if (!ctdb_db->freeze_transaction_started) {
+		DEBUG(DEBUG_ERR,(__location__ " transaction not started\n"));
+		return -1;
+	}
+
+	if (w.tid != ctdb_db->freeze_transaction_id) {
+		DEBUG(DEBUG_ERR,(__location__ " incorrect transaction id 0x%x in commit\n", w.tid));
+		return -1;
+	}
+
+	if (tdb_wipe_all(ctdb_db->ltdb->tdb) != 0) {
+		DEBUG(DEBUG_ERR,(__location__ " Failed to wipe database for db '%s'\n",
+			 ctdb_db->db_name));
+		return -1;
+	}
+
+	if (ctdb_db_volatile(ctdb_db)) {
+		talloc_free(ctdb_db->delete_queue);
+		talloc_free(ctdb_db->fetch_queue);
+		ctdb_db->delete_queue = trbt_create(ctdb_db, 0);
+		if (ctdb_db->delete_queue == NULL) {
+			DEBUG(DEBUG_ERR, (__location__ " Failed to re-create "
+					  "the delete queue.\n"));
+			return -1;
+		}
+		ctdb_db->fetch_queue = trbt_create(ctdb_db, 0);
+		if (ctdb_db->fetch_queue == NULL) {
+			DEBUG(DEBUG_ERR, (__location__ " Failed to re-create "
+					  "the fetch queue.\n"));
+			return -1;
+		}
+	}
+
+	return 0;
+}
+
+bool ctdb_db_frozen(struct ctdb_db_context *ctdb_db)
+{
+	if (ctdb_db->freeze_mode != CTDB_FREEZE_FROZEN) {
+		return false;
+	}
+
+	return true;
+}
+
+bool ctdb_db_all_frozen(struct ctdb_context *ctdb)
+{
+	if (ctdb->freeze_mode != CTDB_FREEZE_FROZEN) {
+		return false;
+	}
+	return true;
+}
+
+bool ctdb_db_allow_access(struct ctdb_db_context *ctdb_db)
+{
+	if (ctdb_db->freeze_mode == CTDB_FREEZE_NONE) {
+		/* If database is not frozen, then allow access. */
+		return true;
+	} else if (ctdb_db->freeze_transaction_started) {
+		/* If database is frozen, allow access only if the
+		 * transaction is started.  This is required during
+		 * recovery.
+		 *
+		 * If a node is inactive, then transaction is not started.
+		 */
+		return true;
+	}
+
+	return false;
+}