#include "db_config.h"
#include "db_int.h"
#include "dbinc/txn.h"
static int __db_xa_close __P((char *, int, long));
static int __db_xa_commit __P((XID *, int, long));
static int __db_xa_complete __P((int *, int *, int, long));
static int __db_xa_end __P((XID *, int, long));
static int __db_xa_forget __P((XID *, int, long));
static int __db_xa_open __P((char *, int, long));
static int __db_xa_prepare __P((XID *, int, long));
static int __db_xa_recover __P((XID *, long, int, long));
static int __db_xa_rollback __P((XID *, int, long));
static int __db_xa_start __P((XID *, int, long));
static int __xa_put_txn __P((DB_ENV *, DB_TXN *));
const struct xa_switch_t db_xa_switch = {
"Berkeley DB",
TMNOMIGRATE,
0,
__db_xa_open,
__db_xa_close,
__db_xa_start,
__db_xa_end,
__db_xa_rollback,
__db_xa_prepare,
__db_xa_commit,
__db_xa_recover,
__db_xa_forget,
__db_xa_complete
};
#undef XA_MULTI_THREAD
int
__xa_get_txn(dbenv, txnp, do_init)
DB_ENV *dbenv;
DB_TXN **txnp;
int do_init;
{
#ifdef XA_MULTI_THREAD
DB_TXN *t;
DB_TXNMGR *mgr;
TXN_DETAIL *td;
db_threadid_t tid;
pid_t pid;
#endif
int ret;
ret = 0;
#ifdef XA_MULTI_THREAD
dbenv->thread_id(dbenv, &pid, &tid);
*txnp = NULL;
DB_ASSERT(dbenv, dbenv->tx_handle != NULL);
mgr = dbenv->tx_handle;
MUTEX_LOCK(dbenv, mgr->mutex);
TAILQ_FOREACH(t, &dbenv->xa_txn, xalinks) {
td = t->td;
if (td->pid != pid)
continue;
#ifdef HAVE_SIMPLE_THREAD_TYPE
if (t->tid == tid) {
*txnp = t;
break;
}
#else
if (memcmp(&t->tid, &tid, sizeof(tid)) == 0) {
*txnp = t;
break;
}
#endif
}
MUTEX_UNLOCK(dbenv, mgr->mutex);
if (*txnp == NULL) {
if (!do_init)
ret = EINVAL;
else if ((ret =
__os_malloc(dbenv, sizeof(DB_TXN), txnp)) == 0) {
(*txnp)->tid = tid;
MUTEX_LOCK(dbenv, mgr->mutex);
TAILQ_INSERT_HEAD(&dbenv->xa_txn, *txnp, xalinks);
MUTEX_UNLOCK(dbenv, mgr->mutex);
}
}
#else
COMPQUIET(do_init, 0);
*txnp = TAILQ_FIRST(&dbenv->xa_txn);
if (*txnp == NULL &&
(ret = __os_calloc(dbenv, 1, sizeof(DB_TXN), txnp)) == 0) {
(*txnp)->txnid = TXN_INVALID;
TAILQ_INSERT_HEAD(&dbenv->xa_txn, *txnp, xalinks);
}
#endif
return (ret);
}
static int
__xa_put_txn(dbenv, txnp)
DB_ENV *dbenv;
DB_TXN *txnp;
{
#ifdef XA_MULTI_THREAD
DB_TXNMGR *mgr;
mgr = dbenv->tx_handle;
MUTEX_LOCK(dbenv, mgr->mutex);
TAILQ_REMOVE(&dbenv->xa_txn, txnp, xalinks);
MUTEX_UNLOCK(dbenv, mgr->mutex);
__os_free(dbenv, txnp);
#else
COMPQUIET(dbenv, NULL);
txnp->txnid = TXN_INVALID;
#endif
return (0);
}
#ifdef XA_MULTI_THREAD
#define XA_FLAGS \
(DB_CREATE | DB_INIT_LOCK | DB_INIT_LOG | DB_INIT_MPOOL | \
DB_INIT_TXN | DB_THREAD)
#else
#define XA_FLAGS \
(DB_CREATE | DB_INIT_LOCK | DB_INIT_LOG | DB_INIT_MPOOL | \
DB_INIT_TXN)
#endif
static int
__db_xa_open(xa_info, rmid, arg_flags)
char *xa_info;
int rmid;
long arg_flags;
{
DB_ENV *dbenv;
u_long flags;
flags = (u_long)arg_flags;
if (LF_ISSET(TMASYNC))
return (XAER_ASYNC);
if (flags != TMNOFLAGS)
return (XAER_INVAL);
if (__db_rmid_to_env(rmid, &dbenv) == 0)
return (XA_OK);
if (db_env_create(&dbenv, 0) != 0)
return (XAER_RMERR);
if (dbenv->open(dbenv, xa_info, XA_FLAGS, 0) != 0)
goto err;
if (__db_map_rmid(rmid, dbenv) != 0)
goto err;
TAILQ_INIT(&dbenv->xa_txn);
return (XA_OK);
err: (void)dbenv->close(dbenv, 0);
return (XAER_RMERR);
}
static int
__db_xa_close(xa_info, rmid, arg_flags)
char *xa_info;
int rmid;
long arg_flags;
{
DB_ENV *dbenv;
DB_TXN *t;
int ret, t_ret;
u_long flags;
COMPQUIET(xa_info, NULL);
flags = (u_long)arg_flags;
if (LF_ISSET(TMASYNC))
return (XAER_ASYNC);
if (flags != TMNOFLAGS)
return (XAER_INVAL);
if (__db_rmid_to_env(rmid, &dbenv) != 0)
return (XA_OK);
if ((t = TAILQ_FIRST(&dbenv->xa_txn)) != NULL &&
t->txnid != TXN_INVALID)
return (XAER_PROTO);
ret = __db_unmap_rmid(rmid);
while ((t = TAILQ_FIRST(&dbenv->xa_txn)) != NULL) {
TAILQ_REMOVE(&dbenv->xa_txn, t, xalinks);
__os_free(dbenv, t);
}
if ((t_ret = dbenv->close(dbenv, 0)) != 0 && ret == 0)
ret = t_ret;
return (ret == 0 ? XA_OK : XAER_RMERR);
}
static int
__db_xa_start(xid, rmid, arg_flags)
XID *xid;
int rmid;
long arg_flags;
{
DB_ENV *dbenv;
DB_TXN *txnp;
TXN_DETAIL *td;
roff_t off;
u_long flags;
int is_known;
flags = (u_long)arg_flags;
#define OK_FLAGS (TMJOIN | TMRESUME | TMNOWAIT | TMASYNC | TMNOFLAGS)
if (LF_ISSET(~OK_FLAGS))
return (XAER_INVAL);
if (LF_ISSET(TMJOIN) && LF_ISSET(TMRESUME))
return (XAER_INVAL);
if (LF_ISSET(TMASYNC))
return (XAER_ASYNC);
if (__db_rmid_to_env(rmid, &dbenv) != 0)
return (XAER_PROTO);
is_known = __db_xid_to_txn(dbenv, xid, &off) == 0;
if (is_known && !LF_ISSET(TMRESUME) && !LF_ISSET(TMJOIN))
return (XAER_DUPID);
if (!is_known && LF_ISSET(TMRESUME | TMJOIN))
return (XAER_NOTA);
if (is_known) {
td = R_ADDR(&dbenv->tx_handle->reginfo, off);
if (td->xa_status == TXN_XA_SUSPENDED &&
!LF_ISSET(TMRESUME | TMJOIN))
return (XAER_PROTO);
if (td->xa_status == TXN_XA_DEADLOCKED)
return (XA_RBDEADLOCK);
if (td->xa_status == TXN_XA_ABORTED)
return (XA_RBOTHER);
if (__xa_get_txn(dbenv, &txnp, 1) != 0)
return (XAER_RMERR);
if (__txn_continue(dbenv, txnp, td) != 0)
return (XAER_RMERR);
td->xa_status = TXN_XA_STARTED;
} else {
if (__xa_get_txn(dbenv, &txnp, 1) != 0)
return (XAER_RMERR);
if (__txn_xa_begin(dbenv, txnp))
return (XAER_RMERR);
(void)__db_map_xid(dbenv, xid, txnp->td);
td = txnp->td;
td->xa_status = TXN_XA_STARTED;
}
return (XA_OK);
}
static int
__db_xa_end(xid, rmid, flags)
XID *xid;
int rmid;
long flags;
{
DB_ENV *dbenv;
DB_TXN *txn;
TXN_DETAIL *td;
roff_t off;
if (flags != TMNOFLAGS && !LF_ISSET(TMSUSPEND | TMSUCCESS | TMFAIL))
return (XAER_INVAL);
if (__db_rmid_to_env(rmid, &dbenv) != 0)
return (XAER_PROTO);
if (__db_xid_to_txn(dbenv, xid, &off) != 0)
return (XAER_NOTA);
if (__xa_get_txn(dbenv, &txn, 0) != 0)
return (XAER_RMERR);
td = R_ADDR(&dbenv->tx_handle->reginfo, off);
if (td != txn->td)
return (XAER_PROTO);
if (td->xa_status == TXN_XA_DEADLOCKED)
return (XA_RBDEADLOCK);
if (td->status == TXN_ABORTED)
return (XA_RBOTHER);
if (td->xa_status != TXN_XA_STARTED)
return (XAER_PROTO);
if (LF_ISSET(TMSUSPEND))
td->xa_status = TXN_XA_SUSPENDED;
else
td->xa_status = TXN_XA_ENDED;
(void)__xa_put_txn(dbenv, txn);
return (XA_OK);
}
static int
__db_xa_prepare(xid, rmid, arg_flags)
XID *xid;
int rmid;
long arg_flags;
{
DB_ENV *dbenv;
DB_TXN *txnp;
TXN_DETAIL *td;
roff_t off;
u_long flags;
flags = (u_long)arg_flags;
if (LF_ISSET(TMASYNC))
return (XAER_ASYNC);
if (flags != TMNOFLAGS)
return (XAER_INVAL);
if (__db_rmid_to_env(rmid, &dbenv) != 0)
return (XAER_PROTO);
if (__db_xid_to_txn(dbenv, xid, &off) != 0)
return (XAER_NOTA);
td = R_ADDR(&dbenv->tx_handle->reginfo, off);
if (td->xa_status == TXN_XA_DEADLOCKED)
return (XA_RBDEADLOCK);
if (td->xa_status != TXN_XA_ENDED && td->xa_status != TXN_XA_SUSPENDED)
return (XAER_PROTO);
if (__xa_get_txn(dbenv, &txnp, 0) != 0)
return (XAER_PROTO);
if (__txn_continue(dbenv, txnp, td) != 0)
return (XAER_RMERR);
if (txnp->prepare(txnp, (u_int8_t *)xid->data) != 0)
return (XAER_RMERR);
td->xa_status = TXN_XA_PREPARED;
(void)__xa_put_txn(dbenv, txnp);
return (XA_OK);
}
static int
__db_xa_commit(xid, rmid, arg_flags)
XID *xid;
int rmid;
long arg_flags;
{
DB_ENV *dbenv;
DB_TXN *txnp;
TXN_DETAIL *td;
roff_t off;
u_long flags;
flags = (u_long)arg_flags;
if (LF_ISSET(TMASYNC))
return (XAER_ASYNC);
#undef OK_FLAGS
#define OK_FLAGS (TMNOFLAGS | TMNOWAIT | TMONEPHASE)
if (LF_ISSET(~OK_FLAGS))
return (XAER_INVAL);
if (__db_rmid_to_env(rmid, &dbenv) != 0)
return (XAER_PROTO);
if (__db_xid_to_txn(dbenv, xid, &off) != 0)
return (XAER_NOTA);
td = R_ADDR(&dbenv->tx_handle->reginfo, off);
if (td->xa_status == TXN_XA_DEADLOCKED)
return (XA_RBDEADLOCK);
if (td->xa_status == TXN_XA_ABORTED)
return (XA_RBOTHER);
if (LF_ISSET(TMONEPHASE) &&
td->xa_status != TXN_XA_ENDED && td->xa_status != TXN_XA_SUSPENDED)
return (XAER_PROTO);
if (!LF_ISSET(TMONEPHASE) && td->xa_status != TXN_XA_PREPARED)
return (XAER_PROTO);
if (__xa_get_txn(dbenv, &txnp, 0) != 0)
return (XAER_RMERR);
if (__txn_continue(dbenv, txnp, td) != 0)
return (XAER_RMERR);
if (txnp->commit(txnp, 0) != 0)
return (XAER_RMERR);
(void)__xa_put_txn(dbenv, txnp);
return (XA_OK);
}
static int
__db_xa_recover(xids, count, rmid, flags)
XID *xids;
long count, flags;
int rmid;
{
DB_ENV *dbenv;
u_int32_t newflags;
long rval;
if (__db_rmid_to_env(rmid, &dbenv) != 0)
return (XAER_PROTO);
if (LF_ISSET(TMSTARTRSCAN))
newflags = DB_FIRST;
else if (LF_ISSET(TMENDRSCAN))
newflags = DB_LAST;
else
newflags = DB_NEXT;
rval = 0;
if (__txn_get_prepared(dbenv, xids, NULL, count, &rval, newflags) != 0)
return (XAER_RMERR);
else
return (rval);
}
static int
__db_xa_rollback(xid, rmid, arg_flags)
XID *xid;
int rmid;
long arg_flags;
{
DB_ENV *dbenv;
DB_TXN *txnp;
TXN_DETAIL *td;
roff_t off;
u_long flags;
flags = (u_long)arg_flags;
if (LF_ISSET(TMASYNC))
return (XAER_ASYNC);
if (flags != TMNOFLAGS)
return (XAER_INVAL);
if (__db_rmid_to_env(rmid, &dbenv) != 0)
return (XAER_PROTO);
if (__db_xid_to_txn(dbenv, xid, &off) != 0)
return (XAER_NOTA);
td = R_ADDR(&dbenv->tx_handle->reginfo, off);
if (td->xa_status == TXN_XA_DEADLOCKED)
return (XA_RBDEADLOCK);
if (td->xa_status == TXN_XA_ABORTED)
return (XA_RBOTHER);
if (td->xa_status != TXN_XA_ENDED &&
td->xa_status != TXN_XA_SUSPENDED &&
td->xa_status != TXN_XA_PREPARED)
return (XAER_PROTO);
if (__xa_get_txn(dbenv, &txnp, 0) != 0)
return (XAER_RMERR);
if (__txn_continue(dbenv, txnp, td) != 0)
return (XAER_RMERR);
if (txnp->abort(txnp) != 0)
return (XAER_RMERR);
(void)__xa_put_txn(dbenv, txnp);
return (XA_OK);
}
static int
__db_xa_forget(xid, rmid, arg_flags)
XID *xid;
int rmid;
long arg_flags;
{
DB_ENV *dbenv;
roff_t off;
u_long flags;
flags = (u_long)arg_flags;
if (LF_ISSET(TMASYNC))
return (XAER_ASYNC);
if (flags != TMNOFLAGS)
return (XAER_INVAL);
if (__db_rmid_to_env(rmid, &dbenv) != 0)
return (XAER_PROTO);
if (__db_xid_to_txn(dbenv, xid, &off) != 0)
return (XA_OK);
__db_unmap_xid(dbenv, xid, off);
return (XA_OK);
}
static int
__db_xa_complete(handle, retval, rmid, flags)
int *handle, *retval, rmid;
long flags;
{
COMPQUIET(handle, NULL);
COMPQUIET(retval, NULL);
COMPQUIET(rmid, 0);
COMPQUIET(flags, 0);
return (XAER_INVAL);
}