samba-mirror/ctdb/server/ctdb_persistent.c

/* 
   persistent store logic

   Copyright (C) Andrew Tridgell  2007
   Copyright (C) Ronnie Sahlberg  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/filesys.h"
#include "system/network.h"
#include "system/time.h"
#include "system/wait.h"

#include <talloc.h>
#include <tevent.h>

#include "lib/tdb_wrap/tdb_wrap.h"
#include "lib/util/debug.h"
#include "lib/util/samba_util.h"

#include "ctdb_private.h"

#include "common/reqid.h"
#include "common/common.h"
#include "common/logging.h"

struct ctdb_persistent_state {
	struct ctdb_context *ctdb;
	struct ctdb_db_context *ctdb_db; /* used by trans3_commit */
	struct ctdb_client *client; /* used by trans3_commit */
	struct ctdb_req_control_old *c;
	const char *errormsg;
	uint32_t num_pending;
	int32_t status;
	uint32_t num_failed, num_sent;
};

/*
  1) all nodes fail, and all nodes reply
  2) some nodes fail, all nodes reply
  3) some nodes timeout
  4) all nodes succeed
 */

/*
  called when a node has acknowledged a ctdb_control_update_record call
 */
static void ctdb_persistent_callback(struct ctdb_context *ctdb,
				     int32_t status, TDB_DATA data, 
				     const char *errormsg,
				     void *private_data)
{
	struct ctdb_persistent_state *state = talloc_get_type(private_data, 
							      struct ctdb_persistent_state);

	if (ctdb->recovery_mode != CTDB_RECOVERY_NORMAL) {
		DEBUG(DEBUG_INFO, ("ctdb_persistent_callback: ignoring reply "
				   "during recovery\n"));
		return;
	}

	if (status != 0) {
		DEBUG(DEBUG_ERR,("ctdb_persistent_callback failed with status %d (%s)\n",
			 status, errormsg?errormsg:"no error message given"));
		state->status = status;
		state->errormsg = errormsg;
		state->num_failed++;

		/*
		 * If a node failed to complete the update_record control,
		 * then either a recovery is already running or something
		 * bad is going on. So trigger a recovery and let the
		 * recovery finish the transaction, sending back the reply
		 * for the trans3_commit control to the client.
		 */
		ctdb->recovery_mode = CTDB_RECOVERY_ACTIVE;
		return;
	}

	state->num_pending--;

	if (state->num_pending != 0) {
		return;
	}

	ctdb_request_control_reply(state->ctdb, state->c, NULL, 0, state->errormsg);
	talloc_free(state);
}

/*
  called if persistent store times out
 */
static void ctdb_persistent_store_timeout(struct tevent_context *ev,
					  struct tevent_timer *te,
					  struct timeval t, void *private_data)
{
	struct ctdb_persistent_state *state = talloc_get_type(private_data, struct ctdb_persistent_state);

	if (state->ctdb->recovery_mode != CTDB_RECOVERY_NORMAL) {
		DEBUG(DEBUG_INFO, ("ctdb_persistent_store_timeout: ignoring "
				   "timeout during recovery\n"));
		return;
	}

	ctdb_request_control_reply(state->ctdb, state->c, NULL, 1,
				   "timeout in ctdb_persistent_state");

	talloc_free(state);
}

/**
 * Finish pending trans3 commit controls, i.e. send
 * reply to the client. This is called by the end-recovery
 * control to fix the situation when a recovery interrupts
 * the usual progress of a transaction.
 */
void ctdb_persistent_finish_trans3_commits(struct ctdb_context *ctdb)
{
	struct ctdb_db_context *ctdb_db;

	if (ctdb->recovery_mode != CTDB_RECOVERY_NORMAL) {
		DEBUG(DEBUG_INFO, ("ctdb_persistent_finish_trans3_commits: "
				   "skipping execution when recovery is "
				   "active\n"));
		return;
	}

	for (ctdb_db = ctdb->db_list; ctdb_db; ctdb_db = ctdb_db->next) {
		struct ctdb_persistent_state *state;

		if (ctdb_db->persistent_state == NULL) {
			continue;
		}

		state = ctdb_db->persistent_state;

		ctdb_request_control_reply(ctdb, state->c, NULL, 2,
					   "trans3 commit ended by recovery");

		/* The destructor sets ctdb_db->persistent_state to NULL. */
		talloc_free(state);
	}
}

static int ctdb_persistent_state_destructor(struct ctdb_persistent_state *state)
{
	if (state->client != NULL) {
		state->client->db_id = 0;
	}

	if (state->ctdb_db != NULL) {
		state->ctdb_db->persistent_state = NULL;
	}

	return 0;
}

/*
 * Store a set of persistent records.
 * This is used to roll out a transaction to all nodes.
 */
int32_t ctdb_control_trans3_commit(struct ctdb_context *ctdb,
				   struct ctdb_req_control_old *c,
				   TDB_DATA recdata, bool *async_reply)
{
	struct ctdb_client *client;
	struct ctdb_persistent_state *state;
	int i;
	struct ctdb_marshall_buffer *m = (struct ctdb_marshall_buffer *)recdata.dptr;
	struct ctdb_db_context *ctdb_db;

	if (ctdb->recovery_mode != CTDB_RECOVERY_NORMAL) {
		DEBUG(DEBUG_INFO,("rejecting ctdb_control_trans3_commit when recovery active\n"));
		return -1;
	}

	client = reqid_find(ctdb->idr, c->client_id, struct ctdb_client);
	if (client == NULL) {
		DEBUG(DEBUG_ERR,(__location__ " can not match persistent_store "
				 "to a client. Returning error\n"));
		return -1;
	}

	if (client->db_id != 0) {
		DEBUG(DEBUG_ERR,(__location__ " ERROR: trans3_commit: "
				 "client-db_id[0x%08x] != 0 "
				 "(client_id[0x%08x]): trans3_commit active?\n",
				 client->db_id, client->client_id));
		return -1;
	}

	ctdb_db = find_ctdb_db(ctdb, m->db_id);
	if (ctdb_db == NULL) {
		DEBUG(DEBUG_ERR,(__location__ " ctdb_control_trans3_commit: "
				 "Unknown database db_id[0x%08x]\n", m->db_id));
		return -1;
	}

	if (ctdb_db->persistent_state != NULL) {
		DEBUG(DEBUG_ERR, (__location__ " Error: "
				  "ctdb_control_trans3_commit "
				  "called while a transaction commit is "
				  "active. db_id[0x%08x]\n", m->db_id));
		return -1;
	}

	ctdb_db->persistent_state = talloc_zero(ctdb_db,
						struct ctdb_persistent_state);
	CTDB_NO_MEMORY(ctdb, ctdb_db->persistent_state);

	client->db_id = m->db_id;

	state = ctdb_db->persistent_state;
	state->ctdb = ctdb;
	state->ctdb_db = ctdb_db;
	state->c    = c;
	state->client = client;

	talloc_set_destructor(state, ctdb_persistent_state_destructor);

	for (i = 0; i < ctdb->vnn_map->size; i++) {
		struct ctdb_node *node = ctdb->nodes[ctdb->vnn_map->map[i]];
		int ret;

		/* only send to active nodes */
		if (node->flags & NODE_FLAGS_INACTIVE) {
			continue;
		}

		ret = ctdb_daemon_send_control(ctdb, node->pnn, 0,
					       CTDB_CONTROL_UPDATE_RECORD,
					       c->client_id, 0, recdata,
					       ctdb_persistent_callback,
					       state);
		if (ret == -1) {
			DEBUG(DEBUG_ERR,("Unable to send "
					 "CTDB_CONTROL_UPDATE_RECORD "
					 "to pnn %u\n", node->pnn));
			talloc_free(state);
			return -1;
		}

		state->num_pending++;
		state->num_sent++;
	}

	if (state->num_pending == 0) {
		talloc_free(state);
		return 0;
	}

	/* we need to wait for the replies */
	*async_reply = true;

	/* need to keep the control structure around */
	talloc_steal(state, c);

	/* but we won't wait forever */
	tevent_add_timer(ctdb->ev, state,
			 timeval_current_ofs(ctdb->tunable.control_timeout, 0),
			 ctdb_persistent_store_timeout, state);

	return 0;
}


/*
  backwards compatibility:

  start a persistent store operation. passing both the key, header and
  data to the daemon. If the client disconnects before it has issued
  a persistent_update call to the daemon we trigger a full recovery
  to ensure the databases are brought back in sync.
  for now we ignore the recdata that the client has passed to us.
 */
int32_t ctdb_control_start_persistent_update(struct ctdb_context *ctdb, 
				      struct ctdb_req_control_old *c,
				      TDB_DATA recdata)
{
	struct ctdb_client *client = reqid_find(ctdb->idr, c->client_id, struct ctdb_client);

	if (client == NULL) {
		DEBUG(DEBUG_ERR,(__location__ " can not match start_persistent_update to a client. Returning error\n"));
		return -1;
	}

	client->num_persistent_updates++;

	return 0;
}

/* 
  backwards compatibility:

  called to tell ctdbd that it is no longer doing a persistent update 
*/
int32_t ctdb_control_cancel_persistent_update(struct ctdb_context *ctdb, 
					      struct ctdb_req_control_old *c,
					      TDB_DATA recdata)
{
	struct ctdb_client *client = reqid_find(ctdb->idr, c->client_id, struct ctdb_client);

	if (client == NULL) {
		DEBUG(DEBUG_ERR,(__location__ " can not match cancel_persistent_update to a client. Returning error\n"));
		return -1;
	}

	if (client->num_persistent_updates > 0) {
		client->num_persistent_updates--;
	}

	return 0;
}

static int32_t ctdb_get_db_seqnum(struct ctdb_context *ctdb,
				  uint32_t db_id,
				  uint64_t *seqnum)
{
	int32_t ret;
	struct ctdb_db_context *ctdb_db;
	const char *keyname = CTDB_DB_SEQNUM_KEY;
	TDB_DATA key;
	TDB_DATA data;
	TALLOC_CTX *mem_ctx = talloc_new(ctdb);
	struct ctdb_ltdb_header header;

	ctdb_db = find_ctdb_db(ctdb, db_id);
	if (!ctdb_db) {
		DEBUG(DEBUG_ERR,(__location__ " Unknown db 0x%08x\n", db_id));
		ret = -1;
		goto done;
	}

	key.dptr = (uint8_t *)discard_const(keyname);
	key.dsize = strlen(keyname) + 1;

	ret = (int32_t)ctdb_ltdb_fetch(ctdb_db, key, &header, mem_ctx, &data);
	if (ret != 0) {
		goto done;
	}

	if (data.dsize != sizeof(uint64_t)) {
		*seqnum = 0;
		goto done;
	}

	*seqnum = *(uint64_t *)data.dptr;

done:
	talloc_free(mem_ctx);
	return ret;
}

/**
 * Get the sequence number of a persistent database.
 */
int32_t ctdb_control_get_db_seqnum(struct ctdb_context *ctdb,
				   TDB_DATA indata,
				   TDB_DATA *outdata)
{
	uint32_t db_id;
	int32_t ret;
	uint64_t seqnum;

	db_id = *(uint32_t *)indata.dptr;
	ret = ctdb_get_db_seqnum(ctdb, db_id, &seqnum);
	if (ret != 0) {
		goto done;
	}

	outdata->dsize = sizeof(uint64_t);
	outdata->dptr = talloc_memdup(outdata, &seqnum, sizeof(uint64_t));
	if (outdata->dptr == NULL) {
		ret = -1;
	}

done:
	return ret;
}