#include "db_config.h"
#include "db_int.h"
#include "dbinc/db_page.h"
#include "dbinc/btree.h"
#include "dbinc/mp.h"
#include "dbinc/txn.h"
static int __rep_abort_prepared __P((ENV *));
static int __rep_await_condition __P((ENV *,
struct rep_waitgoal *, db_timeout_t));
static int __rep_bt_cmp __P((DB *, const DBT *, const DBT *));
static int __rep_check_applied __P((ENV *,
DB_THREAD_INFO *, DB_COMMIT_INFO *, struct rep_waitgoal *));
static void __rep_config_map __P((ENV *, u_int32_t *, u_int32_t *));
static u_int32_t __rep_conv_vers __P((ENV *, u_int32_t));
static int __rep_read_lsn_history __P((ENV *,
DB_THREAD_INFO *, DB_TXN **, DBC **, u_int32_t,
__rep_lsn_hist_data_args *, struct rep_waitgoal *, u_int32_t));
static int __rep_restore_prepared __P((ENV *));
static int __rep_save_lsn_hist __P((ENV *, DB_THREAD_INFO *, DB_LSN *));
int
__rep_env_create(dbenv)
DB_ENV *dbenv;
{
DB_REP *db_rep;
ENV *env;
int ret;
env = dbenv->env;
if ((ret = __os_calloc(env, 1, sizeof(DB_REP), &db_rep)) != 0)
return (ret);
db_rep->eid = DB_EID_INVALID;
db_rep->bytes = REP_DEFAULT_THROTTLE;
DB_TIMEOUT_TO_TIMESPEC(DB_REP_REQUEST_GAP, &db_rep->request_gap);
DB_TIMEOUT_TO_TIMESPEC(DB_REP_MAX_GAP, &db_rep->max_gap);
db_rep->elect_timeout = 2 * US_PER_SEC;
db_rep->chkpt_delay = 30 * US_PER_SEC;
db_rep->my_priority = DB_REP_DEFAULT_PRIORITY;
db_rep->clock_skew = 1;
db_rep->clock_base = 1;
FLD_SET(db_rep->config, REP_C_AUTOINIT);
FLD_SET(db_rep->config, REP_C_AUTOROLLBACK);
FLD_SET(dbenv->verbose, DB_VERB_REP_SYSTEM);
#ifdef HAVE_REPLICATION_THREADS
if ((ret = __repmgr_env_create(env, db_rep)) != 0) {
__os_free(env, db_rep);
return (ret);
}
#endif
env->rep_handle = db_rep;
return (0);
}
void
__rep_env_destroy(dbenv)
DB_ENV *dbenv;
{
ENV *env;
env = dbenv->env;
if (env->rep_handle != NULL) {
#ifdef HAVE_REPLICATION_THREADS
__repmgr_env_destroy(env, env->rep_handle);
#endif
__os_free(env, env->rep_handle);
env->rep_handle = NULL;
}
}
int
__rep_get_config(dbenv, which, onp)
DB_ENV *dbenv;
u_int32_t which;
int *onp;
{
DB_REP *db_rep;
ENV *env;
REP *rep;
u_int32_t mapped;
env = dbenv->env;
#undef OK_FLAGS
#define OK_FLAGS \
(DB_REP_CONF_AUTOINIT | DB_REP_CONF_AUTOROLLBACK | \
DB_REP_CONF_BULK | DB_REP_CONF_DELAYCLIENT | DB_REP_CONF_INMEM | \
DB_REP_CONF_LEASE | DB_REP_CONF_NOWAIT | \
DB_REPMGR_CONF_2SITE_STRICT | DB_REPMGR_CONF_ELECTIONS)
if (FLD_ISSET(which, ~OK_FLAGS))
return (__db_ferr(env, "DB_ENV->rep_get_config", 0));
db_rep = env->rep_handle;
ENV_NOT_CONFIGURED(
env, db_rep->region, "DB_ENV->rep_get_config", DB_INIT_REP);
mapped = 0;
__rep_config_map(env, &which, &mapped);
if (REP_ON(env)) {
rep = db_rep->region;
if (FLD_ISSET(rep->config, mapped))
*onp = 1;
else
*onp = 0;
} else {
if (FLD_ISSET(db_rep->config, mapped))
*onp = 1;
else
*onp = 0;
}
return (0);
}
int
__rep_set_config(dbenv, which, on)
DB_ENV *dbenv;
u_int32_t which;
int on;
{
DB_LOG *dblp;
DB_REP *db_rep;
DB_THREAD_INFO *ip;
ENV *env;
LOG *lp;
REP *rep;
REP_BULK bulk;
u_int32_t mapped, orig;
int ret, t_ret;
env = dbenv->env;
db_rep = env->rep_handle;
ret = 0;
#undef OK_FLAGS
#define OK_FLAGS \
(DB_REP_CONF_AUTOINIT | DB_REP_CONF_AUTOROLLBACK | \
DB_REP_CONF_BULK | DB_REP_CONF_DELAYCLIENT | DB_REP_CONF_INMEM | \
DB_REP_CONF_LEASE | DB_REP_CONF_NOWAIT | \
DB_REPMGR_CONF_2SITE_STRICT | DB_REPMGR_CONF_ELECTIONS)
#define REPMGR_FLAGS (REP_C_2SITE_STRICT | REP_C_ELECTIONS)
ENV_NOT_CONFIGURED(
env, db_rep->region, "DB_ENV->rep_set_config", DB_INIT_REP);
if (FLD_ISSET(which, ~OK_FLAGS))
return (__db_ferr(env, "DB_ENV->rep_set_config", 0));
mapped = 0;
__rep_config_map(env, &which, &mapped);
if (APP_IS_BASEAPI(env) && FLD_ISSET(mapped, REPMGR_FLAGS)) {
__db_errx(env, DB_STR_A("3548",
"%s cannot configure repmgr settings from base replication application",
"%s"), "DB_ENV->rep_set_config:");
return (EINVAL);
}
if (REP_ON(env)) {
#ifdef HAVE_REPLICATION_THREADS
if ((ret = __repmgr_valid_config(env, mapped)) != 0)
return (ret);
#endif
ENV_ENTER(env, ip);
rep = db_rep->region;
if (FLD_ISSET(mapped, REP_C_INMEM)) {
__db_errx(env, DB_STR_A("3549",
"%s in-memory replication must be configured before DB_ENV->open",
"%s"), "DB_ENV->rep_set_config:");
ENV_LEAVE(env, ip);
return (EINVAL);
}
if (FLD_ISSET(mapped, REP_C_LEASE)) {
if (F_ISSET(rep, REP_F_START_CALLED)) {
__db_errx(env, DB_STR("3550",
"DB_ENV->rep_set_config: leases must be "
"configured before DB_ENV->rep_start"));
ret = EINVAL;
}
if (on == 0) {
__db_errx(env, DB_STR("3551",
"DB_ENV->rep_set_config: leases cannot be turned off"));
ret = EINVAL;
}
if (ret != 0) {
ENV_LEAVE(env, ip);
return (ret);
}
}
MUTEX_LOCK(env, rep->mtx_clientdb);
REP_SYSTEM_LOCK(env);
orig = rep->config;
if (on)
FLD_SET(rep->config, mapped);
else
FLD_CLR(rep->config, mapped);
dblp = env->lg_handle;
lp = dblp->reginfo.primary;
if (FLD_ISSET(rep->config, REP_C_BULK) &&
!FLD_ISSET(orig, REP_C_BULK))
db_rep->bulk = R_ADDR(&dblp->reginfo, lp->bulk_buf);
REP_SYSTEM_UNLOCK(env);
if (FLD_ISSET(orig, REP_C_BULK) &&
!FLD_ISSET(rep->config, REP_C_BULK) && lp->bulk_off != 0) {
memset(&bulk, 0, sizeof(bulk));
if (db_rep->bulk == NULL)
bulk.addr =
R_ADDR(&dblp->reginfo, lp->bulk_buf);
else
bulk.addr = db_rep->bulk;
bulk.offp = &lp->bulk_off;
bulk.len = lp->bulk_len;
bulk.type = REP_BULK_LOG;
bulk.eid = DB_EID_BROADCAST;
bulk.flagsp = &lp->bulk_flags;
ret = __rep_send_bulk(env, &bulk, 0);
}
MUTEX_UNLOCK(env, rep->mtx_clientdb);
ENV_LEAVE(env, ip);
#ifdef HAVE_REPLICATION_THREADS
if (!FLD_ISSET(orig, REP_C_ELECTIONS) &&
FLD_ISSET(rep->config, REP_C_ELECTIONS) &&
(t_ret = __repmgr_turn_on_elections(env)) != 0 && ret == 0)
ret = t_ret;
#endif
} else {
if (on)
FLD_SET(db_rep->config, mapped);
else
FLD_CLR(db_rep->config, mapped);
}
if (ret == 0 && FLD_ISSET(mapped, REPMGR_FLAGS))
APP_SET_REPMGR(env);
return (ret);
}
static void
__rep_config_map(env, inflagsp, outflagsp)
ENV *env;
u_int32_t *inflagsp, *outflagsp;
{
COMPQUIET(env, NULL);
if (FLD_ISSET(*inflagsp, DB_REP_CONF_AUTOINIT)) {
FLD_SET(*outflagsp, REP_C_AUTOINIT);
FLD_CLR(*inflagsp, DB_REP_CONF_AUTOINIT);
}
if (FLD_ISSET(*inflagsp, DB_REP_CONF_AUTOROLLBACK)) {
FLD_SET(*outflagsp, REP_C_AUTOROLLBACK);
FLD_CLR(*inflagsp, DB_REP_CONF_AUTOROLLBACK);
}
if (FLD_ISSET(*inflagsp, DB_REP_CONF_BULK)) {
FLD_SET(*outflagsp, REP_C_BULK);
FLD_CLR(*inflagsp, DB_REP_CONF_BULK);
}
if (FLD_ISSET(*inflagsp, DB_REP_CONF_DELAYCLIENT)) {
FLD_SET(*outflagsp, REP_C_DELAYCLIENT);
FLD_CLR(*inflagsp, DB_REP_CONF_DELAYCLIENT);
}
if (FLD_ISSET(*inflagsp, DB_REP_CONF_INMEM)) {
FLD_SET(*outflagsp, REP_C_INMEM);
FLD_CLR(*inflagsp, DB_REP_CONF_INMEM);
}
if (FLD_ISSET(*inflagsp, DB_REP_CONF_LEASE)) {
FLD_SET(*outflagsp, REP_C_LEASE);
FLD_CLR(*inflagsp, DB_REP_CONF_LEASE);
}
if (FLD_ISSET(*inflagsp, DB_REP_CONF_NOWAIT)) {
FLD_SET(*outflagsp, REP_C_NOWAIT);
FLD_CLR(*inflagsp, DB_REP_CONF_NOWAIT);
}
if (FLD_ISSET(*inflagsp, DB_REPMGR_CONF_2SITE_STRICT)) {
FLD_SET(*outflagsp, REP_C_2SITE_STRICT);
FLD_CLR(*inflagsp, DB_REPMGR_CONF_2SITE_STRICT);
}
if (FLD_ISSET(*inflagsp, DB_REPMGR_CONF_ELECTIONS)) {
FLD_SET(*outflagsp, REP_C_ELECTIONS);
FLD_CLR(*inflagsp, DB_REPMGR_CONF_ELECTIONS);
}
DB_ASSERT(env, *inflagsp == 0);
}
int
__rep_start_pp(dbenv, dbt, flags)
DB_ENV *dbenv;
DBT *dbt;
u_int32_t flags;
{
DB_REP *db_rep;
ENV *env;
env = dbenv->env;
db_rep = env->rep_handle;
ENV_REQUIRES_CONFIG_XX(
env, rep_handle, "DB_ENV->rep_start", DB_INIT_REP);
if (APP_IS_REPMGR(env)) {
__db_errx(env, DB_STR("3552",
"DB_ENV->rep_start: cannot call from Replication Manager application"));
return (EINVAL);
}
switch (LF_ISSET(DB_REP_CLIENT | DB_REP_MASTER)) {
case DB_REP_CLIENT:
case DB_REP_MASTER:
break;
default:
__db_errx(env, DB_STR("3553",
"DB_ENV->rep_start: must specify DB_REP_CLIENT or DB_REP_MASTER"));
return (EINVAL);
}
if (db_rep->send == NULL) {
__db_errx(env, DB_STR("3554",
"DB_ENV->rep_start: must be called after DB_ENV->rep_set_transport"));
return (EINVAL);
}
return (__rep_start_int(env, dbt, flags));
}
int
__rep_start_int(env, dbt, flags)
ENV *env;
DBT *dbt;
u_int32_t flags;
{
DB *dbp;
DB_LOG *dblp;
DB_LOGC *logc;
DB_LSN lsn, perm_lsn;
DB_REP *db_rep;
DB_THREAD_INFO *ip;
DB_TXNREGION *region;
LOG *lp;
REGENV *renv;
REGINFO *infop;
REP *rep;
db_timeout_t tmp;
u_int32_t new_gen, oldvers, pending_event, role;
int interrupting, locked, ret, role_chg, start_th, t_ret;
db_rep = env->rep_handle;
rep = db_rep->region;
infop = env->reginfo;
renv = infop->primary;
interrupting = locked = 0;
pending_event = DB_EVENT_NO_SUCH_EVENT;
role = LF_ISSET(DB_REP_CLIENT | DB_REP_MASTER);
start_th = 0;
if (IS_USING_LEASES(env) && rep->lease_timeout == 0) {
__db_errx(env, DB_STR("3555",
"DB_ENV->rep_start: must call DB_ENV->rep_set_timeout for leases first"));
return (EINVAL);
}
ENV_ENTER(env, ip);
MUTEX_LOCK(env, rep->mtx_repstart);
start_th = 1;
if ((ret = __log_flush(env, NULL)) != 0)
goto out;
REP_SYSTEM_LOCK(env);
role_chg = (!F_ISSET(rep, REP_F_MASTER) && role == DB_REP_MASTER) ||
(!F_ISSET(rep, REP_F_CLIENT) && role == DB_REP_CLIENT);
if (role == DB_REP_CLIENT && !role_chg) {
REP_SYSTEM_UNLOCK(env);
if ((ret = __dbt_usercopy(env, dbt)) == 0)
(void)__rep_send_message(env,
DB_EID_BROADCAST, REP_NEWCLIENT, NULL, dbt, 0, 0);
goto out;
}
if (FLD_ISSET(rep->lockout_flags, REP_LOCKOUT_MSG)) {
RPRINT(env, (env, DB_VERB_REP_MISC,
"Thread already in msg lockout"));
REP_SYSTEM_UNLOCK(env);
goto out;
} else if ((ret = __rep_lockout_msg(env, rep, 0)) != 0)
goto errunlock;
if (IN_INTERNAL_INIT(rep) && role == DB_REP_MASTER) {
__db_errx(env, DB_STR("3556",
"DB_ENV->rep_start: Cannot become master during internal init"));
ret = DB_REP_UNAVAIL;
goto errunlock;
}
if (role_chg) {
if ((ret = __rep_lockout_api(env, rep)) != 0)
goto errunlock;
locked = 1;
}
dblp = env->lg_handle;
lp = dblp->reginfo.primary;
if (role == DB_REP_MASTER) {
if (role_chg) {
(void)__memp_set_config(
env->dbenv, DB_MEMP_SYNC_INTERRUPT, 1);
interrupting = 1;
if ((ret = __rep_preclose(env)) != 0)
goto errunlock;
new_gen = rep->gen + 1;
if (rep->egen > rep->gen)
new_gen = rep->egen;
SET_GEN(new_gen);
if (IS_USING_LEASES(env) &&
rep->config_nsites > 1 &&
!F_ISSET(rep, REP_F_MASTERELECT)) {
__db_errx(env, DB_STR("3557",
"rep_start: Cannot become master without being elected when using leases."));
ret = EINVAL;
goto errunlock;
}
if (F_ISSET(rep, REP_F_MASTERELECT)) {
__rep_elect_done(env, rep);
F_CLR(rep, REP_F_MASTERELECT);
} else if (FLD_ISSET(rep->config, REP_C_INMEM))
RPRINT(env, (env, DB_VERB_REP_MISC,
"Appointed new master while running in-memory replication."));
if (rep->egen <= rep->gen)
rep->egen = rep->gen + 1;
RPRINT(env, (env, DB_VERB_REP_MISC,
"New master gen %lu, egen %lu",
(u_long)rep->gen, (u_long)rep->egen));
if (!FLD_ISSET(rep->config, REP_C_INMEM) &&
(ret = __rep_write_gen(env, rep, rep->gen)) != 0)
goto errunlock;
}
if (IS_USING_LEASES(env) &&
(role_chg || !IS_REP_STARTED(env))) {
if ((ret = __rep_islease_granted(env))) {
__db_errx(env, DB_STR("3558",
"rep_start: Cannot become master with outstanding lease granted."));
ret = EINVAL;
goto errunlock;
}
if ((ret = __log_cursor(env, &logc)) != 0)
goto errunlock;
ret = __rep_log_backup(env, logc, &perm_lsn,
REP_REC_PERM);
(void)__logc_close(logc);
if (ret == 0)
lp->max_perm_lsn = perm_lsn;
else if (ret == DB_NOTFOUND)
INIT_LSN(lp->max_perm_lsn);
else
goto errunlock;
tmp = (db_timeout_t)((double)rep->lease_timeout /
((double)rep->clock_skew /
(double)rep->clock_base));
DB_TIMEOUT_TO_TIMESPEC(tmp, &rep->lease_duration);
if ((ret = __rep_lease_table_alloc(env,
rep->config_nsites)) != 0)
goto errunlock;
}
rep->master_id = rep->eid;
STAT_INC(env, rep,
master_change, rep->stat.st_master_changes, rep->eid);
#ifdef DIAGNOSTIC
if (!F_ISSET(rep, REP_F_GROUP_ESTD))
RPRINT(env, (env, DB_VERB_REP_MISC,
"Establishing group as master."));
#endif
F_CLR(rep, REP_F_CLIENT | REP_F_ABBREVIATED |
REP_F_MASTERELECT | REP_F_SKIPPED_APPLY | REP_F_DELAY |
REP_F_LEASE_EXPIRED | REP_F_NEWFILE);
F_SET(rep, REP_F_MASTER | REP_F_GROUP_ESTD |
REP_F_NIMDBS_LOADED);
rep->sync_state = SYNC_OFF;
oldvers = lp->persist.version;
VPRINT(env, (env, DB_VERB_REP_MISC,
"rep_start: Old log version was %lu", (u_long)oldvers));
if (lp->persist.version != DB_LOGVERSION) {
if ((ret = __env_init_rec(env, DB_LOGVERSION)) != 0)
goto errunlock;
}
rep->version = DB_REPVERSION;
FLD_CLR(rep->lockout_flags,
REP_LOCKOUT_ARCHIVE | REP_LOCKOUT_MSG);
REP_SYSTEM_UNLOCK(env);
LOG_SYSTEM_LOCK(env);
lsn = lp->lsn;
LOG_SYSTEM_UNLOCK(env);
(void)__rep_send_message(env,
DB_EID_BROADCAST, REP_NEWMASTER, &lsn, NULL, 0, 0);
ret = 0;
if (role_chg) {
pending_event = DB_EVENT_REP_MASTER;
region = env->tx_handle->reginfo.primary;
if (region->stat.st_nrestores == 0 &&
(t_ret = __rep_restore_prepared(env)) != 0 &&
ret == 0)
ret = t_ret;
if (region->stat.st_nrestores != 0) {
if ((t_ret = __dbreg_mark_restored(env)) != 0 &&
ret == 0)
ret = t_ret;
} else {
ret = __dbreg_invalidate_files(env, 0);
if ((t_ret = __rep_closefiles(env)) != 0 &&
ret == 0)
ret = t_ret;
}
REP_SYSTEM_LOCK(env);
F_SET(rep, REP_F_SYS_DB_OP);
REP_SYSTEM_UNLOCK(env);
if ((t_ret = __txn_recycle_id(env, 0)) != 0 && ret == 0)
ret = t_ret;
if ((t_ret = __rep_save_lsn_hist(env, ip, &lsn)) != 0 &&
ret == 0)
ret = t_ret;
REP_SYSTEM_LOCK(env);
rep->gen_base_lsn = lsn;
rep->master_envid = renv->envid;
F_CLR(rep, REP_F_SYS_DB_OP);
CLR_LOCKOUT_BDB(rep);
locked = 0;
REP_SYSTEM_UNLOCK(env);
(void)__memp_set_config(
env->dbenv, DB_MEMP_SYNC_INTERRUPT, 0);
interrupting = 0;
}
} else {
rep->master_id = DB_EID_INVALID;
DB_ASSERT(env, !FLD_ISSET(rep->elect_flags, ~REP_E_TALLY));
DB_ASSERT(env, !F_ISSET(rep,
REP_F_ABBREVIATED | REP_F_DELAY | REP_F_NEWFILE));
DB_ASSERT(env, rep->sync_state == SYNC_OFF);
if ((ret = __log_get_oldversion(env, &oldvers)) != 0)
goto errunlock;
RPRINT(env, (env, DB_VERB_REP_MISC,
"rep_start: Found old version log %d", oldvers));
if (oldvers >= DB_LOGVERSION_MIN) {
__log_set_version(env, oldvers);
if ((ret = __env_init_rec(env, oldvers)) != 0)
goto errunlock;
oldvers = __rep_conv_vers(env, oldvers);
DB_ASSERT(env, oldvers != DB_REPVERSION_INVALID);
rep->version = oldvers;
}
F_CLR(rep, REP_F_MASTER | REP_F_MASTERELECT);
F_SET(rep, REP_F_CLIENT);
if (IS_USING_LEASES(env) && !IS_REP_STARTED(env)) {
if ((ret = __rep_lease_expire(env)) != 0)
goto errunlock;
tmp = (db_timeout_t)((double)rep->lease_timeout *
((double)rep->clock_skew /
(double)rep->clock_base));
DB_TIMEOUT_TO_TIMESPEC(tmp, &rep->lease_duration);
if (rep->lease_off != INVALID_ROFF) {
MUTEX_LOCK(env, renv->mtx_regenv);
__env_alloc_free(infop,
R_ADDR(infop, rep->lease_off));
MUTEX_UNLOCK(env, renv->mtx_regenv);
rep->lease_off = INVALID_ROFF;
}
}
REP_SYSTEM_UNLOCK(env);
if ((ret = __rep_abort_prepared(env)) != 0)
goto errlock;
if ((ret = __db_create_internal(&dbp, env, 0)) != 0)
goto errlock;
MUTEX_LOCK(env, rep->mtx_clientdb);
(void)__db_remove(dbp, ip, NULL, REPDBNAME,
NULL, DB_FORCE);
MUTEX_UNLOCK(env, rep->mtx_clientdb);
pending_event = DB_EVENT_REP_CLIENT;
REP_SYSTEM_LOCK(env);
FLD_CLR(rep->lockout_flags, REP_LOCKOUT_MSG);
if (locked) {
CLR_LOCKOUT_BDB(rep);
locked = 0;
}
if (F_ISSET(env, ENV_PRIVATE))
SET_GEN(0);
REP_SYSTEM_UNLOCK(env);
if ((ret = __dbt_usercopy(env, dbt)) != 0)
goto out;
(void)__rep_send_message(env,
DB_EID_BROADCAST, REP_NEWCLIENT, NULL, dbt, 0, 0);
}
if (0) {
errlock: REP_SYSTEM_LOCK(env);
errunlock: FLD_CLR(rep->lockout_flags, REP_LOCKOUT_MSG);
if (locked)
CLR_LOCKOUT_BDB(rep);
if (interrupting)
(void)__memp_set_config(
env->dbenv, DB_MEMP_SYNC_INTERRUPT, 0);
REP_SYSTEM_UNLOCK(env);
}
out:
if (ret == 0) {
REP_SYSTEM_LOCK(env);
F_SET(rep, REP_F_START_CALLED);
REP_SYSTEM_UNLOCK(env);
}
if (pending_event != DB_EVENT_NO_SUCH_EVENT)
__rep_fire_event(env, pending_event, NULL);
if (start_th)
MUTEX_UNLOCK(env, rep->mtx_repstart);
__dbt_userfree(env, dbt, NULL, NULL);
ENV_LEAVE(env, ip);
return (ret);
}
static int
__rep_save_lsn_hist(env, ip, lsnp)
ENV *env;
DB_THREAD_INFO *ip;
DB_LSN *lsnp;
{
DB_REP *db_rep;
REP *rep;
REGENV *renv;
DB_TXN *txn;
DB *dbp;
DBT key_dbt, data_dbt;
__rep_lsn_hist_key_args key;
__rep_lsn_hist_data_args data;
u_int8_t key_buf[__REP_LSN_HIST_KEY_SIZE];
u_int8_t data_buf[__REP_LSN_HIST_DATA_SIZE];
db_timespec now;
int ret, t_ret;
db_rep = env->rep_handle;
rep = db_rep->region;
renv = env->reginfo->primary;
txn = NULL;
ret = 0;
if ((ret = __txn_begin(env, ip, NULL, &txn, DB_IGNORE_LEASE)) != 0)
return (ret);
if ((dbp = db_rep->lsn_db) == NULL &&
(ret = __rep_open_sysdb(env,
ip, txn, REPLSNHIST, DB_CREATE, &dbp)) != 0)
goto err;
key.version = REP_LSN_HISTORY_FMT_VERSION;
key.gen = rep->gen;
__rep_lsn_hist_key_marshal(env, &key, key_buf);
data.envid = renv->envid;
data.lsn = *lsnp;
__os_gettime(env, &now, 0);
data.hist_sec = (u_int32_t)now.tv_sec;
data.hist_nsec = (u_int32_t)now.tv_nsec;
__rep_lsn_hist_data_marshal(env, &data, data_buf);
DB_INIT_DBT(key_dbt, key_buf, sizeof(key_buf));
DB_INIT_DBT(data_dbt, data_buf, sizeof(data_buf));
ret = __db_put(dbp, ip, txn, &key_dbt, &data_dbt, 0);
err:
if (dbp != NULL &&
(t_ret = __db_close(dbp, txn, DB_NOSYNC)) != 0 && ret == 0)
ret = t_ret;
db_rep->lsn_db = NULL;
DB_ASSERT(env, txn != NULL);
if ((t_ret = __db_txn_auto_resolve(env, txn, 0, ret)) != 0 && ret == 0)
ret = t_ret;
return (ret);
}
int
__rep_open_sysdb(env, ip, txn, dbname, flags, dbpp)
ENV *env;
DB_THREAD_INFO *ip;
DB_TXN *txn;
const char *dbname;
u_int32_t flags;
DB **dbpp;
{
DB_REP *db_rep;
REP *rep;
DB *dbp;
char *fname;
u_int32_t myflags;
int ret, t_ret;
db_rep = env->rep_handle;
rep = db_rep->region;
if ((ret = __db_create_internal(&dbp, env, 0)) != 0)
return (ret);
myflags = DB_INTERNAL_PERSISTENT_DB |
(F_ISSET(env, ENV_THREAD) ? DB_THREAD : 0);
if ((ret = __db_open(dbp, ip, txn,
REPSYSDBNAME, dbname, DB_BTREE, myflags, 0, PGNO_BASE_MD)) == 0)
goto found;
if (ret != ENOENT)
goto err;
ret = __db_close(dbp, txn, DB_NOSYNC);
dbp = NULL;
if (ret != 0 || (ret = __db_create_internal(&dbp, env, 0)) != 0)
goto err;
if ((ret = __db_open(dbp, ip, txn,
NULL, dbname, DB_BTREE, myflags, 0, PGNO_BASE_MD)) == 0)
goto found;
if (ret != ENOENT)
goto err;
ret = __db_close(dbp, txn, DB_NOSYNC);
dbp = NULL;
if (ret != 0)
goto err;
if (LF_ISSET(DB_CREATE)) {
if ((ret = __db_create_internal(&dbp, env, 0)) != 0)
goto err;
if ((ret = __db_set_pagesize(dbp, REPSYSDBPGSZ)) != 0)
goto err;
FLD_SET(myflags, DB_CREATE);
fname = FLD_ISSET(rep->config, REP_C_INMEM) ?
NULL : REPSYSDBNAME;
if ((ret = __db_open(dbp, ip, txn, fname,
dbname, DB_BTREE, myflags, 0, PGNO_BASE_MD)) == 0)
goto found;
} else
ret = ENOENT;
err:
if (dbp != NULL && (t_ret = __db_close(dbp, txn, DB_NOSYNC)) != 0 &&
(ret == 0 || ret == ENOENT))
ret = t_ret;
return (ret);
found:
*dbpp = dbp;
return (0);
}
int
__rep_client_dbinit(env, startup, which)
ENV *env;
int startup;
repdb_t which;
{
DB *dbp, **rdbpp;
DB_REP *db_rep;
DB_THREAD_INFO *ip;
REP *rep;
int ret, t_ret;
u_int32_t flags;
const char *fname, *name, *subdb;
db_rep = env->rep_handle;
rep = db_rep->region;
dbp = NULL;
if (which == REP_DB) {
name = REPDBNAME;
rdbpp = &db_rep->rep_db;
} else {
name = REPPAGENAME;
rdbpp = &db_rep->file_dbp;
}
if (*rdbpp != NULL)
return (0);
ENV_GET_THREAD_INFO(env, ip);
fname = name;
subdb = NULL;
if (FLD_ISSET(rep->config, REP_C_INMEM)) {
fname = NULL;
subdb = name;
}
if (startup) {
if ((ret = __db_create_internal(&dbp, env, 0)) != 0)
goto err;
if (FLD_ISSET(rep->config, REP_C_INMEM))
(void)__db_set_flags(dbp, DB_TXN_NOT_DURABLE);
(void)__db_remove(dbp, ip, NULL, fname, subdb, DB_FORCE);
}
if ((ret = __db_create_internal(&dbp, env, 0)) != 0)
goto err;
if (which == REP_DB &&
(ret = __bam_set_bt_compare(dbp, __rep_bt_cmp)) != 0)
goto err;
if ((ret = __db_set_flags(dbp, DB_TXN_NOT_DURABLE)) != 0)
goto err;
flags = DB_NO_AUTO_COMMIT | DB_CREATE | DB_INTERNAL_TEMPORARY_DB |
(F_ISSET(env, ENV_THREAD) ? DB_THREAD : 0);
if ((ret = __db_open(dbp, ip, NULL, fname, subdb,
(which == REP_DB ? DB_BTREE : DB_RECNO),
flags, 0, PGNO_BASE_MD)) != 0)
goto err;
*rdbpp = dbp;
if (0) {
err: if (dbp != NULL &&
(t_ret = __db_close(dbp, NULL, DB_NOSYNC)) != 0 && ret == 0)
ret = t_ret;
*rdbpp = NULL;
}
return (ret);
}
static int
__rep_bt_cmp(dbp, dbt1, dbt2)
DB *dbp;
const DBT *dbt1, *dbt2;
{
DB_LSN lsn1, lsn2;
__rep_control_args *rp1, *rp2;
COMPQUIET(dbp, NULL);
rp1 = dbt1->data;
rp2 = dbt2->data;
(void)__ua_memcpy(&lsn1, &rp1->lsn, sizeof(DB_LSN));
(void)__ua_memcpy(&lsn2, &rp2->lsn, sizeof(DB_LSN));
if (lsn1.file > lsn2.file)
return (1);
if (lsn1.file < lsn2.file)
return (-1);
if (lsn1.offset > lsn2.offset)
return (1);
if (lsn1.offset < lsn2.offset)
return (-1);
return (0);
}
static int
__rep_abort_prepared(env)
ENV *env;
{
#define PREPLISTSIZE 50
DB_LOG *dblp;
DB_PREPLIST prep[PREPLISTSIZE], *p;
DB_TXNMGR *mgr;
DB_TXNREGION *region;
LOG *lp;
int ret;
long count, i;
u_int32_t op;
mgr = env->tx_handle;
region = mgr->reginfo.primary;
dblp = env->lg_handle;
lp = dblp->reginfo.primary;
if (region->stat.st_nrestores == 0)
return (0);
op = DB_FIRST;
do {
if ((ret = __txn_recover(env,
prep, PREPLISTSIZE, &count, op)) != 0)
return (ret);
for (i = 0; i < count; i++) {
p = &prep[i];
if ((ret = __txn_abort(p->txn)) != 0)
return (ret);
env->rep_handle->region->op_cnt--;
env->rep_handle->region->max_prep_lsn = lp->lsn;
region->stat.st_nrestores--;
}
op = DB_NEXT;
} while (count == PREPLISTSIZE);
return (0);
}
static int
__rep_restore_prepared(env)
ENV *env;
{
DBT rec;
DB_LOGC *logc;
DB_LSN ckp_lsn, lsn;
DB_REP *db_rep;
DB_TXNHEAD *txninfo;
REP *rep;
__txn_ckp_args *ckp_args;
__txn_regop_args *regop_args;
__txn_prepare_args *prep_args;
int ret, t_ret;
u_int32_t hi_txn, low_txn, rectype, status, txnid, txnop;
db_rep = env->rep_handle;
rep = db_rep->region;
if (IS_ZERO_LSN(rep->max_prep_lsn)) {
VPRINT(env, (env, DB_VERB_REP_MISC,
"restore_prep: No prepares. Skip."));
return (0);
}
txninfo = NULL;
ckp_args = NULL;
prep_args = NULL;
regop_args = NULL;
ZERO_LSN(ckp_lsn);
ZERO_LSN(lsn);
if ((ret = __log_cursor(env, &logc)) != 0)
return (ret);
memset(&rec, 0, sizeof(DBT));
if ((ret = __logc_get(logc, &lsn, &rec, DB_FIRST)) != 0) {
__db_errx(env, DB_STR("3559", "First record not found"));
goto err;
}
if (rep->max_prep_lsn.file < lsn.file) {
VPRINT(env, (env, DB_VERB_REP_MISC,
"restore_prep: Prepare resolved. Skip"));
ZERO_LSN(rep->max_prep_lsn);
goto done;
}
if ((ret = __txn_getckp(env, &lsn)) == 0) {
if ((ret = __logc_get(logc, &lsn, &rec, DB_SET)) != 0) {
__db_errx(env, DB_STR_A("3560",
"Checkpoint record at LSN [%lu][%lu] not found",
"%lu %lu"), (u_long)lsn.file, (u_long)lsn.offset);
goto err;
}
if ((ret = __txn_ckp_read(
env, rec.data, &ckp_args)) == 0) {
ckp_lsn = ckp_args->ckp_lsn;
__os_free(env, ckp_args);
}
if (ret != 0) {
__db_errx(env, DB_STR_A("3561",
"Invalid checkpoint record at [%lu][%lu]",
"%lu %lu"), (u_long)lsn.file, (u_long)lsn.offset);
goto err;
}
if ((ret = __logc_get(logc, &ckp_lsn, &rec, DB_SET)) != 0) {
__db_errx(env, DB_STR_A("3562",
"Checkpoint LSN record [%lu][%lu] not found",
"%lu %lu"),
(u_long)ckp_lsn.file, (u_long)ckp_lsn.offset);
goto err;
}
} else if ((ret = __logc_get(logc, &lsn, &rec, DB_FIRST)) != 0) {
if (ret == DB_NOTFOUND) {
ret = 0;
goto done;
}
__db_errx(env, DB_STR("3563",
"Attempt to get first log record failed"));
goto err;
}
do {
LOGCOPY_32(env, &low_txn,
(u_int8_t *)rec.data + sizeof(u_int32_t));
if (low_txn != 0)
break;
} while ((ret = __logc_get(logc, &lsn, &rec, DB_NEXT)) == 0);
if (ret == DB_NOTFOUND) {
ret = 0;
goto done;
} else if (ret != 0)
goto err;
if ((ret = __logc_get(logc, &lsn, &rec, DB_LAST)) != 0) {
__db_errx(env, DB_STR("3564",
"Final log record not found"));
goto err;
}
do {
LOGCOPY_32(env, &hi_txn,
(u_int8_t *)rec.data + sizeof(u_int32_t));
if (hi_txn != 0)
break;
} while ((ret = __logc_get(logc, &lsn, &rec, DB_PREV)) == 0);
if (ret == DB_NOTFOUND) {
ret = 0;
goto done;
} else if (ret != 0)
goto err;
if ((ret = __db_txnlist_init(env,
NULL, low_txn, hi_txn, NULL, &txninfo)) != 0)
goto err;
F_SET(env->lg_handle, DBLOG_RECOVER);
for (ret = __logc_get(logc, &lsn, &rec, DB_LAST);
ret == 0 && LOG_COMPARE(&lsn, &ckp_lsn) > 0;
ret = __logc_get(logc, &lsn, &rec, DB_PREV)) {
LOGCOPY_32(env, &rectype, rec.data);
switch (rectype) {
case DB___txn_regop:
if ((ret = __txn_regop_read(
env, rec.data, ®op_args)) != 0)
goto err;
txnid = regop_args->txnp->txnid;
txnop = regop_args->opcode;
__os_free(env, regop_args);
ret = __db_txnlist_find(env,
txninfo, txnid, &status);
if (ret == DB_NOTFOUND)
ret = __db_txnlist_add(env, txninfo,
txnid, txnop, &lsn);
else if (ret != 0)
goto err;
break;
case DB___txn_prepare:
if ((ret = __txn_prepare_read(
env, rec.data, &prep_args)) != 0)
goto err;
ret = __db_txnlist_find(env, txninfo,
prep_args->txnp->txnid, &status);
if (ret == DB_NOTFOUND) {
if (prep_args->opcode == TXN_ABORT)
ret = __db_txnlist_add(env, txninfo,
prep_args->txnp->txnid,
prep_args->opcode, &lsn);
else if ((ret =
__rep_process_txn(env, &rec)) == 0) {
rep->op_cnt++;
ret = __txn_restore_txn(env,
&lsn, prep_args);
}
} else if (ret != 0)
goto err;
__os_free(env, prep_args);
break;
default:
continue;
}
}
if (ret == DB_NOTFOUND)
ret = 0;
done:
err: t_ret = __logc_close(logc);
F_CLR(env->lg_handle, DBLOG_RECOVER);
if (txninfo != NULL)
__db_txnlist_end(env, txninfo);
return (ret == 0 ? t_ret : ret);
}
int
__rep_get_limit(dbenv, gbytesp, bytesp)
DB_ENV *dbenv;
u_int32_t *gbytesp, *bytesp;
{
DB_REP *db_rep;
DB_THREAD_INFO *ip;
ENV *env;
REP *rep;
env = dbenv->env;
db_rep = env->rep_handle;
ENV_NOT_CONFIGURED(
env, db_rep->region, "DB_ENV->rep_get_limit", DB_INIT_REP);
if (REP_ON(env)) {
rep = db_rep->region;
ENV_ENTER(env, ip);
REP_SYSTEM_LOCK(env);
if (gbytesp != NULL)
*gbytesp = rep->gbytes;
if (bytesp != NULL)
*bytesp = rep->bytes;
REP_SYSTEM_UNLOCK(env);
ENV_LEAVE(env, ip);
} else {
if (gbytesp != NULL)
*gbytesp = db_rep->gbytes;
if (bytesp != NULL)
*bytesp = db_rep->bytes;
}
return (0);
}
int
__rep_set_limit(dbenv, gbytes, bytes)
DB_ENV *dbenv;
u_int32_t gbytes, bytes;
{
DB_REP *db_rep;
DB_THREAD_INFO *ip;
ENV *env;
REP *rep;
env = dbenv->env;
db_rep = env->rep_handle;
ENV_NOT_CONFIGURED(
env, db_rep->region, "DB_ENV->rep_set_limit", DB_INIT_REP);
if (bytes > GIGABYTE) {
gbytes += bytes / GIGABYTE;
bytes = bytes % GIGABYTE;
}
if (REP_ON(env)) {
rep = db_rep->region;
ENV_ENTER(env, ip);
REP_SYSTEM_LOCK(env);
rep->gbytes = gbytes;
rep->bytes = bytes;
REP_SYSTEM_UNLOCK(env);
ENV_LEAVE(env, ip);
} else {
db_rep->gbytes = gbytes;
db_rep->bytes = bytes;
}
return (0);
}
int
__rep_set_nsites_pp(dbenv, n)
DB_ENV *dbenv;
u_int32_t n;
{
DB_REP *db_rep;
DB_THREAD_INFO *ip;
ENV *env;
int ret;
env = dbenv->env;
db_rep = env->rep_handle;
ENV_NOT_CONFIGURED(
env, db_rep->region, "DB_ENV->rep_set_nsites", DB_INIT_REP);
if (APP_IS_REPMGR(env)) {
__db_errx(env, DB_STR("3565",
"DB_ENV->rep_set_nsites: cannot call from Replication Manager application"));
return (EINVAL);
}
if ((ret = __rep_set_nsites_int(env, n)) == 0)
APP_SET_BASEAPI(env);
return (ret);
}
int
__rep_set_nsites_int(env, n)
ENV *env;
u_int32_t n;
{
DB_REP *db_rep;
REP *rep;
int ret;
db_rep = env->rep_handle;
ret = 0;
if (REP_ON(env)) {
rep = db_rep->region;
rep->config_nsites = n;
if (IS_USING_LEASES(env) &&
IS_REP_MASTER(env) && IS_REP_STARTED(env)) {
REP_SYSTEM_LOCK(env);
ret = __rep_lease_table_alloc(env, n);
REP_SYSTEM_UNLOCK(env);
}
} else
db_rep->config_nsites = n;
return (ret);
}
int
__rep_get_nsites(dbenv, n)
DB_ENV *dbenv;
u_int32_t *n;
{
DB_REP *db_rep;
ENV *env;
REP *rep;
env = dbenv->env;
db_rep = env->rep_handle;
ENV_NOT_CONFIGURED(
env, db_rep->region, "DB_ENV->rep_get_nsites", DB_INIT_REP);
if (APP_IS_REPMGR(env))
return (__repmgr_get_nsites(env, n));
if (REP_ON(env)) {
rep = db_rep->region;
*n = rep->config_nsites;
} else
*n = db_rep->config_nsites;
return (0);
}
int
__rep_set_priority(dbenv, priority)
DB_ENV *dbenv;
u_int32_t priority;
{
DB_REP *db_rep;
ENV *env;
REP *rep;
u_int32_t prev;
int ret;
env = dbenv->env;
db_rep = env->rep_handle;
ENV_NOT_CONFIGURED(
env, db_rep->region, "DB_ENV->rep_set_priority", DB_INIT_REP);
ret = 0;
if (REP_ON(env)) {
rep = db_rep->region;
prev = rep->priority;
rep->priority = priority;
#ifdef HAVE_REPLICATION_THREADS
ret = __repmgr_chg_prio(env, prev, priority);
#endif
} else
db_rep->my_priority = priority;
return (ret);
}
int
__rep_get_priority(dbenv, priority)
DB_ENV *dbenv;
u_int32_t *priority;
{
DB_REP *db_rep;
ENV *env;
REP *rep;
env = dbenv->env;
db_rep = env->rep_handle;
ENV_NOT_CONFIGURED(
env, db_rep->region, "DB_ENV->rep_get_priority", DB_INIT_REP);
if (REP_ON(env)) {
rep = db_rep->region;
*priority = rep->priority;
} else
*priority = db_rep->my_priority;
return (0);
}
int
__rep_set_timeout(dbenv, which, timeout)
DB_ENV *dbenv;
int which;
db_timeout_t timeout;
{
DB_REP *db_rep;
DB_THREAD_INFO *ip;
ENV *env;
REP *rep;
int repmgr_timeout, ret;
env = dbenv->env;
db_rep = env->rep_handle;
rep = db_rep->region;
ret = 0;
repmgr_timeout = 0;
if (timeout == 0 && (which == DB_REP_CONNECTION_RETRY ||
which == DB_REP_ELECTION_TIMEOUT || which == DB_REP_LEASE_TIMEOUT ||
which == DB_REP_ELECTION_RETRY)) {
__db_errx(env, DB_STR("3566", "timeout value must be > 0"));
return (EINVAL);
}
if (which == DB_REP_ACK_TIMEOUT || which == DB_REP_CONNECTION_RETRY ||
which == DB_REP_ELECTION_RETRY ||
which == DB_REP_HEARTBEAT_MONITOR ||
which == DB_REP_HEARTBEAT_SEND)
repmgr_timeout = 1;
ENV_NOT_CONFIGURED(
env, db_rep->region, "DB_ENV->rep_set_timeout", DB_INIT_REP);
if (APP_IS_BASEAPI(env) && repmgr_timeout) {
__db_errx(env, DB_STR_A("3567",
"%scannot set Replication Manager timeout from base replication application",
"%s"), "DB_ENV->rep_set_timeout:");
return (EINVAL);
}
if (which == DB_REP_LEASE_TIMEOUT && IS_REP_STARTED(env)) {
ret = EINVAL;
__db_errx(env, DB_STR_A("3568",
"%s: lease timeout must be set before DB_ENV->rep_start.",
"%s"), "DB_ENV->rep_set_timeout");
return (EINVAL);
}
switch (which) {
case DB_REP_CHECKPOINT_DELAY:
if (REP_ON(env))
rep->chkpt_delay = timeout;
else
db_rep->chkpt_delay = timeout;
break;
case DB_REP_ELECTION_TIMEOUT:
if (REP_ON(env))
rep->elect_timeout = timeout;
else
db_rep->elect_timeout = timeout;
break;
case DB_REP_FULL_ELECTION_TIMEOUT:
if (REP_ON(env))
rep->full_elect_timeout = timeout;
else
db_rep->full_elect_timeout = timeout;
break;
case DB_REP_LEASE_TIMEOUT:
if (REP_ON(env))
rep->lease_timeout = timeout;
else
db_rep->lease_timeout = timeout;
break;
#ifdef HAVE_REPLICATION_THREADS
case DB_REP_ACK_TIMEOUT:
if (REP_ON(env))
rep->ack_timeout = timeout;
else
db_rep->ack_timeout = timeout;
break;
case DB_REP_CONNECTION_RETRY:
if (REP_ON(env))
rep->connection_retry_wait = timeout;
else
db_rep->connection_retry_wait = timeout;
break;
case DB_REP_ELECTION_RETRY:
if (REP_ON(env))
rep->election_retry_wait = timeout;
else
db_rep->election_retry_wait = timeout;
break;
case DB_REP_HEARTBEAT_MONITOR:
if (REP_ON(env))
rep->heartbeat_monitor_timeout = timeout;
else
db_rep->heartbeat_monitor_timeout = timeout;
break;
case DB_REP_HEARTBEAT_SEND:
if (REP_ON(env))
rep->heartbeat_frequency = timeout;
else
db_rep->heartbeat_frequency = timeout;
break;
#endif
default:
__db_errx(env, DB_STR("3569",
"Unknown timeout type argument to DB_ENV->rep_set_timeout"));
ret = EINVAL;
}
if (ret == 0 && repmgr_timeout)
APP_SET_REPMGR(env);
return (ret);
}
int
__rep_get_timeout(dbenv, which, timeout)
DB_ENV *dbenv;
int which;
db_timeout_t *timeout;
{
DB_REP *db_rep;
ENV *env;
REP *rep;
env = dbenv->env;
db_rep = env->rep_handle;
rep = db_rep->region;
ENV_NOT_CONFIGURED(
env, db_rep->region, "DB_ENV->rep_get_timeout", DB_INIT_REP);
switch (which) {
case DB_REP_CHECKPOINT_DELAY:
*timeout = REP_ON(env) ?
rep->chkpt_delay : db_rep->chkpt_delay;
break;
case DB_REP_ELECTION_TIMEOUT:
*timeout = REP_ON(env) ?
rep->elect_timeout : db_rep->elect_timeout;
break;
case DB_REP_FULL_ELECTION_TIMEOUT:
*timeout = REP_ON(env) ?
rep->full_elect_timeout : db_rep->full_elect_timeout;
break;
case DB_REP_LEASE_TIMEOUT:
*timeout = REP_ON(env) ?
rep->lease_timeout : db_rep->lease_timeout;
break;
#ifdef HAVE_REPLICATION_THREADS
case DB_REP_ACK_TIMEOUT:
*timeout = REP_ON(env) ?
rep->ack_timeout : db_rep->ack_timeout;
break;
case DB_REP_CONNECTION_RETRY:
*timeout = REP_ON(env) ?
rep->connection_retry_wait : db_rep->connection_retry_wait;
break;
case DB_REP_ELECTION_RETRY:
*timeout = REP_ON(env) ?
rep->election_retry_wait : db_rep->election_retry_wait;
break;
case DB_REP_HEARTBEAT_MONITOR:
*timeout = REP_ON(env) ? rep->heartbeat_monitor_timeout :
db_rep->heartbeat_monitor_timeout;
break;
case DB_REP_HEARTBEAT_SEND:
*timeout = REP_ON(env) ?
rep->heartbeat_frequency : db_rep->heartbeat_frequency;
break;
#endif
default:
__db_errx(env, DB_STR("3570",
"unknown timeout type argument to DB_ENV->rep_get_timeout"));
return (EINVAL);
}
return (0);
}
int
__rep_get_request(dbenv, minp, maxp)
DB_ENV *dbenv;
db_timeout_t *minp, *maxp;
{
DB_REP *db_rep;
DB_THREAD_INFO *ip;
ENV *env;
REP *rep;
env = dbenv->env;
db_rep = env->rep_handle;
ENV_NOT_CONFIGURED(
env, db_rep->region, "DB_ENV->rep_get_request", DB_INIT_REP);
if (REP_ON(env)) {
rep = db_rep->region;
ENV_ENTER(env, ip);
REP_SYSTEM_LOCK(env);
if (minp != NULL)
DB_TIMESPEC_TO_TIMEOUT((*minp), &rep->request_gap, 0);
if (maxp != NULL)
DB_TIMESPEC_TO_TIMEOUT((*maxp), &rep->max_gap, 0);
REP_SYSTEM_UNLOCK(env);
ENV_LEAVE(env, ip);
} else {
if (minp != NULL)
DB_TIMESPEC_TO_TIMEOUT((*minp),
&db_rep->request_gap, 0);
if (maxp != NULL)
DB_TIMESPEC_TO_TIMEOUT((*maxp), &db_rep->max_gap, 0);
}
return (0);
}
int
__rep_set_request(dbenv, min, max)
DB_ENV *dbenv;
db_timeout_t min, max;
{
DB_LOG *dblp;
DB_REP *db_rep;
DB_THREAD_INFO *ip;
ENV *env;
LOG *lp;
REP *rep;
env = dbenv->env;
db_rep = env->rep_handle;
ENV_NOT_CONFIGURED(
env, db_rep->region, "DB_ENV->rep_set_request", DB_INIT_REP);
if (min == 0 || max < min) {
__db_errx(env, DB_STR("3571",
"DB_ENV->rep_set_request: Invalid min or max values"));
return (EINVAL);
}
if (REP_ON(env)) {
rep = db_rep->region;
ENV_ENTER(env, ip);
REP_SYSTEM_LOCK(env);
DB_TIMEOUT_TO_TIMESPEC(min, &rep->request_gap);
DB_TIMEOUT_TO_TIMESPEC(max, &rep->max_gap);
REP_SYSTEM_UNLOCK(env);
MUTEX_LOCK(env, rep->mtx_clientdb);
dblp = env->lg_handle;
if (dblp != NULL && (lp = dblp->reginfo.primary) != NULL) {
DB_TIMEOUT_TO_TIMESPEC(min, &lp->wait_ts);
}
MUTEX_UNLOCK(env, rep->mtx_clientdb);
ENV_LEAVE(env, ip);
} else {
DB_TIMEOUT_TO_TIMESPEC(min, &db_rep->request_gap);
DB_TIMEOUT_TO_TIMESPEC(max, &db_rep->max_gap);
}
return (0);
}
int
__rep_set_transport_pp(dbenv, eid, f_send)
DB_ENV *dbenv;
int eid;
int (*f_send) __P((DB_ENV *,
const DBT *, const DBT *, const DB_LSN *, int, u_int32_t));
{
DB_REP *db_rep;
DB_THREAD_INFO *ip;
ENV *env;
int ret;
env = dbenv->env;
db_rep = env->rep_handle;
ret = 0;
ENV_NOT_CONFIGURED(
env, db_rep->region, "DB_ENV->rep_set_transport", DB_INIT_REP);
if (APP_IS_REPMGR(env)) {
__db_errx(env, DB_STR("3572",
"DB_ENV->rep_set_transport: cannot call from "
"Replication Manager application"));
return (EINVAL);
}
if (f_send == NULL) {
__db_errx(env, DB_STR("3573",
"DB_ENV->rep_set_transport: no send function specified"));
return (EINVAL);
}
if (eid < 0) {
__db_errx(env, DB_STR("3574",
"DB_ENV->rep_set_transport: eid must be greater than or equal to 0"));
return (EINVAL);
}
if ((ret = __rep_set_transport_int(env, eid, f_send)) == 0)
APP_SET_BASEAPI(env);
return (ret);
}
int
__rep_set_transport_int(env, eid, f_send)
ENV *env;
int eid;
int (*f_send) __P((DB_ENV *,
const DBT *, const DBT *, const DB_LSN *, int, u_int32_t));
{
DB_REP *db_rep;
REP *rep;
db_rep = env->rep_handle;
db_rep->send = f_send;
if (REP_ON(env)) {
rep = db_rep->region;
rep->eid = eid;
} else
db_rep->eid = eid;
return (0);
}
int
__rep_get_clockskew(dbenv, fast_clockp, slow_clockp)
DB_ENV *dbenv;
u_int32_t *fast_clockp, *slow_clockp;
{
DB_REP *db_rep;
DB_THREAD_INFO *ip;
ENV *env;
REP *rep;
env = dbenv->env;
db_rep = env->rep_handle;
ENV_NOT_CONFIGURED(
env, db_rep->region, "DB_ENV->rep_get_clockskew", DB_INIT_REP);
if (REP_ON(env)) {
rep = db_rep->region;
ENV_ENTER(env, ip);
REP_SYSTEM_LOCK(env);
*fast_clockp = rep->clock_skew;
*slow_clockp = rep->clock_base;
REP_SYSTEM_UNLOCK(env);
ENV_LEAVE(env, ip);
} else {
*fast_clockp = db_rep->clock_skew;
*slow_clockp = db_rep->clock_base;
}
return (0);
}
int
__rep_set_clockskew(dbenv, fast_clock, slow_clock)
DB_ENV *dbenv;
u_int32_t fast_clock, slow_clock;
{
DB_REP *db_rep;
DB_THREAD_INFO *ip;
ENV *env;
REP *rep;
int ret;
env = dbenv->env;
db_rep = env->rep_handle;
ret = 0;
ENV_NOT_CONFIGURED(
env, db_rep->region, "DB_ENV->rep_set_clockskew", DB_INIT_REP);
if (fast_clock == 0 || slow_clock == 0) {
if (slow_clock != 0 || fast_clock != 0) {
__db_errx(env, DB_STR("3575",
"DB_ENV->rep_set_clockskew: Zero only valid for "
"when used for both arguments"));
return (EINVAL);
}
fast_clock = 1;
slow_clock = 1;
}
if (fast_clock < slow_clock) {
__db_errx(env, DB_STR("3576",
"DB_ENV->rep_set_clockskew: slow_clock value is "
"larger than fast_clock_value"));
return (EINVAL);
}
if (REP_ON(env)) {
rep = db_rep->region;
if (IS_REP_STARTED(env)) {
__db_errx(env, DB_STR("3577",
"DB_ENV->rep_set_clockskew: must be called before DB_ENV->rep_start"));
return (EINVAL);
}
ENV_ENTER(env, ip);
REP_SYSTEM_LOCK(env);
rep->clock_skew = fast_clock;
rep->clock_base = slow_clock;
REP_SYSTEM_UNLOCK(env);
ENV_LEAVE(env, ip);
} else {
db_rep->clock_skew = fast_clock;
db_rep->clock_base = slow_clock;
}
return (ret);
}
int
__rep_flush(dbenv)
DB_ENV *dbenv;
{
DBT rec;
DB_LOGC *logc;
DB_LSN lsn;
DB_REP *db_rep;
DB_THREAD_INFO *ip;
ENV *env;
int ret, t_ret;
env = dbenv->env;
db_rep = env->rep_handle;
ENV_REQUIRES_CONFIG_XX(
env, rep_handle, "DB_ENV->rep_flush", DB_INIT_REP);
if (IS_REP_CLIENT(env))
return (0);
if (db_rep->send == NULL) {
__db_errx(env, DB_STR("3578",
"DB_ENV->rep_flush: must be called after DB_ENV->rep_set_transport"));
return (EINVAL);
}
ENV_ENTER(env, ip);
if ((ret = __log_cursor(env, &logc)) != 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;
(void)__rep_send_message(env,
DB_EID_BROADCAST, REP_LOG, &lsn, &rec, 0, 0);
err: if ((t_ret = __logc_close(logc)) != 0 && ret == 0)
ret = t_ret;
ENV_LEAVE(env, ip);
return (ret);
}
int
__rep_sync(dbenv, flags)
DB_ENV *dbenv;
u_int32_t flags;
{
DB_LOG *dblp;
DB_LSN lsn;
DB_REP *db_rep;
DB_THREAD_INFO *ip;
ENV *env;
LOG *lp;
REP *rep;
int master, ret;
u_int32_t repflags, type;
env = dbenv->env;
db_rep = env->rep_handle;
COMPQUIET(flags, 0);
ENV_REQUIRES_CONFIG_XX(
env, rep_handle, "DB_ENV->rep_sync", DB_INIT_REP);
if (db_rep->send == NULL) {
__db_errx(env, DB_STR("3579",
"DB_ENV->rep_sync: must be called after DB_ENV->rep_set_transport"));
return (EINVAL);
}
dblp = env->lg_handle;
lp = dblp->reginfo.primary;
rep = db_rep->region;
ret = 0;
ENV_ENTER(env, ip);
MUTEX_LOCK(env, rep->mtx_clientdb);
lsn = lp->verify_lsn;
MUTEX_UNLOCK(env, rep->mtx_clientdb);
REP_SYSTEM_LOCK(env);
master = rep->master_id;
if (master == DB_EID_INVALID) {
REP_SYSTEM_UNLOCK(env);
(void)__rep_send_message(env, DB_EID_BROADCAST,
REP_MASTER_REQ, NULL, NULL, 0, 0);
goto out;
}
if (!F_ISSET(rep, REP_F_DELAY)) {
REP_SYSTEM_UNLOCK(env);
goto out;
}
DB_ASSERT(env,
!IS_USING_LEASES(env) || __rep_islease_granted(env) == 0);
F_CLR(rep, REP_F_DELAY);
if (IS_ZERO_LSN(lsn) && !FLD_ISSET(rep->config, REP_C_AUTOINIT)) {
FLD_CLR(rep->lockout_flags, REP_LOCKOUT_ARCHIVE);
CLR_RECOVERY_SETTINGS(rep);
ret = DB_REP_JOIN_FAILURE;
REP_SYSTEM_UNLOCK(env);
goto out;
}
REP_SYSTEM_UNLOCK(env);
if (IS_ZERO_LSN(lsn)) {
DB_ASSERT(env, rep->sync_state == SYNC_UPDATE);
type = REP_UPDATE_REQ;
repflags = 0;
} else {
DB_ASSERT(env, rep->sync_state == SYNC_VERIFY);
type = REP_VERIFY_REQ;
repflags = DB_REP_ANYWHERE;
}
(void)__rep_send_message(env, master, type, &lsn, NULL, 0, repflags);
out: ENV_LEAVE(env, ip);
return (ret);
}
int
__rep_txn_applied(env, ip, commit_info, timeout)
ENV *env;
DB_THREAD_INFO *ip;
DB_COMMIT_INFO *commit_info;
db_timeout_t timeout;
{
REP *rep;
db_timespec limit, now, t;
db_timeout_t duration;
struct rep_waitgoal reason;
int locked, ret, t_ret;
if (commit_info->gen == 0) {
__db_errx(env, DB_STR("3580",
"non-replication commit token in replication env"));
return (EINVAL);
}
rep = env->rep_handle->region;
VPRINT(env, (env, DB_VERB_REP_MISC,
"checking txn_applied: gen %lu, envid %lu, LSN [%lu][%lu]",
(u_long)commit_info->gen, (u_long)commit_info->envid,
(u_long)commit_info->lsn.file, (u_long)commit_info->lsn.offset));
locked = 0;
__os_gettime(env, &limit, 1);
TIMESPEC_ADD_DB_TIMEOUT(&limit, timeout);
retry:
if ((ret = __op_handle_enter(env)) != 0)
goto out;
ret = __rep_check_applied(env, ip, commit_info, &reason);
t_ret = __env_db_rep_exit(env);
if (ret == DB_TIMEOUT && t_ret == 0 && F_ISSET(rep, REP_F_CLIENT)) {
__os_gettime(env, &now, 1);
if (timespeccmp(&now, &limit, <)) {
t = limit;
timespecsub(&t, &now);
DB_TIMESPEC_TO_TIMEOUT(duration, &t, 1);
if (reason.why == AWAIT_LSN ||
reason.why == AWAIT_HISTORY) {
MUTEX_LOCK(env, rep->mtx_clientdb);
locked = 1;
}
REP_SYSTEM_LOCK(env);
ret = __rep_check_goal(env, &reason);
if (locked) {
MUTEX_UNLOCK(env, rep->mtx_clientdb);
locked = 0;
}
if (ret == DB_TIMEOUT) {
ret = __rep_await_condition(env,
&reason, duration);
DB_ASSERT(env, ret != DB_TIMEOUT);
}
REP_SYSTEM_UNLOCK(env);
if (ret != 0)
goto out;
goto retry;
}
}
if (t_ret != 0 &&
(ret == 0 || ret == DB_TIMEOUT || ret == DB_NOTFOUND))
ret = t_ret;
out:
return (ret);
}
static int
__rep_await_condition(env, reasonp, duration)
ENV *env;
struct rep_waitgoal *reasonp;
db_timeout_t duration;
{
REGENV *renv;
REGINFO *infop;
REP *rep;
struct __rep_waiter *waiter;
int ret;
rep = env->rep_handle->region;
infop = env->reginfo;
renv = infop->primary;
if ((waiter = SH_TAILQ_FIRST(&rep->free_waiters,
__rep_waiter)) == NULL) {
MUTEX_LOCK(env, renv->mtx_regenv);
if ((ret = __env_alloc(env->reginfo,
sizeof(struct __rep_waiter), &waiter)) == 0) {
memset(waiter, 0, sizeof(*waiter));
if ((ret = __mutex_alloc(env, MTX_REP_WAITER,
DB_MUTEX_SELF_BLOCK, &waiter->mtx_repwait)) != 0)
__env_alloc_free(infop, waiter);
}
MUTEX_UNLOCK(env, renv->mtx_regenv);
if (ret != 0)
return (ret);
MUTEX_LOCK(env, waiter->mtx_repwait);
} else
SH_TAILQ_REMOVE(&rep->free_waiters,
waiter, links, __rep_waiter);
waiter->flags = 0;
waiter->goal = *reasonp;
SH_TAILQ_INSERT_HEAD(&rep->waiters,
waiter, links, __rep_waiter);
VPRINT(env, (env, DB_VERB_REP_MISC,
"waiting for condition %d", (int)reasonp->why));
REP_SYSTEM_UNLOCK(env);
MUTEX_WAIT(env, waiter->mtx_repwait, duration);
REP_SYSTEM_LOCK(env);
if (!F_ISSET(waiter, REP_F_WOKEN))
SH_TAILQ_REMOVE(&rep->waiters, waiter, links, __rep_waiter);
SH_TAILQ_INSERT_HEAD(&rep->free_waiters, waiter, links, __rep_waiter);
return (0);
}
static int
__rep_check_applied(env, ip, commit_info, reasonp)
ENV *env;
DB_THREAD_INFO *ip;
DB_COMMIT_INFO *commit_info;
struct rep_waitgoal *reasonp;
{
DB_LOG *dblp;
DB_REP *db_rep;
LOG *lp;
REP *rep;
DB_TXN *txn;
DBC *dbc;
__rep_lsn_hist_data_args hist, hist2;
DB_LSN lsn;
u_int32_t gen;
int ret, t_ret;
db_rep = env->rep_handle;
rep = db_rep->region;
dblp = env->lg_handle;
lp = dblp->reginfo.primary;
gen = rep->gen;
txn = NULL;
dbc = NULL;
if (F_ISSET(rep, REP_F_MASTER)) {
LOG_SYSTEM_LOCK(env);
lsn = lp->lsn;
LOG_SYSTEM_UNLOCK(env);
} else {
MUTEX_LOCK(env, rep->mtx_clientdb);
lsn = lp->max_perm_lsn;
MUTEX_UNLOCK(env, rep->mtx_clientdb);
}
if (commit_info->gen == gen) {
ret = __rep_read_lsn_history(env,
ip, &txn, &dbc, gen, &hist, reasonp, DB_SET);
if (ret == DB_NOTFOUND) {
ret = DB_TIMEOUT;
reasonp->why = AWAIT_HISTORY;
reasonp->u.lsn = lsn;
}
if (ret != 0)
goto out;
if (commit_info->envid != hist.envid) {
ret = DB_NOTFOUND;
goto out;
}
if (LOG_COMPARE(&commit_info->lsn, &lsn) > 0) {
ret = DB_TIMEOUT;
reasonp->why = AWAIT_LSN;
reasonp->u.lsn = commit_info->lsn;
goto out;
}
if (LOG_COMPARE(&commit_info->lsn, &hist.lsn) >= 0) {
ret = 0;
goto out;
}
ret = DB_NOTFOUND;
} else if (commit_info->gen < gen || gen == 0) {
ret = __rep_read_lsn_history(env,
ip, &txn, &dbc, commit_info->gen, &hist, reasonp, DB_SET);
t_ret = __rep_read_lsn_history(env,
ip, &txn, &dbc, commit_info->gen, &hist2, reasonp, DB_NEXT);
if (ret == DB_NOTFOUND) {
if (t_ret == 0)
goto out;
if (t_ret == DB_NOTFOUND) {
ret = DB_TIMEOUT;
reasonp->why = AWAIT_HISTORY;
reasonp->u.lsn = lsn;
goto out;
}
ret = t_ret;
goto out;
}
if (ret != 0)
goto out;
if (commit_info->envid != hist.envid) {
ret = DB_NOTFOUND;
goto out;
}
if ((ret = t_ret) == DB_NOTFOUND) {
ret = DB_TIMEOUT;
reasonp->why = AWAIT_HISTORY;
reasonp->u.lsn = lsn;
} else if (ret != 0)
goto out;
if (LOG_COMPARE(&commit_info->lsn, &hist.lsn) >= 0 &&
LOG_COMPARE(&commit_info->lsn, &hist2.lsn) < 0)
ret = 0;
else
ret = DB_NOTFOUND;
} else {
if (F_ISSET(rep, REP_F_CLIENT) &&
LOG_COMPARE(&commit_info->lsn, &lsn) > 0) {
reasonp->why = AWAIT_GEN;
reasonp->u.gen = commit_info->gen;
return (DB_TIMEOUT);
}
return (DB_NOTFOUND);
}
out:
if (dbc != NULL &&
(t_ret = __dbc_close(dbc)) != 0 && ret == 0)
ret = t_ret;
if (txn != NULL &&
(t_ret = __db_txn_auto_resolve(env, txn, 1, ret)) != 0 && ret == 0)
ret = t_ret;
return (ret);
}
static int
__rep_read_lsn_history(env, ip, txn, dbc, gen, gen_infop, reasonp, flags)
ENV *env;
DB_THREAD_INFO *ip;
DB_TXN **txn;
DBC **dbc;
u_int32_t gen;
__rep_lsn_hist_data_args *gen_infop;
struct rep_waitgoal *reasonp;
u_int32_t flags;
{
DB_REP *db_rep;
REP *rep;
DB *dbp;
__rep_lsn_hist_key_args key;
u_int8_t key_buf[__REP_LSN_HIST_KEY_SIZE];
u_int8_t data_buf[__REP_LSN_HIST_DATA_SIZE];
DBT key_dbt, data_dbt;
u_int32_t desired_gen;
int ret, tries;
db_rep = env->rep_handle;
rep = db_rep->region;
ret = 0;
DB_ASSERT(env, flags == DB_SET || flags == DB_NEXT);
desired_gen = flags == DB_SET ? gen : gen + 1;
REP_SYSTEM_LOCK(env);
if (rep->gen == desired_gen && !IS_ZERO_LSN(rep->gen_base_lsn)) {
gen_infop->lsn = rep->gen_base_lsn;
gen_infop->envid = rep->master_envid;
goto unlock;
}
REP_SYSTEM_UNLOCK(env);
tries = 0;
retry:
if (*txn == NULL &&
(ret = __txn_begin(env, ip, NULL, txn, 0)) != 0)
return (ret);
if ((dbp = db_rep->lsn_db) == NULL) {
if ((ret = __rep_open_sysdb(env,
ip, *txn, REPLSNHIST, 0, &dbp)) != 0) {
if (ret == ENOENT &&
!F_ISSET(rep, REP_F_NIMDBS_LOADED | REP_F_MASTER)) {
ret = DB_TIMEOUT;
reasonp->why = AWAIT_NIMDB;
}
goto err;
}
db_rep->lsn_db = dbp;
}
if (*dbc == NULL &&
(ret = __db_cursor(dbp, ip, *txn, dbc, 0)) != 0)
goto err;
if (flags == DB_SET) {
key.version = REP_LSN_HISTORY_FMT_VERSION;
key.gen = gen;
__rep_lsn_hist_key_marshal(env, &key, key_buf);
}
DB_INIT_DBT(key_dbt, key_buf, __REP_LSN_HIST_KEY_SIZE);
key_dbt.ulen = __REP_LSN_HIST_KEY_SIZE;
F_SET(&key_dbt, DB_DBT_USERMEM);
memset(&data_dbt, 0, sizeof(data_dbt));
data_dbt.data = data_buf;
data_dbt.ulen = __REP_LSN_HIST_DATA_SIZE;
F_SET(&data_dbt, DB_DBT_USERMEM);
if ((ret = __dbc_get(*dbc, &key_dbt, &data_dbt, flags)) != 0) {
if ((ret == DB_LOCK_DEADLOCK || ret == DB_LOCK_NOTGRANTED) &&
++tries < 5) {
ret = __dbc_close(*dbc);
*dbc = NULL;
if (ret != 0)
goto err;
ret = __txn_abort(*txn);
*txn = NULL;
if (ret != 0)
goto err;
__os_yield(env, 0, 10000);
goto retry;
}
goto err;
}
if ((ret = __rep_lsn_hist_key_unmarshal(env,
&key, key_buf, __REP_LSN_HIST_KEY_SIZE, NULL)) != 0)
goto err;
ret = __rep_lsn_hist_data_unmarshal(env,
gen_infop, data_buf, __REP_LSN_HIST_DATA_SIZE, NULL);
REP_SYSTEM_LOCK(env);
if (rep->gen == key.gen) {
rep->gen_base_lsn = gen_infop->lsn;
rep->master_envid = gen_infop->envid;
}
unlock:
REP_SYSTEM_UNLOCK(env);
err:
return (ret);
}
static u_int32_t
__rep_conv_vers(env, log_ver)
ENV *env;
u_int32_t log_ver;
{
COMPQUIET(env, NULL);
if (log_ver == DB_LOGVERSION_53)
return (DB_REPVERSION_53);
if (log_ver == DB_LOGVERSION_52)
return (DB_REPVERSION_52);
if (log_ver == DB_LOGVERSION_51)
return (DB_REPVERSION_51);
if (log_ver == DB_LOGVERSION_48p2)
return (DB_REPVERSION_48);
if (log_ver == DB_LOGVERSION_48)
return (DB_REPVERSION_48);
if (log_ver == DB_LOGVERSION_47)
return (DB_REPVERSION_47);
if (log_ver == DB_LOGVERSION_46)
return (DB_REPVERSION_46);
if (log_ver == DB_LOGVERSION_45)
return (DB_REPVERSION_45);
if (log_ver == DB_LOGVERSION_44)
return (DB_REPVERSION_44);
if (log_ver == DB_LOGVERSION)
return (DB_REPVERSION);
return (DB_REPVERSION_INVALID);
}