#include "db_config.h"
#include "db_int.h"
#include "dbinc/db_page.h"
#include "dbinc/btree.h"
#include "dbinc/log.h"
#include "dbinc/mp.h"
#include "dbinc/txn.h"
static int __rep_abort_prepared __P((ENV *));
static int __rep_bt_cmp __P((DB *, const DBT *, const DBT *));
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_restore_prepared __P((ENV *));
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;
#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_BULK | DB_REP_CONF_DELAYCLIENT | \
DB_REP_CONF_NOAUTOINIT | DB_REP_CONF_NOWAIT)
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;
env = dbenv->env;
db_rep = env->rep_handle;
ret = 0;
#undef OK_FLAGS
#define OK_FLAGS \
(DB_REP_CONF_BULK | DB_REP_CONF_DELAYCLIENT | DB_REP_CONF_LEASE | \
DB_REP_CONF_NOAUTOINIT | DB_REP_CONF_NOWAIT | DB_REPMGR_CONF_2SITE_STRICT)
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 (REP_ON(env)) {
ENV_ENTER(env, ip);
rep = db_rep->region;
if (FLD_ISSET(mapped, REP_C_LEASE)) {
if (F_ISSET(rep, REP_F_START_CALLED)) {
__db_errx(env,
"DB_ENV->rep_set_config: leases must be configured before DB_ENV->rep_start");
ret = EINVAL;
}
if (on == 0) {
__db_errx(env,
"DB_ENV->rep_set_config: leases cannot be turned off");
ret = EINVAL;
}
if (ret != 0)
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);
} else {
if (on)
FLD_SET(db_rep->config, mapped);
else
FLD_CLR(db_rep->config, mapped);
}
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_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_LEASE)) {
FLD_SET(*outflagsp, REP_C_LEASE);
FLD_CLR(*inflagsp, DB_REP_CONF_LEASE);
}
if (FLD_ISSET(*inflagsp, DB_REP_CONF_NOAUTOINIT)) {
FLD_SET(*outflagsp, REP_C_NOAUTOINIT);
FLD_CLR(*inflagsp, DB_REP_CONF_NOAUTOINIT);
}
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);
}
}
int
__rep_start(dbenv, dbt, flags)
DB_ENV *dbenv;
DBT *dbt;
u_int32_t flags;
{
DB *dbp;
DB_LOG *dblp;
DB_LSN lsn;
DB_REP *db_rep;
DB_THREAD_INFO *ip;
DB_TXNREGION *region;
ENV *env;
LOG *lp;
REGINFO *infop;
REP *rep;
db_timeout_t tmp;
u_int32_t oldvers, pending_event, repflags, role;
int announce, locked, ret, role_chg;
int t_ret;
env = dbenv->env;
ENV_REQUIRES_CONFIG_XX(
env, rep_handle, "DB_ENV->rep_start", DB_INIT_REP);
db_rep = env->rep_handle;
rep = db_rep->region;
infop = env->reginfo;
locked = 0;
pending_event = DB_EVENT_NO_SUCH_EVENT;
switch (role = LF_ISSET(DB_REP_CLIENT | DB_REP_MASTER)) {
case DB_REP_CLIENT:
case DB_REP_MASTER:
break;
default:
__db_errx(env,
"DB_ENV->rep_start: must specify DB_REP_CLIENT or DB_REP_MASTER");
return (EINVAL);
}
if (db_rep->send == NULL) {
__db_errx(env,
"DB_ENV->rep_set_transport must be called before DB_ENV->rep_start");
return (EINVAL);
}
if (IS_USING_LEASES(env) && rep->lease_timeout == 0) {
__db_errx(env,
"DB_ENV->rep_start: must call DB_ENV->rep_set_timeout for leases first");
return (EINVAL);
}
ENV_ENTER(env, ip);
if ((ret = __log_flush(env, NULL)) != 0)
goto out;
REP_SYSTEM_LOCK(env);
if (F_ISSET(rep, REP_F_READY_MSG)) {
RPRINT(env, DB_VERB_REP_MISC,
(env, "Thread already in lockout"));
REP_SYSTEM_UNLOCK(env);
goto out;
} else if ((ret = __rep_lockout_msg(env, rep, 0)) != 0)
goto errunlock;
role_chg = (!F_ISSET(rep, REP_F_MASTER) && role == DB_REP_MASTER) ||
(!F_ISSET(rep, REP_F_CLIENT) && role == DB_REP_CLIENT);
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) {
if ((ret = __rep_preclose(env)) != 0)
goto errunlock;
rep->gen++;
if (rep->egen > rep->gen)
rep->gen = rep->egen;
if (IS_USING_LEASES(env) &&
!F_ISSET(rep, REP_F_MASTERELECT)) {
__db_errx(env,
"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, 0);
F_CLR(rep, REP_F_MASTERELECT);
}
if (rep->egen <= rep->gen)
rep->egen = rep->gen + 1;
RPRINT(env, DB_VERB_REP_MISC, (env,
"New master gen %lu, egen %lu",
(u_long)rep->gen, (u_long)rep->egen));
if ((ret = __rep_write_gen(env, rep->gen)) != 0)
goto errunlock;
}
if (IS_USING_LEASES(env) &&
(role_chg || !F_ISSET(rep, REP_F_START_CALLED))) {
if ((ret = __rep_islease_granted(env))) {
__db_errx(env,
"rep_start: Cannot become master with outstanding lease granted.");
ret = EINVAL;
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);
INIT_LSN(lp->max_perm_lsn);
if ((ret = __rep_lease_table_alloc(env,
rep->nsites)) != 0)
goto errunlock;
}
rep->master_id = rep->eid;
repflags = F_ISSET(rep, REP_F_READY_API | REP_F_READY_MSG |
REP_F_READY_OP);
#ifdef DIAGNOSTIC
if (!F_ISSET(rep, REP_F_GROUP_ESTD))
RPRINT(env, DB_VERB_REP_MISC, (env,
"Establishing group as master."));
#endif
FLD_SET(repflags, REP_F_MASTER | REP_F_GROUP_ESTD);
rep->flags = repflags;
oldvers = lp->persist.version;
RPRINT(env, DB_VERB_REP_MISC, (env,
"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;
F_CLR(rep, REP_F_READY_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)) != 0 && ret == 0)
ret = t_ret;
}
if ((t_ret = __txn_recycle_id(env)) != 0 && ret == 0)
ret = t_ret;
REP_SYSTEM_LOCK(env);
F_CLR(rep, REP_F_READY_API | REP_F_READY_OP);
locked = 0;
REP_SYSTEM_UNLOCK(env);
(void)__memp_set_config(
dbenv, DB_MEMP_SYNC_INTERRUPT, 0);
}
} else {
announce = role_chg || rep->master_id == DB_EID_INVALID;
if (role_chg)
rep->master_id = DB_EID_INVALID;
repflags = F_ISSET(rep, REP_F_NOARCHIVE | REP_F_READY_MSG |
REP_F_RECOVER_MASK | REP_F_TALLY);
FLD_SET(repflags, REP_F_CLIENT);
if (role_chg) {
if ((ret = __log_get_oldversion(env, &oldvers)) != 0)
goto errunlock;
RPRINT(env, DB_VERB_REP_MISC, (env,
"rep_start: Found old version log %d", oldvers));
if (oldvers >= DB_LOGVERSION_MIN) {
__log_set_version(env, oldvers);
oldvers = __rep_conv_vers(env, oldvers);
DB_ASSERT(
env, oldvers != DB_REPVERSION_INVALID);
rep->version = oldvers;
}
}
rep->flags = repflags;
if (IS_USING_LEASES(env) &&
(role_chg || !F_ISSET(rep, REP_F_START_CALLED))) {
if ((ret = __rep_lease_expire(env, 1)) != 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) {
__env_alloc_free(infop,
R_ADDR(infop, rep->lease_off));
rep->lease_off = INVALID_ROFF;
}
}
REP_SYSTEM_UNLOCK(env);
if ((ret = __rep_abort_prepared(env)) != 0)
goto errlock;
if (role_chg) {
if ((ret = db_create(&dbp, dbenv, 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);
F_CLR(rep, REP_F_READY_MSG);
if (locked) {
F_CLR(rep, REP_F_READY_API | REP_F_READY_OP);
locked = 0;
}
REP_SYSTEM_UNLOCK(env);
if (announce) {
if (F_ISSET(env, ENV_PRIVATE))
rep->gen = 0;
if ((ret = __dbt_usercopy(env, dbt)) != 0)
goto out;
(void)__rep_send_message(env,
DB_EID_BROADCAST, REP_NEWCLIENT, NULL, dbt, 0, 0);
} else
(void)__rep_send_message(env,
DB_EID_BROADCAST, REP_ALIVE_REQ, NULL, NULL, 0, 0);
}
if (0) {
errlock: REP_SYSTEM_LOCK(env);
errunlock: F_CLR(rep, REP_F_READY_MSG);
if (locked)
F_CLR(rep, REP_F_READY_API | REP_F_READY_OP);
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);
__dbt_userfree(env, dbt, NULL, NULL);
ENV_LEAVE(env, ip);
return (ret);
}
int
__rep_client_dbinit(env, startup, which)
ENV *env;
int startup;
repdb_t which;
{
DB *dbp, **rdbpp;
DB_ENV *dbenv;
DB_REP *db_rep;
DB_THREAD_INFO *ip;
REP *rep;
int ret, t_ret;
u_int32_t flags;
const char *name;
dbenv = env->dbenv;
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 = &rep->file_dbp;
}
if (*rdbpp != NULL)
return (0);
ENV_GET_THREAD_INFO(env, ip);
if (startup) {
if ((ret = db_create(&dbp, dbenv, 0)) != 0)
goto err;
(void)__db_remove(dbp, ip, NULL, name, NULL, DB_FORCE);
}
if ((ret = db_create(&dbp, dbenv, 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 |
(F_ISSET(env, ENV_THREAD) ? DB_THREAD : 0);
if ((ret = __db_open(dbp, ip, NULL, name, NULL,
(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_xa_regop_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)) {
RPRINT(env, DB_VERB_REP_MISC,
(env, "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, "First record not found");
goto err;
}
if (rep->max_prep_lsn.file < lsn.file) {
RPRINT(env, DB_VERB_REP_MISC,
(env, "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,
"Checkpoint record at LSN [%lu][%lu] not found",
(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,
"Invalid checkpoint record at [%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,
"Checkpoint LSN record [%lu][%lu] not found",
(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, "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, "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_xa_regop:
if ((ret = __txn_xa_regop_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(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_set_nsites", DB_INIT_REP);
if (REP_ON(env)) {
rep = db_rep->region;
if (rep != NULL && F_ISSET(rep, REP_F_START_CALLED)) {
__db_errx(env,
"DB_ENV->rep_set_nsites: must be called before DB_ENV->rep_start");
return (EINVAL);
}
rep->config_nsites = n;
} else
db_rep->config_nsites = n;
return (0);
}
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;
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;
env = dbenv->env;
db_rep = env->rep_handle;
if (REP_ON(env)) {
rep = db_rep->region;
rep->priority = priority;
} else
db_rep->my_priority = priority;
return (0);
}
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;
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;
ENV *env;
REP *rep;
int ret;
env = dbenv->env;
db_rep = env->rep_handle;
rep = db_rep->region;
ret = 0;
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) && F_ISSET(rep, REP_F_START_CALLED)) {
ret = EINVAL;
__db_errx(env,
"DB_ENV->rep_set_timeout: lease timeout must be set before DB_ENV->rep_start.");
goto out;
}
if (REP_ON(env))
rep->lease_timeout = timeout;
else
db_rep->lease_timeout = timeout;
break;
#ifdef HAVE_REPLICATION_THREADS
case DB_REP_ACK_TIMEOUT:
db_rep->ack_timeout = timeout;
break;
case DB_REP_CONNECTION_RETRY:
db_rep->connection_retry_wait = timeout;
break;
case DB_REP_ELECTION_RETRY:
db_rep->election_retry_wait = timeout;
break;
case DB_REP_HEARTBEAT_MONITOR:
db_rep->heartbeat_monitor_timeout = timeout;
break;
case DB_REP_HEARTBEAT_SEND:
db_rep->heartbeat_frequency = timeout;
break;
#endif
default:
__db_errx(env,
"Unknown timeout type argument to DB_ENV->rep_set_timeout");
ret = EINVAL;
}
out:
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;
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 = db_rep->ack_timeout;
break;
case DB_REP_ELECTION_RETRY:
*timeout = db_rep->election_retry_wait;
break;
case DB_REP_CONNECTION_RETRY:
*timeout = db_rep->connection_retry_wait;
break;
#endif
default:
__db_errx(env,
"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_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(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;
REP *rep;
env = dbenv->env;
if (f_send == NULL) {
__db_errx(env,
"DB_ENV->rep_set_transport: no send function specified");
return (EINVAL);
}
if (eid < 0) {
__db_errx(env,
"DB_ENV->rep_set_transport: eid must be greater than or equal to 0");
return (EINVAL);
}
db_rep = env->rep_handle;
db_rep->send = f_send;
if (REP_ON(env)) {
rep = db_rep->region;
ENV_ENTER(env, ip);
REP_SYSTEM_LOCK(env);
rep->eid = eid;
REP_SYSTEM_UNLOCK(env);
ENV_LEAVE(env, ip);
} 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;
ENV *env;
REP *rep;
env = dbenv->env;
db_rep = env->rep_handle;
if (REP_ON(env)) {
rep = db_rep->region;
*fast_clockp = rep->clock_skew;
*slow_clockp = rep->clock_base;
} 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_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_ENV->rep_set_clockskew: slow_clock value is larger than fast_clock_value");
return (EINVAL);
}
if (REP_ON(env)) {
rep = db_rep->region;
if (F_ISSET(rep, REP_F_START_CALLED)) {
__db_errx(env,
"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_THREAD_INFO *ip;
ENV *env;
int ret, t_ret;
env = dbenv->env;
ENV_REQUIRES_CONFIG_XX(
env, rep_handle, "DB_ENV->rep_flush", DB_INIT_REP);
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;
COMPQUIET(flags, 0);
ENV_REQUIRES_CONFIG_XX(
env, rep_handle, "DB_ENV->rep_sync", DB_INIT_REP);
dblp = env->lg_handle;
lp = dblp->reginfo.primary;
db_rep = env->rep_handle;
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_NOAUTOINIT)) {
F_CLR(rep, REP_F_NOARCHIVE | REP_F_RECOVER_MASK);
ret = DB_REP_JOIN_FAILURE;
REP_SYSTEM_UNLOCK(env);
goto out;
}
REP_SYSTEM_UNLOCK(env);
if (IS_ZERO_LSN(lsn)) {
DB_ASSERT(env, F_ISSET(rep, REP_F_RECOVER_UPDATE));
type = REP_UPDATE_REQ;
repflags = 0;
} else {
DB_ASSERT(env, F_ISSET(rep, REP_F_RECOVER_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);
}
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_44)
return (DB_REPVERSION_44);
if (log_ver == DB_LOGVERSION_45)
return (DB_REPVERSION_45);
if (log_ver == DB_LOGVERSION_46)
return (DB_REPVERSION_46);
if (log_ver == DB_LOGVERSION_47)
return (DB_REPVERSION_47);
return (DB_REPVERSION_INVALID);
}