#include "db_config.h"
#ifndef lint
static const char revid[] = "$Id: db_am.c,v 1.2 2004/03/30 01:21:24 jtownsen Exp $";
#endif
#ifndef NO_SYSTEM_INCLUDES
#include <sys/types.h>
#include <string.h>
#endif
#include "db_int.h"
#include "dbinc/db_page.h"
#include "dbinc/db_shash.h"
#include "dbinc/btree.h"
#include "dbinc/hash.h"
#include "dbinc/lock.h"
#include "dbinc/log.h"
#include "dbinc/mp.h"
#include "dbinc/qam.h"
static int __db_append_primary __P((DBC *, DBT *, DBT *));
static int __db_secondary_get __P((DB *, DB_TXN *, DBT *, DBT *, u_int32_t));
static int __db_secondary_close __P((DB *, u_int32_t));
#ifdef DEBUG
static int __db_cprint_item __P((DBC *));
#endif
int
__db_cursor_int(dbp, txn, dbtype, root, is_opd, lockerid, dbcp)
DB *dbp;
DB_TXN *txn;
DBTYPE dbtype;
db_pgno_t root;
int is_opd;
u_int32_t lockerid;
DBC **dbcp;
{
DBC *dbc, *adbc;
DBC_INTERNAL *cp;
DB_ENV *dbenv;
int allocated, ret;
dbenv = dbp->dbenv;
allocated = 0;
MUTEX_THREAD_LOCK(dbenv, dbp->mutexp);
for (dbc = TAILQ_FIRST(&dbp->free_queue);
dbc != NULL; dbc = TAILQ_NEXT(dbc, links))
if (dbtype == dbc->dbtype) {
TAILQ_REMOVE(&dbp->free_queue, dbc, links);
F_CLR(dbc, ~DBC_OWN_LID);
break;
}
MUTEX_THREAD_UNLOCK(dbenv, dbp->mutexp);
if (dbc == NULL) {
if ((ret = __os_calloc(dbenv, 1, sizeof(DBC), &dbc)) != 0)
return (ret);
allocated = 1;
dbc->flags = 0;
dbc->dbp = dbp;
if (LOCKING_ON(dbenv)) {
if (!DB_IS_THREADED(dbp) &&
(adbc = TAILQ_FIRST(&dbp->active_queue)) != NULL)
dbc->lid = adbc->lid;
else {
if ((ret = __lock_id(dbenv, &dbc->lid)) != 0)
goto err;
F_SET(dbc, DBC_OWN_LID);
}
if (CDB_LOCKING(dbenv) &&
F_ISSET(dbp, DB_AM_SECONDARY))
memcpy(dbc->lock.fileid,
dbp->s_primary->fileid, DB_FILE_ID_LEN);
else
memcpy(dbc->lock.fileid,
dbp->fileid, DB_FILE_ID_LEN);
if (CDB_LOCKING(dbenv)) {
if (F_ISSET(dbenv, DB_ENV_CDB_ALLDB)) {
DB_ASSERT(sizeof(db_pgno_t) ==
sizeof(u_int32_t));
dbc->lock_dbt.size = sizeof(u_int32_t);
dbc->lock_dbt.data = &dbc->lock.pgno;
dbc->lock.pgno = 0;
} else {
dbc->lock_dbt.size = DB_FILE_ID_LEN;
dbc->lock_dbt.data = dbc->lock.fileid;
}
} else {
dbc->lock.type = DB_PAGE_LOCK;
dbc->lock_dbt.size = sizeof(dbc->lock);
dbc->lock_dbt.data = &dbc->lock;
}
}
switch (dbtype) {
case DB_BTREE:
case DB_RECNO:
if ((ret = __bam_c_init(dbc, dbtype)) != 0)
goto err;
break;
case DB_HASH:
if ((ret = __ham_c_init(dbc)) != 0)
goto err;
break;
case DB_QUEUE:
if ((ret = __qam_c_init(dbc)) != 0)
goto err;
break;
case DB_UNKNOWN:
default:
ret = __db_unknown_type(dbenv, "DB->cursor", dbtype);
goto err;
}
cp = dbc->internal;
}
dbc->dbtype = dbtype;
RESET_RET_MEM(dbc);
if ((dbc->txn = txn) == NULL) {
if (lockerid != DB_LOCK_INVALIDID)
dbc->locker = lockerid;
else
dbc->locker = dbc->lid;
} else {
dbc->locker = txn->txnid;
txn->cursors++;
}
if (F_ISSET(dbp, DB_AM_SECONDARY))
dbc->c_get = __db_c_secondary_get_pp;
if (is_opd)
F_SET(dbc, DBC_OPD);
if (F_ISSET(dbp, DB_AM_RECOVER))
F_SET(dbc, DBC_RECOVER);
if (F_ISSET(dbp, DB_AM_COMPENSATE))
F_SET(dbc, DBC_COMPENSATE);
cp = dbc->internal;
cp->opd = NULL;
cp->indx = 0;
cp->page = NULL;
cp->pgno = PGNO_INVALID;
cp->root = root;
switch (dbtype) {
case DB_BTREE:
case DB_RECNO:
if ((ret = __bam_c_refresh(dbc)) != 0)
goto err;
break;
case DB_HASH:
case DB_QUEUE:
break;
case DB_UNKNOWN:
default:
ret = __db_unknown_type(dbenv, "DB->cursor", dbp->type);
goto err;
}
MUTEX_THREAD_LOCK(dbenv, dbp->mutexp);
TAILQ_INSERT_TAIL(&dbp->active_queue, dbc, links);
F_SET(dbc, DBC_ACTIVE);
MUTEX_THREAD_UNLOCK(dbenv, dbp->mutexp);
*dbcp = dbc;
return (0);
err: if (allocated)
__os_free(dbenv, dbc);
return (ret);
}
#ifdef DEBUG
int
__db_cprint(dbp)
DB *dbp;
{
DBC *dbc;
int ret, t_ret;
ret = 0;
MUTEX_THREAD_LOCK(dbp->dbenv, dbp->mutexp);
fprintf(stderr, "Active queue:\n");
for (dbc = TAILQ_FIRST(&dbp->active_queue);
dbc != NULL; dbc = TAILQ_NEXT(dbc, links))
if ((t_ret = __db_cprint_item(dbc)) != 0 && ret == 0)
ret = t_ret;
fprintf(stderr, "Join queue:\n");
for (dbc = TAILQ_FIRST(&dbp->join_queue);
dbc != NULL; dbc = TAILQ_NEXT(dbc, links))
if ((t_ret = __db_cprint_item(dbc)) != 0 && ret == 0)
ret = t_ret;
fprintf(stderr, "Free queue:\n");
for (dbc = TAILQ_FIRST(&dbp->free_queue);
dbc != NULL; dbc = TAILQ_NEXT(dbc, links))
if ((t_ret = __db_cprint_item(dbc)) != 0 && ret == 0)
ret = t_ret;
MUTEX_THREAD_UNLOCK(dbp->dbenv, dbp->mutexp);
return (ret);
}
static
int __db_cprint_item(dbc)
DBC *dbc;
{
static const FN fn[] = {
{ DBC_ACTIVE, "active" },
{ DBC_COMPENSATE, "compensate" },
{ DBC_OPD, "off-page-dup" },
{ DBC_RECOVER, "recover" },
{ DBC_RMW, "read-modify-write" },
{ DBC_TRANSIENT, "transient" },
{ DBC_WRITECURSOR, "write cursor" },
{ DBC_WRITER, "short-term write cursor" },
{ 0, NULL }
};
DB *dbp;
DBC_INTERNAL *cp;
const char *s;
dbp = dbc->dbp;
cp = dbc->internal;
s = __db_dbtype_to_string(dbc->dbtype);
fprintf(stderr, "%s/%#lx: opd: %#lx\n",
s, P_TO_ULONG(dbc), P_TO_ULONG(cp->opd));
fprintf(stderr, "\ttxn: %#lx lid: %lu locker: %lu\n",
P_TO_ULONG(dbc->txn), (u_long)dbc->lid, (u_long)dbc->locker);
fprintf(stderr, "\troot: %lu page/index: %lu/%lu",
(u_long)cp->root, (u_long)cp->pgno, (u_long)cp->indx);
__db_prflags(dbc->flags, fn, stderr);
fprintf(stderr, "\n");
switch (dbp->type) {
case DB_BTREE:
case DB_RECNO:
__bam_cprint(dbc);
break;
case DB_HASH:
__ham_cprint(dbc);
break;
case DB_UNKNOWN:
DB_ASSERT(dbp->type != DB_UNKNOWN);
case DB_QUEUE:
default:
break;
}
return (0);
}
#endif
int
__db_put(dbp, txn, key, data, flags)
DB *dbp;
DB_TXN *txn;
DBT *key, *data;
u_int32_t flags;
{
DBC *dbc;
DBT tdata;
DB_ENV *dbenv;
int ret, t_ret;
dbenv = dbp->dbenv;
if ((ret = __db_cursor(dbp, txn, &dbc, DB_WRITELOCK)) != 0)
return (ret);
DEBUG_LWRITE(dbc, txn, "DB->put", key, data, flags);
SET_RET_MEM(dbc, dbp);
F_SET(dbc, DBC_TRANSIENT);
switch (flags) {
case DB_APPEND:
tdata = *data;
switch (dbp->type) {
case DB_QUEUE:
if ((ret = __qam_append(dbc, key, &tdata)) != 0)
goto err;
break;
case DB_RECNO:
if ((ret = __ram_append(dbc, key, &tdata)) != 0)
goto err;
break;
case DB_BTREE:
case DB_HASH:
case DB_UNKNOWN:
default:
DB_ASSERT(
dbp->type == DB_QUEUE || dbp->type == DB_RECNO);
ret = __db_ferr(dbenv, "DB->put", 0);
goto err;
}
DB_ASSERT(!F_ISSET(dbp, DB_AM_SECONDARY));
if (LIST_FIRST(&dbp->s_secondaries) != NULL)
ret = __db_append_primary(dbc, key, &tdata);
FREE_IF_NEEDED(dbp, &tdata);
goto done;
case DB_NOOVERWRITE:
flags = 0;
memset(&tdata, 0, sizeof(tdata));
F_SET(&tdata, DB_DBT_USERMEM | DB_DBT_PARTIAL);
if ((ret = __db_c_get(dbc, key, &tdata,
DB_SET | (STD_LOCKING(dbc) ? DB_RMW : 0))) == 0)
ret = DB_KEYEXIST;
else if (ret == DB_NOTFOUND || ret == DB_KEYEMPTY)
ret = 0;
break;
default:
break;
}
if (ret == 0)
ret = __db_c_put(dbc,
key, data, flags == 0 ? DB_KEYLAST : flags);
err:
done:
if ((t_ret = __db_c_close(dbc)) != 0 && ret == 0)
ret = t_ret;
return (ret);
}
int
__db_del(dbp, txn, key, flags)
DB *dbp;
DB_TXN *txn;
DBT *key;
u_int32_t flags;
{
DBC *dbc;
DBT data, lkey;
u_int32_t f_init, f_next;
int ret, t_ret;
if ((ret = __db_cursor(dbp, txn, &dbc, DB_WRITELOCK)) != 0)
goto err;
DEBUG_LWRITE(dbc, txn, "DB->del", key, NULL, flags);
COMPQUIET(flags, 0);
memset(&lkey, 0, sizeof(lkey));
F_SET(&lkey, DB_DBT_USERMEM | DB_DBT_PARTIAL);
memset(&data, 0, sizeof(data));
F_SET(&data, DB_DBT_USERMEM | DB_DBT_PARTIAL);
f_init = DB_SET;
f_next = DB_NEXT_DUP;
if (STD_LOCKING(dbc)) {
f_init |= DB_RMW;
f_next |= DB_RMW;
}
if ((ret = __db_c_get(dbc, key, &data, f_init)) != 0)
goto err;
if (dbp->type == DB_HASH)
if (LIST_FIRST(&dbp->s_secondaries) == NULL &&
!F_ISSET(dbp, DB_AM_SECONDARY) &&
dbc->internal->opd == NULL) {
ret = __ham_quick_delete(dbc);
goto done;
}
for (;;) {
if ((ret = __db_c_del(dbc, 0)) != 0)
break;
if ((ret = __db_c_get(dbc, &lkey, &data, f_next)) != 0) {
if (ret == DB_NOTFOUND)
ret = 0;
break;
}
}
done:
err:
if ((t_ret = __db_c_close(dbc)) != 0 && ret == 0)
ret = t_ret;
return (ret);
}
int
__db_sync(dbp)
DB *dbp;
{
int ret, t_ret;
ret = 0;
if (F_ISSET(dbp, DB_AM_RDONLY))
return (0);
if (dbp->type == DB_RECNO)
ret = __ram_writeback(dbp);
if (F_ISSET(dbp, DB_AM_INMEM))
return (ret);
if (dbp->type == DB_QUEUE)
ret = __qam_sync(dbp);
else
if ((t_ret = __memp_fsync(dbp->mpf)) != 0 && ret == 0)
ret = t_ret;
return (ret);
}
int
__db_associate(dbp, txn, sdbp, callback, flags)
DB *dbp, *sdbp;
DB_TXN *txn;
int (*callback) __P((DB *, const DBT *, const DBT *, DBT *));
u_int32_t flags;
{
DB_ENV *dbenv;
DBC *pdbc, *sdbc;
DBT skey, key, data;
int build, ret, t_ret;
dbenv = dbp->dbenv;
pdbc = sdbc = NULL;
ret = 0;
sdbp->s_callback = callback;
sdbp->s_primary = dbp;
sdbp->stored_get = sdbp->get;
sdbp->get = __db_secondary_get;
sdbp->stored_close = sdbp->close;
sdbp->close = __db_secondary_close;
F_SET(sdbp, DB_AM_SECONDARY);
build = 0;
if (LF_ISSET(DB_CREATE)) {
if ((ret = __db_cursor(sdbp, txn, &sdbc, 0)) != 0)
goto err;
memset(&key, 0, sizeof(DBT));
memset(&data, 0, sizeof(DBT));
F_SET(&key, DB_DBT_PARTIAL | DB_DBT_USERMEM);
F_SET(&data, DB_DBT_PARTIAL | DB_DBT_USERMEM);
if ((ret = __db_c_get(sdbc, &key, &data,
(STD_LOCKING(sdbc) ? DB_RMW : 0) |
DB_FIRST)) == DB_NOTFOUND) {
build = 1;
ret = 0;
}
if ((t_ret = __db_c_close(sdbc)) != 0 && ret == 0)
ret = t_ret;
sdbc = NULL;
if (ret != 0)
goto err;
}
MUTEX_THREAD_LOCK(dbenv, dbp->mutexp);
DB_ASSERT(sdbp->s_refcnt == 0);
sdbp->s_refcnt = 1;
LIST_INSERT_HEAD(&dbp->s_secondaries, sdbp, s_links);
MUTEX_THREAD_UNLOCK(dbenv, dbp->mutexp);
if (build) {
if ((ret = __db_cursor(sdbp, txn, &sdbc,
CDB_LOCKING(sdbp->dbenv) ? DB_WRITECURSOR : 0)) != 0)
goto err;
if ((ret = __db_cursor_int(dbp,
txn, dbp->type, PGNO_INVALID, 0, sdbc->locker, &pdbc)) != 0)
goto err;
dbp->associate_lid = sdbc->locker;
memset(&key, 0, sizeof(DBT));
memset(&data, 0, sizeof(DBT));
while ((ret = __db_c_get(pdbc, &key, &data, DB_NEXT)) == 0) {
memset(&skey, 0, sizeof(DBT));
if ((ret = callback(sdbp, &key, &data, &skey)) != 0) {
if (ret == DB_DONOTINDEX)
continue;
goto err;
}
if ((ret = __db_c_put(sdbc,
&skey, &key, DB_UPDATE_SECONDARY)) != 0) {
FREE_IF_NEEDED(sdbp, &skey);
goto err;
}
FREE_IF_NEEDED(sdbp, &skey);
}
if (ret == DB_NOTFOUND)
ret = 0;
}
err: if (sdbc != NULL && (t_ret = __db_c_close(sdbc)) != 0 && ret == 0)
ret = t_ret;
if (pdbc != NULL && (t_ret = __db_c_close(pdbc)) != 0 && ret == 0)
ret = t_ret;
dbp->associate_lid = DB_LOCK_INVALIDID;
return (ret);
}
static int
__db_secondary_get(sdbp, txn, skey, data, flags)
DB *sdbp;
DB_TXN *txn;
DBT *skey, *data;
u_int32_t flags;
{
DB_ASSERT(F_ISSET(sdbp, DB_AM_SECONDARY));
return (__db_pget(sdbp, txn, skey, NULL, data, flags));
}
static int
__db_secondary_close(sdbp, flags)
DB *sdbp;
u_int32_t flags;
{
DB *primary;
int doclose;
doclose = 0;
primary = sdbp->s_primary;
MUTEX_THREAD_LOCK(primary->dbenv, primary->mutexp);
DB_ASSERT(sdbp->s_refcnt != 0);
if (--sdbp->s_refcnt == 0) {
LIST_REMOVE(sdbp, s_links);
doclose = 1;
}
MUTEX_THREAD_UNLOCK(primary->dbenv, primary->mutexp);
return (doclose ? __db_close(sdbp, NULL, flags) : 0);
}
static int
__db_append_primary(dbc, key, data)
DBC *dbc;
DBT *key, *data;
{
DB *dbp, *sdbp;
DBC *sdbc, *pdbc;
DBT oldpkey, pkey, pdata, skey;
int cmp, ret, t_ret;
dbp = dbc->dbp;
sdbp = NULL;
ret = 0;
pdbc = NULL;
if (F_ISSET(data, DB_DBT_PARTIAL) || F_ISSET(key, DB_DBT_PARTIAL)) {
if ((ret = __db_c_idup(dbc, &pdbc, DB_POSITION)) != 0)
return (ret);
memset(&pkey, 0, sizeof(DBT));
memset(&pdata, 0, sizeof(DBT));
if ((ret = __db_c_get(pdbc, &pkey, &pdata, DB_CURRENT)) != 0)
goto err;
key = &pkey;
data = &pdata;
}
for (sdbp = __db_s_first(dbp);
sdbp != NULL && ret == 0; ret = __db_s_next(&sdbp)) {
memset(&skey, 0, sizeof(DBT));
if ((ret = sdbp->s_callback(sdbp, key, data, &skey)) != 0) {
if (ret == DB_DONOTINDEX)
continue;
else
goto err;
}
if ((ret = __db_cursor_int(sdbp, dbc->txn, sdbp->type,
PGNO_INVALID, 0, dbc->locker, &sdbc)) != 0) {
FREE_IF_NEEDED(sdbp, &skey);
goto err;
}
if (CDB_LOCKING(sdbp->dbenv)) {
DB_ASSERT(sdbc->mylock.off == LOCK_INVALID);
F_SET(sdbc, DBC_WRITER);
}
if (!F_ISSET(sdbp, DB_AM_DUP)) {
memset(&oldpkey, 0, sizeof(DBT));
F_SET(&oldpkey, DB_DBT_MALLOC);
ret = __db_c_get(sdbc, &skey, &oldpkey,
DB_SET | (STD_LOCKING(dbc) ? DB_RMW : 0));
if (ret == 0) {
cmp = __bam_defcmp(sdbp, &oldpkey, key);
__os_ufree(sdbp->dbenv,
oldpkey.data);
if (cmp != 0) {
__db_err(sdbp->dbenv, "%s%s",
"Append results in a non-unique secondary key in",
" an index not configured to support duplicates");
ret = EINVAL;
goto err1;
}
} else if (ret != DB_NOTFOUND && ret != DB_KEYEMPTY)
goto err1;
}
ret = __db_c_put(sdbc, &skey, key, DB_UPDATE_SECONDARY);
err1: FREE_IF_NEEDED(sdbp, &skey);
if ((t_ret = __db_c_close(sdbc)) != 0 && ret == 0)
ret = t_ret;
if (ret != 0)
goto err;
}
err: if (pdbc != NULL && (t_ret = __db_c_close(pdbc)) != 0 && ret == 0)
ret = t_ret;
if (sdbp != NULL && (t_ret = __db_s_done(sdbp)) != 0 && ret == 0)
ret = t_ret;
return (ret);
}