Logo Search packages:      
Sourcecode: db3 version File versions  Download package

bt_cursor.c

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

#include "db_config.h"

#ifndef lint
static const char revid[] = "$Id: bt_cursor.c,v 11.88 2001/01/11 18:19:49 bostic Exp $";
#endif /* not lint */

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

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

#include "db_int.h"
#include "db_page.h"
#include "db_shash.h"
#include "btree.h"
#include "lock.h"
#include "qam.h"
#include "common_ext.h"

static int  __bam_c_close __P((DBC *, db_pgno_t, int *));
static int  __bam_c_del __P((DBC *));
static int  __bam_c_destroy __P((DBC *));
static int  __bam_c_first __P((DBC *));
static int  __bam_c_get __P((DBC *, DBT *, DBT *, u_int32_t, db_pgno_t *));
static int  __bam_c_getstack __P((DBC *));
static int  __bam_c_last __P((DBC *));
static int  __bam_c_next __P((DBC *, int));
static int  __bam_c_physdel __P((DBC *));
static int  __bam_c_prev __P((DBC *));
static int  __bam_c_put __P((DBC *, DBT *, DBT *, u_int32_t, db_pgno_t *));
static void __bam_c_reset __P((BTREE_CURSOR *));
static int  __bam_c_search __P((DBC *, const DBT *, u_int32_t, int *));
static int  __bam_c_writelock __P((DBC *));
static int  __bam_getboth_finddatum __P((DBC *, DBT *));
static int  __bam_getbothc __P((DBC *, DBT *));
static int  __bam_isopd __P((DBC *, db_pgno_t *));

/*
 * Acquire a new page/lock.  If we hold a page/lock, discard the page, and
 * lock-couple the lock.
 *
 * !!!
 * We have to handle both where we have a lock to lock-couple and where we
 * don't -- we don't duplicate locks when we duplicate cursors if we are
 * running in a transaction environment as there's no point if locks are
 * never discarded.  This means that the cursor may or may not hold a lock.
 */
#undef      ACQUIRE
#define     ACQUIRE(dbc, mode, lpgno, lock, fpgno, pagep, ret) {\
      if ((pagep) != NULL) {                                \
            ret = memp_fput((dbc)->dbp->mpf, pagep, 0);           \
            pagep = NULL;                                   \
      } else                                                \
            ret = 0;                                  \
      if ((ret) == 0 && STD_LOCKING(dbc))                   \
            ret = __db_lget(dbc,                            \
                (lock).off == LOCK_INVALID ? 0 : LCK_COUPLE,      \
                lpgno, mode, 0, &lock);                     \
      else                                            \
            (lock).off = LOCK_INVALID;                      \
      if ((ret) == 0)                                       \
            ret = memp_fget((dbc)->dbp->mpf, &(fpgno), 0, &(pagep));\
}

/* Acquire a new page/lock for a cursor. */
#undef      ACQUIRE_CUR
#define     ACQUIRE_CUR(dbc, mode, ret) {                         \
      BTREE_CURSOR *__cp = (BTREE_CURSOR *)(dbc)->internal;       \
      ACQUIRE(dbc, mode,                                    \
          __cp->pgno, __cp->lock, __cp->pgno, __cp->page, ret);   \
      if ((ret) == 0)                                       \
            __cp->lock_mode = (mode);                       \
}

/*
 * Acquire a new page/lock for a cursor, and move the cursor on success.
 * The reason that this is a separate macro is because we don't want to
 * set the pgno/indx fields in the cursor until we actually have the lock,
 * otherwise the cursor adjust routines will adjust the cursor even though
 * we're not really on the page.
 */
#undef      ACQUIRE_CUR_SET
#define     ACQUIRE_CUR_SET(dbc, mode, p, ret) {                        \
      BTREE_CURSOR *__cp = (BTREE_CURSOR *)(dbc)->internal;       \
      ACQUIRE(dbc, mode, p, __cp->lock, p, __cp->page, ret);            \
      if ((ret) == 0) {                               \
            __cp->pgno = p;                           \
            __cp->indx = 0;                           \
            __cp->lock_mode = (mode);                       \
      }                                               \
}

/*
 * Acquire a write lock if we don't already have one.
 *
 * !!!
 * See ACQUIRE macro on why we handle cursors that don't have locks.
 */
#undef      ACQUIRE_WRITE_LOCK
#define     ACQUIRE_WRITE_LOCK(dbc, ret) {                              \
      BTREE_CURSOR *__cp = (BTREE_CURSOR *)(dbc)->internal;       \
      ret = 0;                                        \
      if (STD_LOCKING(dbc) &&                               \
          __cp->lock_mode != DB_LOCK_WRITE &&                     \
          ((ret) = __db_lget(dbc,                           \
          __cp->lock.off == LOCK_INVALID ? 0 : LCK_COUPLE,        \
          __cp->pgno, DB_LOCK_WRITE, 0, &__cp->lock)) == 0)       \
            __cp->lock_mode = DB_LOCK_WRITE;                \
}

/* Discard the current page/lock. */
#undef      DISCARD
#define     DISCARD(dbc, ldiscard, lock, pagep, ret) {                  \
      int __t_ret;                                          \
      if ((pagep) != NULL) {                                \
            ret = memp_fput((dbc)->dbp->mpf, pagep, 0);           \
            pagep = NULL;                                   \
      } else                                                \
            ret = 0;                                  \
      if ((lock).off != LOCK_INVALID) {                     \
            __t_ret = ldiscard ?                            \
                __LPUT((dbc), lock): __TLPUT((dbc), lock);        \
            if (__t_ret != 0 && (ret) == 0)                       \
                  ret = __t_ret;                            \
            (lock).off = LOCK_INVALID;                      \
      }                                               \
}

/* Discard the current page/lock for a cursor. */
#undef      DISCARD_CUR
#define     DISCARD_CUR(dbc, ret) {                               \
      BTREE_CURSOR *__cp = (BTREE_CURSOR *)(dbc)->internal;       \
      DISCARD(dbc, 0, __cp->lock, __cp->page, ret);               \
      if ((ret) == 0)                                       \
            __cp->lock_mode = DB_LOCK_NG;                   \
}

/* If on-page item is a deleted record. */
#undef      IS_DELETED
#define     IS_DELETED(page, indx)                                \
      B_DISSET(GET_BKEYDATA(page,                           \
          (indx) + (TYPE(page) == P_LBTREE ? O_INDX : 0))->type)
#undef      IS_CUR_DELETED
#define     IS_CUR_DELETED(dbc)                                   \
      IS_DELETED((dbc)->internal->page, (dbc)->internal->indx)

/*
 * Test to see if two cursors could point to duplicates of the same key.
 * In the case of off-page duplicates they are they same, as the cursors
 * will be in the same off-page duplicate tree.  In the case of on-page
 * duplicates, the key index offsets must be the same.  For the last test,
 * as the original cursor may not have a valid page pointer, we use the
 * current cursor's.
 */
#undef      IS_DUPLICATE
#define     IS_DUPLICATE(dbc, i1, i2)                             \
          (((PAGE *)(dbc)->internal->page)->inp[i1] ==            \
           ((PAGE *)(dbc)->internal->page)->inp[i2])
#undef      IS_CUR_DUPLICATE
#define     IS_CUR_DUPLICATE(dbc, orig_pgno, orig_indx)                 \
      (F_ISSET(dbc, DBC_OPD) ||                             \
          (orig_pgno == (dbc)->internal->pgno &&                  \
          IS_DUPLICATE(dbc, (dbc)->internal->indx, orig_indx)))

/*
 * __bam_c_reset --
 *    Initialize internal cursor structure.
 */
static void
__bam_c_reset(cp)
      BTREE_CURSOR *cp;
{
      cp->csp = cp->sp;
      cp->lock.off = LOCK_INVALID;
      cp->lock_mode = DB_LOCK_NG;
      cp->recno = RECNO_OOB;
      cp->order = INVALID_ORDER;
      cp->flags = 0;
}

/*
 * __bam_c_init --
 *    Initialize the access private portion of a cursor
 *
 * PUBLIC: int __bam_c_init __P((DBC *, DBTYPE));
 */
int
__bam_c_init(dbc, dbtype)
      DBC *dbc;
      DBTYPE dbtype;
{
      BTREE *t;
      BTREE_CURSOR *cp;
      DB *dbp;
      int ret;
      u_int32_t minkey;

      dbp = dbc->dbp;

      /* Allocate/initialize the internal structure. */
      if (dbc->internal == NULL) {
            if ((ret = __os_malloc(dbp->dbenv,
                sizeof(BTREE_CURSOR), NULL, &cp)) != 0)
                  return (ret);
            dbc->internal = (DBC_INTERNAL *)cp;

            cp->sp = cp->csp = cp->stack;
            cp->esp = cp->stack + sizeof(cp->stack) / sizeof(cp->stack[0]);
      } else
            cp = (BTREE_CURSOR *)dbc->internal;
      __bam_c_reset(cp);

      /* Initialize methods. */
      dbc->c_close = __db_c_close;
      dbc->c_count = __db_c_count;
      dbc->c_del = __db_c_del;
      dbc->c_dup = __db_c_dup;
      dbc->c_get = __db_c_get;
      dbc->c_put = __db_c_put;
      if (dbtype == DB_BTREE) {
            dbc->c_am_close = __bam_c_close;
            dbc->c_am_del = __bam_c_del;
            dbc->c_am_destroy = __bam_c_destroy;
            dbc->c_am_get = __bam_c_get;
            dbc->c_am_put = __bam_c_put;
            dbc->c_am_writelock = __bam_c_writelock;
      } else {
            dbc->c_am_close = __bam_c_close;
            dbc->c_am_del = __ram_c_del;
            dbc->c_am_destroy = __bam_c_destroy;
            dbc->c_am_get = __ram_c_get;
            dbc->c_am_put = __ram_c_put;
            dbc->c_am_writelock = __bam_c_writelock;
      }

      /*
       * The btree leaf page data structures require that two key/data pairs
       * (or four items) fit on a page, but other than that there's no fixed
       * requirement.  The btree off-page duplicates only require two items,
       * to be exact, but requiring four for them as well seems reasonable.
       *
       * Recno uses the btree bt_ovflsize value -- it's close enough.
       */
      t = dbp->bt_internal;
      minkey = F_ISSET(dbc, DBC_OPD) ? 2 : t->bt_minkey;
      cp->ovflsize = B_MINKEY_TO_OVFLSIZE(minkey, dbp->pgsize);

      return (0);
}

/*
 * __bam_c_refresh
 *    Set things up properly for cursor re-use.
 *
 * PUBLIC: int __bam_c_refresh __P((DBC *));
 */
int
__bam_c_refresh(dbc)
      DBC *dbc;
{
      BTREE_CURSOR *cp;
      DB *dbp;

      dbp = dbc->dbp;
      cp = (BTREE_CURSOR *)dbc->internal;
      __bam_c_reset(cp);

      /*
       * If our caller set the root page number, it's because the root was
       * known.  This is always the case for off page dup cursors.  Else,
       * pull it out of our internal information.
       */
      if (cp->root == PGNO_INVALID)
            cp->root = ((BTREE *)dbp->bt_internal)->bt_root;

      /* Initialize for record numbers. */
      if (F_ISSET(dbc, DBC_OPD) ||
          dbc->dbtype == DB_RECNO || F_ISSET(dbp, DB_BT_RECNUM)) {
            F_SET(cp, C_RECNUM);

            /*
             * All btrees that support record numbers, optionally standard
             * recno trees, and all off-page duplicate recno trees have
             * mutable record numbers.
             */
            if ((F_ISSET(dbc, DBC_OPD) && dbc->dbtype == DB_RECNO) ||
                F_ISSET(dbp, DB_BT_RECNUM | DB_RE_RENUMBER))
                  F_SET(cp, C_RENUMBER);
      }

      return (0);
}

/*
 * __bam_c_close --
 *    Close down the cursor.
 */
static int
__bam_c_close(dbc, root_pgno, rmroot)
      DBC *dbc;
      db_pgno_t root_pgno;
      int *rmroot;
{
      BTREE_CURSOR *cp, *cp_opd, *cp_c;
      DB *dbp;
      DBC *dbc_opd, *dbc_c;
      PAGE *h;
      u_int32_t num;
      int cdb_lock, ret, t_ret;

      dbp = dbc->dbp;
      cp = (BTREE_CURSOR *)dbc->internal;
      cp_opd = (dbc_opd = cp->opd) == NULL ?
          NULL : (BTREE_CURSOR *)dbc_opd->internal;
      cdb_lock = ret = 0;

      /*
       * There are 3 ways this function is called:
       *
       * 1. Closing a primary cursor: we get called with a pointer to a
       *    primary cursor that has a NULL opd field.  This happens when
       *    closing a btree/recno database cursor without an associated
       *    off-page duplicate tree.
       *
       * 2. Closing a primary and an off-page duplicate cursor stack: we
       *    get called with a pointer to the primary cursor which has a
       *    non-NULL opd field.  This happens when closing a btree cursor
       *    into database with an associated off-page btree/recno duplicate
       *    tree. (It can't be a primary recno database, recno databases
       *    don't support duplicates.)
       *
       * 3. Closing an off-page duplicate cursor stack: we get called with
       *    a pointer to the off-page duplicate cursor.  This happens when
       *    closing a non-btree database that has an associated off-page
       *    btree/recno duplicate tree or for a btree database when the
       *    opd tree is not empty (root_pgno == PGNO_INVALID).
       *
       * If either the primary or off-page duplicate cursor deleted a btree
       * key/data pair, check to see if the item is still referenced by a
       * different cursor.  If it is, confirm that cursor's delete flag is
       * set and leave it to that cursor to do the delete.
       *
       * NB: The test for == 0 below is correct.  Our caller already removed
       * our cursor argument from the active queue, we won't find it when we
       * search the queue in __bam_ca_delete().
       * NB: It can't be true that both the primary and off-page duplicate
       * cursors have deleted a btree key/data pair.  Either the primary
       * cursor may have deleted an item and there's no off-page duplicate
       * cursor, or there's an off-page duplicate cursor and it may have
       * deleted an item.
       *
       * Primary recno databases aren't an issue here.  Recno keys are either
       * deleted immediately or never deleted, and do not have to be handled
       * here.
       *
       * Off-page duplicate recno databases are an issue here, cases #2 and
       * #3 above can both be off-page recno databases.  The problem is the
       * same as the final problem for off-page duplicate btree databases.
       * If we no longer need the off-page duplicate tree, we want to remove
       * it.  For off-page duplicate btrees, we are done with the tree when
       * we delete the last item it contains, i.e., there can be no further
       * references to it when it's empty.  For off-page duplicate recnos,
       * we remove items from the tree as the application calls the remove
       * function, so we are done with the tree when we close the last cursor
       * that references it.
       *
       * We optionally take the root page number from our caller.  If the
       * primary database is a btree, we can get it ourselves because dbc
       * is the primary cursor.  If the primary database is not a btree,
       * the problem is that we may be dealing with a stack of pages.  The
       * cursor we're using to do the delete points at the bottom of that
       * stack and we need the top of the stack.
       */
      if (F_ISSET(cp, C_DELETED)) {
            dbc_c = dbc;
            switch (dbc->dbtype) {
            case DB_BTREE:                      /* Case #1, #3. */
                  if (__bam_ca_delete(dbp, cp->pgno, cp->indx, 1) == 0)
                        goto lock;
                  goto done;
            case DB_RECNO:
                  if (!F_ISSET(dbc, DBC_OPD))   /* Case #1. */
                        goto done;
                                          /* Case #3. */
                  if (__ram_ca_delete(dbp, cp->root) == 0)
                        goto lock;
                  goto done;
            default:
                  return (__db_unknown_type(dbp->dbenv,
                        "__bam_c_close", dbc->dbtype));
            }
      }

      if (dbc_opd == NULL)
            goto done;

      if (F_ISSET(cp_opd, C_DELETED)) {         /* Case #2. */
            /*
             * We will not have been provided a root page number.  Acquire
             * one from the primary database.
             */
            if ((ret = memp_fget(dbp->mpf, &cp->pgno, 0, &h)) != 0)
                  goto err;
            root_pgno = GET_BOVERFLOW(h, cp->indx + O_INDX)->pgno;
            if ((ret = memp_fput(dbp->mpf, h, 0)) != 0)
                  goto err;

            dbc_c = dbc_opd;
            switch (dbc_opd->dbtype) {
            case DB_BTREE:
                  if (__bam_ca_delete(
                      dbp, cp_opd->pgno, cp_opd->indx, 1) == 0)
                        goto lock;
                  goto done;
            case DB_RECNO:
                  if (__ram_ca_delete(dbp, cp_opd->root) == 0)
                        goto lock;
                  goto done;
            default:
                  return (__db_unknown_type(dbp->dbenv,
                        "__bam_c_close", dbc->dbtype));
            }
      }
      goto done;

lock: cp_c = (BTREE_CURSOR *)dbc_c->internal;

      /*
       * If this is CDB, upgrade the lock if necessary.  While we acquired
       * the write lock to logically delete the record, we released it when
       * we returned from that call, and so may not be holding a write lock
       * at the moment.  NB: to get here in CDB we must either be holding a
       * write lock or be the only cursor that is permitted to acquire write
       * locks.  The reason is that there can never be more than a single CDB
       * write cursor (that cursor cannot be dup'd), and so that cursor must
       * be closed and the item therefore deleted before any other cursor
       * could acquire a reference to this item.
       *
       * Note that dbc may be an off-page dup cursor;  this is the sole
       * instance in which an OPD cursor does any locking, but it's necessary
       * because we may be closed by ourselves without a parent cursor
       * handy, and we have to do a lock upgrade on behalf of somebody.
       * If this is the case, the OPD has been given the parent's locking
       * info in __db_c_get--the OPD is also a WRITEDUP.
       */
      if (CDB_LOCKING(dbp->dbenv)) {
            DB_ASSERT(!F_ISSET(dbc, DBC_OPD) || F_ISSET(dbc, DBC_WRITEDUP));
            if (!F_ISSET(dbc, DBC_WRITER)) {
                  if ((ret =
                      lock_get(dbp->dbenv, dbc->locker, DB_LOCK_UPGRADE,
                      &dbc->lock_dbt, DB_LOCK_WRITE, &dbc->mylock)) != 0)
                        goto err;
                  cdb_lock = 1;
            }

            cp_c->lock.off = LOCK_INVALID;
            if ((ret =
                memp_fget(dbp->mpf, &cp_c->pgno, 0, &cp_c->page)) != 0)
                  goto err;

            goto delete;
      }

      /*
       * The variable dbc_c has been initialized to reference the cursor in
       * which we're going to do the delete.  Initialize the cursor's page
       * and lock structures as necessary.
       *
       * First, we may not need to acquire any locks.  If we're in case #3,
       * that is, the primary database isn't a btree database, our caller
       * is responsible for acquiring any necessary locks before calling us.
       */
      if (F_ISSET(dbc, DBC_OPD)) {
            cp_c->lock.off = LOCK_INVALID;
            if ((ret =
                memp_fget(dbp->mpf, &cp_c->pgno, 0, &cp_c->page)) != 0)
                  goto err;
            goto delete;
      }

      /*
       * Otherwise, acquire a write lock.  If the cursor that did the initial
       * logical deletion (and which had a write lock) is not the same as the
       * cursor doing the physical deletion (which may have only ever had a
       * read lock on the item), we need to upgrade.  The confusion comes as
       * follows:
       *
       *    C1    created, acquires item read lock
       *    C2    dup C1, create C2, also has item read lock.
       *    C1    acquire write lock, delete item
       *    C1    close
       *    C2    close, needs a write lock to physically delete item.
       *
       * If we're in a TXN, we know that C2 will be able to acquire the write
       * lock, because no locker other than the one shared by C1 and C2 can
       * acquire a write lock -- the original write lock C1 acquire was never
       * discarded.
       *
       * If we're not in a TXN, it's nastier.  Other cursors might acquire
       * read locks on the item after C1 closed, discarding its write lock,
       * and such locks would prevent C2 from acquiring a read lock.  That's
       * OK, though, we'll simply wait until we can acquire a read lock, or
       * we'll deadlock.  (Which better not happen, since we're not in a TXN.)
       *
       * Lock the primary database page, regardless of whether we're deleting
       * an item on a primary database page or an off-page duplicates page.
       */
      ACQUIRE(dbc, DB_LOCK_WRITE,
          cp->pgno, cp_c->lock, cp_c->pgno, cp_c->page, ret);
      if (ret != 0)
            goto err;

delete:     /*
       * If the delete occurred in a btree, delete the on-page physical item
       * referenced by the cursor.
       */
      if (dbc_c->dbtype == DB_BTREE && (ret = __bam_c_physdel(dbc_c)) != 0)
            goto err;

      /*
       * If we're not working in an off-page duplicate tree, then we're
       * done.
       */
      if (!F_ISSET(dbc_c, DBC_OPD) || root_pgno == PGNO_INVALID)
            goto done;

      /*
       * We may have just deleted the last element in the off-page duplicate
       * tree, and closed the last cursor in the tree.  For an off-page btree
       * there are no other cursors in the tree by definition, if the tree is
       * empty.  For an off-page recno we know we have closed the last cursor
       * in the tree because the __ram_ca_delete call above returned 0 only
       * in that case.  So, if the off-page duplicate tree is empty at this
       * point, we want to remove it.
       */
      if ((ret = memp_fget(dbp->mpf, &root_pgno, 0, &h)) != 0)
            goto err;
      if ((num = NUM_ENT(h)) == 0) {
            if ((ret = __db_free(dbc, h)) != 0)
                  goto err;
      } else {
            if ((ret = memp_fput(dbp->mpf, h, 0)) != 0)
                  goto err;
            goto done;
      }

      /*
       * When removing the tree, we have to do one of two things.  If this is
       * case #2, that is, the primary tree is a btree, delete the key that's
       * associated with the tree from the btree leaf page.  We know we are
       * the only reference to it and we already have the correct lock.  We
       * detect this case because the cursor that was passed to us references
       * an off-page duplicate cursor.
       *
       * If this is case #3, that is, the primary tree isn't a btree, pass
       * the information back to our caller, it's their job to do cleanup on
       * the primary page.
       */
      if (dbc_opd != NULL) {
            cp->lock.off = LOCK_INVALID;
            if ((ret = memp_fget(dbp->mpf, &cp->pgno, 0, &cp->page)) != 0)
                  goto err;
            if ((ret = __bam_c_physdel(dbc)) != 0)
                  goto err;
      } else
            *rmroot = 1;
err:
done: /*
       * Discard the page references and locks, and confirm that the stack
       * has been emptied.
       */
      if (dbc_opd != NULL) {
            DISCARD_CUR(dbc_opd, t_ret);
            if (t_ret != 0 && ret == 0)
                  ret = t_ret;
      }
      DISCARD_CUR(dbc, t_ret);
      if (t_ret != 0 && ret == 0)
            ret = t_ret;

      /* Downgrade any CDB lock we acquired. */
      if (cdb_lock)
            (void)__lock_downgrade(
                dbp->dbenv, &dbc->mylock, DB_LOCK_IWRITE, 0);

      return (ret);
}

/*
 * __bam_c_destroy --
 *    Close a single cursor -- internal version.
 */
static int
__bam_c_destroy(dbc)
      DBC *dbc;
{
      /* Discard the structures. */
      __os_free(dbc->internal, sizeof(BTREE_CURSOR));

      return (0);
}

/*
 * __bam_c_count --
 *    Return a count of on and off-page duplicates.
 *
 * PUBLIC: int __bam_c_count __P((DBC *, db_recno_t *));
 */
int
__bam_c_count(dbc, recnop)
      DBC *dbc;
      db_recno_t *recnop;
{
      BTREE_CURSOR *cp;
      DB *dbp;
      db_indx_t indx, top;
      db_recno_t recno;
      int ret;

      dbp = dbc->dbp;
      cp = (BTREE_CURSOR *)dbc->internal;

      /*
       * Called with the top-level cursor that may reference an off-page
       * duplicates page.  If it's a set of on-page duplicates, get the
       * page and count.  Otherwise, get the root page of the off-page
       * duplicate tree, and use the count.  We don't have to acquire any
       * new locks, we have to have a read lock to even get here.
       */
      if (cp->opd == NULL) {
            if ((ret = memp_fget(dbp->mpf, &cp->pgno, 0, &cp->page)) != 0)
                  return (ret);

            /*
             * Move back to the beginning of the set of duplicates and
             * then count forward.
             */
            for (indx = cp->indx;; indx -= P_INDX)
                  if (indx == 0 ||
                      !IS_DUPLICATE(dbc, indx, indx - P_INDX))
                        break;
            for (recno = 1, top = NUM_ENT(cp->page) - P_INDX;
                indx < top; ++recno, indx += P_INDX)
                  if (!IS_DUPLICATE(dbc, indx, indx + P_INDX))
                        break;
            *recnop = recno;
      } else {
            if ((ret = memp_fget(dbp->mpf,
                &cp->opd->internal->root, 0, &cp->page)) != 0)
                  return (ret);

            *recnop = RE_NREC(cp->page);
      }

      ret = memp_fput(dbp->mpf, cp->page, 0);
      cp->page = NULL;

      return (ret);
}

/*
 * __bam_c_del --
 *    Delete using a cursor.
 */
static int
__bam_c_del(dbc)
      DBC *dbc;
{
      BTREE_CURSOR *cp;
      DB *dbp;
      int ret, t_ret;

      dbp = dbc->dbp;
      cp = (BTREE_CURSOR *)dbc->internal;
      ret = 0;

      /* If the item was already deleted, return failure. */
      if (F_ISSET(cp, C_DELETED))
            return (DB_KEYEMPTY);

      /*
       * This code is always called with a page lock but no page.
       */
      DB_ASSERT(cp->page == NULL);

      /*
       * We don't physically delete the record until the cursor moves, so
       * we have to have a long-lived write lock on the page instead of a
       * a long-lived read lock.  Note, we have to have a read lock to even
       * get here.
       *
       * If we're maintaining record numbers, we lock the entire tree, else
       * we lock the single page.
       */
      if (F_ISSET(cp, C_RECNUM)) {
            if ((ret = __bam_c_getstack(dbc)) != 0)
                  goto err;
            cp->page = cp->csp->page;
      } else {
            ACQUIRE_CUR(dbc, DB_LOCK_WRITE, ret);
            if (ret != 0)
                  goto err;
      }

      /* Log the change. */
      if (DB_LOGGING(dbc) &&
          (ret = __bam_cdel_log(dbp->dbenv, dbc->txn, &LSN(cp->page), 0,
          dbp->log_fileid, PGNO(cp->page), &LSN(cp->page), cp->indx)) != 0)
            goto err;

      /* Set the intent-to-delete flag on the page. */
      if (TYPE(cp->page) == P_LBTREE)
            B_DSET(GET_BKEYDATA(cp->page, cp->indx + O_INDX)->type);
      else
            B_DSET(GET_BKEYDATA(cp->page, cp->indx)->type);

      /* Mark the page dirty. */
      ret = memp_fset(dbp->mpf, cp->page, DB_MPOOL_DIRTY);

err:  /*
       * If we've been successful so far and the tree has record numbers,
       * adjust the record counts.  Either way, release acquired page(s).
       */
      if (F_ISSET(cp, C_RECNUM)) {
            if (ret == 0)
                  ret = __bam_adjust(dbc, -1);
            (void)__bam_stkrel(dbc, 0);
      } else
            if (cp->page != NULL &&
                (t_ret = memp_fput(dbp->mpf, cp->page, 0)) != 0 && ret == 0)
                  ret = t_ret;

      cp->page = NULL;

      /* Update the cursors last, after all chance of failure is past. */
      if (ret == 0)
            (void)__bam_ca_delete(dbp, cp->pgno, cp->indx, 1);

      return (ret);
}

/*
 * __bam_c_dup --
 *    Duplicate a btree cursor, such that the new one holds appropriate
 *    locks for the position of the original.
 *
 * PUBLIC: int __bam_c_dup __P((DBC *, DBC *));
 */
int
__bam_c_dup(orig_dbc, new_dbc)
      DBC *orig_dbc, *new_dbc;
{
      BTREE_CURSOR *orig, *new;
      int ret;

      orig = (BTREE_CURSOR *)orig_dbc->internal;
      new = (BTREE_CURSOR *)new_dbc->internal;

      /*
       * If we're holding a lock we need to acquire a copy of it, unless
       * we're in a transaction.  We don't need to copy any lock we're
       * holding inside a transaction because all the locks are retained
       * until the transaction commits or aborts.
       */
      if (orig->lock.off != LOCK_INVALID && orig_dbc->txn == NULL) {
            if ((ret = __db_lget(new_dbc,
                0, new->pgno, new->lock_mode, 0, &new->lock)) != 0)
                  return (ret);
      }
      new->ovflsize = orig->ovflsize;
      new->recno = orig->recno;
      new->flags = orig->flags;

      return (0);
}

/*
 * __bam_c_get --
 *    Get using a cursor (btree).
 */
static int
__bam_c_get(dbc, key, data, flags, pgnop)
      DBC *dbc;
      DBT *key, *data;
      u_int32_t flags;
      db_pgno_t *pgnop;
{
      BTREE_CURSOR *cp;
      DB *dbp;
      db_pgno_t orig_pgno;
      db_indx_t orig_indx;
      int exact, newopd, ret;

      dbp = dbc->dbp;
      cp = (BTREE_CURSOR *)dbc->internal;
      orig_pgno = cp->pgno;
      orig_indx = cp->indx;

      newopd = 0;
      switch (flags) {
      case DB_CURRENT:
            /* It's not possible to return a deleted record. */
            if (F_ISSET(cp, C_DELETED)) {
                  ret = DB_KEYEMPTY;
                  goto err;
            }

            /*
             * Acquire the current page.  We have at least a read-lock
             * already.  The caller may have set DB_RMW asking for a
             * write lock, but upgrading to a write lock has no better
             * chance of succeeding now instead of later, so don't try.
             */
            if ((ret = memp_fget(dbp->mpf, &cp->pgno, 0, &cp->page)) != 0)
                  goto err;
            break;
      case DB_FIRST:
            newopd = 1;
            if ((ret = __bam_c_first(dbc)) != 0)
                  goto err;
            break;
      case DB_GET_BOTH:
            /*
             * There are two ways to get here based on DBcursor->c_get
             * with the DB_GET_BOTH flag set:
             *
             * 1. Searching a sorted off-page duplicate tree: do a tree
             * search.
             *
             * 2. Searching btree: do a tree search.  If it returns a
             * reference to off-page duplicate tree, return immediately
             * and let our caller deal with it.  If the search doesn't
             * return a reference to off-page duplicate tree, start an
             * on-page search.
             */
            if (F_ISSET(dbc, DBC_OPD)) {
                  if ((ret = __bam_c_search(
                      dbc, data, DB_GET_BOTH, &exact)) != 0)
                        goto err;
                  if (!exact) {
                        ret = DB_NOTFOUND;
                        goto err;
                  }
            } else {
                  if ((ret = __bam_c_search(
                      dbc, key, DB_GET_BOTH, &exact)) != 0)
                        return (ret);
                  if (!exact) {
                        ret = DB_NOTFOUND;
                        goto err;
                  }

                  if (pgnop != NULL && __bam_isopd(dbc, pgnop)) {
                        newopd = 1;
                        break;
                  }
                  if ((ret = __bam_getboth_finddatum(dbc, data)) != 0)
                        goto err;
            }
            break;
      case DB_GET_BOTHC:
            if ((ret = __bam_getbothc(dbc, data)) != 0)
                  goto err;
            break;
      case DB_LAST:
            newopd = 1;
            if ((ret = __bam_c_last(dbc)) != 0)
                  goto err;
            break;
      case DB_NEXT:
            newopd = 1;
            if (cp->pgno == PGNO_INVALID) {
                  if ((ret = __bam_c_first(dbc)) != 0)
                        goto err;
            } else
                  if ((ret = __bam_c_next(dbc, 1)) != 0)
                        goto err;
            break;
      case DB_NEXT_DUP:
            if ((ret = __bam_c_next(dbc, 1)) != 0)
                  goto err;
            if (!IS_CUR_DUPLICATE(dbc, orig_pgno, orig_indx)) {
                  ret = DB_NOTFOUND;
                  goto err;
            }
            break;
      case DB_NEXT_NODUP:
            newopd = 1;
            if (cp->pgno == PGNO_INVALID) {
                  if ((ret = __bam_c_first(dbc)) != 0)
                        goto err;
            } else
                  do {
                        if ((ret = __bam_c_next(dbc, 1)) != 0)
                              goto err;
                  } while (IS_CUR_DUPLICATE(dbc, orig_pgno, orig_indx));
            break;
      case DB_PREV:
            newopd = 1;
            if (cp->pgno == PGNO_INVALID) {
                  if ((ret = __bam_c_last(dbc)) != 0)
                        goto err;
            } else
                  if ((ret = __bam_c_prev(dbc)) != 0)
                        goto err;
            break;
      case DB_PREV_NODUP:
            newopd = 1;
            if (cp->pgno == PGNO_INVALID) {
                  if ((ret = __bam_c_last(dbc)) != 0)
                        goto err;
            } else
                  do {
                        if ((ret = __bam_c_prev(dbc)) != 0)
                              goto err;
                  } while (IS_CUR_DUPLICATE(dbc, orig_pgno, orig_indx));
            break;
      case DB_SET:
      case DB_SET_RECNO:
            newopd = 1;
            if ((ret = __bam_c_search(dbc, key, flags, &exact)) != 0)
                  goto err;
            break;
      case DB_SET_RANGE:
            newopd = 1;
            if ((ret = __bam_c_search(dbc, key, flags, &exact)) != 0)
                  goto err;

            /*
             * As we didn't require an exact match, the search function
             * may have returned an entry past the end of the page.  Or,
             * we may be referencing a deleted record.  If so, move to
             * the next entry.
             */
            if (cp->indx == NUM_ENT(cp->page) || IS_CUR_DELETED(dbc))
                  if ((ret = __bam_c_next(dbc, 0)) != 0)
                        goto err;
            break;
      default:
            ret = __db_unknown_flag(dbp->dbenv, "__bam_c_get", flags);
            goto err;
      }

      /*
       * We may have moved to an off-page duplicate tree.  Return that
       * information to our caller.
       */
      if (newopd && pgnop != NULL)
            (void)__bam_isopd(dbc, pgnop);

      /* Don't return the key, it was passed to us */
      if (flags == DB_SET)
            F_SET(key, DB_DBT_ISSET);

err:  /*
       * Regardless of whether we were successful or not, if the cursor
       * moved, clear the delete flag, DBcursor->c_get never references
       * a deleted key, if it moved at all.
       */
      if (F_ISSET(cp, C_DELETED)
          && (cp->pgno != orig_pgno || cp->indx != orig_indx))
            F_CLR(cp, C_DELETED);

      return (ret);
}

/*
 * __bam_getbothc --
 *    Search for a matching data item on a join.
 */
static int
__bam_getbothc(dbc, data)
      DBC *dbc;
      DBT *data;
{
      BTREE_CURSOR *cp;
      DB *dbp;
      int cmp, exact, ret;

      dbp = dbc->dbp;
      cp = (BTREE_CURSOR *)dbc->internal;

      /*
       * Acquire the current page.  We have at least a read-lock
       * already.  The caller may have set DB_RMW asking for a
       * write lock, but upgrading to a write lock has no better
       * chance of succeeding now instead of later, so don't try.
       */
      if ((ret = memp_fget(dbp->mpf, &cp->pgno, 0, &cp->page)) != 0)
            return (ret);

      /*
       * An off-page duplicate cursor.  Search the remaining duplicates
       * for one which matches (do a normal btree search, then verify
       * that the retrieved record is greater than the original one).
       */
      if (F_ISSET(dbc, DBC_OPD)) {
            /*
             * Check to make sure the desired item comes strictly after
             * the current position;  if it doesn't, return DB_NOTFOUND.
             */
            if ((ret = __bam_cmp(dbp, data, cp->page, cp->indx,
                dbp->dup_compare == NULL ? __bam_defcmp : dbp->dup_compare,
                &cmp)) != 0)
                  return (ret);

            if (cmp <= 0)
                  return (DB_NOTFOUND);

            /* Discard the current page, we're going to do a full search. */
            if ((ret = memp_fput(dbp->mpf, cp->page, 0)) != 0)
                  return (ret);
            cp->page = NULL;

            return (__bam_c_search(dbc, data, DB_GET_BOTH, &exact));
      }

      /*
       * We're doing a DBC->c_get(DB_GET_BOTHC) and we're already searching
       * a set of on-page duplicates (either sorted or unsorted).  Continue
       * a linear search from after the current position.
       *
       * (Note that we could have just finished a "set" of one duplicate,
       * i.e. not a duplicate at all, but the following check will always
       * return DB_NOTFOUND in this case, which is the desired behavior.)
       */
      if (cp->indx + P_INDX >= NUM_ENT(cp->page) ||
          !IS_DUPLICATE(dbc, cp->indx, cp->indx + P_INDX))
            return (DB_NOTFOUND);
      cp->indx += P_INDX;

      return (__bam_getboth_finddatum(dbc, data));
}

/*
 * __bam_getboth_finddatum --
 *    Find a matching on-page data item.
 */
static int
__bam_getboth_finddatum(dbc, data)
      DBC *dbc;
      DBT *data;
{
      BTREE_CURSOR *cp;
      DB *dbp;
      db_indx_t base, lim, top;
      int cmp, ret;

      dbp = dbc->dbp;
      cp = (BTREE_CURSOR *)dbc->internal;

      /*
       * Called (sometimes indirectly) from DBC->get to search on-page data
       * item(s) for a matching value.  If the original flag was DB_GET_BOTH,
       * the cursor argument is set to the first data item for the key.  If
       * the original flag was DB_GET_BOTHC, the cursor argument is set to
       * the first data item that we can potentially return.  In both cases,
       * there may or may not be additional duplicate data items to search.
       *
       * If the duplicates are not sorted, do a linear search.
       *
       * If the duplicates are sorted, do a binary search.  The reason for
       * this is that large pages and small key/data pairs result in large
       * numbers of on-page duplicates before they get pushed off-page.
       */
      if (dbp->dup_compare == NULL) {
            for (;; cp->indx += P_INDX) {
                  if (!IS_CUR_DELETED(dbc) &&
                      (ret = __bam_cmp(dbp, data, cp->page,
                      cp->indx + O_INDX, __bam_defcmp, &cmp)) != 0)
                        return (ret);
                  if (cmp == 0)
                        return (0);

                  if (cp->indx + P_INDX >= NUM_ENT(cp->page) ||
                      !IS_DUPLICATE(dbc, cp->indx, cp->indx + P_INDX))
                        break;
            }
      } else {
            /*
             * Find the top and bottom of the duplicate set.  Binary search
             * requires at least two items, don't loop if there's only one.
             */
            for (base = top = cp->indx;
                top < NUM_ENT(cp->page); top += P_INDX)
                  if (!IS_DUPLICATE(dbc, cp->indx, top))
                        break;
            if (base == (top - P_INDX)) {
                  if  ((ret = __bam_cmp(dbp, data,
                      cp->page, cp->indx + O_INDX,
                      dbp->dup_compare, &cmp)) != 0)
                         return (ret);
                  return (cmp == 0 ? 0 : DB_NOTFOUND);
            }

            for (lim =
                (top - base) / (db_indx_t)P_INDX; lim != 0; lim >>= 1) {
                  cp->indx = base + ((lim >> 1) * P_INDX);
                  if ((ret = __bam_cmp(dbp, data, cp->page,
                      cp->indx + O_INDX, dbp->dup_compare, &cmp)) != 0)
                        return (ret);
                  if (cmp == 0) {
                        if (!IS_CUR_DELETED(dbc))
                              return (0);
                        break;
                  }
                  if (cmp > 0) {
                        base = cp->indx + P_INDX;
                        --lim;
                  }
            }
      }
      return (DB_NOTFOUND);
}

/*
 * __bam_c_put --
 *    Put using a cursor.
 */
static int
__bam_c_put(dbc, key, data, flags, pgnop)
      DBC *dbc;
      DBT *key, *data;
      u_int32_t flags;
      db_pgno_t *pgnop;
{
      BTREE_CURSOR *cp;
      DB *dbp;
      DBT dbt;
      u_int32_t iiop;
      int cmp, exact, needkey, ret, stack;
      void *arg;

      dbp = dbc->dbp;
      cp = (BTREE_CURSOR *)dbc->internal;

split:      needkey = ret = stack = 0;
      switch (flags) {
      case DB_AFTER:
      case DB_BEFORE:
      case DB_CURRENT:
            needkey = 1;
            iiop = flags;

            /*
             * If the Btree has record numbers (and we're not replacing an
             * existing record), we need a complete stack so that we can
             * adjust the record counts.  The check for flags == DB_CURRENT
             * is superfluous but left in for clarity.  (If C_RECNUM is set
             * we know that flags must be DB_CURRENT, as DB_AFTER/DB_BEFORE
             * are illegal in a Btree unless it's configured for duplicates
             * and you cannot configure a Btree for both record renumbering
             * and duplicates.)
             */
            if (flags == DB_CURRENT &&
                F_ISSET(cp, C_RECNUM) && F_ISSET(cp, C_DELETED)) {
                  if ((ret = __bam_c_getstack(dbc)) != 0)
                        goto err;
                  /*
                   * Initialize the cursor from the stack.  Don't take
                   * the page number or page index, they should already
                   * be set.
                   */
                  cp->page = cp->csp->page;
                  cp->lock = cp->csp->lock;
                  cp->lock_mode = cp->csp->lock_mode;

                  stack = 1;
                  break;
            }

            /* Acquire the current page with a write lock. */
            ACQUIRE_WRITE_LOCK(dbc, ret);
            if (ret != 0)
                  goto err;
            if ((ret = memp_fget(dbp->mpf, &cp->pgno, 0, &cp->page)) != 0)
                  goto err;
            break;
      case DB_KEYFIRST:
      case DB_KEYLAST:
      case DB_NODUPDATA:
            /*
             * Searching off-page, sorted duplicate tree: do a tree search
             * for the correct item; __bam_c_search returns the smallest
             * slot greater than the key, use it.
             */
            if (F_ISSET(dbc, DBC_OPD)) {
                  if ((ret =
                      __bam_c_search(dbc, data, flags, &exact)) != 0)
                        goto err;
                  stack = 1;

                  /* Disallow "sorted" duplicate duplicates. */
                  if (exact) {
                        ret = __db_duperr(dbp, flags);
                        goto err;
                  }
                  iiop = DB_BEFORE;
                  break;
            }

            /* Searching a btree. */
            if ((ret = __bam_c_search(dbc, key,
                flags == DB_KEYFIRST || dbp->dup_compare != NULL ?
                DB_KEYFIRST : DB_KEYLAST, &exact)) != 0)
                  goto err;
            stack = 1;

            /*
             * If we don't have an exact match, __bam_c_search returned
             * the smallest slot greater than the key, use it.
             */
            if (!exact) {
                  iiop = DB_KEYFIRST;
                  break;
            }

            /*
             * If duplicates aren't supported, replace the current item.
             * (If implementing the DB->put function, our caller already
             * checked the DB_NOOVERWRITE flag.)
             */
            if (!F_ISSET(dbp, DB_AM_DUP)) {
                  iiop = DB_CURRENT;
                  break;
            }

            /*
             * If we find a matching entry, it may be an off-page duplicate
             * tree.  Return the page number to our caller, we need a new
             * cursor.
             */
            if (pgnop != NULL && __bam_isopd(dbc, pgnop))
                  goto done;

            /* If the duplicates aren't sorted, move to the right slot. */
            if (dbp->dup_compare == NULL) {
                  if (flags == DB_KEYFIRST)
                        iiop = DB_BEFORE;
                  else
                        for (;; cp->indx += P_INDX)
                              if (cp->indx + P_INDX >=
                                  NUM_ENT(cp->page) ||
                                  !IS_DUPLICATE(dbc, cp->indx,
                                  cp->indx + P_INDX)) {
                                    iiop = DB_AFTER;
                                    break;
                              }
                  break;
            }

            /*
             * We know that we're looking at the first of a set of sorted
             * on-page duplicates.  Walk the list to find the right slot.
             */
            for (;; cp->indx += P_INDX) {
                  if ((ret = __bam_cmp(dbp, data, cp->page,
                      cp->indx + O_INDX, dbp->dup_compare, &cmp)) !=0)
                        return (ret);
                  if (cmp < 0) {
                        iiop = DB_BEFORE;
                        break;
                  }

                  /* Disallow "sorted" duplicate duplicates. */
                  if (cmp == 0) {
                        if (IS_DELETED(cp->page, cp->indx)) {
                              iiop = DB_CURRENT;
                              break;
                        }
                        ret = __db_duperr(dbp, flags);
                        goto err;
                  }

                  if (cp->indx + P_INDX >= NUM_ENT(cp->page) ||
                      ((PAGE *)cp->page)->inp[cp->indx] !=
                      ((PAGE *)cp->page)->inp[cp->indx + P_INDX]) {
                        iiop = DB_AFTER;
                        break;
                  }
            }
            break;
      default:
            ret = __db_unknown_flag(dbp->dbenv, "__bam_c_put", flags);
            goto err;
      }

      switch (ret = __bam_iitem(dbc, key, data, iiop, 0)) {
      case 0:
            break;
      case DB_NEEDSPLIT:
            /*
             * To split, we need a key for the page.  Either use the key
             * argument or get a copy of the key from the page.
             */
            if (flags == DB_AFTER ||
                flags == DB_BEFORE || flags == DB_CURRENT) {
                  memset(&dbt, 0, sizeof(DBT));
                  if ((ret = __db_ret(dbp, cp->page, 0, &dbt,
                      &dbc->rkey.data, &dbc->rkey.ulen)) != 0)
                        goto err;
                  arg = &dbt;
            } else
                  arg = F_ISSET(dbc, DBC_OPD) ? data : key;

            /*
             * Discard any locks and pinned pages (the locks are discarded
             * even if we're running with transactions, as they lock pages
             * that we're sorry we ever acquired).  If stack is set and the
             * cursor entries are valid, they point to the same entries as
             * the stack, don't free them twice.
             */
            if (stack)
                  ret = __bam_stkrel(dbc, STK_CLRDBC | STK_NOLOCK);
            else
                  DISCARD_CUR(dbc, ret);
            if (ret != 0)
                  goto err;

            /* Split the tree. */
            if ((ret = __bam_split(dbc, arg)) != 0)
                  return (ret);

            goto split;
      default:
            goto err;
      }

err:
done: /*
       * Discard any pages pinned in the tree and their locks, except for
       * the leaf page.  Note, the leaf page participated in any stack we
       * acquired, and so we have to adjust the stack as necessary.  If
       * there was only a single page on the stack, we don't have to free
       * further stack pages.
       */
      if (stack && BT_STK_POP(cp) != NULL)
            (void)__bam_stkrel(dbc, 0);

      /*
       * Regardless of whether we were successful or not, clear the delete
       * flag.  If we're successful, we either moved the cursor or the item
       * is no longer deleted.  If we're not successful, then we're just a
       * copy, no need to have the flag set.
       */
      F_CLR(cp, C_DELETED);

      return (ret);
}

/*
 * __bam_c_rget --
 *    Return the record number for a cursor.
 *
 * PUBLIC: int __bam_c_rget __P((DBC *, DBT *, u_int32_t));
 */
int
__bam_c_rget(dbc, data, flags)
      DBC *dbc;
      DBT *data;
      u_int32_t flags;
{
      BTREE_CURSOR *cp;
      DB *dbp;
      DBT dbt;
      db_recno_t recno;
      int exact, ret;

      COMPQUIET(flags, 0);
      dbp = dbc->dbp;
      cp = (BTREE_CURSOR *)dbc->internal;

      /*
       * Get the page with the current item on it.
       * Get a copy of the key.
       * Release the page, making sure we don't release it twice.
       */
      if ((ret = memp_fget(dbp->mpf, &cp->pgno, 0, &cp->page)) != 0)
            return (ret);
      memset(&dbt, 0, sizeof(DBT));
      if ((ret = __db_ret(dbp, cp->page,
          cp->indx, &dbt, &dbc->rkey.data, &dbc->rkey.ulen)) != 0)
            goto err;
      ret = memp_fput(dbp->mpf, cp->page, 0);
      cp->page = NULL;
      if (ret != 0)
            return (ret);

      if ((ret = __bam_search(dbc, &dbt,
          F_ISSET(dbc, DBC_RMW) ? S_FIND_WR : S_FIND,
          1, &recno, &exact)) != 0)
            goto err;

      ret = __db_retcopy(dbp, data,
          &recno, sizeof(recno), &dbc->rdata.data, &dbc->rdata.ulen);

      /* Release the stack. */
err:  __bam_stkrel(dbc, 0);

      return (ret);
}

/*
 * __bam_c_writelock --
 *    Upgrade the cursor to a write lock.
 */
static int
__bam_c_writelock(dbc)
      DBC *dbc;
{
      BTREE_CURSOR *cp;
      int ret;

      cp = (BTREE_CURSOR *)dbc->internal;

      if (cp->lock_mode == DB_LOCK_WRITE)
            return (0);

      /*
       * When writing to an off-page duplicate tree, we need to have the
       * appropriate page in the primary tree locked.  The general DBC
       * code calls us first with the primary cursor so we can acquire the
       * appropriate lock.
       */
      ACQUIRE_WRITE_LOCK(dbc, ret);
      return (ret);
}

/*
 * __bam_c_first --
 *    Return the first record.
 */
static int
__bam_c_first(dbc)
      DBC *dbc;
{
      BTREE_CURSOR *cp;
      DB *dbp;
      db_pgno_t pgno;
      int ret;

      dbp = dbc->dbp;
      cp = (BTREE_CURSOR *)dbc->internal;
      ret = 0;

      /* Walk down the left-hand side of the tree. */
      for (pgno = cp->root;;) {
            ACQUIRE_CUR_SET(dbc, DB_LOCK_READ, pgno, ret);
            if (ret != 0)
                  return (ret);

            /* If we find a leaf page, we're done. */
            if (ISLEAF(cp->page))
                  break;

            pgno = GET_BINTERNAL(cp->page, 0)->pgno;
      }

      /* If we want a write lock instead of a read lock, get it now. */
      if (F_ISSET(dbc, DBC_RMW)) {
            ACQUIRE_WRITE_LOCK(dbc, ret);
            if (ret != 0)
                  return (ret);
      }

      /* If on an empty page or a deleted record, move to the next one. */
      if (NUM_ENT(cp->page) == 0 || IS_CUR_DELETED(dbc))
            if ((ret = __bam_c_next(dbc, 0)) != 0)
                  return (ret);

      return (0);
}

/*
 * __bam_c_last --
 *    Return the last record.
 */
static int
__bam_c_last(dbc)
      DBC *dbc;
{
      BTREE_CURSOR *cp;
      DB *dbp;
      db_pgno_t pgno;
      int ret;

      dbp = dbc->dbp;
      cp = (BTREE_CURSOR *)dbc->internal;
      ret = 0;

      /* Walk down the right-hand side of the tree. */
      for (pgno = cp->root;;) {
            ACQUIRE_CUR_SET(dbc, DB_LOCK_READ, pgno, ret);
            if (ret != 0)
                  return (ret);

            /* If we find a leaf page, we're done. */
            if (ISLEAF(cp->page))
                  break;

            pgno =
                GET_BINTERNAL(cp->page, NUM_ENT(cp->page) - O_INDX)->pgno;
      }

      /* If we want a write lock instead of a read lock, get it now. */
      if (F_ISSET(dbc, DBC_RMW)) {
            ACQUIRE_WRITE_LOCK(dbc, ret);
            if (ret != 0)
                  return (ret);
      }

      cp->indx = NUM_ENT(cp->page) == 0 ? 0 :
          NUM_ENT(cp->page) -
          (TYPE(cp->page) == P_LBTREE ? P_INDX : O_INDX);

      /* If on an empty page or a deleted record, move to the previous one. */
      if (NUM_ENT(cp->page) == 0 || IS_CUR_DELETED(dbc))
            if ((ret = __bam_c_prev(dbc)) != 0)
                  return (ret);

      return (0);
}

/*
 * __bam_c_next --
 *    Move to the next record.
 */
static int
__bam_c_next(dbc, initial_move)
      DBC *dbc;
      int initial_move;
{
      BTREE_CURSOR *cp;
      DB *dbp;
      db_indx_t adjust;
      db_lockmode_t lock_mode;
      db_pgno_t pgno;
      int ret;

      dbp = dbc->dbp;
      cp = (BTREE_CURSOR *)dbc->internal;
      ret = 0;

      /*
       * We're either moving through a page of duplicates or a btree leaf
       * page.
       *
       * !!!
       * This code handles empty pages and pages with only deleted entries.
       */
      if (F_ISSET(dbc, DBC_OPD)) {
            adjust = O_INDX;
            lock_mode = DB_LOCK_NG;
      } else {
            adjust = dbc->dbtype == DB_BTREE ? P_INDX : O_INDX;
            lock_mode =
                F_ISSET(dbc, DBC_RMW) ? DB_LOCK_WRITE : DB_LOCK_READ;
      }
      if (cp->page == NULL) {
            ACQUIRE_CUR(dbc, lock_mode, ret);
            if (ret != 0)
                  return (ret);
      }

      if (initial_move)
            cp->indx += adjust;

      for (;;) {
            /*
             * If at the end of the page, move to a subsequent page.
             *
             * !!!
             * Check for >= NUM_ENT.  If the original search landed us on
             * NUM_ENT, we may have incremented indx before the test.
             */
            if (cp->indx >= NUM_ENT(cp->page)) {
                  if ((pgno
                      = NEXT_PGNO(cp->page)) == PGNO_INVALID)
                        return (DB_NOTFOUND);

                  ACQUIRE_CUR_SET(dbc, lock_mode, pgno, ret);
                  if (ret != 0)
                        return (ret);
                  continue;
            }
            if (IS_CUR_DELETED(dbc)) {
                  cp->indx += adjust;
                  continue;
            }
            break;
      }
      return (0);
}

/*
 * __bam_c_prev --
 *    Move to the previous record.
 */
static int
__bam_c_prev(dbc)
      DBC *dbc;
{
      BTREE_CURSOR *cp;
      DB *dbp;
      db_indx_t adjust;
      db_lockmode_t lock_mode;
      db_pgno_t pgno;
      int ret;

      dbp = dbc->dbp;
      cp = (BTREE_CURSOR *)dbc->internal;
      ret = 0;

      /*
       * We're either moving through a page of duplicates or a btree leaf
       * page.
       *
       * !!!
       * This code handles empty pages and pages with only deleted entries.
       */
      if (F_ISSET(dbc, DBC_OPD)) {
            adjust = O_INDX;
            lock_mode = DB_LOCK_NG;
      } else {
            adjust = dbc->dbtype == DB_BTREE ? P_INDX : O_INDX;
            lock_mode =
                F_ISSET(dbc, DBC_RMW) ? DB_LOCK_WRITE : DB_LOCK_READ;
      }
      if (cp->page == NULL) {
            ACQUIRE_CUR(dbc, lock_mode, ret);
            if (ret != 0)
                  return (ret);
      }

      for (;;) {
            /* If at the beginning of the page, move to a previous one. */
            if (cp->indx == 0) {
                  if ((pgno =
                      PREV_PGNO(cp->page)) == PGNO_INVALID)
                        return (DB_NOTFOUND);

                  ACQUIRE_CUR_SET(dbc, lock_mode, pgno, ret);
                  if (ret != 0)
                        return (ret);

                  if ((cp->indx = NUM_ENT(cp->page)) == 0)
                        continue;
            }

            /* Ignore deleted records. */
            cp->indx -= adjust;
            if (IS_CUR_DELETED(dbc))
                  continue;

            break;
      }
      return (0);
}

/*
 * __bam_c_search --
 *    Move to a specified record.
 */
static int
__bam_c_search(dbc, key, flags, exactp)
      DBC *dbc;
      const DBT *key;
      u_int32_t flags;
      int *exactp;
{
      BTREE *t;
      BTREE_CURSOR *cp;
      DB *dbp;
      PAGE *h;
      db_indx_t indx;
      db_pgno_t bt_lpgno;
      db_recno_t recno;
      u_int32_t sflags;
      int cmp, ret;

      dbp = dbc->dbp;
      cp = (BTREE_CURSOR *)dbc->internal;
      t = dbp->bt_internal;
      ret = 0;

      /*
       * Find an entry in the database.  Discard any lock we currently hold,
       * we're going to search the tree.
       */
      DISCARD_CUR(dbc, ret);
      if (ret != 0)
            return (ret);

      switch (flags) {
      case DB_SET_RECNO:
            if ((ret = __ram_getno(dbc, key, &recno, 0)) != 0)
                  return (ret);
            sflags = (F_ISSET(dbc, DBC_RMW) ? S_FIND_WR : S_FIND) | S_EXACT;
            if ((ret = __bam_rsearch(dbc, &recno, sflags, 1, exactp)) != 0)
                  return (ret);
            break;
      case DB_SET:
      case DB_GET_BOTH:
            sflags = (F_ISSET(dbc, DBC_RMW) ? S_FIND_WR : S_FIND) | S_EXACT;
            goto search;
      case DB_SET_RANGE:
            sflags =
                (F_ISSET(dbc, DBC_RMW) ? S_WRITE : S_READ) | S_DUPFIRST;
            goto search;
      case DB_KEYFIRST:
            sflags = S_KEYFIRST;
            goto fast_search;
      case DB_KEYLAST:
      case DB_NODUPDATA:
            sflags = S_KEYLAST;
fast_search:      /*
             * If the application has a history of inserting into the first
             * or last pages of the database, we check those pages first to
             * avoid doing a full search.
             *
             * If the tree has record numbers, we need a complete stack so
             * that we can adjust the record counts, so fast_search isn't
             * possible.
             */
            if (F_ISSET(cp, C_RECNUM))
                  goto search;

            /*
             * !!!
             * We do not mutex protect the t->bt_lpgno field, which means
             * that it can only be used in an advisory manner.  If we find
             * page we can use, great.  If we don't, we don't care, we do
             * it the slow way instead.  Regardless, copy it into a local
             * variable, otherwise we might acquire a lock for a page and
             * then read a different page because it changed underfoot.
             */
            bt_lpgno = t->bt_lpgno;

            /*
             * If the tree has no history of insertion, do it the slow way.
             */
            if (bt_lpgno == PGNO_INVALID)
                  goto search;

            /* Lock and retrieve the page on which we last inserted. */
            h = NULL;
            ACQUIRE(dbc,
                DB_LOCK_WRITE, bt_lpgno, cp->lock, bt_lpgno, h, ret);
            if (ret != 0)
                  goto fast_miss;

            /*
             * It's okay if the page type isn't right or it's empty, it
             * just means that the world changed.
             */
            if (TYPE(h) != P_LBTREE || NUM_ENT(h) == 0)
                  goto fast_miss;

            /*
             * What we do here is test to see if we're at the beginning or
             * end of the tree and if the new item sorts before/after the
             * first/last page entry.  We don't try and catch inserts into
             * the middle of the tree (although we could, as long as there
             * were two keys on the page and we saved both the index and
             * the page number of the last insert).
             */
            if (h->next_pgno == PGNO_INVALID) {
                  indx = NUM_ENT(h) - P_INDX;
                  if ((ret = __bam_cmp(dbp,
                      key, h, indx, t->bt_compare, &cmp)) != 0)
                        return (ret);

                  if (cmp < 0)
                        goto try_begin;
                  if (cmp > 0) {
                        indx += P_INDX;
                        goto fast_hit;
                  }

                  /*
                   * Found a duplicate.  If doing DB_KEYLAST, we're at
                   * the correct position, otherwise, move to the first
                   * of the duplicates.  If we're looking at off-page
                   * duplicates, duplicate duplicates aren't permitted,
                   * so we're done.
                   */
                  if (flags == DB_KEYLAST)
                        goto fast_hit;
                  for (;
                      indx > 0 && h->inp[indx - P_INDX] == h->inp[indx];
                      indx -= P_INDX)
                        ;
                  goto fast_hit;
            }
try_begin:  if (h->prev_pgno == PGNO_INVALID) {
                  indx = 0;
                  if ((ret = __bam_cmp(dbp,
                      key, h, indx, t->bt_compare, &cmp)) != 0)
                        return (ret);

                  if (cmp > 0)
                        goto fast_miss;
                  if (cmp < 0)
                        goto fast_hit;

                  /*
                   * Found a duplicate.  If doing DB_KEYFIRST, we're at
                   * the correct position, otherwise, move to the last
                   * of the duplicates.  If we're looking at off-page
                   * duplicates, duplicate duplicates aren't permitted,
                   * so we're done.
                   */
                  if (flags == DB_KEYFIRST)
                        goto fast_hit;
                  for (;
                      indx < (db_indx_t)(NUM_ENT(h) - P_INDX) &&
                      h->inp[indx] == h->inp[indx + P_INDX];
                      indx += P_INDX)
                        ;
                  goto fast_hit;
            }
            goto fast_miss;

fast_hit:   /* Set the exact match flag, we may have found a duplicate. */
            *exactp = cmp == 0;

            /*
             * Insert the entry in the stack.  (Our caller is likely to
             * call __bam_stkrel() after our return.)
             */
            BT_STK_CLR(cp);
            BT_STK_ENTER(dbp->dbenv,
                cp, h, indx, cp->lock, cp->lock_mode, ret);
            if (ret != 0)
                  return (ret);
            break;

fast_miss:  /*
             * This was not the right page, so we do not need to retain
             * the lock even in the presence of transactions.
             */
            DISCARD(dbc, 1, cp->lock, h, ret);
            if (ret != 0)
                  return (ret);

search:           if ((ret =
                __bam_search(dbc, key, sflags, 1, NULL, exactp)) != 0)
                  return (ret);
            break;
      default:
            return (__db_unknown_flag(dbp->dbenv, "__bam_c_search", flags));
      }

      /* Initialize the cursor from the stack. */
      cp->page = cp->csp->page;
      cp->pgno = cp->csp->page->pgno;
      cp->indx = cp->csp->indx;
      cp->lock = cp->csp->lock;
      cp->lock_mode = cp->csp->lock_mode;

      /*
       * If we inserted a key into the first or last slot of the tree,
       * remember where it was so we can do it more quickly next time.
       */
      if (TYPE(cp->page) == P_LBTREE &&
          (flags == DB_KEYFIRST || flags == DB_KEYLAST))
            t->bt_lpgno =
                (NEXT_PGNO(cp->page) == PGNO_INVALID &&
                cp->indx >= NUM_ENT(cp->page)) ||
                (PREV_PGNO(cp->page) == PGNO_INVALID &&
                cp->indx == 0) ? cp->pgno : PGNO_INVALID;
      return (0);
}

/*
 * __bam_c_physdel --
 *    Physically remove an item from the page.
 */
static int
__bam_c_physdel(dbc)
      DBC *dbc;
{
      BTREE_CURSOR *cp;
      DB *dbp;
      DBT key;
      DB_LOCK lock;
      PAGE *h;
      db_pgno_t pgno;
      int delete_page, empty_page, exact, level, ret;

      dbp = dbc->dbp;
      cp = (BTREE_CURSOR *)dbc->internal;
      delete_page = empty_page = ret = 0;

      /* If the page is going to be emptied, consider deleting it. */
      delete_page = empty_page =
          NUM_ENT(cp->page) == (TYPE(cp->page) == P_LBTREE ? 2 : 1);

      /*
       * Check if the application turned off reverse splits.  Applications
       * can't turn off reverse splits in off-page duplicate trees, that
       * space will never be reused unless the exact same key is specified.
       */
      if (delete_page &&
          !F_ISSET(dbc, DBC_OPD) && F_ISSET(dbp, DB_BT_REVSPLIT))
            delete_page = 0;

      /*
       * We never delete the last leaf page.  (Not really true -- we delete
       * the last leaf page of off-page duplicate trees, but that's handled
       * by our caller, not down here.)
       */
      if (delete_page && cp->pgno == cp->root)
            delete_page = 0;

      /*
       * To delete a leaf page other than an empty root page, we need a
       * copy of a key from the page.  Use the 0th page index since it's
       * the last key the page held.
       */
      if (delete_page) {
            memset(&key, 0, sizeof(DBT));
            if ((ret = __db_ret(dbp, cp->page,
                0, &key, &dbc->rkey.data, &dbc->rkey.ulen)) != 0)
                  return (ret);
      }

      /*
       * Delete the items.  If page isn't empty, we adjust the cursors.
       *
       * !!!
       * The following operations to delete a page may deadlock.  The easy
       * scenario is if we're deleting an item because we're closing cursors
       * because we've already deadlocked and want to call txn_abort().  If
       * we fail due to deadlock, we'll leave a locked, possibly empty page
       * in the tree, which won't be empty long because we'll undo the delete
       * when we undo the transaction's modifications.
       *
       * !!!
       * Delete the key item first, otherwise the on-page duplicate checks
       * in __bam_ditem() won't work!
       */
      if (TYPE(cp->page) == P_LBTREE) {
            if ((ret = __bam_ditem(dbc, cp->page, cp->indx)) != 0)
                  return (ret);
            if (!empty_page)
                  if ((ret = __bam_ca_di(dbc,
                      PGNO(cp->page), cp->indx, -1)) != 0)
                        return (ret);
      }
      if ((ret = __bam_ditem(dbc, cp->page, cp->indx)) != 0)
            return (ret);
      if (!empty_page)
            if ((ret = __bam_ca_di(dbc, PGNO(cp->page), cp->indx, -1)) != 0)
                  return (ret);

      /* If we're not going to try and delete the page, we're done. */
      if (!delete_page)
            return (0);

      /*
       * Call __bam_search to reacquire the empty leaf page, but this time
       * get both the leaf page and it's parent, locked.  Jump back up the
       * tree, until we have the top pair of pages that we want to delete.
       * Once we have the top page that we want to delete locked, lock the
       * underlying pages and check to make sure they're still empty.  If
       * they are, delete them.
       */
      for (level = LEAFLEVEL;; ++level) {
            /* Acquire a page and its parent, locked. */
            if ((ret = __bam_search(
                dbc, &key, S_WRPAIR, level, NULL, &exact)) != 0)
                  return (ret);

            /*
             * If we reach the root or the parent page isn't going to be
             * empty when we delete one record, stop.
             */
            h = cp->csp[-1].page;
            if (h->pgno == cp->root || NUM_ENT(h) != 1)
                  break;

            /* Discard the stack, retaining no locks. */
            (void)__bam_stkrel(dbc, STK_NOLOCK);
      }

      /*
       * Move the stack pointer one after the last entry, we may be about
       * to push more items onto the page stack.
       */
      ++cp->csp;

      /*
       * cp->csp[-2].page is now the parent page, which we may or may not be
       * going to delete, and cp->csp[-1].page is the first page we know we
       * are going to delete.  Walk down the chain of pages, acquiring pages
       * until we've acquired a leaf page.  Generally, this shouldn't happen;
       * we should only see a single internal page with one item and a single
       * leaf page with no items.  The scenario where we could see something
       * else is if reverse splits were turned off for awhile and then turned
       * back on.  That could result in all sorts of strangeness, e.g., empty
       * pages in the tree, trees that looked like linked lists, and so on.
       *
       * !!!
       * Sheer paranoia: if we find any pages that aren't going to be emptied
       * by the delete, someone else added an item while we were walking the
       * tree, and we discontinue the delete.  Shouldn't be possible, but we
       * check regardless.
       */
      for (h = cp->csp[-1].page;;) {
            if (ISLEAF(h)) {
                  if (NUM_ENT(h) != 0)
                        break;
                  break;
            } else
                  if (NUM_ENT(h) != 1)
                        break;

            /*
             * Get the next page, write lock it and push it onto the stack.
             * We know it's index 0, because it can only have one element.
             */
            switch (TYPE(h)) {
            case P_IBTREE:
                  pgno = GET_BINTERNAL(h, 0)->pgno;
                  break;
            case P_IRECNO:
                  pgno = GET_RINTERNAL(h, 0)->pgno;
                  break;
            default:
                  return (__db_pgfmt(dbp, PGNO(h)));
            }

            if ((ret =
                __db_lget(dbc, 0, pgno, DB_LOCK_WRITE, 0, &lock)) != 0)
                  break;
            if ((ret = memp_fget(dbp->mpf, &pgno, 0, &h)) != 0)
                  break;
            BT_STK_PUSH(dbp->dbenv, cp, h, 0, lock, DB_LOCK_WRITE, ret);
            if (ret != 0)
                  break;
      }

      /* Adjust the cursor stack to reference the last page on the stack. */
      BT_STK_POP(cp);

      /*
       * If everything worked, delete the stack, otherwise, release the
       * stack and page locks without further damage.
       */
      if (ret == 0)
            ret = __bam_dpages(dbc, cp->sp);
      else
            (void)__bam_stkrel(dbc, 0);

      return (ret);
}

/*
 * __bam_c_getstack --
 *    Acquire a full stack for a cursor.
 */
static int
__bam_c_getstack(dbc)
      DBC *dbc;
{
      BTREE_CURSOR *cp;
      DB *dbp;
      DBT dbt;
      PAGE *h;
      int exact, ret, t_ret;

      dbp = dbc->dbp;
      cp = (BTREE_CURSOR *)dbc->internal;

      /*
       * Get the page with the current item on it.  The caller of this
       * routine has to already hold a read lock on the page, so there
       * is no additional lock to acquire.
       */
      if ((ret = memp_fget(dbp->mpf, &cp->pgno, 0, &h)) != 0)
            return (ret);

      /* Get a copy of a key from the page. */
      memset(&dbt, 0, sizeof(DBT));
      if ((ret = __db_ret(dbp,
          h, 0, &dbt, &dbc->rkey.data, &dbc->rkey.ulen)) != 0)
            goto err;

      /* Get a write-locked stack for the page. */
      exact = 0;
      ret = __bam_search(dbc, &dbt, S_KEYFIRST, 1, NULL, &exact);

err:  /* Discard the key and the page. */
      if ((t_ret = memp_fput(dbp->mpf, h, 0)) != 0 && ret == 0)
            ret = t_ret;

      return (ret);
}

/*
 * __bam_isopd --
 *    Return if the cursor references an off-page duplicate tree via its
 *    page number.
 */
static int
__bam_isopd(dbc, pgnop)
      DBC *dbc;
      db_pgno_t *pgnop;
{
      BOVERFLOW *bo;

      if (TYPE(dbc->internal->page) != P_LBTREE)
            return (0);

      bo = GET_BOVERFLOW(dbc->internal->page, dbc->internal->indx + O_INDX);
      if (B_TYPE(bo->type) == B_DUPLICATE) {
            *pgnop = bo->pgno;
            return (1);
      }
      return (0);
}

Generated by  Doxygen 1.6.0   Back to index