#include "db_config.h"
#include "db_int.h"
#include "dbinc/crypto.h"
#include "dbinc/hmac.h"
#include "dbinc/db_page.h"
#include "dbinc/hash.h"
#include "dbinc/lock.h"
#include "dbinc/mp.h"
#include "dbinc/txn.h"
#define LOG_FLAGS(txn) \
(DB_LOG_COMMIT | (F_ISSET(txn, TXN_SYNC) ? \
DB_FLUSH : (F_ISSET(txn, TXN_WRITE_NOSYNC) ? \
DB_LOG_WRNOSYNC : 0)))
typedef enum {
TXN_OP_ABORT,
TXN_OP_COMMIT,
TXN_OP_DISCARD,
TXN_OP_PREPARE
} txnop_t;
static int __txn_abort_pp __P((DB_TXN *));
static int __txn_applied __P((ENV *,
DB_THREAD_INFO *, DB_COMMIT_INFO *, db_timeout_t));
static void __txn_build_token __P((DB_TXN *, DB_LSN *));
static int __txn_begin_int __P((DB_TXN *));
static int __txn_close_cursors __P((DB_TXN *));
static int __txn_commit_pp __P((DB_TXN *, u_int32_t));
static int __txn_discard __P((DB_TXN *, u_int32_t));
static int __txn_dispatch_undo
__P((ENV *, DB_TXN *, DBT *, DB_LSN *, DB_TXNHEAD *));
static int __txn_end __P((DB_TXN *, int));
static int __txn_isvalid __P((const DB_TXN *, txnop_t));
static int __txn_undo __P((DB_TXN *));
static int __txn_set_commit_token __P((DB_TXN *txn, DB_TXN_TOKEN *));
static void __txn_set_txn_lsnp __P((DB_TXN *, DB_LSN **, DB_LSN **));
#define TxnAlloc "Unable to allocate a transaction handle"
int
__txn_begin_pp(dbenv, parent, txnpp, flags)
DB_ENV *dbenv;
DB_TXN *parent, **txnpp;
u_int32_t flags;
{
DB_THREAD_INFO *ip;
ENV *env;
int rep_check, ret;
env = dbenv->env;
ENV_REQUIRES_CONFIG(env, env->tx_handle, "txn_begin", DB_INIT_TXN);
if ((ret = __db_fchk(env,
"txn_begin", flags,
DB_IGNORE_LEASE |DB_READ_COMMITTED | DB_READ_UNCOMMITTED |
DB_TXN_FAMILY | DB_TXN_NOSYNC | DB_TXN_SNAPSHOT | DB_TXN_SYNC |
DB_TXN_WAIT | DB_TXN_WRITE_NOSYNC | DB_TXN_NOWAIT |
DB_TXN_BULK)) != 0)
return (ret);
if ((ret = __db_fcchk(env, "txn_begin", flags,
DB_TXN_WRITE_NOSYNC | DB_TXN_NOSYNC, DB_TXN_SYNC)) != 0)
return (ret);
if ((ret = __db_fcchk(env, "txn_begin",
flags, DB_TXN_WRITE_NOSYNC, DB_TXN_NOSYNC)) != 0)
return (ret);
if (parent != NULL && LF_ISSET(DB_TXN_FAMILY)) {
__db_errx(env, DB_STR("4521",
"Family transactions cannot have parents"));
return (EINVAL);
} else if (IS_REAL_TXN(parent) &&
!F_ISSET(parent, TXN_SNAPSHOT) && LF_ISSET(DB_TXN_SNAPSHOT)) {
__db_errx(env, DB_STR("4522",
"Child transaction snapshot setting must match parent"));
return (EINVAL);
}
ENV_ENTER(env, ip);
rep_check = IS_ENV_REPLICATED(env) &&
!IS_REAL_TXN(parent) && !LF_ISSET(DB_TXN_FAMILY);
if (rep_check && (ret = __op_rep_enter(env, 0, 1)) != 0)
goto err;
ret = __txn_begin(env, ip, parent, txnpp, flags);
if (ret != 0 && rep_check)
(void)__op_rep_exit(env);
err: ENV_LEAVE(env, ip);
return (ret);
}
int
__txn_begin(env, ip, parent, txnpp, flags)
ENV *env;
DB_THREAD_INFO *ip;
DB_TXN *parent, **txnpp;
u_int32_t flags;
{
DB_ENV *dbenv;
DB_LOCKREGION *region;
DB_TXN *txn;
TXN_DETAIL *ptd, *td;
int ret;
if (F_ISSET(env, ENV_FORCE_TXN_BULK))
flags |= DB_TXN_BULK;
*txnpp = NULL;
if ((ret = __os_calloc(env, 1, sizeof(DB_TXN), &txn)) != 0) {
__db_errx(env, TxnAlloc);
return (ret);
}
dbenv = env->dbenv;
txn->mgrp = env->tx_handle;
txn->parent = parent;
if (parent != NULL && F_ISSET(parent, TXN_FAMILY))
parent = NULL;
TAILQ_INIT(&txn->kids);
TAILQ_INIT(&txn->events);
STAILQ_INIT(&txn->logs);
TAILQ_INIT(&txn->my_cursors);
TAILQ_INIT(&txn->femfs);
txn->flags = TXN_MALLOC;
txn->thread_info =
ip != NULL ? ip : (parent != NULL ? parent->thread_info : NULL);
if (LF_ISSET(DB_TXN_SYNC))
F_SET(txn, TXN_SYNC);
else if (LF_ISSET(DB_TXN_NOSYNC))
F_SET(txn, TXN_NOSYNC);
else if (LF_ISSET(DB_TXN_WRITE_NOSYNC))
F_SET(txn, TXN_WRITE_NOSYNC);
else if (F_ISSET(dbenv, DB_ENV_TXN_NOSYNC))
F_SET(txn, TXN_NOSYNC);
else if (F_ISSET(dbenv, DB_ENV_TXN_WRITE_NOSYNC))
F_SET(txn, TXN_WRITE_NOSYNC);
else
F_SET(txn, TXN_SYNC);
if (LF_ISSET(DB_TXN_NOWAIT) ||
(F_ISSET(dbenv, DB_ENV_TXN_NOWAIT) && !LF_ISSET(DB_TXN_WAIT)))
F_SET(txn, TXN_NOWAIT);
if (LF_ISSET(DB_READ_COMMITTED))
F_SET(txn, TXN_READ_COMMITTED);
if (LF_ISSET(DB_READ_UNCOMMITTED))
F_SET(txn, TXN_READ_UNCOMMITTED);
if (LF_ISSET(DB_TXN_FAMILY))
F_SET(txn, TXN_FAMILY | TXN_INFAMILY | TXN_READONLY);
if (LF_ISSET(DB_TXN_SNAPSHOT) || F_ISSET(dbenv, DB_ENV_TXN_SNAPSHOT) ||
(parent != NULL && F_ISSET(parent, TXN_SNAPSHOT)))
F_SET(txn, TXN_SNAPSHOT);
if (LF_ISSET(DB_IGNORE_LEASE))
F_SET(txn, TXN_IGNORE_LEASE);
if (LF_ISSET(DB_TXN_BULK) && parent == NULL && !REP_ON(txn->mgrp->env))
F_SET(txn, TXN_BULK);
if ((ret = __txn_begin_int(txn)) != 0)
goto err;
td = txn->td;
if (parent != NULL) {
ptd = parent->td;
TAILQ_INSERT_HEAD(&parent->kids, txn, klinks);
SH_TAILQ_INSERT_HEAD(&ptd->kids, td, klinks, __txn_detail);
}
if (LOCKING_ON(env)) {
region = env->lk_handle->reginfo.primary;
if (parent != NULL) {
ret = __lock_inherit_timeout(env,
parent->locker, txn->locker);
if (ret == EINVAL) {
parent = NULL;
ret = 0;
}
if (ret != 0)
goto err;
}
if (parent == NULL && region->tx_timeout != 0)
if ((ret = __lock_set_timeout(env, txn->locker,
region->tx_timeout, DB_SET_TXN_TIMEOUT)) != 0)
goto err;
}
*txnpp = txn;
PERFMON2(env, txn, begin, txn->txnid, flags);
return (0);
err:
__os_free(env, txn);
return (ret);
}
int
__txn_recycle_id(env, locked)
ENV *env;
int locked;
{
DB_LSN null_lsn;
DB_TXNMGR *mgr;
DB_TXNREGION *region;
TXN_DETAIL *td;
u_int32_t *ids;
int nids, ret;
mgr = env->tx_handle;
region = mgr->reginfo.primary;
if ((ret = __os_malloc(env,
sizeof(u_int32_t) * region->curtxns, &ids)) != 0) {
__db_errx(env, DB_STR("4523",
"Unable to allocate transaction recycle buffer"));
return (ret);
}
nids = 0;
SH_TAILQ_FOREACH(td, ®ion->active_txn, links, __txn_detail)
ids[nids++] = td->txnid;
region->last_txnid = TXN_MINIMUM - 1;
region->cur_maxid = TXN_MAXIMUM;
if (nids != 0)
__db_idspace(ids, nids,
®ion->last_txnid, ®ion->cur_maxid);
__os_free(env, ids);
if (LOGGING_ON(env)) {
if (locked)
TXN_SYSTEM_UNLOCK(env);
ret = __txn_recycle_log(env, NULL, &null_lsn,
0, region->last_txnid + 1, region->cur_maxid);
if (locked && ret != 0)
TXN_SYSTEM_LOCK(env);
}
return (ret);
}
static int
__txn_begin_int(txn)
DB_TXN *txn;
{
DB_ENV *dbenv;
DB_TXNMGR *mgr;
DB_TXNREGION *region;
ENV *env;
TXN_DETAIL *td;
u_int32_t id;
int inserted, ret;
mgr = txn->mgrp;
env = mgr->env;
dbenv = env->dbenv;
region = mgr->reginfo.primary;
td = NULL;
inserted = 0;
TXN_SYSTEM_LOCK(env);
if (!F_ISSET(txn, TXN_COMPENSATE) && F_ISSET(region, TXN_IN_RECOVERY)) {
__db_errx(env, DB_STR("4524",
"operation not permitted during recovery"));
ret = EINVAL;
goto err;
}
if (region->last_txnid == TXN_MAXIMUM &&
region->cur_maxid != TXN_MAXIMUM)
region->last_txnid = TXN_MINIMUM - 1;
if ((ret =
__env_alloc(&mgr->reginfo, sizeof(TXN_DETAIL), &td)) != 0) {
__db_errx(env, DB_STR("4525",
"Unable to allocate memory for transaction detail"));
goto err;
}
id = ++region->last_txnid;
#ifdef HAVE_STATISTICS
STAT_INC(env, txn, nbegins, region->stat.st_nbegins, id);
STAT_INC(env, txn, nactive, region->stat.st_nactive, id);
if (region->stat.st_nactive > region->stat.st_maxnactive)
STAT_SET(env, txn, maxnactive,
region->stat.st_maxnactive, region->stat.st_nactive, id);
#endif
td->txnid = id;
dbenv->thread_id(dbenv, &td->pid, &td->tid);
ZERO_LSN(td->last_lsn);
ZERO_LSN(td->begin_lsn);
SH_TAILQ_INIT(&td->kids);
if (txn->parent != NULL && !F_ISSET(txn->parent, TXN_FAMILY))
td->parent = R_OFFSET(&mgr->reginfo, txn->parent->td);
else
td->parent = INVALID_ROFF;
td->name = INVALID_ROFF;
MAX_LSN(td->read_lsn);
MAX_LSN(td->visible_lsn);
td->mvcc_ref = 0;
td->mvcc_mtx = MUTEX_INVALID;
td->status = TXN_RUNNING;
td->flags = F_ISSET(txn, TXN_NOWAIT) ? TXN_DTL_NOWAIT : 0;
td->nlog_dbs = 0;
td->nlog_slots = TXN_NSLOTS;
td->log_dbs = R_OFFSET(&mgr->reginfo, td->slots);
td->xa_ref = 1;
td->xa_br_status = TXN_XA_IDLE;
SH_TAILQ_INSERT_HEAD(®ion->active_txn, td, links, __txn_detail);
region->curtxns++;
if (F_ISSET(txn, TXN_BULK))
((DB_TXNREGION *)env->tx_handle->reginfo.primary)->n_bulk_txn++;
inserted = 1;
if (region->last_txnid == region->cur_maxid) {
if ((ret = __txn_recycle_id(env, 1)) != 0)
goto err;
} else
TXN_SYSTEM_UNLOCK(env);
txn->txnid = id;
txn->td = td;
if (LOCKING_ON(env) && (ret =
__lock_getlocker(env->lk_handle, id, 1, &txn->locker)) != 0)
goto err;
txn->abort = __txn_abort_pp;
txn->commit = __txn_commit_pp;
txn->discard = __txn_discard;
txn->get_name = __txn_get_name;
txn->get_priority = __txn_get_priority;
txn->id = __txn_id;
txn->prepare = __txn_prepare;
txn->set_commit_token = __txn_set_commit_token;
txn->set_txn_lsnp = __txn_set_txn_lsnp;
txn->set_name = __txn_set_name;
txn->set_priority = __txn_set_priority;
txn->set_timeout = __txn_set_timeout;
if (LOCKING_ON(env) && (ret = __txn_set_priority(txn,
txn->parent == NULL ?
TXN_PRIORITY_DEFAULT : txn->parent->locker->priority)) != 0)
goto err;
else
td->priority = 0;
if (txn->parent != NULL) {
if (LOCKING_ON(env) && (ret = __lock_addfamilylocker(env,
txn->parent->txnid, txn->txnid,
F_ISSET(txn->parent, TXN_FAMILY))) != 0)
goto err;
if (F_ISSET(txn->parent, TXN_FAMILY)) {
txn->parent = NULL;
F_SET(txn, TXN_INFAMILY);
}
}
if (F_ISSET(txn, TXN_MALLOC)) {
MUTEX_LOCK(env, mgr->mutex);
TAILQ_INSERT_TAIL(&mgr->txn_chain, txn, links);
MUTEX_UNLOCK(env, mgr->mutex);
}
return (0);
err: if (inserted) {
TXN_SYSTEM_LOCK(env);
SH_TAILQ_REMOVE(®ion->active_txn, td, links, __txn_detail);
region->curtxns--;
if (F_ISSET(txn, TXN_BULK))
((DB_TXNREGION *)
env->tx_handle->reginfo.primary)->n_bulk_txn--;
}
if (td != NULL)
__env_alloc_free(&mgr->reginfo, td);
TXN_SYSTEM_UNLOCK(env);
return (ret);
}
int
__txn_continue(env, txn, td, ip, add_to_list)
ENV *env;
DB_TXN *txn;
TXN_DETAIL *td;
DB_THREAD_INFO *ip;
int add_to_list;
{
DB_LOCKREGION *region;
DB_TXNMGR *mgr;
int ret;
ret = 0;
mgr = txn->mgrp = env->tx_handle;
txn->parent = NULL;
txn->thread_info = ip;
txn->txnid = td->txnid;
txn->name = NULL;
txn->td = td;
td->xa_ref++;
txn->txn_list = NULL;
TAILQ_INIT(&txn->kids);
TAILQ_INIT(&txn->events);
STAILQ_INIT(&txn->logs);
TAILQ_INIT(&txn->my_cursors);
TAILQ_INIT(&txn->femfs);
if (add_to_list) {
MUTEX_LOCK(env, mgr->mutex);
TAILQ_INSERT_TAIL(&mgr->txn_chain, txn, links);
MUTEX_UNLOCK(env, mgr->mutex);
}
txn->token_buffer = 0;
txn->cursors = 0;
txn->abort = __txn_abort_pp;
txn->commit = __txn_commit_pp;
txn->discard = __txn_discard;
txn->get_name = __txn_get_name;
txn->get_priority = __txn_get_priority;
txn->id = __txn_id;
txn->prepare = __txn_prepare;
txn->set_commit_token = __txn_set_commit_token;
txn->set_name = __txn_set_name;
txn->set_priority = __txn_set_priority;
txn->set_timeout = __txn_set_timeout;
txn->set_txn_lsnp = __txn_set_txn_lsnp;
txn->flags = TXN_MALLOC |
(F_ISSET(td, TXN_DTL_NOWAIT) ? TXN_NOWAIT : 0);
txn->xa_thr_status = TXN_XA_THREAD_NOTA;
if (F_ISSET(td, TXN_DTL_RESTORED))
F_SET(txn, TXN_RESTORED);
else
if ((ret = __lock_getlocker(env->lk_handle,
txn->txnid, 0, &txn->locker)) == 0)
ret = __txn_set_priority(txn, td->priority);
if (LOCKING_ON(env)) {
region = env->lk_handle->reginfo.primary;
if (region->tx_timeout != 0 &&
(ret = __lock_set_timeout(env, txn->locker,
region->tx_timeout, DB_SET_TXN_TIMEOUT)) != 0)
return (ret);
txn->lock_timeout = region->tx_timeout;
}
return (ret);
}
static int
__txn_commit_pp(txn, flags)
DB_TXN *txn;
u_int32_t flags;
{
DB_THREAD_INFO *ip;
ENV *env;
int rep_check, ret, t_ret;
env = txn->mgrp->env;
rep_check = IS_ENV_REPLICATED(env) &&
txn->parent == NULL && IS_REAL_TXN(txn);
ENV_ENTER(env, ip);
ret = __txn_commit(txn, flags);
if (rep_check && (t_ret = __op_rep_exit(env)) != 0 && ret == 0)
ret = t_ret;
ENV_LEAVE(env, ip);
return (ret);
}
int
__txn_commit(txn, flags)
DB_TXN *txn;
u_int32_t flags;
{
DBT list_dbt;
DB_LOCKREQ request;
DB_TXN *kid;
ENV *env;
REGENV *renv;
REGINFO *infop;
TXN_DETAIL *td;
DB_LSN token_lsn;
u_int32_t id;
int ret, t_ret;
env = txn->mgrp->env;
td = txn->td;
PERFMON2(env, txn, commit, txn->txnid, flags);
DB_ASSERT(env, txn->xa_thr_status == TXN_XA_THREAD_NOTA ||
td->xa_ref == 1);
if (F_ISSET(txn, TXN_DEADLOCK)) {
ret = __db_txn_deadlock_err(env, txn);
goto err;
}
if ((ret = __txn_close_cursors(txn)) != 0)
goto err;
if ((ret = __txn_isvalid(txn, TXN_OP_COMMIT)) != 0)
return (ret);
if (txn->parent == NULL && IS_REP_MASTER(env) &&
IS_USING_LEASES(env) && !F_ISSET(txn, TXN_IGNORE_LEASE) &&
!IS_ZERO_LSN(td->last_lsn) &&
(ret = __rep_lease_check(env, 1)) != 0) {
DB_ASSERT(env, ret != DB_NOTFOUND);
goto err;
}
infop = env->reginfo;
renv = infop->primary;
id = renv->envid;
if (__db_fchk(env, "DB_TXN->commit", flags,
DB_TXN_NOSYNC | DB_TXN_SYNC | DB_TXN_WRITE_NOSYNC) != 0)
flags = DB_TXN_SYNC;
if (__db_fcchk(env, "DB_TXN->commit", flags,
DB_TXN_SYNC, DB_TXN_NOSYNC | DB_TXN_WRITE_NOSYNC) != 0)
flags = DB_TXN_SYNC;
if (LF_ISSET(DB_TXN_WRITE_NOSYNC)) {
F_CLR(txn, TXN_SYNC_FLAGS);
F_SET(txn, TXN_WRITE_NOSYNC);
}
if (LF_ISSET(DB_TXN_NOSYNC)) {
F_CLR(txn, TXN_SYNC_FLAGS);
F_SET(txn, TXN_NOSYNC);
}
if (LF_ISSET(DB_TXN_SYNC)) {
F_CLR(txn, TXN_SYNC_FLAGS);
F_SET(txn, TXN_SYNC);
}
DB_ASSERT(env, F_ISSET(txn, TXN_SYNC_FLAGS));
while ((kid = TAILQ_FIRST(&txn->kids)) != NULL)
if ((ret = __txn_commit(kid, flags)) != 0)
while ((kid = TAILQ_FIRST(&txn->kids)) != NULL)
if ((t_ret = __txn_abort(kid)) != 0)
return (__env_panic(env, t_ret));
ZERO_LSN(token_lsn);
if (DBENV_LOGGING(env) && (!IS_ZERO_LSN(td->last_lsn) ||
STAILQ_FIRST(&txn->logs) != NULL)) {
if (txn->parent == NULL) {
if ((ret =
__txn_doevents(env, txn, TXN_COMMIT, 1)) != 0)
goto err;
memset(&request, 0, sizeof(request));
if (LOCKING_ON(env)) {
request.op = DB_LOCK_PUT_READ;
if (IS_REP_MASTER(env) &&
!IS_ZERO_LSN(td->last_lsn)) {
memset(&list_dbt, 0, sizeof(list_dbt));
request.obj = &list_dbt;
}
ret = __lock_vec(env,
txn->locker, 0, &request, 1, NULL);
}
if (ret == 0 && !IS_ZERO_LSN(td->last_lsn)) {
ret = __txn_flush_fe_files(txn);
if (ret == 0)
ret = __txn_regop_log(env, txn,
&td->visible_lsn, LOG_FLAGS(txn),
TXN_COMMIT,
(int32_t)time(NULL), id,
request.obj);
if (ret == 0)
token_lsn = td->last_lsn =
td->visible_lsn;
#ifdef DIAGNOSTIC
if (ret == 0) {
DB_LSN s_lsn;
DB_ASSERT(env, __log_current_lsn_int(
env, &s_lsn, NULL, NULL) == 0);
DB_ASSERT(env, LOG_COMPARE(
&td->visible_lsn, &s_lsn) <= 0);
COMPQUIET(s_lsn.file, 0);
}
#endif
}
if (request.obj != NULL && request.obj->data != NULL)
__os_free(env, request.obj->data);
if (ret != 0)
goto err;
} else {
if (!IS_ZERO_LSN(td->last_lsn) &&
(ret = __txn_child_log(env, txn->parent,
&((TXN_DETAIL *)txn->parent->td)->last_lsn,
0, txn->txnid, &td->last_lsn)) != 0) {
goto err;
}
if (STAILQ_FIRST(&txn->logs) != NULL) {
STAILQ_CONCAT(&txn->logs, &txn->parent->logs);
txn->parent->logs = txn->logs;
STAILQ_INIT(&txn->logs);
}
F_SET(txn->parent, TXN_CHILDCOMMIT);
}
}
if (txn->token_buffer != NULL && ret == 0 && DBENV_LOGGING(env))
__txn_build_token(txn, &token_lsn);
if (txn->txn_list != NULL) {
__db_txnlist_end(env, txn->txn_list);
txn->txn_list = NULL;
}
if (ret != 0)
goto err;
if (txn->parent == NULL && IS_REP_MASTER(env) &&
IS_USING_LEASES(env) && !F_ISSET(txn, TXN_IGNORE_LEASE) &&
!IS_ZERO_LSN(td->last_lsn) &&
(ret = __rep_lease_check(env, 1)) != 0)
return (__env_panic(env, ret));
__txn_reset_fe_watermarks(txn);
return (__txn_end(txn, 1));
err:
if (td->status == TXN_PREPARED)
return (__env_panic(env, ret));
if ((t_ret = __txn_abort(txn)) != 0)
ret = t_ret;
return (ret);
}
static int
__txn_close_cursors(txn)
DB_TXN *txn;
{
int ret, tret;
DBC *dbc;
ret = tret = 0;
dbc = NULL;
if (txn == NULL)
return (0);
while ((dbc = TAILQ_FIRST(&txn->my_cursors)) != NULL) {
DB_ASSERT(dbc->env, txn == dbc->txn);
TAILQ_REMOVE(&(txn->my_cursors), dbc, txn_cursors);
dbc->txn_cursors.tqe_next = NULL;
dbc->txn_cursors.tqe_prev = NULL;
if (F_ISSET(dbc, DBC_ACTIVE))
ret = __dbc_close(dbc);
dbc->txn = NULL;
if (ret != 0) {
__db_err(dbc->env, ret, "__dbc_close");
if (tret == 0)
tret = ret;
}
}
txn->my_cursors.tqh_first = NULL;
txn->my_cursors.tqh_last = NULL;
return (tret);
}
static int
__txn_set_commit_token(txn, tokenp)
DB_TXN *txn;
DB_TXN_TOKEN *tokenp;
{
ENV *env;
env = txn->mgrp->env;
ENV_REQUIRES_CONFIG(env,
env->lg_handle, "DB_TXN->set_commit_token", DB_INIT_LOG);
if (txn->parent != NULL) {
__db_errx(env, DB_STR("4526",
"commit token unavailable for nested txn"));
return (EINVAL);
}
if (IS_REP_CLIENT(env)) {
__db_errx(env, DB_STR("4527",
"may not be called on a replication client"));
return (EINVAL);
}
txn->token_buffer = tokenp;
#ifdef DIAGNOSTIC
memset(tokenp, 0, DB_TXN_TOKEN_SIZE);
#endif
return (0);
}
static void
__txn_build_token(txn, lsnp)
DB_TXN *txn;
DB_LSN *lsnp;
{
ENV *env;
REGENV *renv;
u_int8_t *bp;
u_int32_t gen, version;
bp = txn->token_buffer->buf;
env = txn->mgrp->env;
renv = env->reginfo->primary;
version = REP_COMMIT_TOKEN_FMT_VERSION;
gen = REP_ON(env) ? env->rep_handle->region->gen : 0;
DB_HTONL_COPYOUT(env, bp, version);
DB_HTONL_COPYOUT(env, bp, gen);
DB_HTONL_COPYOUT(env, bp, renv->envid);
DB_HTONL_COPYOUT(env, bp, lsnp->file);
DB_HTONL_COPYOUT(env, bp, lsnp->offset);
}
static int
__txn_abort_pp(txn)
DB_TXN *txn;
{
DB_THREAD_INFO *ip;
ENV *env;
int rep_check, ret, t_ret;
env = txn->mgrp->env;
rep_check = IS_ENV_REPLICATED(env) &&
txn->parent == NULL && IS_REAL_TXN(txn);
ENV_ENTER(env, ip);
ret = __txn_abort(txn);
if (rep_check && (t_ret = __op_rep_exit(env)) != 0 && ret == 0)
ret = t_ret;
ENV_LEAVE(env, ip);
return (ret);
}
int
__txn_abort(txn)
DB_TXN *txn;
{
DB_LOCKREQ request;
DB_TXN *kid;
ENV *env;
REGENV *renv;
REGINFO *infop;
TXN_DETAIL *td;
u_int32_t id;
int ret;
env = txn->mgrp->env;
td = txn->td;
if (txn->xa_thr_status != TXN_XA_THREAD_NOTA && td->xa_ref > 1) {
td->status = TXN_NEED_ABORT;
return (0);
}
PERFMON1(env, txn, abort, txn->txnid);
if ((ret = __txn_close_cursors(txn)) != 0)
return (__env_panic(env, ret));
if ((ret = __txn_isvalid(txn, TXN_OP_ABORT)) != 0)
return (__env_panic(env, ret));
__txn_reset_fe_watermarks(txn);
while ((kid = TAILQ_FIRST(&txn->kids)) != NULL)
if ((ret = __txn_abort(kid)) != 0)
return (ret);
infop = env->reginfo;
renv = infop->primary;
id = renv->envid;
if (IS_ZERO_LSN(td->last_lsn) && STAILQ_FIRST(&txn->logs) == NULL) {
if (txn->txn_list == NULL)
goto done;
else
goto undo;
}
if (LOCKING_ON(env)) {
if (txn->locker == NULL &&
(ret = __lock_getlocker(env->lk_handle,
txn->txnid, 1, &txn->locker)) != 0)
return (__env_panic(env, ret));
if ((ret = __txn_doevents(env, txn, TXN_ABORT, 1)) != 0)
return (__env_panic(env, ret));
if ((ret = __lock_set_timeout(env,
txn->locker, 0, DB_SET_TXN_TIMEOUT)) != 0)
return (__env_panic(env, ret));
if ((ret = __lock_set_timeout(env,
txn->locker, 0, DB_SET_LOCK_TIMEOUT)) != 0)
return (__env_panic(env, ret));
request.op = DB_LOCK_UPGRADE_WRITE;
request.obj = NULL;
if ((ret = __lock_vec(
env, txn->locker, 0, &request, 1, NULL)) != 0)
return (__env_panic(env, ret));
}
undo: if ((ret = __txn_undo(txn)) != 0)
return (__env_panic(env, ret));
done: if (DBENV_LOGGING(env) && td->status == TXN_PREPARED &&
(ret = __txn_regop_log(env, txn, &td->last_lsn,
LOG_FLAGS(txn), TXN_ABORT, (int32_t)time(NULL), id, NULL)) != 0)
return (__env_panic(env, ret));
return (__txn_end(txn, 0));
}
static int
__txn_discard(txn, flags)
DB_TXN *txn;
u_int32_t flags;
{
DB_THREAD_INFO *ip;
ENV *env;
int rep_check, ret, t_ret;
env = txn->mgrp->env;
rep_check = IS_ENV_REPLICATED(env) &&
txn->parent == NULL && IS_REAL_TXN(txn);
ENV_ENTER(env, ip);
ret = __txn_discard_int(txn, flags);
if (rep_check && (t_ret = __op_rep_exit(env)) != 0 && ret == 0)
ret = t_ret;
ENV_LEAVE(env, ip);
return (ret);
}
int
__txn_discard_int(txn, flags)
DB_TXN *txn;
u_int32_t flags;
{
DB_TXNMGR *mgr;
ENV *env;
int ret;
COMPQUIET(flags, 0);
mgr = txn->mgrp;
env = mgr->env;
if ((ret = __txn_close_cursors(txn)) != 0)
return (ret);
if ((ret = __txn_isvalid(txn, TXN_OP_DISCARD)) != 0)
return (ret);
DB_ASSERT(env, TAILQ_FIRST(&txn->kids) == NULL);
MUTEX_LOCK(env, mgr->mutex);
mgr->n_discards++;
if (F_ISSET(txn, TXN_MALLOC)) {
TAILQ_REMOVE(&mgr->txn_chain, txn, links);
}
MUTEX_UNLOCK(env, mgr->mutex);
if (F_ISSET(txn, TXN_MALLOC) &&
txn->xa_thr_status != TXN_XA_THREAD_ASSOCIATED)
__os_free(env, txn);
return (0);
}
int
__txn_prepare(txn, gid)
DB_TXN *txn;
u_int8_t *gid;
{
DBT list_dbt, gid_dbt;
DB_LOCKREQ request;
DB_THREAD_INFO *ip;
DB_TXN *kid;
ENV *env;
TXN_DETAIL *td;
u_int32_t lflags;
int ret;
env = txn->mgrp->env;
td = txn->td;
PERFMON2(env, txn, prepare, txn->txnid, gid);
DB_ASSERT(env, txn->xa_thr_status == TXN_XA_THREAD_NOTA ||
td->xa_ref == 1);
ENV_ENTER(env, ip);
if ((ret = __txn_close_cursors(txn)) != 0)
goto err;
if ((ret = __txn_isvalid(txn, TXN_OP_PREPARE)) != 0)
goto err;
if (F_ISSET(txn, TXN_DEADLOCK)) {
ret = __db_txn_deadlock_err(env, txn);
goto err;
}
while ((kid = TAILQ_FIRST(&txn->kids)) != NULL)
if ((ret = __txn_commit(kid, DB_TXN_NOSYNC)) != 0)
goto err;
memcpy(td->gid, gid, DB_GID_SIZE);
if ((ret = __txn_doevents(env, txn, TXN_PREPARE, 1)) != 0)
goto err;
memset(&request, 0, sizeof(request));
if (LOCKING_ON(env)) {
request.op = DB_LOCK_PUT_READ;
if (!IS_ZERO_LSN(td->last_lsn)) {
memset(&list_dbt, 0, sizeof(list_dbt));
request.obj = &list_dbt;
}
if ((ret = __lock_vec(env,
txn->locker, 0, &request, 1, NULL)) != 0)
goto err;
}
if (DBENV_LOGGING(env)) {
memset(&gid_dbt, 0, sizeof(gid));
gid_dbt.data = gid;
gid_dbt.size = DB_GID_SIZE;
lflags = DB_LOG_COMMIT | DB_FLUSH;
if ((ret = __txn_prepare_log(env,
txn, &td->last_lsn, lflags, TXN_PREPARE,
&gid_dbt, &td->begin_lsn, request.obj)) != 0)
__db_err(env, ret, DB_STR("4528",
"DB_TXN->prepare: log_write failed"));
if (request.obj != NULL && request.obj->data != NULL)
__os_free(env, request.obj->data);
if (ret != 0)
goto err;
}
MUTEX_LOCK(env, txn->mgrp->mutex);
td->status = TXN_PREPARED;
MUTEX_UNLOCK(env, txn->mgrp->mutex);
err: ENV_LEAVE(env, ip);
return (ret);
}
u_int32_t
__txn_id(txn)
DB_TXN *txn;
{
return (txn->txnid);
}
int
__txn_get_name(txn, namep)
DB_TXN *txn;
const char **namep;
{
*namep = txn->name;
return (0);
}
int
__txn_set_name(txn, name)
DB_TXN *txn;
const char *name;
{
DB_THREAD_INFO *ip;
DB_TXNMGR *mgr;
ENV *env;
TXN_DETAIL *td;
size_t len;
int ret;
char *p;
mgr = txn->mgrp;
env = mgr->env;
td = txn->td;
len = strlen(name) + 1;
if ((ret = __os_realloc(env, len, &txn->name)) != 0)
return (ret);
memcpy(txn->name, name, len);
ENV_ENTER(env, ip);
TXN_SYSTEM_LOCK(env);
if (td->name != INVALID_ROFF) {
__env_alloc_free(
&mgr->reginfo, R_ADDR(&mgr->reginfo, td->name));
td->name = INVALID_ROFF;
}
if ((ret = __env_alloc(&mgr->reginfo, len, &p)) != 0) {
TXN_SYSTEM_UNLOCK(env);
__db_errx(env, DB_STR("4529",
"Unable to allocate memory for transaction name"));
__os_free(env, txn->name);
txn->name = NULL;
ENV_LEAVE(env, ip);
return (ret);
}
TXN_SYSTEM_UNLOCK(env);
td->name = R_OFFSET(&mgr->reginfo, p);
memcpy(p, name, len);
#ifdef DIAGNOSTIC
if (DBENV_LOGGING(env))
(void)__log_printf(env, txn, "transaction %#lx named %s",
(u_long)txn->txnid, name);
#endif
ENV_LEAVE(env, ip);
return (0);
}
int
__txn_get_priority(txn, priorityp)
DB_TXN *txn;
u_int32_t *priorityp;
{
if (txn->locker == NULL)
return EINVAL;
*priorityp = txn->locker->priority;
return (0);
}
int
__txn_set_priority(txn, priority)
DB_TXN *txn;
u_int32_t priority;
{
if (txn->locker == NULL)
return EINVAL;
txn->locker->priority = priority;
((TXN_DETAIL *)txn->td)->priority = priority;
return (0);
}
int
__txn_set_timeout(txn, timeout, op)
DB_TXN *txn;
db_timeout_t timeout;
u_int32_t op;
{
DB_THREAD_INFO *ip;
ENV *env;
int ret;
env = txn->mgrp->env;
if (op != DB_SET_TXN_TIMEOUT && op != DB_SET_LOCK_TIMEOUT)
return (__db_ferr(env, "DB_TXN->set_timeout", 0));
ENV_ENTER(env, ip);
ret = __lock_set_timeout( env, txn->locker, timeout, op);
ENV_LEAVE(txn->mgrp->env, ip);
return (ret);
}
static int
__txn_isvalid(txn, op)
const DB_TXN *txn;
txnop_t op;
{
DB_TXNMGR *mgr;
DB_TXNREGION *region;
ENV *env;
TXN_DETAIL *td;
mgr = txn->mgrp;
env = mgr->env;
region = mgr->reginfo.primary;
if (!F_ISSET(txn, TXN_COMPENSATE) &&
F_ISSET(region, TXN_IN_RECOVERY)) {
__db_errx(env, DB_STR("4530",
"operation not permitted during recovery"));
goto err;
}
if (txn->cursors != 0) {
__db_errx(env, DB_STR("4531",
"transaction has active cursors"));
goto err;
}
td = txn->td;
switch (op) {
case TXN_OP_DISCARD:
if (txn->txnid != td->txnid)
return (0);
if (td->status != TXN_PREPARED &&
!F_ISSET(td, TXN_DTL_RESTORED)) {
__db_errx(env, DB_STR("4532",
"not a restored transaction"));
return (__env_panic(env, EINVAL));
}
return (0);
case TXN_OP_PREPARE:
if (txn->parent != NULL) {
__db_errx(env, DB_STR("4533",
"Prepare disallowed on child transactions"));
return (EINVAL);
}
break;
case TXN_OP_ABORT:
case TXN_OP_COMMIT:
default:
break;
}
switch (td->status) {
case TXN_PREPARED:
if (op == TXN_OP_PREPARE) {
__db_errx(env, DB_STR("4534",
"transaction already prepared"));
return (EINVAL);
}
break;
case TXN_RUNNING:
case TXN_NEED_ABORT:
break;
case TXN_ABORTED:
case TXN_COMMITTED:
default:
__db_errx(env, DB_STR_A("4535",
"transaction already %s", "%s"),
td->status == TXN_COMMITTED ?
DB_STR_P("committed") : DB_STR_P("aborted"));
goto err;
}
return (0);
err:
return (__env_panic(env, EINVAL));
}
static int
__txn_end(txn, is_commit)
DB_TXN *txn;
int is_commit;
{
DB_LOCKREQ request;
DB_TXNLOGREC *lr;
DB_TXNMGR *mgr;
DB_TXNREGION *region;
ENV *env;
TXN_DETAIL *ptd, *td;
db_mutex_t mvcc_mtx;
int do_closefiles, ret;
mgr = txn->mgrp;
env = mgr->env;
region = mgr->reginfo.primary;
do_closefiles = 0;
if ((ret = __txn_doevents(env,
txn, is_commit ? TXN_COMMIT : TXN_ABORT, 0)) != 0)
return (__env_panic(env, ret));
td = txn->td;
if (td->nlog_dbs != 0 &&
(ret = __txn_dref_fname(env, txn)) != 0 && ret != EIO)
return (__env_panic(env, ret));
if (td->mvcc_ref != 0 && IS_MAX_LSN(td->visible_lsn)) {
if ((ret = __log_current_lsn_int(env, &td->visible_lsn,
NULL, NULL)) != 0)
return (__env_panic(env, ret));
}
if (LOCKING_ON(env)) {
if (txn->locker == NULL &&
(ret = __lock_getlocker(env->lk_handle,
txn->txnid, 1, &txn->locker)) != 0)
return (__env_panic(env, ret));
request.op = txn->parent == NULL ||
is_commit == 0 ? DB_LOCK_PUT_ALL : DB_LOCK_INHERIT;
request.obj = NULL;
if ((ret = __lock_vec(env,
txn->locker, 0, &request, 1, NULL)) != 0)
return (__env_panic(env, ret));
}
TXN_SYSTEM_LOCK(env);
td->status = is_commit ? TXN_COMMITTED : TXN_ABORTED;
SH_TAILQ_REMOVE(®ion->active_txn, td, links, __txn_detail);
region->curtxns--;
if (F_ISSET(td, TXN_DTL_RESTORED)) {
region->stat.st_nrestores--;
do_closefiles = region->stat.st_nrestores == 0;
}
if (td->name != INVALID_ROFF) {
__env_alloc_free(&mgr->reginfo,
R_ADDR(&mgr->reginfo, td->name));
td->name = INVALID_ROFF;
}
if (td->nlog_slots != TXN_NSLOTS)
__env_alloc_free(&mgr->reginfo,
R_ADDR(&mgr->reginfo, td->log_dbs));
if (txn->parent != NULL) {
ptd = txn->parent->td;
SH_TAILQ_REMOVE(&ptd->kids, td, klinks, __txn_detail);
} else if ((mvcc_mtx = td->mvcc_mtx) != MUTEX_INVALID) {
MUTEX_LOCK(env, mvcc_mtx);
if (td->mvcc_ref != 0) {
SH_TAILQ_INSERT_HEAD(®ion->mvcc_txn,
td, links, __txn_detail);
F_SET(td, TXN_DTL_SNAPSHOT);
#ifdef HAVE_STATISTICS
STAT_INC(env, txn,
nsnapshot, region->stat.st_nsnapshot, txn->txnid);
if (region->stat.st_nsnapshot >
region->stat.st_maxnsnapshot)
STAT_SET(env, txn, maxnsnapshot,
region->stat.st_maxnsnapshot,
region->stat.st_nsnapshot,
txn->txnid);
#endif
td = NULL;
}
MUTEX_UNLOCK(env, mvcc_mtx);
if (td != NULL)
if ((ret = __mutex_free(env, &td->mvcc_mtx)) != 0)
return (__env_panic(env, ret));
}
if (td != NULL)
__env_alloc_free(&mgr->reginfo, td);
#ifdef HAVE_STATISTICS
if (is_commit)
STAT_INC(env,
txn, ncommits, region->stat.st_ncommits, txn->txnid);
else
STAT_INC(env,
txn, naborts, region->stat.st_naborts, txn->txnid);
STAT_DEC(env, txn, nactive, region->stat.st_nactive, txn->txnid);
#endif
if (F_ISSET(txn, TXN_BULK))
((DB_TXNREGION *)env->tx_handle->reginfo.primary)->n_bulk_txn--;
TXN_SYSTEM_UNLOCK(env);
if (LOCKING_ON(env) && (ret =
__lock_freelocker(env->lk_handle, txn->locker)) != 0)
return (__env_panic(env, ret));
if (txn->parent != NULL)
TAILQ_REMOVE(&txn->parent->kids, txn, klinks);
while ((lr = STAILQ_FIRST(&txn->logs)) != NULL) {
STAILQ_REMOVE(&txn->logs, lr, __txn_logrec, links);
__os_free(env, lr);
}
if (txn->name != NULL) {
__os_free(env, txn->name);
txn->name = NULL;
}
if (F_ISSET(txn, TXN_MALLOC) &&
txn->xa_thr_status != TXN_XA_THREAD_ASSOCIATED) {
MUTEX_LOCK(env, mgr->mutex);
TAILQ_REMOVE(&mgr->txn_chain, txn, links);
MUTEX_UNLOCK(env, mgr->mutex);
__os_free(env, txn);
}
if (do_closefiles) {
(void)__dbreg_invalidate_files(env, 1);
(void)__dbreg_close_files(env, 1);
if (IS_REP_MASTER(env))
F_CLR(env->rep_handle, DBREP_OPENFILES);
F_CLR(env->lg_handle, DBLOG_OPENFILES);
mgr->n_discards = 0;
(void)__txn_checkpoint(env, 0, 0,
DB_CKP_INTERNAL | DB_FORCE);
}
return (0);
}
static int
__txn_dispatch_undo(env, txn, rdbt, key_lsn, txnlist)
ENV *env;
DB_TXN *txn;
DBT *rdbt;
DB_LSN *key_lsn;
DB_TXNHEAD *txnlist;
{
int ret;
txnlist->td = txn->td;
ret = __db_dispatch(env, &env->recover_dtab,
rdbt, key_lsn, DB_TXN_ABORT, txnlist);
if (ret == DB_SURPRISE_KID) {
F_SET(txn, TXN_CHILDCOMMIT);
ret = 0;
}
if (ret == 0 && F_ISSET(txn, TXN_CHILDCOMMIT) && IS_ZERO_LSN(*key_lsn))
ret = __db_txnlist_lsnget(env, txnlist, key_lsn, 0);
return (ret);
}
static int
__txn_undo(txn)
DB_TXN *txn;
{
DBT rdbt;
DB_LOGC *logc;
DB_LSN key_lsn;
DB_TXN *ptxn;
DB_TXNHEAD *txnlist;
DB_TXNLOGREC *lr;
DB_TXNMGR *mgr;
ENV *env;
int ret, t_ret;
mgr = txn->mgrp;
env = mgr->env;
logc = NULL;
txnlist = NULL;
ret = 0;
if (!LOGGING_ON(env))
return (0);
memset(&rdbt, 0, sizeof(rdbt));
for (ptxn = txn->parent; ptxn != NULL && ptxn->parent != NULL;)
ptxn = ptxn->parent;
if (ptxn != NULL && ptxn->txn_list != NULL)
txnlist = ptxn->txn_list;
else if (txn->txn_list != NULL)
txnlist = txn->txn_list;
else if ((ret = __db_txnlist_init(env,
txn->thread_info, 0, 0, NULL, &txnlist)) != 0)
return (ret);
else if (ptxn != NULL)
ptxn->txn_list = txnlist;
STAILQ_FOREACH(lr, &txn->logs, links) {
rdbt.data = lr->data;
rdbt.size = 0;
LSN_NOT_LOGGED(key_lsn);
ret =
__txn_dispatch_undo(env, txn, &rdbt, &key_lsn, txnlist);
if (ret != 0) {
__db_err(env, ret, DB_STR("4536",
"DB_TXN->abort: in-memory log undo failed"));
goto err;
}
}
key_lsn = ((TXN_DETAIL *)txn->td)->last_lsn;
if (!IS_ZERO_LSN(key_lsn) &&
(ret = __log_cursor(env, &logc)) != 0)
goto err;
while (!IS_ZERO_LSN(key_lsn)) {
if ((ret = __logc_get(logc, &key_lsn, &rdbt, DB_SET)) == 0) {
ret = __txn_dispatch_undo(env,
txn, &rdbt, &key_lsn, txnlist);
}
if (ret != 0) {
__db_err(env, ret, DB_STR_A("4537",
"DB_TXN->abort: log undo failed for LSN: %lu %lu",
"%lu %lu"), (u_long)key_lsn.file,
(u_long)key_lsn.offset);
goto err;
}
}
err: if (logc != NULL && (t_ret = __logc_close(logc)) != 0 && ret == 0)
ret = t_ret;
if (ptxn == NULL && txnlist != NULL)
__db_txnlist_end(env, txnlist);
return (ret);
}
int
__txn_activekids(env, rectype, txn)
ENV *env;
u_int32_t rectype;
DB_TXN *txn;
{
if (F_ISSET(txn, TXN_COMPENSATE) || rectype == DB___txn_child)
return (0);
if (TAILQ_FIRST(&txn->kids) != NULL) {
__db_errx(env, DB_STR("4538",
"Child transaction is active"));
return (EPERM);
}
return (0);
}
int
__txn_force_abort(env, buffer)
ENV *env;
u_int8_t *buffer;
{
DB_CIPHER *db_cipher;
HDR hdr, *hdrp;
u_int32_t offset, opcode, sum_len;
u_int8_t *bp, *key;
size_t hdrsize, rec_len;
int ret;
db_cipher = env->crypto_handle;
hdrsize = CRYPTO_ON(env) ? HDR_CRYPTO_SZ : HDR_NORMAL_SZ;
hdrp = (HDR *)buffer;
memcpy(&hdr.prev, buffer + SSZ(HDR, prev), sizeof(hdr.prev));
memcpy(&hdr.len, buffer + SSZ(HDR, len), sizeof(hdr.len));
if (LOG_SWAPPED(env))
__log_hdrswap(&hdr, CRYPTO_ON(env));
rec_len = hdr.len - hdrsize;
offset = sizeof(u_int32_t) + sizeof(u_int32_t) + sizeof(DB_LSN);
if (CRYPTO_ON(env)) {
key = db_cipher->mac_key;
sum_len = DB_MAC_KEY;
if ((ret = db_cipher->decrypt(env, db_cipher->data,
&hdrp->iv[0], buffer + hdrsize, rec_len)) != 0)
return (__env_panic(env, ret));
} else {
key = NULL;
sum_len = sizeof(u_int32_t);
}
bp = buffer + hdrsize + offset;
opcode = TXN_ABORT;
LOGCOPY_32(env, bp, &opcode);
if (CRYPTO_ON(env) &&
(ret = db_cipher->encrypt(env,
db_cipher->data, &hdrp->iv[0], buffer + hdrsize, rec_len)) != 0)
return (__env_panic(env, ret));
#ifdef HAVE_LOG_CHECKSUM
__db_chksum(&hdr, buffer + hdrsize, rec_len, key, NULL);
if (LOG_SWAPPED(env))
__log_hdrswap(&hdr, CRYPTO_ON(env));
memcpy(buffer + SSZA(HDR, chksum), hdr.chksum, sum_len);
#endif
return (0);
}
int
__txn_preclose(env)
ENV *env;
{
DB_TXNMGR *mgr;
DB_TXNREGION *region;
int do_closefiles, ret;
mgr = env->tx_handle;
region = mgr->reginfo.primary;
do_closefiles = 0;
TXN_SYSTEM_LOCK(env);
if (region != NULL &&
region->stat.st_nrestores <= mgr->n_discards &&
mgr->n_discards != 0)
do_closefiles = 1;
TXN_SYSTEM_UNLOCK(env);
if (do_closefiles) {
F_SET(env->lg_handle, DBLOG_RECOVER);
ret = __dbreg_close_files(env, 0);
F_CLR(env->lg_handle, DBLOG_RECOVER);
} else
ret = 0;
return (ret);
}
int
__txn_reset(env)
ENV *env;
{
DB_LSN scrap;
DB_TXNREGION *region;
region = env->tx_handle->reginfo.primary;
region->last_txnid = TXN_MINIMUM;
DB_ASSERT(env, LOGGING_ON(env));
return (__txn_recycle_log(env,
NULL, &scrap, 0, TXN_MINIMUM, TXN_MAXIMUM));
}
static void
__txn_set_txn_lsnp(txn, blsnp, llsnp)
DB_TXN *txn;
DB_LSN **blsnp, **llsnp;
{
TXN_DETAIL *td;
td = txn->td;
*llsnp = &td->last_lsn;
while (txn->parent != NULL)
txn = txn->parent;
td = txn->td;
if (IS_ZERO_LSN(td->begin_lsn))
*blsnp = &td->begin_lsn;
}
int
__txn_applied_pp(dbenv, token, timeout, flags)
DB_ENV *dbenv;
DB_TXN_TOKEN *token;
db_timeout_t timeout;
u_int32_t flags;
{
ENV *env;
DB_THREAD_INFO *ip;
DB_COMMIT_INFO commit_info;
u_int8_t *bp;
int ret;
env = dbenv->env;
if (flags != 0)
return (__db_ferr(env, "DB_ENV->txn_applied", 0));
bp = token->buf;
DB_NTOHL_COPYIN(env, commit_info.version, bp);
DB_ASSERT(env, commit_info.version == REP_COMMIT_TOKEN_FMT_VERSION);
DB_NTOHL_COPYIN(env, commit_info.gen, bp);
DB_NTOHL_COPYIN(env, commit_info.envid, bp);
DB_NTOHL_COPYIN(env, commit_info.lsn.file, bp);
DB_NTOHL_COPYIN(env, commit_info.lsn.offset, bp);
if (IS_ZERO_LSN(commit_info.lsn))
return (DB_KEYEMPTY);
ENV_REQUIRES_CONFIG(env,
env->lg_handle, "DB_ENV->txn_applied", DB_INIT_LOG);
ENV_ENTER(env, ip);
ret = __txn_applied(env, ip, &commit_info, timeout);
ENV_LEAVE(env, ip);
return (ret);
}
static int
__txn_applied(env, ip, commit_info, timeout)
ENV *env;
DB_THREAD_INFO *ip;
DB_COMMIT_INFO *commit_info;
db_timeout_t timeout;
{
LOG *lp;
DB_LSN lsn;
REGENV *renv;
if (REP_ON(env))
return (__rep_txn_applied(env, ip, commit_info, timeout));
if (commit_info->gen != 0) {
__db_errx(env, DB_STR("4539",
"replication commit token in non-replication env"));
return (EINVAL);
}
lp = env->lg_handle->reginfo.primary;
LOG_SYSTEM_LOCK(env);
lsn = lp->lsn;
LOG_SYSTEM_UNLOCK(env);
renv = env->reginfo->primary;
if (renv->envid == commit_info->envid &&
LOG_COMPARE(&commit_info->lsn, &lsn) <= 0)
return (0);
return (DB_NOTFOUND);
}