db_am.c

Go to the documentation of this file.

/*-
 * See the file LICENSE for redistribution information.
 *
 * Copyright (c) 1998, 1999, 2000
 *      Sleepycat Software.  All rights reserved.
 */

#include "config.h"

#ifndef lint
static const char revid[] = "$Id: db__am_8c-source.html,v 1.1 2008/06/08 10:17:11 sebdiaz Exp $";
#endif /* not lint */

#ifndef NO_SYSTEM_INCLUDES
#include <sys/types.h>

#include <errno.h>
#include <string.h>
#endif

#include "db_int.h"
#include "db_page.h"
#include "db_shash.h"
#include "btree.h"
#include "hash.h"
#include "qam.h"
#include "lock.h"
#include "mp.h"
#include "txn.h"
#include "db_am.h"
#include "db_ext.h"

#ifdef DEBUG
#include "WordMonitor.h"
#endif /* DEBUG */

/*
 * CDB___db_cursor --
 *      Allocate and return a cursor.
 *
 * PUBLIC: int CDB___db_cursor __P((DB *, DB_TXN *, DBC **, u_int32_t));
 */
int
CDB___db_cursor(dbp, txn, dbcp, flags)
        DB *dbp;
        DB_TXN *txn;
        DBC **dbcp;
        u_int32_t flags;
{
        DB_ENV *dbenv;
        DBC *dbc;
        db_lockmode_t mode;
        u_int32_t op;
        int ret;

        dbenv = dbp->dbenv;

        PANIC_CHECK(dbenv);
        DB_ILLEGAL_BEFORE_OPEN(dbp, "DB->cursor");

        /* Check for invalid flags. */
        if ((ret = CDB___db_cursorchk(dbp, flags, F_ISSET(dbp, DB_AM_RDONLY))) != 0)
                return (ret);

        if ((ret =
            CDB___db_icursor(dbp, txn, dbp->type, PGNO_INVALID, 0, dbcp)) != 0)
                return (ret);
        dbc = *dbcp;

        /*
         * If this is CDB, do all the locking in the interface, which is
         * right here.
         */
        if (LOCKING(dbenv)) {
                op = LF_ISSET(DB_OPFLAGS_MASK);
                mode = (op == DB_WRITELOCK) ? DB_LOCK_WRITE :
                    ((op == DB_WRITECURSOR) ? DB_LOCK_IWRITE : DB_LOCK_READ);
                if ((ret = CDB_lock_get(dbenv, dbc->locker, 0,
                    &dbc->lock_dbt, mode, &dbc->mylock)) != 0) {
                        (void)CDB___db_c_close(dbc);
                        return (ret);
                }
                if (op == DB_WRITECURSOR)
                        F_SET(dbc, DBC_WRITECURSOR);
                if (op == DB_WRITELOCK)
                        F_SET(dbc, DBC_WRITER);
        }

        return (0);
}

/*
 * CDB___db_icursor --
 *      Internal version of CDB___db_cursor.  If dbcp is
 *      non-NULL it is assumed to point to an area to
 *      initialize as a cursor.
 *
 * PUBLIC: int CDB___db_icursor
 * PUBLIC:     __P((DB *, DB_TXN *, DBTYPE, db_pgno_t, int, DBC **));
 */
int
CDB___db_icursor(dbp, txn, dbtype, root, is_opd, dbcp)
        DB *dbp;
        DB_TXN *txn;
        DBTYPE dbtype;
        db_pgno_t root;
        int is_opd;
        DBC **dbcp;
{
        DBC *dbc, *adbc;
        DBC_INTERNAL *cp;
        DB_ENV *dbenv;
        int allocated, ret;

        dbenv = dbp->dbenv;
        allocated = 0;

        /*
         * Take one from the free list if it's available.  Take only the
         * right type.  With off page dups we may have different kinds
         * of cursors on the queue for a single database.
         */
        MUTEX_THREAD_LOCK(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);
                        dbc->flags = 0;
                        break;
                }
        MUTEX_THREAD_UNLOCK(dbp->mutexp);

        if (dbc == NULL) {
                if ((ret = CDB___os_calloc(dbp->dbenv, 1, sizeof(DBC), &dbc)) != 0)
                        return (ret);
                allocated = 1;
                dbc->flags = 0;

                dbc->dbp = dbp;

                /* Set up locking information. */
                if (LOCKING_ON(dbenv)) {
                        /*
                         * If we are not threaded, then there is no need to
                         * create new locker ids.  We know that no one else
                         * is running concurrently using this DB, so we can
                         * take a peek at any cursors on the active queue.
                         */
                        if (!DB_IS_THREADED(dbp) &&
                            (adbc = TAILQ_FIRST(&dbp->active_queue)) != NULL)
                                dbc->lid = adbc->lid;
                        else
                                if ((ret = CDB_lock_id(dbenv, &dbc->lid)) != 0)
                                        goto err;

                        memcpy(dbc->lock.fileid, dbp->fileid, DB_FILE_ID_LEN);
                        if (LOCKING(dbenv)) {
                                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;
                        }
                }
                /* Init the DBC internal structure. */
                switch (dbtype) {
                case DB_BTREE:
                case DB_RECNO:
                        if ((ret = CDB___bam_c_init(dbc, dbtype)) != 0)
                                goto err;
                        break;
                case DB_HASH:
                        if ((ret = CDB___ham_c_init(dbc)) != 0)
                                goto err;
                        break;
                case DB_QUEUE:
                        if ((ret = CDB___qam_c_init(dbc)) != 0)
                                goto err;
                        break;
                default:
                        ret = CDB___db_unknown_type(dbp->dbenv,
                            "CDB___db_icursor", dbtype);
                        goto err;
                }

                cp = dbc->internal;
        }

        /* Refresh the DBC structure. */
        dbc->dbtype = dbtype;

        if ((dbc->txn = txn) == NULL)
                dbc->locker = dbc->lid;
        else
                dbc->locker = txn->txnid;

        if (is_opd)
                F_SET(dbc, DBC_OPD);
        if (F_ISSET(dbp, DB_AM_RECOVER))
                F_SET(dbc, DBC_RECOVER);

        /* Refresh the DBC internal structure. */
        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 = CDB___bam_c_refresh(dbc)) != 0)
                        goto err;
                break;
        case DB_HASH:
        case DB_QUEUE:
                break;
        default:
                ret = CDB___db_unknown_type(dbp->dbenv, "CDB___db_icursor", dbp->type);
                goto err;
        }

        MUTEX_THREAD_LOCK(dbp->mutexp);
        TAILQ_INSERT_TAIL(&dbp->active_queue, dbc, links);
        F_SET(dbc, DBC_ACTIVE);
        MUTEX_THREAD_UNLOCK(dbp->mutexp);

        *dbcp = dbc;
        return (0);

err:    if (allocated)
                CDB___os_free(dbc, sizeof(*dbc));
        return (ret);
}

#ifdef DEBUG
/*
 * CDB___db_cprint --
 *      Display the current cursor list.
 *
 * PUBLIC: int CDB___db_cprint __P((DB *));
 */
int
CDB___db_cprint(dbp)
        DB *dbp;
{
        static const FN fn[] = {
                { DBC_ACTIVE,           "active" },
                { DBC_OPD,              "off-page-dup" },
                { DBC_RECOVER,          "recover" },
                { DBC_RMW,              "read-modify-write" },
                { DBC_WRITECURSOR,      "write cursor" },
                { DBC_WRITEDUP,         "internally dup'ed write cursor" },
                { DBC_WRITER,           "short-term write cursor" },
                { 0,                    NULL }
        };
        DBC *dbc;
        DBC_INTERNAL *cp;
        char *s;

        MUTEX_THREAD_LOCK(dbp->mutexp);
        for (dbc = TAILQ_FIRST(&dbp->active_queue);
            dbc != NULL; dbc = TAILQ_NEXT(dbc, links)) {
                switch (dbc->dbtype) {
                case DB_BTREE:
                        s = "btree";
                        break;
                case DB_HASH:
                        s = "hash";
                        break;
                case DB_RECNO:
                        s = "recno";
                        break;
                case DB_QUEUE:
                        s = "queue";
                        break;
                default:
                        DB_ASSERT(0);
                        return (1);
                }
                cp = dbc->internal;
                fprintf(stderr, "%s/%#0lx: opd: %#0lx\n",
                    s, P_TO_ULONG(dbc), P_TO_ULONG(cp->opd));
                fprintf(stderr, "\ttxn: %#0lx 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);
                CDB___db_prflags(dbc->flags, fn, stderr);
                fprintf(stderr, "\n");

                if (dbp->type == DB_BTREE)
                        CDB___bam_cprint(dbc);
        }
        for (dbc = TAILQ_FIRST(&dbp->free_queue);
            dbc != NULL; dbc = TAILQ_NEXT(dbc, links))
                fprintf(stderr, "free: %#0lx ", P_TO_ULONG(dbc));
        fprintf(stderr, "\n");
        MUTEX_THREAD_UNLOCK(dbp->mutexp);

        return (0);
}
#endif /* DEBUG */

/*
 * db_fd --
 *      Return a file descriptor for flock'ing.
 *
 * PUBLIC: int CDB___db_fd __P((DB *, int *));
 */
int
CDB___db_fd(dbp, fdp)
        DB *dbp;
        int *fdp;
{
        DB_FH *fhp;
        int ret;

        PANIC_CHECK(dbp->dbenv);
        DB_ILLEGAL_BEFORE_OPEN(dbp, "DB->fd");

        /*
         * XXX
         * Truly spectacular layering violation.
         */
        if ((ret = CDB___mp_xxx_fh(dbp->mpf, &fhp)) != 0)
                return (ret);

        if (F_ISSET(fhp, DB_FH_VALID)) {
                *fdp = fhp->fd;
                return (0);
        } else {
                *fdp = -1;
                CDB___db_err(dbp->dbenv, "DB does not have a valid file handle.");
                return (ENOENT);
        }
}

/*
 * CDB___db_get --
 *      Return a key/data pair.
 *
 * PUBLIC: int CDB___db_get __P((DB *, DB_TXN *, DBT *, DBT *, u_int32_t));
 */
int
CDB___db_get(dbp, txn, key, data, flags)
        DB *dbp;
        DB_TXN *txn;
        DBT *key, *data;
        u_int32_t flags;
{
        DBC *dbc;
        int ret, t_ret;

        PANIC_CHECK(dbp->dbenv);
        DB_ILLEGAL_BEFORE_OPEN(dbp, "DB->get");

        if ((ret = CDB___db_getchk(dbp, key, data, flags)) != 0)
                return (ret);

        if ((ret = dbp->cursor(dbp, txn, &dbc, 0)) != 0)
                return (ret);

        DEBUG_LREAD(dbc, txn, "CDB___db_get", key, NULL, flags);

        ret = dbc->c_get(dbc, key, data,
            flags == 0 || flags == DB_RMW ? flags | DB_SET : flags);

        if ((t_ret = CDB___db_c_close(dbc)) != 0 && ret == 0)
                ret = t_ret;

#ifdef DEBUG
        switch(flags) {
        case 0:
          word_monitor_add(DB_MONITOR(dbp->dbenv), WORD_MONITOR_GET, 1);
          break;
        case DB_NEXT:
          word_monitor_add(DB_MONITOR(dbp->dbenv), WORD_MONITOR_GET_NEXT, 1);
          break;
        case DB_SET_RANGE:
          word_monitor_add(DB_MONITOR(dbp->dbenv), WORD_MONITOR_GET_SET_RANGE, 1);
          break;
        default:
          word_monitor_add(DB_MONITOR(dbp->dbenv), WORD_MONITOR_GET_OTHER, 1);
          break;
        }
#endif /* DEBUG */
        return (ret);
}

/*
 * CDB___db_put --
 *      Store a key/data pair.
 *
 * PUBLIC: int CDB___db_put __P((DB *, DB_TXN *, DBT *, DBT *, u_int32_t));
 */
int
CDB___db_put(dbp, txn, key, data, flags)
        DB *dbp;
        DB_TXN *txn;
        DBT *key, *data;
        u_int32_t flags;
{
        DBC *dbc;
        DBT tdata;
        int ret, t_ret;

        PANIC_CHECK(dbp->dbenv);
        DB_ILLEGAL_BEFORE_OPEN(dbp, "DB->put");

        if ((ret = CDB___db_putchk(dbp, key, data,
            flags, F_ISSET(dbp, DB_AM_RDONLY),
            F_ISSET(dbp, DB_AM_DUP) || F_ISSET(key, DB_DBT_DUPOK))) != 0)
                return (ret);

        if ((ret = dbp->cursor(dbp, txn, &dbc, DB_WRITELOCK)) != 0)
                return (ret);

        DEBUG_LWRITE(dbc, txn, "CDB___db_put", key, data, flags);

        if (flags == DB_NOOVERWRITE) {
                flags = 0;
                /*
                 * Set DB_DBT_USERMEM, this might be a threaded application and
                 * the flags checking will catch us.  We don't want the actual
                 * data, so request a partial of length 0.
                 */
                memset(&tdata, 0, sizeof(tdata));
                F_SET(&tdata, DB_DBT_USERMEM | DB_DBT_PARTIAL);

                /*
                 * If we're doing page-level locking, set the read-modify-write
                 * flag, we're going to overwrite immediately.
                 */
                if ((ret = dbc->c_get(dbc, key, &tdata,
                    DB_SET | (STD_LOCKING(dbc) ? DB_RMW : 0))) == 0)
                        ret = DB_KEYEXIST;
                else if (ret == DB_NOTFOUND)
                        ret = 0;
        }
        if (ret == 0)
                ret = dbc->c_put(dbc,
                     key, data, flags == 0 ? DB_KEYLAST : flags);

        if ((t_ret = CDB___db_c_close(dbc)) != 0 && ret == 0)
                ret = t_ret;

#ifdef DEBUG
        word_monitor_add(DB_MONITOR(dbp->dbenv), WORD_MONITOR_PUT, 1);
#endif /* DEBUG */

        return (ret);
}

/*
 * CDB___db_sync --
 *      Flush the database cache.
 *
 * PUBLIC: int CDB___db_sync __P((DB *, u_int32_t));
 */
int
CDB___db_sync(dbp, flags)
        DB *dbp;
        u_int32_t flags;
{
        int ret, t_ret;

        PANIC_CHECK(dbp->dbenv);
        DB_ILLEGAL_BEFORE_OPEN(dbp, "DB->sync");

        if ((ret = CDB___db_syncchk(dbp, flags)) != 0)
                return (ret);

        /* Read-only trees never need to be sync'd. */
        if (F_ISSET(dbp, DB_AM_RDONLY))
                return (0);

        /* If it's a Recno tree, write the backing source text file. */
        if (dbp->type == DB_RECNO)
                ret = CDB___ram_writeback(dbp);

        /* If the tree was never backed by a database file, we're done. */
        if (F_ISSET(dbp, DB_AM_INMEM))
                return (0);

        /* Flush any dirty pages from the cache to the backing file. */
        if ((t_ret = CDB_memp_fsync(dbp->mpf)) != 0 && ret == 0)
                ret = t_ret;
        return (ret);
}

---