rep_method.c

这是linux下运行的mysql软件包,可用于linux 下安装 php + mysql + apach 的网络配置

	    ret == 0 && log_compare(&lsn, &ckp_lsn) > 0;
	    ret = logc->get(logc, &lsn, &rec, DB_PREV)) {
		memcpy(&rectype, rec.data, sizeof(rectype));
		switch (rectype) {
		case DB___txn_regop:
			/*
			 * It's a commit or abort--but we don't care
			 * which!  Just add it to the list of txns
			 * that are resolved.
			 */
			if ((ret = __txn_regop_read(dbenv, rec.data,
			    &regop_args)) != 0)
				goto err;

			ret = __db_txnlist_find(dbenv,
			    txninfo, regop_args->txnid->txnid);
			if (ret == DB_NOTFOUND)
				ret = __db_txnlist_add(dbenv, txninfo,
				    regop_args->txnid->txnid,
				    regop_args->opcode, &lsn);
			__os_free(dbenv, regop_args);
			break;
		case DB___txn_xa_regop:
			/*
			 * It's a prepare.  If we haven't put the
			 * txn on our list yet, it hasn't been
			 * resolved, so apply and restore it.
			 */
			if ((ret = __txn_xa_regop_read(dbenv, rec.data,
			    &prep_args)) != 0)
				goto err;
			ret = __db_txnlist_find(dbenv, txninfo,
			    prep_args->txnid->txnid);
			if (ret == DB_NOTFOUND)
				if ((ret = __rep_process_txn(dbenv, &rec)) == 0)
					ret = __txn_restore_txn(dbenv,
					    &lsn, prep_args);
			__os_free(dbenv, prep_args);
			break;
		default:
			continue;
		}
	}

	/* It's not an error to have hit the beginning of the log. */
	if (ret == DB_NOTFOUND)
		ret = 0;

done:
err:	t_ret = logc->close(logc, 0);

	if (txninfo != NULL)
		__db_txnlist_end(dbenv, txninfo);

	return (ret == 0 ? t_ret : ret);
}

/*
 * __rep_set_limit --
 *	Set a limit on the amount of data that will be sent during a single
 * invocation of __rep_process_message.
 */
static int
__rep_set_limit(dbenv, gbytes, bytes)
	DB_ENV *dbenv;
	u_int32_t gbytes;
	u_int32_t bytes;
{
	DB_REP *db_rep;
	REP *rep;

	PANIC_CHECK(dbenv);

	if ((db_rep = dbenv->rep_handle) == NULL) {
		__db_err(dbenv,
    "DB_ENV->set_rep_limit: database environment not properly initialized");
		return (__db_panic(dbenv, EINVAL));
	}
	rep = db_rep->region;
	MUTEX_LOCK(dbenv, db_rep->mutexp);
	if (bytes > GIGABYTE) {
		gbytes += bytes / GIGABYTE;
		bytes = bytes % GIGABYTE;
	}
	rep->gbytes = gbytes;
	rep->bytes = bytes;
	MUTEX_UNLOCK(dbenv, db_rep->mutexp);

	return (0);
}

/*
 * __rep_set_request --
 *	Set the minimum and maximum number of log records that we wait
 * before retransmitting.
 * UNDOCUMENTED.
 */
static int
__rep_set_request(dbenv, min, max)
	DB_ENV *dbenv;
	u_int32_t min;
	u_int32_t max;
{
	LOG *lp;
	DB_LOG *dblp;
	DB_REP *db_rep;
	REP *rep;

	PANIC_CHECK(dbenv);

	if ((db_rep = dbenv->rep_handle) == NULL) {
		__db_err(dbenv,
    "DB_ENV->set_rep_request: database environment not properly initialized");
		return (__db_panic(dbenv, EINVAL));
	}
	rep = db_rep->region;
	MUTEX_LOCK(dbenv, db_rep->mutexp);
	rep->request_gap = min;
	rep->max_gap = max;
	MUTEX_UNLOCK(dbenv, db_rep->mutexp);
	dblp = dbenv->lg_handle;
	if (dblp != NULL && (lp = dblp->reginfo.primary) != NULL) {
		R_LOCK(dbenv, &dblp->reginfo);
		lp->wait_recs = 0;
		lp->rcvd_recs = 0;
		R_UNLOCK(dbenv, &dblp->reginfo);
	}

	return (0);
}

/*
 * __rep_set_transport --
 *	Set the transport function for replication.
 */
static int
__rep_set_rep_transport(dbenv, eid, f_send)
	DB_ENV *dbenv;
	int eid;
	int (*f_send) __P((DB_ENV *, const DBT *, const DBT *, int, u_int32_t));
{
	DB_REP *db_rep;

	PANIC_CHECK(dbenv);

	if ((db_rep = dbenv->rep_handle) == NULL) {
		__db_err(dbenv,
    "DB_ENV->set_rep_transport: database environment not properly initialized");
		return (__db_panic(dbenv, EINVAL));
	}

	if (f_send == NULL) {
		__db_err(dbenv,
	"DB_ENV->set_rep_transport: no send function specified");
		return (EINVAL);
	}

	if (eid < 0) {
		__db_err(dbenv,
	"DB_ENV->set_rep_transport: eid must be greater than or equal to 0");
		return (EINVAL);
	}

	db_rep->rep_send = f_send;

	dbenv->rep_eid = eid;
	return (0);
}

/*
 * __rep_elect --
 *	Called after master failure to hold/participate in an election for
 *	a new master.
 */
static int
__rep_elect(dbenv, nsites, priority, timeout, eidp)
	DB_ENV *dbenv;
	int nsites, priority;
	u_int32_t timeout;
	int *eidp;
{
	DB_LOG *dblp;
	DB_LSN lsn;
	DB_REP *db_rep;
	REP *rep;
	int in_progress, ret, send_vote, tiebreaker;
	u_int32_t pid, sec, usec;

	PANIC_CHECK(dbenv);
	ENV_REQUIRES_CONFIG(dbenv, dbenv->tx_handle, "rep_elect", DB_INIT_TXN);

	/* Error checking. */
	if (nsites <= 0) {
		__db_err(dbenv,
		    "DB_ENV->rep_elect: nsites must be greater than 0");
		return (EINVAL);
	}
	if (priority < 0) {
		__db_err(dbenv,
		    "DB_ENV->rep_elect: priority may not be negative");
		return (EINVAL);
	}

	db_rep = dbenv->rep_handle;
	rep = db_rep->region;
	dblp = dbenv->lg_handle;

	R_LOCK(dbenv, &dblp->reginfo);
	lsn = ((LOG *)dblp->reginfo.primary)->lsn;
	R_UNLOCK(dbenv, &dblp->reginfo);

	/* Generate a randomized tiebreaker value. */
	__os_id(&pid);
	if ((ret = __os_clock(dbenv, &sec, &usec)) != 0)
		return (ret);
	tiebreaker = pid ^ sec ^ usec ^ (u_int)rand() ^ P_TO_UINT32(&pid);

	if ((ret = __rep_elect_init(dbenv,
	    &lsn, nsites, priority, tiebreaker, &in_progress)) != 0) {
		if (ret == DB_REP_NEWMASTER) {
			ret = 0;
			*eidp = dbenv->rep_eid;
		}
		return (ret);
	}

	if (!in_progress) {
#ifdef DIAGNOSTIC
		if (FLD_ISSET(dbenv->verbose, DB_VERB_REPLICATION))
			__db_err(dbenv, "Beginning an election");
#endif
		if ((ret = __rep_send_message(dbenv,
		    DB_EID_BROADCAST, REP_ELECT, NULL, NULL, 0)) != 0)
			goto err;
		DB_ENV_TEST_RECOVERY(dbenv, DB_TEST_ELECTSEND, ret, NULL);
	}

	/* Now send vote */
	if ((ret =
	    __rep_send_vote(dbenv, &lsn, nsites, priority, tiebreaker)) != 0)
		goto err;
	DB_ENV_TEST_RECOVERY(dbenv, DB_TEST_ELECTVOTE1, ret, NULL);

	ret = __rep_wait(dbenv, timeout, eidp, REP_F_EPHASE1);
	DB_ENV_TEST_RECOVERY(dbenv, DB_TEST_ELECTWAIT1, ret, NULL);
	switch (ret) {
		case 0:
			/* Check if election complete or phase complete. */
			if (*eidp != DB_EID_INVALID)
				return (0);
			goto phase2;
		case DB_TIMEOUT:
			break;
		default:
			goto err;
	}
	/*
	 * If we got here, we haven't heard from everyone, but we've
	 * run out of time, so it's time to decide if we have enough
	 * votes to pick a winner and if so, to send out a vote to
	 * the winner.
	 */
	MUTEX_LOCK(dbenv, db_rep->mutexp);
	send_vote = DB_EID_INVALID;
	if (rep->sites > rep->nsites / 2) {
		/* We think we've seen enough to cast a vote. */
		send_vote = rep->winner;
		if (rep->winner == rep->eid)
			rep->votes++;
		F_CLR(rep, REP_F_EPHASE1);
		F_SET(rep, REP_F_EPHASE2);
	}
	MUTEX_UNLOCK(dbenv, db_rep->mutexp);
	if (send_vote == DB_EID_INVALID) {
		/* We do not have enough votes to elect. */
#ifdef DIAGNOSTIC
		if (FLD_ISSET(dbenv->verbose, DB_VERB_REPLICATION))
			__db_err(dbenv,
			    "Not enough votes to elect: received %d of %d",
			    rep->sites, rep->nsites);
#endif
		ret = DB_REP_UNAVAIL;
		goto err;

	}
#ifdef DIAGNOSTIC
	if (FLD_ISSET(dbenv->verbose, DB_VERB_REPLICATION) &&
	    send_vote != rep->eid)
		__db_err(dbenv, "Sending vote");
#endif

	if (send_vote != rep->eid && (ret = __rep_send_message(dbenv,
	    send_vote, REP_VOTE2, NULL, NULL, 0)) != 0)
		goto err;
	DB_ENV_TEST_RECOVERY(dbenv, DB_TEST_ELECTVOTE2, ret, NULL);

phase2:	ret = __rep_wait(dbenv, timeout, eidp, REP_F_EPHASE2);
	DB_ENV_TEST_RECOVERY(dbenv, DB_TEST_ELECTWAIT2, ret, NULL);
	switch (ret) {
		case 0:
			return (0);
		case DB_TIMEOUT:
			ret = DB_REP_UNAVAIL;
			break;
		default:
			goto err;
	}

DB_TEST_RECOVERY_LABEL
err:	MUTEX_LOCK(dbenv, db_rep->mutexp);
	ELECTION_DONE(rep);
	MUTEX_UNLOCK(dbenv, db_rep->mutexp);

#ifdef DIAGNOSTIC
	if (FLD_ISSET(dbenv->verbose, DB_VERB_REPLICATION))
		__db_err(dbenv, "Ended election with %d", ret);
#endif
	return (ret);
}

/*
 * __rep_elect_init
 *	Initialize an election.  Sets beginp non-zero if the election is
 * already in progress; makes it 0 otherwise.
 */
static int
__rep_elect_init(dbenv, lsnp, nsites, priority, tiebreaker, beginp)
	DB_ENV *dbenv;
	DB_LSN *lsnp;
	int nsites, priority, tiebreaker, *beginp;
{
	DB_REP *db_rep;
	REP *rep;
	int ret, *tally;

	db_rep = dbenv->rep_handle;
	rep = db_rep->region;

	ret = 0;

	/* We may miscount, as we don't hold the replication mutex here. */
	rep->stat.st_elections++;

	/* If we are already a master; simply broadcast that fact and return. */
	if (F_ISSET(dbenv, DB_ENV_REP_MASTER)) {
		(void)__rep_send_message(dbenv,
		    DB_EID_BROADCAST, REP_NEWMASTER, lsnp, NULL, 0);
		rep->stat.st_elections_won++;
		return (DB_REP_NEWMASTER);
	}

	MUTEX_LOCK(dbenv, db_rep->mutexp);
	*beginp = IN_ELECTION(rep);
	if (!*beginp) {
		/*
		 * Make sure that we always initialize all the election fields
		 * before putting ourselves in an election state.  That means
		 * issuing calls that can fail (allocation) before setting all
		 * the variables.
		 */
		if (nsites > rep->asites &&
		    (ret = __rep_grow_sites(dbenv, nsites)) != 0)
			goto err;
		DB_ENV_TEST_RECOVERY(dbenv, DB_TEST_ELECTINIT, ret, NULL);
		rep->nsites = nsites;
		rep->priority = priority;
		rep->votes = 0;
		rep->master_id = DB_EID_INVALID;
		F_SET(rep, REP_F_EPHASE1);

		/* We have always heard from ourselves. */
		rep->sites = 1;
		tally = R_ADDR((REGINFO *)dbenv->reginfo, rep->tally_off);
		tally[0] = rep->eid;

		if (priority != 0) {
			/* Make ourselves the winner to start. */
			rep->winner = rep->eid;
			rep->w_priority = priority;
			rep->w_gen = rep->gen;
			rep->w_lsn = *lsnp;
			rep->w_tiebreaker = tiebreaker;
		} else {
			rep->winner = DB_EID_INVALID;
			rep->w_priority = 0;
			rep->w_gen = 0;
			ZERO_LSN(rep->w_lsn);
			rep->w_tiebreaker = 0;
		}
	}
DB_TEST_RECOVERY_LABEL
err:	MUTEX_UNLOCK(dbenv, db_rep->mutexp);
	return (ret);
}

static int
__rep_wait(dbenv, timeout, eidp, flags)
	DB_ENV *dbenv;
	u_int32_t timeout;
	int *eidp;
	u_int32_t flags;
{
	DB_REP *db_rep;
	REP *rep;
	int done, ret;
	u_int32_t sleeptime;

	done = 0;
	db_rep = dbenv->rep_handle;
	rep = db_rep->region;

	/*
	 * The user specifies an overall timeout function, but checking
	 * is cheap and the timeout may be a generous upper bound.
	 * Sleep repeatedly for the smaller of .5s and timeout/10.
	 */
	sleeptime = (timeout > 5000000) ? 500000 : timeout / 10;
	if (sleeptime == 0)
		sleeptime++;
	while (timeout > 0) {
		if ((ret = __os_sleep(dbenv, 0, sleeptime)) != 0)
			return (ret);
		MUTEX_LOCK(dbenv, db_rep->mutexp);
		done = !F_ISSET(rep, flags) && rep->master_id != DB_EID_INVALID;

		*eidp = rep->master_id;
		MUTEX_UNLOCK(dbenv, db_rep->mutexp);

		if (done)
			return (0);

		if (timeout > sleeptime)
			timeout -= sleeptime;
		else
			timeout = 0;
	}
	return (DB_TIMEOUT);
}

/*
 * __rep_flush --
 *	Re-push the last log record to all clients, in case they've lost
 * messages and don't know it.
 */
static int
__rep_flush(dbenv)
	DB_ENV *dbenv;
{
	DBT rec;
	DB_LOGC *logc;
	DB_LSN lsn;
	int ret, t_ret;

	PANIC_CHECK(dbenv);
	ENV_REQUIRES_CONFIG(dbenv, dbenv->tx_handle, "rep_stat", DB_INIT_TXN);

	if ((ret = dbenv->log_cursor(dbenv, &logc, 0)) != 0)
		return (ret);

	memset(&rec, 0, sizeof(rec));
	memset(&lsn, 0, sizeof(lsn));

	if ((ret = logc->get(logc, &lsn, &rec, DB_LAST)) != 0)
		goto err;

	ret = __rep_send_message(dbenv,
	    DB_EID_BROADCAST, REP_LOG, &lsn, &rec, 0);

err:	if ((t_ret = logc->close(logc, 0)) != 0 && ret == 0)
		ret = t_ret;
	return (ret);
}

/*
 * __rep_stat --
 *	Fetch replication statistics.
 */
static int
__rep_stat(dbenv, statp, flags)
	DB_ENV *dbenv;
	DB_REP_STAT **statp;
	u_int32_t flags;
{
	DB_LOG *dblp;
	DB_REP *db_rep;
	DB_REP_STAT *stats;
	LOG *lp;
	REP *rep;
	u_int32_t queued;
	int ret;

	PANIC_CHECK(dbenv);
	ENV_REQUIRES_CONFIG(dbenv, dbenv->tx_handle, "rep_stat", DB_INIT_TXN);

	db_rep = dbenv->rep_handle;
	rep = db_rep->region;
	dblp = dbenv->lg_handle;
	lp = dblp->reginfo.primary;

	*statp = NULL;
	if ((ret = __db_fchk(dbenv,
	    "DB_ENV->rep_stat", flags, DB_STAT_CLEAR)) != 0)
		return (ret);

	/* Allocate a stat struct to return to the user. */
	if ((ret = __os_umalloc(dbenv, sizeof(DB_REP_STAT), &stats)) != 0)
		return (ret);

	MUTEX_LOCK(dbenv, db_rep->mutexp);
	memcpy(stats, &rep->stat, sizeof(*stats));

	/* Copy out election stats. */
	if (IN_ELECTION(rep)) {
		if (F_ISSET(rep, REP_F_EPHASE1))
			stats->st_election_status = 1;
		else if (F_ISSET(rep, REP_F_EPHASE2))
			stats->st_election_status = 2;

		stats->st_election_nsites = rep->sites;
		stats->st_election_cur_winner = rep->winner;
		stats->st_election_priority = rep->w_priority;
		stats->st_election_gen = rep->w_gen;
		stats->st_election_lsn = rep->w_lsn;
		stats->st_election_votes = rep->votes;
		stats->st_election_tiebreaker = rep->w_tiebreaker;
	}

	/* Copy out other info that's protected by the rep mutex. */
	stats->st_env_id = rep->eid;
	stats->st_env_priority = rep->priority;
	stats->st_nsites = rep->nsites;
	stats->st_master = rep->master_id;
	stats->st_gen = rep->gen;

	if (F_ISSET(rep, REP_F_MASTER))
		stats->st_status = DB_REP_MASTER;
	else if (F_ISSET(rep, REP_F_LOGSONLY))
		stats->st_status = DB_REP_LOGSONLY;
	else if (F_ISSET(rep, REP_F_UPGRADE))
		stats->st_status = DB_REP_CLIENT;
	else
		stats->st_status = 0;

	if (LF_ISSET(DB_STAT_CLEAR)) {
		queued = rep->stat.st_log_queued;
		memset(&rep->stat, 0, sizeof(rep->stat));
		rep->stat.st_log_queued = rep->stat.st_log_queued_total =
		    rep->stat.st_log_queued_max = queued;
	}
	MUTEX_UNLOCK(dbenv, db_rep->mutexp);

	/*
	 * Log-related replication info is stored in the log system and
	 * protected by the log region lock.
	 */
	R_LOCK(dbenv, &dblp->reginfo);
	if (F_ISSET(rep, REP_ISCLIENT)) {
		stats->st_next_lsn = lp->ready_lsn;
		stats->st_waiting_lsn = lp->waiting_lsn;
	} else {
		if (F_ISSET(rep, REP_F_MASTER))
			stats->st_next_lsn = lp->lsn;
		else
			ZERO_LSN(stats->st_next_lsn);
		ZERO_LSN(stats->st_waiting_lsn);
	}
	R_UNLOCK(dbenv, &dblp->reginfo);

	*statp = stats;
	return (0);
}