#include "db_config.h"
#include "db_int.h"
#include "dbinc/crypto.h"
#include "dbinc/db_page.h"
#include "dbinc/db_swap.h"
#include "dbinc/db_am.h"
#include "dbinc/lock.h"
#include "dbinc/mp.h"
#include "dbinc/qam.h"
#include "dbinc/fop.h"
static int __qam_init_meta __P((DB *, QMETA *));
int
__qam_open(dbp, txn, name, base_pgno, mode, flags)
DB *dbp;
DB_TXN *txn;
const char *name;
db_pgno_t base_pgno;
int mode;
u_int32_t flags;
{
DBC *dbc;
DB_ENV *dbenv;
DB_LOCK metalock;
DB_MPOOLFILE *mpf;
QMETA *qmeta;
QUEUE *t;
int ret, t_ret;
dbenv = dbp->dbenv;
mpf = dbp->mpf;
t = dbp->q_internal;
ret = 0;
qmeta = NULL;
if (name == NULL && t->page_ext != 0) {
__db_errx(dbenv,
"Extent size may not be specified for in-memory queue database");
return (EINVAL);
}
if (MULTIVERSION(dbp)) {
__db_errx(dbenv,
"Multiversion queue databases are not supported");
return (EINVAL);
}
dbp->db_am_remove = __qam_remove;
dbp->db_am_rename = __qam_rename;
if ((ret = __db_cursor(dbp, txn, &dbc,
LF_ISSET(DB_CREATE) && CDB_LOCKING(dbenv) ?
DB_WRITECURSOR : 0)) != 0)
return (ret);
if ((ret =
__db_lget(dbc, 0, base_pgno, DB_LOCK_READ, 0, &metalock)) != 0)
goto err;
if ((ret = __memp_fget(mpf, &base_pgno, dbc->txn, 0, &qmeta)) != 0)
goto err;
if (qmeta->dbmeta.magic != DB_QAMMAGIC) {
__db_errx(dbenv, "%s: unexpected file type or format", name);
ret = EINVAL;
goto err;
}
t->page_ext = qmeta->page_ext;
if (t->page_ext != 0 && (ret = __qam_set_ext_data(dbp, name)) != 0)
goto err;
if (mode == 0)
mode = __db_omode("rw-rw----");
t->mode = mode;
t->re_pad = (int)qmeta->re_pad;
t->re_len = qmeta->re_len;
t->rec_page = qmeta->rec_page;
t->q_meta = base_pgno;
t->q_root = base_pgno + 1;
err: if (qmeta != NULL && (t_ret =
__memp_fput(mpf, qmeta, dbc->priority)) != 0 && ret == 0)
ret = t_ret;
if ((t_ret = __LPUT(dbc, metalock)) != 0 && ret == 0)
ret = t_ret;
if ((t_ret = __dbc_close(dbc)) != 0 && ret == 0)
ret = t_ret;
return (ret);
}
int
__qam_set_ext_data(dbp, name)
DB *dbp;
const char *name;
{
QUEUE *t;
int ret;
t = dbp->q_internal;
t->pginfo.db_pagesize = dbp->pgsize;
t->pginfo.flags =
F_ISSET(dbp, (DB_AM_CHKSUM | DB_AM_ENCRYPT | DB_AM_SWAP));
t->pginfo.type = dbp->type;
t->pgcookie.data = &t->pginfo;
t->pgcookie.size = sizeof(DB_PGINFO);
if ((ret = __os_strdup(dbp->dbenv, name, &t->path)) != 0)
return (ret);
t->dir = t->path;
if ((t->name = __db_rpath(t->path)) == NULL) {
t->name = t->path;
t->dir = PATH_DOT;
} else
*t->name++ = '\0';
return (0);
}
int
__qam_metachk(dbp, name, qmeta)
DB *dbp;
const char *name;
QMETA *qmeta;
{
DB_ENV *dbenv;
u_int32_t vers;
int ret;
dbenv = dbp->dbenv;
ret = 0;
vers = qmeta->dbmeta.version;
if (F_ISSET(dbp, DB_AM_SWAP))
M_32_SWAP(vers);
switch (vers) {
case 1:
case 2:
__db_errx(dbenv,
"%s: queue version %lu requires a version upgrade",
name, (u_long)vers);
return (DB_OLD_VERSION);
case 3:
case 4:
break;
default:
__db_errx(dbenv,
"%s: unsupported qam version: %lu", name, (u_long)vers);
return (EINVAL);
}
if (F_ISSET(dbp, DB_AM_SWAP) && (ret = __qam_mswap((PAGE *)qmeta)) != 0)
return (ret);
if (dbp->type != DB_QUEUE && dbp->type != DB_UNKNOWN)
return (EINVAL);
dbp->type = DB_QUEUE;
DB_ILLEGAL_METHOD(dbp, DB_OK_QUEUE);
dbp->pgsize = qmeta->dbmeta.pagesize;
memcpy(dbp->fileid, qmeta->dbmeta.uid, DB_FILE_ID_LEN);
dbp->db_am_rename = __qam_rename;
dbp->db_am_remove = __qam_remove;
return (ret);
}
static int
__qam_init_meta(dbp, meta)
DB *dbp;
QMETA *meta;
{
DB_ENV *dbenv;
QUEUE *t;
dbenv = dbp->dbenv;
t = dbp->q_internal;
memset(meta, 0, sizeof(QMETA));
LSN_NOT_LOGGED(meta->dbmeta.lsn);
meta->dbmeta.pgno = PGNO_BASE_MD;
meta->dbmeta.last_pgno = 0;
meta->dbmeta.magic = DB_QAMMAGIC;
meta->dbmeta.version = DB_QAMVERSION;
meta->dbmeta.pagesize = dbp->pgsize;
if (F_ISSET(dbp, DB_AM_CHKSUM))
FLD_SET(meta->dbmeta.metaflags, DBMETA_CHKSUM);
if (F_ISSET(dbp, DB_AM_ENCRYPT)) {
meta->dbmeta.encrypt_alg =
((DB_CIPHER *)dbenv->crypto_handle)->alg;
DB_ASSERT(dbenv, meta->dbmeta.encrypt_alg != 0);
meta->crypto_magic = meta->dbmeta.magic;
}
meta->dbmeta.type = P_QAMMETA;
meta->re_pad = (u_int32_t)t->re_pad;
meta->re_len = t->re_len;
meta->rec_page = CALC_QAM_RECNO_PER_PAGE(dbp);
meta->cur_recno = 1;
meta->first_recno = 1;
meta->page_ext = t->page_ext;
t->rec_page = meta->rec_page;
memcpy(meta->dbmeta.uid, dbp->fileid, DB_FILE_ID_LEN);
if (QAM_RECNO_PER_PAGE(dbp) < 1) {
__db_errx(dbenv,
"Record size of %lu too large for page size of %lu",
(u_long)t->re_len, (u_long)dbp->pgsize);
return (EINVAL);
}
return (0);
}
int
__qam_new_file(dbp, txn, fhp, name)
DB *dbp;
DB_TXN *txn;
DB_FH *fhp;
const char *name;
{
DBT pdbt;
DB_ENV *dbenv;
DB_MPOOLFILE *mpf;
DB_PGINFO pginfo;
QMETA *meta;
db_pgno_t pgno;
int ret, t_ret;
if (F_ISSET(dbp, DB_AM_INMEM)) {
mpf = dbp->mpf;
pgno = PGNO_BASE_MD;
if ((ret = __memp_fget(mpf, &pgno, txn,
DB_MPOOL_CREATE | DB_MPOOL_DIRTY, &meta)) != 0)
return (ret);
if ((ret = __qam_init_meta(dbp, meta)) != 0)
goto err1;
if ((ret = __db_log_page(dbp,
txn, &meta->dbmeta.lsn, pgno, (PAGE *)meta)) != 0)
goto err1;
err1: if ((t_ret =
__memp_fput(mpf, meta, dbp->priority)) != 0 && ret == 0)
ret = t_ret;
} else {
dbenv = dbp->dbenv;
if ((ret = __os_calloc(dbenv, 1, dbp->pgsize, &meta)) != 0)
return (ret);
if ((ret = __qam_init_meta(dbp, meta)) != 0)
goto err2;
pginfo.db_pagesize = dbp->pgsize;
pginfo.flags =
F_ISSET(dbp, (DB_AM_CHKSUM | DB_AM_ENCRYPT | DB_AM_SWAP));
pginfo.type = DB_QUEUE;
DB_SET_DBT(pdbt, &pginfo, sizeof(pginfo));
if ((ret = __db_pgout(dbenv, PGNO_BASE_MD, meta, &pdbt)) != 0)
goto err2;
ret = __fop_write(dbenv, txn, name,
DB_APP_DATA, fhp, dbp->pgsize, 0, 0, meta, dbp->pgsize, 1,
F_ISSET(dbp, DB_AM_NOT_DURABLE) ? DB_LOG_NOT_DURABLE : 0);
err2: __os_free(dbenv, meta);
}
return (ret);
}