ctdb_recover.c

samba最新软件

	}	    

	DEBUG(3,("finished push of %u records for dbid 0x%x\n",
		 reply->count, reply->db_id));

	ctdb_lock_all_databases_unmark(ctdb);
	return 0;

failed:
	ctdb_lock_all_databases_unmark(ctdb);
	return -1;
}


static int traverse_setdmaster(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *p)
{
	uint32_t *dmaster = (uint32_t *)p;
	struct ctdb_ltdb_header *header = (struct ctdb_ltdb_header *)data.dptr;
	int ret;

	header->dmaster = *dmaster;

	ret = tdb_store(tdb, key, data, TDB_REPLACE);
	if (ret) {
		DEBUG(0,(__location__ " failed to write tdb data back  ret:%d\n",ret));
		return ret;
	}
	return 0;
}

int32_t ctdb_control_set_dmaster(struct ctdb_context *ctdb, TDB_DATA indata)
{
	struct ctdb_control_set_dmaster *p = (struct ctdb_control_set_dmaster *)indata.dptr;
	struct ctdb_db_context *ctdb_db;

	if (ctdb->freeze_mode != CTDB_FREEZE_FROZEN) {
		DEBUG(0,("rejecting ctdb_control_set_dmaster when not frozen\n"));
		return -1;
	}

	ctdb_db = find_ctdb_db(ctdb, p->db_id);
	if (!ctdb_db) {
		DEBUG(0,(__location__ " Unknown db 0x%08x\n", p->db_id));
		return -1;
	}

	if (ctdb_lock_all_databases_mark(ctdb) != 0) {
		DEBUG(0,(__location__ " Failed to get lock on entired db - failing\n"));
		return -1;
	}

	tdb_traverse(ctdb_db->ltdb->tdb, traverse_setdmaster, &p->dmaster);

	ctdb_lock_all_databases_unmark(ctdb);
	
	return 0;
}

struct ctdb_set_recmode_state {
	struct ctdb_req_control *c;
	uint32_t recmode;
};

/*
  called when the 'recovered' event script has finished
 */
static void ctdb_recovered_callback(struct ctdb_context *ctdb, int status, void *p)
{
	struct ctdb_set_recmode_state *state = talloc_get_type(p, struct ctdb_set_recmode_state);

	ctdb_start_monitoring(ctdb);

	if (status == 0) {
		ctdb->recovery_mode = state->recmode;
	} else {
		DEBUG(0,(__location__ " recovered event script failed (status %d)\n", status));
	}

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

/*
  set the recovery mode
 */
int32_t ctdb_control_set_recmode(struct ctdb_context *ctdb, 
				 struct ctdb_req_control *c,
				 TDB_DATA indata, bool *async_reply,
				 const char **errormsg)
{
	uint32_t recmode = *(uint32_t *)indata.dptr;
	int ret;
	struct ctdb_set_recmode_state *state;

	if (ctdb->freeze_mode != CTDB_FREEZE_FROZEN) {
		DEBUG(0,("Attempt to change recovery mode to %u when not frozen\n", 
			 recmode));
		(*errormsg) = "Cannot change recovery mode while not frozen";
		return -1;
	}

	if (recmode != CTDB_RECOVERY_NORMAL ||
	    ctdb->recovery_mode != CTDB_RECOVERY_ACTIVE) {
		ctdb->recovery_mode = recmode;
		return 0;
	}

	/* some special handling when ending recovery mode */
	state = talloc(ctdb, struct ctdb_set_recmode_state);
	CTDB_NO_MEMORY(ctdb, state);

	/* we should not be able to get the lock on the nodes list, as it should be
	   held by the recovery master */
	if (ctdb_recovery_lock(ctdb, false)) {
		DEBUG(0,("ERROR: recovery lock file %s not locked when recovering!\n",
			 ctdb->recovery_lock_file));
		return -1;
	}	

	state->c = talloc_steal(state, c);
	state->recmode = recmode;
	
	ctdb_stop_monitoring(ctdb);

	/* call the events script to tell all subsystems that we have recovered */
	ret = ctdb_event_script_callback(ctdb, 
					 timeval_current_ofs(ctdb->tunable.script_timeout, 0),
					 state, 
					 ctdb_recovered_callback, 
					 state, "recovered");
	if (ret != 0) {
		return ret;
	}
	*async_reply = true;

	return 0;
}

/*
  callback for ctdb_control_max_rsn
 */
static int traverse_max_rsn(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *p)
{
	struct ctdb_ltdb_header *h = (struct ctdb_ltdb_header *)data.dptr;
	uint64_t *max_rsn = (uint64_t *)p;

	if (data.dsize >= sizeof(*h)) {
		(*max_rsn) = MAX(*max_rsn, h->rsn);
	}
	return 0;
}

/*
  get max rsn across an entire db
 */
int32_t ctdb_control_max_rsn(struct ctdb_context *ctdb, TDB_DATA indata, TDB_DATA *outdata)
{
	struct ctdb_db_context *ctdb_db;
	uint32_t db_id = *(uint32_t *)indata.dptr;
	uint64_t max_rsn = 0;
	int ret;

	if (ctdb->freeze_mode != CTDB_FREEZE_FROZEN) {
		DEBUG(0,("rejecting ctdb_control_max_rsn when not frozen\n"));
		return -1;
	}

	ctdb_db = find_ctdb_db(ctdb, db_id);
	if (!ctdb_db) {
		DEBUG(0,(__location__ " Unknown db\n"));
		return -1;
	}

	if (ctdb_lock_all_databases_mark(ctdb) != 0) {
		DEBUG(0,(__location__ " Failed to get lock on entired db - failing\n"));
		return -1;
	}

	ret = tdb_traverse_read(ctdb_db->ltdb->tdb, traverse_max_rsn, &max_rsn);
	if (ret < 0) {
		DEBUG(0,(__location__ " traverse failed in ctdb_control_max_rsn\n"));
		return -1;
	}

	ctdb_lock_all_databases_unmark(ctdb);

	outdata->dptr = (uint8_t *)talloc(outdata, uint64_t);
	if (!outdata->dptr) {
		return -1;
	}
	(*(uint64_t *)outdata->dptr) = max_rsn;
	outdata->dsize = sizeof(uint64_t);

	return 0;
}


/*
  callback for ctdb_control_set_rsn_nonempty
 */
static int traverse_set_rsn_nonempty(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *p)
{
	struct ctdb_ltdb_header *h = (struct ctdb_ltdb_header *)data.dptr;
	uint64_t *rsn = (uint64_t *)p;

	if (data.dsize > sizeof(*h)) {
		h->rsn = *rsn;
		if (tdb_store(tdb, key, data, TDB_REPLACE) != 0) {
			return -1;
		}
	}
	return 0;
}

/*
  set rsn for all non-empty records in a database to a given rsn
 */
int32_t ctdb_control_set_rsn_nonempty(struct ctdb_context *ctdb, TDB_DATA indata, TDB_DATA *outdata)
{
	struct ctdb_control_set_rsn_nonempty *p = (struct ctdb_control_set_rsn_nonempty *)indata.dptr;
	struct ctdb_db_context *ctdb_db;
	int ret;

	if (ctdb->freeze_mode != CTDB_FREEZE_FROZEN) {
		DEBUG(0,("rejecting ctdb_control_set_rsn_nonempty when not frozen\n"));
		return -1;
	}

	ctdb_db = find_ctdb_db(ctdb, p->db_id);
	if (!ctdb_db) {
		DEBUG(0,(__location__ " Unknown db\n"));
		return -1;
	}

	if (ctdb_lock_all_databases_mark(ctdb) != 0) {
		DEBUG(0,(__location__ " Failed to get lock on entired db - failing\n"));
		return -1;
	}

	ret = tdb_traverse(ctdb_db->ltdb->tdb, traverse_set_rsn_nonempty, &p->rsn);
	if (ret < 0) {
		DEBUG(0,(__location__ " traverse failed in ctdb_control_set_rsn_nonempty\n"));
		return -1;
	}

	ctdb_lock_all_databases_unmark(ctdb);

	return 0;
}


/*
  callback for ctdb_control_delete_low_rsn
 */
static int traverse_delete_low_rsn(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data, void *p)
{
	struct ctdb_ltdb_header *h = (struct ctdb_ltdb_header *)data.dptr;
	uint64_t *rsn = (uint64_t *)p;

	if (data.dsize < sizeof(*h) || h->rsn < *rsn) {
		if (tdb_delete(tdb, key) != 0) {
			return -1;
		}
	}
	return 0;
}

/*
  delete any records with a rsn < the given rsn
 */
int32_t ctdb_control_delete_low_rsn(struct ctdb_context *ctdb, TDB_DATA indata, TDB_DATA *outdata)
{
	struct ctdb_control_delete_low_rsn *p = (struct ctdb_control_delete_low_rsn *)indata.dptr;
	struct ctdb_db_context *ctdb_db;
	int ret;

	if (ctdb->freeze_mode != CTDB_FREEZE_FROZEN) {
		DEBUG(0,("rejecting ctdb_control_delete_low_rsn when not frozen\n"));
		return -1;
	}

	ctdb_db = find_ctdb_db(ctdb, p->db_id);
	if (!ctdb_db) {
		DEBUG(0,(__location__ " Unknown db\n"));
		return -1;
	}

	if (ctdb_lock_all_databases_mark(ctdb) != 0) {
		DEBUG(0,(__location__ " Failed to get lock on entired db - failing\n"));
		return -1;
	}

	ret = tdb_traverse(ctdb_db->ltdb->tdb, traverse_delete_low_rsn, &p->rsn);
	if (ret < 0) {
		DEBUG(0,(__location__ " traverse failed in ctdb_control_delete_low_rsn\n"));
		return -1;
	}

	ctdb_lock_all_databases_unmark(ctdb);

	return 0;
}


/*
  try and get the recovery lock in shared storage - should only work
  on the recovery master recovery daemon. Anywhere else is a bug
 */
bool ctdb_recovery_lock(struct ctdb_context *ctdb, bool keep)
{
	struct flock lock;

	if (ctdb->recovery_lock_fd != -1) {
		close(ctdb->recovery_lock_fd);
	}
	ctdb->recovery_lock_fd = open(ctdb->recovery_lock_file, O_RDWR|O_CREAT, 0600);
	if (ctdb->recovery_lock_fd == -1) {
		DEBUG(0,("Unable to open %s - (%s)\n", 
			 ctdb->recovery_lock_file, strerror(errno)));
		return false;
	}

	lock.l_type = F_WRLCK;
	lock.l_whence = SEEK_SET;
	lock.l_start = 0;
	lock.l_len = 1;
	lock.l_pid = 0;

	if (fcntl(ctdb->recovery_lock_fd, F_SETLK, &lock) != 0) {
		return false;
	}

	if (!keep) {
		close(ctdb->recovery_lock_fd);
		ctdb->recovery_lock_fd = -1;
	}

	return true;
}