* BerkeleyDB API, but much simplified.
*/
/*
- * Copyright 2011-2013 Howard Chu, Symas Corp.
+ * Copyright 2011-2014 Howard Chu, Symas Corp.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
#endif
#include <sys/types.h>
#include <sys/stat.h>
-#include <sys/param.h>
#ifdef _WIN32
#include <windows.h>
+/** getpid() returns int; MinGW defines pid_t but MinGW64 typedefs it
+ * as int64 which is wrong. MSVC doesn't define it at all, so just
+ * don't use it.
+ */
+#define MDB_PID_T int
+#ifdef __GNUC__
+# include <sys/param.h>
#else
+# define LITTLE_ENDIAN 1234
+# define BIG_ENDIAN 4321
+# define BYTE_ORDER LITTLE_ENDIAN
+# ifndef SSIZE_MAX
+# define SSIZE_MAX INT_MAX
+# endif
+#endif
+#else
+#define MDB_PID_T pid_t
+#include <sys/param.h>
#include <sys/uio.h>
#include <sys/mman.h>
#ifdef HAVE_SYS_FILE_H
#include <fcntl.h>
#endif
-#include <assert.h>
#include <errno.h>
#include <limits.h>
#include <stddef.h>
/** @defgroup internal MDB Internals
* @{
*/
-/** @defgroup compat Windows Compatibility Macros
+/** @defgroup compat Compatibility Macros
* A bunch of macros to minimize the amount of platform-specific ifdefs
* needed throughout the rest of the code. When the features this library
* needs are similar enough to POSIX to be hidden in a one-or-two line
* replacement, this macro approach is used.
* @{
*/
+
+ /** Wrapper around __func__, which is a C99 feature */
+#if __STDC_VERSION__ >= 199901L
+# define mdb_func_ __func__
+#elif __GNUC__ >= 2 || _MSC_VER >= 1300
+# define mdb_func_ __FUNCTION__
+#else
+/* If a debug message says <mdb_unknown>(), update the #if statements above */
+# define mdb_func_ "<mdb_unknown>"
+#endif
+
#ifdef _WIN32
#define MDB_USE_HASH 1
#define MDB_PIDLOCK 0
*/
# define DPRINTF(args) ((void) ((mdb_debug) && DPRINTF0 args))
# define DPRINTF0(fmt, ...) \
- fprintf(stderr, "%s:%d " fmt "\n", __func__, __LINE__, __VA_ARGS__)
+ fprintf(stderr, "%s:%d " fmt "\n", mdb_func_, __LINE__, __VA_ARGS__)
#else
# define DPRINTF(args) ((void) 0)
#endif
(((mc)->mc_flags & C_SUB) ? -(int)(mc)->mc_dbi : (int)(mc)->mc_dbi)
/** @} */
- /** A default memory page size.
- * The actual size is platform-dependent, but we use this for
- * boot-strapping. We probably should not be using this any more.
- * The #GET_PAGESIZE() macro is used to get the actual size.
+ /** @brief The maximum size of a database page.
+ *
+ * This is 32k, since it must fit in #MDB_page.%mp_upper.
+ *
+ * LMDB will use database pages < OS pages if needed.
+ * That causes more I/O in write transactions: The OS must
+ * know (read) the whole page before writing a partial page.
*
* Note that we don't currently support Huge pages. On Linux,
* regular data files cannot use Huge pages, and in general
* pressure from other processes is high. So until OSs have
* actual paging support for Huge pages, they're not viable.
*/
-#define MDB_PAGESIZE 4096
+#define MAX_PAGESIZE 0x8000
/** The minimum number of keys required in a database page.
* Setting this to a larger value will place a smaller bound on the
/** The version number for a database's lockfile format. */
#define MDB_LOCK_VERSION 1
- /** @brief The maximum size of a key in the database.
+ /** @brief The max size of a key we can write, or 0 for dynamic max.
*
- * The library rejects bigger keys, and cannot deal with records
- * with bigger keys stored by a library with bigger max keysize.
+ * Define this as 0 to compute the max from the page size. 511
+ * is default for backwards compat: liblmdb <= 0.9.10 can break
+ * when modifying a DB with keys/dupsort data bigger than its max.
*
- * We require that keys all fit onto a regular page. This limit
- * could be raised a bit further if needed; to something just
- * under #MDB_PAGESIZE / #MDB_MINKEYS.
- *
- * Note that data items in an #MDB_DUPSORT database are actually keys
- * of a subDB, so they're also limited to this size.
+ * Data items in an #MDB_DUPSORT database are also limited to
+ * this size, since they're actually keys of a sub-DB. Keys and
+ * #MDB_DUPSORT data items must fit on a node in a regular page.
*/
#ifndef MDB_MAXKEYSIZE
#define MDB_MAXKEYSIZE 511
+#endif
+
+ /** The maximum size of a key we can write to the environment. */
+#if MDB_MAXKEYSIZE
+#define ENV_MAXKEY(env) (MDB_MAXKEYSIZE)
+#else
+#define ENV_MAXKEY(env) ((env)->me_maxkey)
#endif
/** @brief The maximum size of a data item.
#define MAXDATASIZE 0xffffffffUL
#if MDB_DEBUG
+ /** Key size which fits in a #DKBUF.
+ * @ingroup debug
+ */
+#define DKBUF_MAXKEYSIZE ((MDB_MAXKEYSIZE) > 0 ? (MDB_MAXKEYSIZE) : 511)
/** A key buffer.
* @ingroup debug
* This is used for printing a hex dump of a key's contents.
*/
-#define DKBUF char kbuf[(MDB_MAXKEYSIZE*2+1)]
+#define DKBUF char kbuf[DKBUF_MAXKEYSIZE*2+1]
/** Display a key in hex.
* @ingroup debug
* Invoke a function to display a key in hex.
/** Test if the flags \b f are set in a flag word \b w. */
#define F_ISSET(w, f) (((w) & (f)) == (f))
+ /** Round \b n up to an even number. */
+#define EVEN(n) (((n) + 1U) & -2) /* sign-extending -2 to match n+1U */
+
/** Used for offsets within a single page.
* Since memory pages are typically 4 or 8KB in size, 12-13 bits,
* this is plenty.
*/
txnid_t mrb_txnid;
/** The process ID of the process owning this reader txn. */
- pid_t mrb_pid;
+ MDB_PID_T mrb_pid;
/** The thread ID of the thread owning this txn. */
pthread_t mrb_tid;
} MDB_rxbody;
#define P_DIRTY 0x10 /**< dirty page, also set for #P_SUBP pages */
#define P_LEAF2 0x20 /**< for #MDB_DUPFIXED records */
#define P_SUBP 0x40 /**< for #MDB_DUPSORT sub-pages */
+#define P_LOOSE 0x4000 /**< page was dirtied then freed, can be reused */
#define P_KEEP 0x8000 /**< leave this page alone during spill */
/** @} */
uint16_t mp_flags; /**< @ref mdb_page */
#define OVPAGES(size, psize) ((PAGEHDRSZ-1 + (size)) / (psize) + 1)
/** Header for a single key/data pair within a page.
- * We guarantee 2-byte alignment for nodes.
+ * Used in pages of type #P_BRANCH and #P_LEAF without #P_LEAF2.
+ * We guarantee 2-byte alignment for 'MDB_node's.
*/
typedef struct MDB_node {
/** lo and hi are used for data size on leaf nodes and for
* They are in host byte order in case that lets some
* accesses be optimized into a 32-bit word access.
*/
-#define mn_lo mn_offset[BYTE_ORDER!=LITTLE_ENDIAN]
-#define mn_hi mn_offset[BYTE_ORDER==LITTLE_ENDIAN] /**< part of dsize or pgno */
- unsigned short mn_offset[2]; /**< storage for #mn_lo and #mn_hi */
+#if BYTE_ORDER == LITTLE_ENDIAN
+ unsigned short mn_lo, mn_hi; /**< part of data size or pgno */
+#else
+ unsigned short mn_hi, mn_lo;
+#endif
/** @defgroup mdb_node Node Flags
* @ingroup internal
* Flags for node headers.
txnid_t mm_txnid; /**< txnid that committed this page */
} MDB_meta;
- /** Buffer for a stack-allocated dirty page.
+ /** Buffer for a stack-allocated meta page.
* The members define size and alignment, and silence type
* aliasing warnings. They are not used directly; that could
* mean incorrectly using several union members in parallel.
*/
-typedef union MDB_pagebuf {
- char mb_raw[MDB_PAGESIZE];
+typedef union MDB_metabuf {
MDB_page mb_page;
struct {
char mm_pad[PAGEHDRSZ];
MDB_meta mm_meta;
} mb_metabuf;
-} MDB_pagebuf;
+} MDB_metabuf;
/** Auxiliary DB info.
* The information here is mostly static/read-only. There is
* @{
*/
#define MDB_TXN_RDONLY 0x01 /**< read-only transaction */
-#define MDB_TXN_ERROR 0x02 /**< an error has occurred */
+#define MDB_TXN_ERROR 0x02 /**< txn is unusable after an error */
#define MDB_TXN_DIRTY 0x04 /**< must write, even if dirty list is empty */
#define MDB_TXN_SPILLS 0x08 /**< txn or a parent has spilled pages */
/** @} */
unsigned int mt_flags; /**< @ref mdb_txn */
- /** dirty_list room: Array size - #dirty pages visible to this txn.
+ /** #dirty_list room: Array size - \#dirty pages visible to this txn.
* Includes ancestor txns' dirty pages not hidden by other txns'
* dirty/spilled pages. Thus commit(nested txn) has room to merge
* dirty_list into mt_parent after freeing hidden mt_parent pages.
typedef struct MDB_pgstate {
pgno_t *mf_pghead; /**< Reclaimed freeDB pages, or NULL before use */
txnid_t mf_pglast; /**< ID of last used record, or 0 if !mf_pghead */
+ MDB_page *mf_pgloose; /**< Dirty pages that can be reused */
} MDB_pgstate;
/** The database environment. */
#define MDB_ENV_ACTIVE 0x20000000U
/** me_txkey is set */
#define MDB_ENV_TXKEY 0x10000000U
- /** Have liveness lock in reader table */
-#define MDB_LIVE_READER 0x08000000U
uint32_t me_flags; /**< @ref mdb_env */
- unsigned int me_psize; /**< size of a page, from #GET_PAGESIZE */
+ unsigned int me_psize; /**< DB page size, inited from me_os_psize */
+ unsigned int me_os_psize; /**< OS page size, from #GET_PAGESIZE */
unsigned int me_maxreaders; /**< size of the reader table */
unsigned int me_numreaders; /**< max numreaders set by this env */
MDB_dbi me_numdbs; /**< number of DBs opened */
MDB_dbi me_maxdbs; /**< size of the DB table */
- pid_t me_pid; /**< process ID of this env */
+ MDB_PID_T me_pid; /**< process ID of this env */
char *me_path; /**< path to the DB files */
char *me_map; /**< the memory map of the data file */
MDB_txninfo *me_txns; /**< the memory map of the lock file or NULL */
MDB_meta *me_metas[2]; /**< pointers to the two meta pages */
+ void *me_pbuf; /**< scratch area for DUPSORT put() */
MDB_txn *me_txn; /**< current write transaction */
size_t me_mapsize; /**< size of the data memory map */
off_t me_size; /**< current file size */
MDB_pgstate me_pgstate; /**< state of old pages from freeDB */
# define me_pglast me_pgstate.mf_pglast
# define me_pghead me_pgstate.mf_pghead
+# define me_pgloose me_pgstate.mf_pgloose
MDB_page *me_dpages; /**< list of malloc'd blocks for re-use */
/** IDL of pages that became unused in a write txn */
MDB_IDL me_free_pgs;
int me_maxfree_1pg;
/** Max size of a node on a page */
unsigned int me_nodemax;
+#if !(MDB_MAXKEYSIZE)
+ unsigned int me_maxkey; /**< max size of a key */
+#endif
+ int me_live_reader; /**< have liveness lock in reader table */
#ifdef _WIN32
int me_pidquery; /**< Used in OpenProcess */
HANDLE me_rmutex; /* Windows mutexes don't reside in shared mem */
sem_t *me_rmutex; /* Shared mutexes are not supported */
sem_t *me_wmutex;
#endif
+ void *me_userctx; /**< User-settable context */
+ MDB_assert_func *me_assert_func; /**< Callback for assertion failures */
};
/** Nested transaction */
#define MDB_COMMIT_PAGES IOV_MAX
#endif
- /* max bytes to write in one call */
+ /** max bytes to write in one call */
#define MAX_WRITE (0x80000000U >> (sizeof(ssize_t) == 4))
+ /** Check \b txn and \b dbi arguments to a function */
+#define TXN_DBI_EXIST(txn, dbi) \
+ ((txn) && (dbi) < (txn)->mt_numdbs && ((txn)->mt_dbflags[dbi] & DB_VALID))
+
static int mdb_page_alloc(MDB_cursor *mc, int num, MDB_page **mp);
static int mdb_page_new(MDB_cursor *mc, uint32_t flags, int num, MDB_page **mp);
static int mdb_page_touch(MDB_cursor *mc);
static MDB_node *mdb_node_search(MDB_cursor *mc, MDB_val *key, int *exactp);
static int mdb_node_add(MDB_cursor *mc, indx_t indx,
MDB_val *key, MDB_val *data, pgno_t pgno, unsigned int flags);
-static void mdb_node_del(MDB_page *mp, indx_t indx, int ksize);
+static void mdb_node_del(MDB_cursor *mc, int ksize);
static void mdb_node_shrink(MDB_page *mp, indx_t indx);
static int mdb_node_move(MDB_cursor *csrc, MDB_cursor *cdst);
static int mdb_node_read(MDB_txn *txn, MDB_node *leaf, MDB_val *data);
static void mdb_cursor_pop(MDB_cursor *mc);
static int mdb_cursor_push(MDB_cursor *mc, MDB_page *mp);
-static int mdb_cursor_del0(MDB_cursor *mc, MDB_node *leaf);
+static int mdb_cursor_del0(MDB_cursor *mc);
+static int mdb_del0(MDB_txn *txn, MDB_dbi dbi, MDB_val *key, MDB_val *data, unsigned flags);
static int mdb_cursor_sibling(MDB_cursor *mc, int move_right);
static int mdb_cursor_next(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op);
static int mdb_cursor_prev(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op);
"MDB_INCOMPATIBLE: Operation and DB incompatible, or DB flags changed",
"MDB_BAD_RSLOT: Invalid reuse of reader locktable slot",
"MDB_BAD_TXN: Transaction cannot recover - it must be aborted",
- "MDB_BAD_VALSIZE: Too big key/data, key is empty, or wrong DUPFIXED size",
+ "MDB_BAD_VALSIZE: Unsupported size of key/DB name/data, or wrong DUPFIXED size",
};
char *
return strerror(err);
}
+/** assert(3) variant in cursor context */
+#define mdb_cassert(mc, expr) mdb_assert0((mc)->mc_txn->mt_env, expr, #expr)
+/** assert(3) variant in transaction context */
+#define mdb_tassert(mc, expr) mdb_assert0((txn)->mt_env, expr, #expr)
+/** assert(3) variant in environment context */
+#define mdb_eassert(env, expr) mdb_assert0(env, expr, #expr)
+
+#ifndef NDEBUG
+# define mdb_assert0(env, expr, expr_txt) ((expr) ? (void)0 : \
+ mdb_assert_fail(env, expr_txt, mdb_func_, __FILE__, __LINE__))
+
+static void
+mdb_assert_fail(MDB_env *env, const char *expr_txt,
+ const char *func, const char *file, int line)
+{
+ char buf[400];
+ sprintf(buf, "%.100s:%d: Assertion '%.200s' failed in %.40s()",
+ file, line, expr_txt, func);
+ if (env->me_assert_func)
+ env->me_assert_func(env, buf);
+ fprintf(stderr, "%s\n", buf);
+ abort();
+}
+#else
+# define mdb_assert0(env, expr, expr_txt) ((void) 0)
+#endif /* NDEBUG */
+
#if MDB_DEBUG
+/** Return the page number of \b mp which may be sub-page, for debug output */
+static pgno_t
+mdb_dbg_pgno(MDB_page *mp)
+{
+ pgno_t ret;
+ COPY_PGNO(ret, mp->mp_pgno);
+ return ret;
+}
+
/** Display a key in hexadecimal and return the address of the result.
* @param[in] key the key to display
* @param[in] buf the buffer to write into. Should always be #DKBUF.
if (!key)
return "";
- if (key->mv_size > MDB_MAXKEYSIZE)
+ if (key->mv_size > DKBUF_MAXKEYSIZE)
return "MDB_MAXKEYSIZE";
/* may want to make this a dynamic check: if the key is mostly
* printable characters, print it as-is instead of converting to hex.
return buf;
}
+static const char *
+mdb_leafnode_type(MDB_node *n)
+{
+ static char *const tp[2][2] = {{"", ": DB"}, {": sub-page", ": sub-DB"}};
+ return F_ISSET(n->mn_flags, F_BIGDATA) ? ": overflow page" :
+ tp[F_ISSET(n->mn_flags, F_DUPDATA)][F_ISSET(n->mn_flags, F_SUBDATA)];
+}
+
/** Display all the keys in the page. */
void
mdb_page_list(MDB_page *mp)
{
+ pgno_t pgno = mdb_dbg_pgno(mp);
+ const char *type, *state = (mp->mp_flags & P_DIRTY) ? ", dirty" : "";
MDB_node *node;
- unsigned int i, nkeys, nsize;
+ unsigned int i, nkeys, nsize, total = 0;
MDB_val key;
DKBUF;
+ switch (mp->mp_flags & (P_BRANCH|P_LEAF|P_LEAF2|P_META|P_OVERFLOW|P_SUBP)) {
+ case P_BRANCH: type = "Branch page"; break;
+ case P_LEAF: type = "Leaf page"; break;
+ case P_LEAF|P_SUBP: type = "Sub-page"; break;
+ case P_LEAF|P_LEAF2: type = "LEAF2 page"; break;
+ case P_LEAF|P_LEAF2|P_SUBP: type = "LEAF2 sub-page"; break;
+ case P_OVERFLOW:
+ fprintf(stderr, "Overflow page %"Z"u pages %u%s\n",
+ pgno, mp->mp_pages, state);
+ return;
+ case P_META:
+ fprintf(stderr, "Meta-page %"Z"u txnid %"Z"u\n",
+ pgno, ((MDB_meta *)METADATA(mp))->mm_txnid);
+ return;
+ default:
+ fprintf(stderr, "Bad page %"Z"u flags 0x%u\n", pgno, mp->mp_flags);
+ return;
+ }
+
nkeys = NUMKEYS(mp);
- fprintf(stderr, "Page %"Z"u numkeys %d\n", mp->mp_pgno, nkeys);
+ fprintf(stderr, "%s %"Z"u numkeys %d%s\n", type, pgno, nkeys, state);
+
for (i=0; i<nkeys; i++) {
+ if (IS_LEAF2(mp)) { /* LEAF2 pages have no mp_ptrs[] or node headers */
+ key.mv_size = nsize = mp->mp_pad;
+ key.mv_data = LEAF2KEY(mp, i, nsize);
+ total += nsize;
+ fprintf(stderr, "key %d: nsize %d, %s\n", i, nsize, DKEY(&key));
+ continue;
+ }
node = NODEPTR(mp, i);
key.mv_size = node->mn_ksize;
key.mv_data = node->mn_data;
- nsize = NODESIZE + NODEKSZ(node) + sizeof(indx_t);
+ nsize = NODESIZE + key.mv_size;
if (IS_BRANCH(mp)) {
fprintf(stderr, "key %d: page %"Z"u, %s\n", i, NODEPGNO(node),
DKEY(&key));
+ total += nsize;
} else {
if (F_ISSET(node->mn_flags, F_BIGDATA))
nsize += sizeof(pgno_t);
else
nsize += NODEDSZ(node);
- fprintf(stderr, "key %d: nsize %d, %s\n", i, nsize, DKEY(&key));
+ total += nsize;
+ nsize += sizeof(indx_t);
+ fprintf(stderr, "key %d: nsize %d, %s%s\n",
+ i, nsize, DKEY(&key), mdb_leafnode_type(node));
}
+ total = EVEN(total);
}
+ fprintf(stderr, "Total: header %d + contents %d + unused %d\n",
+ IS_LEAF2(mp) ? PAGEHDRSZ : mp->mp_lower, total, SIZELEFT(mp));
}
void
/** Count all the pages in each DB and in the freelist
* and make sure it matches the actual number of pages
* being used.
+ * All named DBs must be open for a correct count.
*/
static void mdb_audit(MDB_txn *txn)
{
mdb_cursor_init(&mc, txn, FREE_DBI, NULL);
while ((rc = mdb_cursor_get(&mc, &key, &data, MDB_NEXT)) == 0)
freecount += *(MDB_ID *)data.mv_data;
+ mdb_tassert(txn, rc == MDB_NOTFOUND);
count = 0;
for (i = 0; i<txn->mt_numdbs; i++) {
MDB_xcursor mx;
+ if (!(txn->mt_dbflags[i] & DB_VALID))
+ continue;
mdb_cursor_init(&mc, txn, i, &mx);
if (txn->mt_dbs[i].md_root == P_INVALID)
continue;
txn->mt_dbs[i].md_leaf_pages +
txn->mt_dbs[i].md_overflow_pages;
if (txn->mt_dbs[i].md_flags & MDB_DUPSORT) {
- mdb_page_search(&mc, NULL, MDB_PS_FIRST);
- do {
+ rc = mdb_page_search(&mc, NULL, MDB_PS_FIRST);
+ for (; rc == MDB_SUCCESS; rc = mdb_cursor_sibling(&mc, 1)) {
unsigned j;
MDB_page *mp;
mp = mc.mc_pg[mc.mc_top];
}
}
}
- while (mdb_cursor_sibling(&mc, 1) == 0);
+ mdb_tassert(txn, rc == MDB_NOTFOUND);
}
}
if (freecount + count + 2 /* metapages */ != txn->mt_next_pgno) {
{
MDB_env *env = txn->mt_env;
MDB_page *ret = env->me_dpages;
- size_t sz = env->me_psize;
+ size_t psize = env->me_psize, sz = psize, off;
+ /* For ! #MDB_NOMEMINIT, psize counts how much to init.
+ * For a single page alloc, we init everything after the page header.
+ * For multi-page, we init the final page; if the caller needed that
+ * many pages they will be filling in at least up to the last page.
+ */
if (num == 1) {
if (ret) {
VGMEMP_ALLOC(env, ret, sz);
env->me_dpages = ret->mp_next;
return ret;
}
+ psize -= off = PAGEHDRSZ;
} else {
sz *= num;
+ off = sz - psize;
}
if ((ret = malloc(sz)) != NULL) {
VGMEMP_ALLOC(env, ret, sz);
+ if (!(env->me_flags & MDB_NOMEMINIT)) {
+ memset((char *)ret + off, 0, psize);
+ ret->mp_pad = 0;
+ }
+ } else {
+ txn->mt_flags |= MDB_TXN_ERROR;
}
return ret;
}
-
/** Free a single page.
* Saves single pages to a list, for future reuse.
* (This is not used for multi-page overflow pages.)
dl[0].mid = 0;
}
+/** Loosen a single page.
+ * Saves single pages to a list for future reuse
+ * in this same txn. It has been pulled from the freeDB
+ * and already resides on the dirty list, but has been
+ * deleted. Use these pages first before pulling again
+ * from the freeDB.
+ */
+static void
+mdb_page_loose(MDB_env *env, MDB_page *mp)
+{
+ pgno_t *pp = (pgno_t *)mp->mp_ptrs;
+ *pp = mp->mp_pgno;
+ mp->mp_next = env->me_pgloose;
+ env->me_pgloose = mp;
+ mp->mp_flags |= P_LOOSE;
+}
+
/** Set or clear P_KEEP in dirty, non-overflow, non-sub pages watched by txn.
* @param[in] mc A cursor handle for the current operation.
* @param[in] pflags Flags of the pages to update:
break;
}
+ /* Loose pages shouldn't be spilled */
+ for (dp = txn->mt_env->me_pgloose; dp; dp=dp->mp_next) {
+ if ((dp->mp_flags & Mask) == pflags)
+ dp->mp_flags ^= P_KEEP;
+ }
+
if (all) {
/* Mark dirty root pages */
for (i=0; i<txn->mt_numdbs; i++) {
mdb_page_dirty(MDB_txn *txn, MDB_page *mp)
{
MDB_ID2 mid;
- int (*insert)(MDB_ID2L, MDB_ID2 *);
+ int rc, (*insert)(MDB_ID2L, MDB_ID2 *);
if (txn->mt_env->me_flags & MDB_WRITEMAP) {
insert = mdb_mid2l_append;
}
mid.mid = mp->mp_pgno;
mid.mptr = mp;
- insert(txn->mt_u.dirty_list, &mid);
+ rc = insert(txn->mt_u.dirty_list, &mid);
+ mdb_tassert(txn, rc == 0);
txn->mt_dirty_room--;
}
#else
enum { Paranoid = 0, Max_retries = INT_MAX /*infinite*/ };
#endif
- int rc, n2 = num-1, retry = Max_retries;
+ int rc, retry = num * 20;
MDB_txn *txn = mc->mc_txn;
MDB_env *env = txn->mt_env;
pgno_t pgno, *mop = env->me_pghead;
- unsigned i, j, k, mop_len = mop ? mop[0] : 0;
+ unsigned i, j, k, mop_len = mop ? mop[0] : 0, n2 = num-1;
MDB_page *np;
txnid_t oldest = 0, last;
MDB_cursor_op op;
MDB_cursor m2;
+ /* If there are any loose pages, just use them */
+ if (num == 1 && env->me_pgloose) {
+ pgno_t *pp;
+ np = env->me_pgloose;
+ env->me_pgloose = np->mp_next;
+ pp = (pgno_t *)np->mp_ptrs;
+ np->mp_pgno = *pp;
+ *mp = np;
+ return MDB_SUCCESS;
+ }
+
*mp = NULL;
/* If our dirty list is already full, we can't do anything */
- if (txn->mt_dirty_room == 0)
- return MDB_TXN_FULL;
+ if (txn->mt_dirty_room == 0) {
+ rc = MDB_TXN_FULL;
+ goto fail;
+ }
for (op = MDB_FIRST;; op = MDB_NEXT) {
MDB_val key, data;
/* Seek a big enough contiguous page range. Prefer
* pages at the tail, just truncating the list.
*/
- if (mop_len >= (unsigned)num) {
+ if (mop_len > n2) {
i = mop_len;
do {
pgno = mop[i];
if (mop[i-n2] == pgno+n2)
goto search_done;
- } while (--i >= (unsigned)num);
- if (Max_retries < INT_MAX && --retry < 0)
+ } while (--i > n2);
+ if (--retry < 0)
break;
}
if (rc) {
if (rc == MDB_NOTFOUND)
break;
- return rc;
+ goto fail;
}
last = *(txnid_t*)key.mv_data;
if (oldest <= last)
idl = (MDB_ID *) data.mv_data;
i = idl[0];
if (!mop) {
- if (!(env->me_pghead = mop = mdb_midl_alloc(i)))
- return ENOMEM;
+ if (!(env->me_pghead = mop = mdb_midl_alloc(i))) {
+ rc = ENOMEM;
+ goto fail;
+ }
} else {
if ((rc = mdb_midl_need(&env->me_pghead, i)) != 0)
- return rc;
+ goto fail;
mop = env->me_pghead;
}
env->me_pglast = last;
pgno = txn->mt_next_pgno;
if (pgno + num >= env->me_maxpg) {
DPUTS("DB size maxed out");
- return MDB_MAP_FULL;
+ rc = MDB_MAP_FULL;
+ goto fail;
}
search_done:
if (env->me_flags & MDB_WRITEMAP) {
np = (MDB_page *)(env->me_map + env->me_psize * pgno);
} else {
- if (!(np = mdb_page_malloc(txn, num)))
- return ENOMEM;
+ if (!(np = mdb_page_malloc(txn, num))) {
+ rc = ENOMEM;
+ goto fail;
+ }
}
if (i) {
mop[0] = mop_len -= num;
*mp = np;
return MDB_SUCCESS;
+
+fail:
+ txn->mt_flags |= MDB_TXN_ERROR;
+ return rc;
}
/** Copy the used portions of a non-overflow page.
* If a page being referenced was spilled to disk in this txn, bring
* it back and make it dirty/writable again.
* @param[in] txn the transaction handle.
- * @param[in] mp the page being referenced.
+ * @param[in] mp the page being referenced. It must not be dirty.
* @param[out] ret the writable page, if any. ret is unchanged if
* mp wasn't spilled.
*/
np = NULL;
rc = mdb_page_unspill(txn, mp, &np);
if (rc)
- return rc;
+ goto fail;
if (np)
goto done;
}
if ((rc = mdb_midl_need(&txn->mt_free_pgs, 1)) ||
(rc = mdb_page_alloc(mc, 1, &np)))
- return rc;
+ goto fail;
pgno = np->mp_pgno;
DPRINTF(("touched db %d page %"Z"u -> %"Z"u", DDBI(mc),
mp->mp_pgno, pgno));
- assert(mp->mp_pgno != pgno);
+ mdb_cassert(mc, mp->mp_pgno != pgno);
mdb_midl_xappend(txn->mt_free_pgs, mp->mp_pgno);
/* Update the parent page, if any, to point to the new page */
if (mc->mc_top) {
if (x <= dl[0].mid && dl[x].mid == pgno) {
if (mp != dl[x].mptr) { /* bad cursor? */
mc->mc_flags &= ~(C_INITIALIZED|C_EOF);
+ txn->mt_flags |= MDB_TXN_ERROR;
return MDB_CORRUPTED;
}
return 0;
}
}
- assert(dl[0].mid < MDB_IDL_UM_MAX);
+ mdb_cassert(mc, dl[0].mid < MDB_IDL_UM_MAX);
/* No - copy it */
np = mdb_page_malloc(txn, 1);
if (!np)
return ENOMEM;
mid.mid = pgno;
mid.mptr = np;
- mdb_mid2l_insert(dl, &mid);
+ rc = mdb_mid2l_insert(dl, &mid);
+ mdb_cassert(mc, rc == 0);
} else {
return 0;
}
if (m2->mc_pg[mc->mc_top] == mp) {
m2->mc_pg[mc->mc_top] = np;
if ((mc->mc_db->md_flags & MDB_DUPSORT) &&
+ IS_LEAF(np) &&
m2->mc_ki[mc->mc_top] == mc->mc_ki[mc->mc_top])
{
MDB_node *leaf = NODEPTR(np, mc->mc_ki[mc->mc_top]);
}
}
return 0;
+
+fail:
+ txn->mt_flags |= MDB_TXN_ERROR;
+ return rc;
}
int
* lock on the lockfile, set at an offset equal to the pid.
*/
static int
-mdb_reader_pid(MDB_env *env, enum Pidlock_op op, pid_t pid)
+mdb_reader_pid(MDB_env *env, enum Pidlock_op op, MDB_PID_T pid)
{
#if !(MDB_PIDLOCK) /* Currently the same as defined(_WIN32) */
int ret = 0;
if (r->mr_pid != env->me_pid || r->mr_txnid != (txnid_t)-1)
return MDB_BAD_RSLOT;
} else {
- pid_t pid = env->me_pid;
+ MDB_PID_T pid = env->me_pid;
pthread_t tid = pthread_self();
- if (!(env->me_flags & MDB_LIVE_READER)) {
+ if (!env->me_live_reader) {
rc = mdb_reader_pid(env, Pidset, pid);
- if (rc) {
- UNLOCK_MUTEX_R(env);
+ if (rc)
return rc;
- }
- env->me_flags |= MDB_LIVE_READER;
+ env->me_live_reader = 1;
}
LOCK_MUTEX_R(env);
int rc, maxfree_1pg = env->me_maxfree_1pg, more = 1;
txnid_t pglast = 0, head_id = 0;
pgno_t freecnt = 0, *free_pgs, *mop;
- ssize_t head_room = 0, total_room = 0, mop_len;
+ ssize_t head_room = 0, total_room = 0, mop_len, clean_limit;
mdb_cursor_init(&mc, txn, FREE_DBI, NULL);
return rc;
}
+ /* Dispose of loose pages. Usually they will have all
+ * been used up by the time we get here.
+ */
+ if (env->me_pgloose) {
+ MDB_page *mp = env->me_pgloose;
+ pgno_t *pp;
+ /* Just return them to freeDB */
+ if (env->me_pghead) {
+ int i, j;
+ mop = env->me_pghead;
+ while(mp) {
+ pgno_t pg;
+ pp = (pgno_t *)mp->mp_ptrs;
+ pg = *pp;
+ j = mop[0] + 1;
+ for (i = mop[0]; i && mop[i] < pg; i--)
+ mop[j--] = mop[i];
+ mop[j] = pg;
+ mop[0] += 1;
+ mp = mp->mp_next;
+ }
+ } else {
+ /* Oh well, they were wasted. Put on freelist */
+ while(mp) {
+ pp = (pgno_t *)mp->mp_ptrs;
+ mdb_midl_append(&txn->mt_free_pgs, *pp);
+ mp = mp->mp_next;
+ }
+ }
+ env->me_pgloose = NULL;
+ }
+
+ /* MDB_RESERVE cancels meminit in ovpage malloc (when no WRITEMAP) */
+ clean_limit = (env->me_flags & (MDB_NOMEMINIT|MDB_WRITEMAP))
+ ? SSIZE_MAX : maxfree_1pg;
+
for (;;) {
/* Come back here after each Put() in case freelist changed */
MDB_val key, data;
+ pgno_t *pgs;
+ ssize_t j;
/* If using records from freeDB which we have not yet
* deleted, delete them and any we reserved for me_pghead.
return rc;
pglast = head_id = *(txnid_t *)key.mv_data;
total_room = head_room = 0;
- assert(pglast <= env->me_pglast);
+ mdb_tassert(txn, pglast <= env->me_pglast);
rc = mdb_cursor_del(&mc, 0);
if (rc)
return rc;
rc = mdb_cursor_put(&mc, &key, &data, MDB_RESERVE);
if (rc)
return rc;
- *(MDB_ID *)data.mv_data = 0; /* IDL is initially empty */
+ /* IDL is initially empty, zero out at least the length */
+ pgs = (pgno_t *)data.mv_data;
+ j = head_room > clean_limit ? head_room : 0;
+ do {
+ pgs[j] = 0;
+ } while (--j >= 0);
total_room += head_room;
}
mop += mop_len;
rc = mdb_cursor_first(&mc, &key, &data);
for (; !rc; rc = mdb_cursor_next(&mc, &key, &data, MDB_NEXT)) {
- unsigned flags = MDB_CURRENT;
txnid_t id = *(txnid_t *)key.mv_data;
ssize_t len = (ssize_t)(data.mv_size / sizeof(MDB_ID)) - 1;
MDB_ID save;
- assert(len >= 0 && id <= env->me_pglast);
+ mdb_tassert(txn, len >= 0 && id <= env->me_pglast);
key.mv_data = &id;
if (len > mop_len) {
len = mop_len;
data.mv_size = (len + 1) * sizeof(MDB_ID);
- flags = 0;
}
data.mv_data = mop -= len;
save = mop[0];
mop[0] = len;
- rc = mdb_cursor_put(&mc, &key, &data, flags);
+ rc = mdb_cursor_put(&mc, &key, &data, MDB_CURRENT);
mop[0] = save;
if (rc || !(mop_len -= len))
break;
unsigned int i;
MDB_env *env;
- assert(txn != NULL);
- assert(txn->mt_env != NULL);
+ if (txn == NULL || txn->mt_env == NULL)
+ return EINVAL;
if (txn->mt_child) {
rc = mdb_txn_commit(txn->mt_child);
if (yp == dst[x].mid)
free(dst[x--].mptr);
}
- assert(i == x);
+ mdb_tassert(txn, i == x);
dst[0].mid = len;
free(txn->mt_u.dirty_list);
parent->mt_dirty_room = txn->mt_dirty_room;
static int
mdb_env_read_header(MDB_env *env, MDB_meta *meta)
{
- MDB_pagebuf pbuf;
+ MDB_metabuf pbuf;
MDB_page *p;
MDB_meta *m;
int i, rc, off;
+ enum { Size = sizeof(pbuf) };
/* We don't know the page size yet, so use a minimum value.
* Read both meta pages so we can use the latest one.
OVERLAPPED ov;
memset(&ov, 0, sizeof(ov));
ov.Offset = off;
- rc = ReadFile(env->me_fd,&pbuf,MDB_PAGESIZE,&len,&ov) ? (int)len : -1;
+ rc = ReadFile(env->me_fd, &pbuf, Size, &len, &ov) ? (int)len : -1;
if (rc == -1 && ErrCode() == ERROR_HANDLE_EOF)
rc = 0;
#else
- rc = pread(env->me_fd, &pbuf, MDB_PAGESIZE, off);
+ rc = pread(env->me_fd, &pbuf, Size, off);
#endif
- if (rc != MDB_PAGESIZE) {
+ if (rc != Size) {
if (rc == 0 && off == 0)
return ENOENT;
rc = rc < 0 ? (int) ErrCode() : MDB_INVALID;
int r2;
#endif
- assert(txn != NULL);
- assert(txn->mt_env != NULL);
-
toggle = txn->mt_txnid & 1;
DPRINTF(("writing meta page %d for root page %"Z"u",
toggle, txn->mt_dbs[MAIN_DBI].md_root));
mp->mm_last_pg = txn->mt_next_pgno - 1;
mp->mm_txnid = txn->mt_txnid;
if (!(env->me_flags & (MDB_NOMETASYNC|MDB_NOSYNC))) {
+ unsigned meta_size = env->me_psize;
rc = (env->me_flags & MDB_MAPASYNC) ? MS_ASYNC : MS_SYNC;
ptr = env->me_map;
- if (toggle)
- ptr += env->me_psize;
- if (MDB_MSYNC(ptr, env->me_psize, rc)) {
+ if (toggle) {
+#ifndef _WIN32 /* POSIX msync() requires ptr = start of OS page */
+ if (meta_size < env->me_os_psize)
+ meta_size += meta_size;
+ else
+#endif
+ ptr += meta_size;
+ }
+ if (MDB_MSYNC(ptr, meta_size, rc)) {
rc = ErrCode();
goto fail;
}
e->me_wmutex = SEM_FAILED;
#endif
e->me_pid = getpid();
+ GET_PAGESIZE(e->me_os_psize);
VGMEMP_CREATE(e,0,0);
*env = e;
return MDB_SUCCESS;
return i;
DPUTS("new mdbenv");
newenv = 1;
- GET_PAGESIZE(env->me_psize);
+ env->me_psize = env->me_os_psize;
+ if (env->me_psize > MAX_PAGESIZE)
+ env->me_psize = MAX_PAGESIZE;
} else {
env->me_psize = meta.mm_psize;
}
env->me_mapsize = minsize;
}
- rc = mdb_env_map(env, meta.mm_address, newenv);
+ rc = mdb_env_map(env, meta.mm_address, newenv || env->me_mapsize != meta.mm_mapsize);
if (rc)
return rc;
return i;
}
}
- env->me_maxfree_1pg = (env->me_psize - PAGEHDRSZ) / sizeof(pgno_t) - 1;
- env->me_nodemax = (env->me_psize - PAGEHDRSZ) / MDB_MINKEYS;
+ env->me_maxfree_1pg = (env->me_psize - PAGEHDRSZ) / sizeof(pgno_t) - 1;
+ env->me_nodemax = (((env->me_psize - PAGEHDRSZ) / MDB_MINKEYS) & -2)
+ - sizeof(indx_t);
+#if !(MDB_MAXKEYSIZE)
+ env->me_maxkey = env->me_nodemax - (NODESIZE + sizeof(MDB_db));
+#endif
env->me_maxpg = env->me_mapsize / env->me_psize;
+
#if MDB_DEBUG
{
int toggle = mdb_env_pick_meta(env);
case DLL_THREAD_DETACH:
for (i=0; i<mdb_tls_nkeys; i++) {
MDB_reader *r = pthread_getspecific(mdb_tls_keys[i]);
- mdb_env_reader_dest(r);
+ if (r) {
+ mdb_env_reader_dest(r);
+ }
}
break;
case DLL_PROCESS_DETACH: break;
#pragma comment(linker, "/INCLUDE:_tls_used")
#pragma comment(linker, "/INCLUDE:mdb_tls_cbp")
#pragma const_seg(".CRT$XLB")
-extern const PIMAGE_TLS_CALLBACK mdb_tls_callback;
+extern const PIMAGE_TLS_CALLBACK mdb_tls_cbp;
const PIMAGE_TLS_CALLBACK mdb_tls_cbp = mdb_tls_callback;
#pragma const_seg()
#else /* WIN32 */
rsize = (env->me_maxreaders-1) * sizeof(MDB_reader) + sizeof(MDB_txninfo);
if (size < rsize && *excl > 0) {
#ifdef _WIN32
- if (SetFilePointer(env->me_lfd, rsize, NULL, FILE_BEGIN) != rsize
+ if (SetFilePointer(env->me_lfd, rsize, NULL, FILE_BEGIN) != (DWORD)rsize
|| !SetEndOfFile(env->me_lfd))
goto fail_errno;
#else
* at runtime. Changing other flags requires closing the
* environment and re-opening it with the new flags.
*/
-#define CHANGEABLE (MDB_NOSYNC|MDB_NOMETASYNC|MDB_MAPASYNC)
-#define CHANGELESS (MDB_FIXEDMAP|MDB_NOSUBDIR|MDB_RDONLY|MDB_WRITEMAP|MDB_NOTLS|MDB_NOLOCK|MDB_NORDAHEAD)
+#define CHANGEABLE (MDB_NOSYNC|MDB_NOMETASYNC|MDB_MAPASYNC|MDB_NOMEMINIT)
+#define CHANGELESS (MDB_FIXEDMAP|MDB_NOSUBDIR|MDB_RDONLY|MDB_WRITEMAP| \
+ MDB_NOTLS|MDB_NOLOCK|MDB_NORDAHEAD)
+
+#if VALID_FLAGS & PERSISTENT_FLAGS & (CHANGEABLE|CHANGELESS)
+# error "Persistent DB flags & env flags overlap, but both go in mm_flags"
+#endif
int
mdb_env_open(MDB_env *env, const char *path, unsigned int flags, mdb_mode_t mode)
DPRINTF(("opened dbenv %p", (void *) env));
if (excl > 0) {
rc = mdb_env_share_locks(env, &excl);
+ if (rc)
+ goto leave;
}
+ if (!((flags & MDB_RDONLY) ||
+ (env->me_pbuf = calloc(1, env->me_psize))))
+ rc = ENOMEM;
}
leave:
for (i = env->me_maxdbs; --i > MAIN_DBI; )
free(env->me_dbxs[i].md_name.mv_data);
+ free(env->me_pbuf);
free(env->me_dbflags);
free(env->me_dbxs);
free(env->me_path);
if (env->me_fd != INVALID_HANDLE_VALUE)
(void) close(env->me_fd);
if (env->me_txns) {
- pid_t pid = env->me_pid;
+ MDB_PID_T pid = env->me_pid;
/* Clearing readers is done in this function because
* me_txkey with its destructor must be disabled first.
*/
#endif
}
+/** Compare two items pointing at size_t's of unknown alignment. */
+#ifdef MISALIGNED_OK
+# define mdb_cmp_clong mdb_cmp_long
+#else
+# define mdb_cmp_clong mdb_cmp_cint
+#endif
+
/** Compare two items lexically */
static int
mdb_cmp_memn(const MDB_val *a, const MDB_val *b)
nkeys = NUMKEYS(mp);
-#if MDB_DEBUG
- {
- pgno_t pgno;
- COPY_PGNO(pgno, mp->mp_pgno);
DPRINTF(("searching %u keys in %s %spage %"Z"u",
nkeys, IS_LEAF(mp) ? "leaf" : "branch", IS_SUBP(mp) ? "sub-" : "",
- pgno));
- }
-#endif
-
- assert(nkeys > 0);
+ mdb_dbg_pgno(mp)));
low = IS_LEAF(mp) ? 0 : 1;
high = nkeys - 1;
node = NODEPTR(mp, i);
}
if (exactp)
- *exactp = (rc == 0);
+ *exactp = (rc == 0 && nkeys > 0);
/* store the key index */
mc->mc_ki[mc->mc_top] = i;
if (i >= nkeys)
DDBI(mc), (void *) mc));
if (mc->mc_snum >= CURSOR_STACK) {
- assert(mc->mc_snum < CURSOR_STACK);
+ mc->mc_txn->mt_flags |= MDB_TXN_ERROR;
return MDB_CURSOR_FULL;
}
p = (MDB_page *)(env->me_map + env->me_psize * pgno);
} else {
DPRINTF(("page %"Z"u not found", pgno));
- assert(p != NULL);
+ txn->mt_flags |= MDB_TXN_ERROR;
return MDB_PAGE_NOTFOUND;
}
indx_t i;
DPRINTF(("branch page %"Z"u has %u keys", mp->mp_pgno, NUMKEYS(mp)));
- assert(NUMKEYS(mp) > 1);
+ mdb_cassert(mc, NUMKEYS(mp) > 1);
DPRINTF(("found index 0 to page %"Z"u", NODEPGNO(NODEPTR(mp, 0))));
if (flags & (MDB_PS_FIRST|MDB_PS_LAST)) {
else {
i = mc->mc_ki[mc->mc_top];
if (!exact) {
- assert(i > 0);
+ mdb_cassert(mc, i > 0);
i--;
}
}
DPRINTF(("following index %u for key [%s]", i, DKEY(key)));
}
- assert(i < NUMKEYS(mp));
+ mdb_cassert(mc, i < NUMKEYS(mp));
node = NODEPTR(mp, i);
if ((rc = mdb_page_get(mc->mc_txn, NODEPGNO(node), &mp, NULL)) != 0)
if (!IS_LEAF(mp)) {
DPRINTF(("internal error, index points to a %02X page!?",
mp->mp_flags));
+ mc->mc_txn->mt_flags |= MDB_TXN_ERROR;
return MDB_CORRUPTED;
}
}
}
- assert(root > 1);
+ mdb_cassert(mc, root > 1);
if (!mc->mc_pg[0] || mc->mc_pg[0]->mp_pgno != root)
if ((rc = mdb_page_get(mc->mc_txn, root, &mc->mc_pg[0], NULL)) != 0)
return rc;
iy = dl[x];
dl[x] = ix;
} else {
- assert(x > 1);
+ mdb_cassert(mc, x > 1);
j = ++(dl[0].mid);
dl[j] = ix; /* Unsorted. OK when MDB_TXN_ERROR. */
txn->mt_flags |= MDB_TXN_ERROR;
int exact = 0;
DKBUF;
- assert(key);
- assert(data);
DPRINTF(("===> get db %u key [%s]", dbi, DKEY(key)));
- if (txn == NULL || !dbi || dbi >= txn->mt_numdbs || !(txn->mt_dbflags[dbi] & DB_VALID))
+ if (!key || !data || dbi == FREE_DBI || !TXN_DBI_EXIST(txn, dbi))
return EINVAL;
if (txn->mt_flags & MDB_TXN_ERROR)
return MDB_BAD_TXN;
- if (key->mv_size > MDB_MAXKEYSIZE) {
- return MDB_BAD_VALSIZE;
- }
-
mdb_cursor_init(&mc, txn, dbi, &mx);
return mdb_cursor_set(&mc, key, data, MDB_SET, &exact);
}
DPRINTF(("just moving to %s index key %u",
move_right ? "right" : "left", mc->mc_ki[mc->mc_top]));
}
- assert(IS_BRANCH(mc->mc_pg[mc->mc_top]));
+ mdb_cassert(mc, IS_BRANCH(mc->mc_pg[mc->mc_top]));
indx = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]);
if ((rc = mdb_page_get(mc->mc_txn, NODEPGNO(indx), &mp, NULL)) != 0) {
return MDB_NOTFOUND;
}
- assert(mc->mc_flags & C_INITIALIZED);
+ mdb_cassert(mc, mc->mc_flags & C_INITIALIZED);
mp = mc->mc_pg[mc->mc_top];
}
}
- DPRINTF(("cursor_next: top page is %"Z"u in cursor %p", mp->mp_pgno, (void *) mc));
+ DPRINTF(("cursor_next: top page is %"Z"u in cursor %p",
+ mdb_dbg_pgno(mp), (void *) mc));
if (mc->mc_flags & C_DEL)
goto skip;
skip:
DPRINTF(("==> cursor points to page %"Z"u with %u keys, key index %u",
- mp->mp_pgno, NUMKEYS(mp), mc->mc_ki[mc->mc_top]));
+ mdb_dbg_pgno(mp), NUMKEYS(mp), mc->mc_ki[mc->mc_top]));
if (IS_LEAF2(mp)) {
key->mv_size = mc->mc_db->md_pad;
return MDB_SUCCESS;
}
- assert(IS_LEAF(mp));
+ mdb_cassert(mc, IS_LEAF(mp));
leaf = NODEPTR(mp, mc->mc_ki[mc->mc_top]);
if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
MDB_node *leaf;
int rc;
- assert(mc->mc_flags & C_INITIALIZED);
+ mdb_cassert(mc, mc->mc_flags & C_INITIALIZED);
mp = mc->mc_pg[mc->mc_top];
}
}
- DPRINTF(("cursor_prev: top page is %"Z"u in cursor %p", mp->mp_pgno, (void *) mc));
+ DPRINTF(("cursor_prev: top page is %"Z"u in cursor %p",
+ mdb_dbg_pgno(mp), (void *) mc));
if (mc->mc_ki[mc->mc_top] == 0) {
DPUTS("=====> move to prev sibling page");
mc->mc_flags &= ~C_EOF;
DPRINTF(("==> cursor points to page %"Z"u with %u keys, key index %u",
- mp->mp_pgno, NUMKEYS(mp), mc->mc_ki[mc->mc_top]));
+ mdb_dbg_pgno(mp), NUMKEYS(mp), mc->mc_ki[mc->mc_top]));
if (IS_LEAF2(mp)) {
key->mv_size = mc->mc_db->md_pad;
return MDB_SUCCESS;
}
- assert(IS_LEAF(mp));
+ mdb_cassert(mc, IS_LEAF(mp));
leaf = NODEPTR(mp, mc->mc_ki[mc->mc_top]);
if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
MDB_node *leaf = NULL;
DKBUF;
- assert(mc);
- assert(key);
if (key->mv_size == 0)
return MDB_BAD_VALSIZE;
if (!mc->mc_top) {
/* There are no other pages */
mc->mc_ki[mc->mc_top] = 0;
- if (op == MDB_SET_RANGE) {
+ if (op == MDB_SET_RANGE && !exactp) {
rc = 0;
goto set1;
} else
return rc;
mp = mc->mc_pg[mc->mc_top];
- assert(IS_LEAF(mp));
+ mdb_cassert(mc, IS_LEAF(mp));
set2:
leaf = mdb_node_search(mc, key, exactp);
if ((rc = mdb_cursor_sibling(mc, 1)) != MDB_SUCCESS)
return rc; /* no entries matched */
mp = mc->mc_pg[mc->mc_top];
- assert(IS_LEAF(mp));
+ mdb_cassert(mc, IS_LEAF(mp));
leaf = NODEPTR(mp, 0);
}
mc->mc_flags &= ~C_EOF;
if (IS_LEAF2(mp)) {
- key->mv_size = mc->mc_db->md_pad;
- key->mv_data = LEAF2KEY(mp, mc->mc_ki[mc->mc_top], key->mv_size);
+ if (op == MDB_SET_RANGE || op == MDB_SET_KEY) {
+ key->mv_size = mc->mc_db->md_pad;
+ key->mv_data = LEAF2KEY(mp, mc->mc_ki[mc->mc_top], key->mv_size);
+ }
return MDB_SUCCESS;
}
if (op == MDB_GET_BOTH || rc > 0)
return MDB_NOTFOUND;
rc = 0;
+ *data = d2;
}
} else {
if (rc != MDB_SUCCESS)
return rc;
}
- assert(IS_LEAF(mc->mc_pg[mc->mc_top]));
+ mdb_cassert(mc, IS_LEAF(mc->mc_pg[mc->mc_top]));
leaf = NODEPTR(mc->mc_pg[mc->mc_top], 0);
mc->mc_flags |= C_INITIALIZED;
if (rc != MDB_SUCCESS)
return rc;
}
- assert(IS_LEAF(mc->mc_pg[mc->mc_top]));
+ mdb_cassert(mc, IS_LEAF(mc->mc_pg[mc->mc_top]));
}
mc->mc_ki[mc->mc_top] = NUMKEYS(mc->mc_pg[mc->mc_top]) - 1;
int exact = 0;
int (*mfunc)(MDB_cursor *mc, MDB_val *key, MDB_val *data);
- assert(mc);
+ if (mc == NULL)
+ return EINVAL;
if (mc->mc_txn->mt_flags & MDB_TXN_ERROR)
return MDB_BAD_TXN;
case MDB_SET_RANGE:
if (key == NULL) {
rc = EINVAL;
- } else if (key->mv_size > MDB_MAXKEYSIZE) {
- rc = MDB_BAD_VALSIZE;
- } else if (op == MDB_SET_RANGE)
- rc = mdb_cursor_set(mc, key, data, op, NULL);
- else
- rc = mdb_cursor_set(mc, key, data, op, &exact);
+ } else {
+ rc = mdb_cursor_set(mc, key, data, op,
+ op == MDB_SET_RANGE ? NULL : &exact);
+ }
break;
case MDB_GET_MULTIPLE:
if (data == NULL || !(mc->mc_flags & C_INITIALIZED)) {
unsigned int flags)
{
enum { MDB_NO_ROOT = MDB_LAST_ERRCODE+10 }; /* internal code */
+ MDB_env *env;
MDB_node *leaf = NULL;
- MDB_val xdata, *rdata, dkey;
- MDB_page *fp;
+ MDB_page *fp, *mp;
+ uint16_t fp_flags;
+ MDB_val xdata, *rdata, dkey, olddata;
MDB_db dummy;
- int do_sub = 0, insert = 0;
+ int do_sub = 0, insert_key, insert_data;
unsigned int mcount = 0, dcount = 0, nospill;
size_t nsize;
int rc, rc2;
- MDB_pagebuf pbuf;
- char dbuf[MDB_MAXKEYSIZE+1];
unsigned int nflags;
DKBUF;
+ if (mc == NULL || key == NULL)
+ return EINVAL;
+
+ env = mc->mc_txn->mt_env;
+
/* Check this first so counter will always be zero on any
* early failures.
*/
if (mc->mc_txn->mt_flags & (MDB_TXN_RDONLY|MDB_TXN_ERROR))
return (mc->mc_txn->mt_flags & MDB_TXN_RDONLY) ? EACCES : MDB_BAD_TXN;
- if (flags != MDB_CURRENT && (key->mv_size == 0 || key->mv_size > MDB_MAXKEYSIZE))
- return MDB_BAD_VALSIZE;
-
- if (F_ISSET(mc->mc_db->md_flags, MDB_DUPSORT) && data->mv_size > MDB_MAXKEYSIZE)
+ if (key->mv_size-1 >= ENV_MAXKEY(env))
return MDB_BAD_VALSIZE;
#if SIZE_MAX > MAXDATASIZE
- if (data->mv_size > MAXDATASIZE)
+ if (data->mv_size > ((mc->mc_db->md_flags & MDB_DUPSORT) ? ENV_MAXKEY(env) : MAXDATASIZE))
+ return MDB_BAD_VALSIZE;
+#else
+ if ((mc->mc_db->md_flags & MDB_DUPSORT) && data->mv_size > ENV_MAXKEY(env))
return MDB_BAD_VALSIZE;
#endif
return rc2;
}
- /* The key already exists */
- if (rc == MDB_SUCCESS) {
+ insert_key = insert_data = rc;
+ if (insert_key) {
+ /* The key does not exist */
+ DPRINTF(("inserting key at index %i", mc->mc_ki[mc->mc_top]));
+ if ((mc->mc_db->md_flags & MDB_DUPSORT) &&
+ LEAFSIZE(key, data) > env->me_nodemax)
+ {
+ /* Too big for a node, insert in sub-DB */
+ fp_flags = P_LEAF|P_DIRTY;
+ fp = env->me_pbuf;
+ fp->mp_pad = data->mv_size; /* used if MDB_DUPFIXED */
+ fp->mp_lower = fp->mp_upper = olddata.mv_size = PAGEHDRSZ;
+ goto prep_subDB;
+ }
+ } else {
/* there's only a key anyway, so this is a no-op */
if (IS_LEAF2(mc->mc_pg[mc->mc_top])) {
+ char *ptr;
unsigned int ksize = mc->mc_db->md_pad;
if (key->mv_size != ksize)
return MDB_BAD_VALSIZE;
- if (flags == MDB_CURRENT) {
- char *ptr = LEAF2KEY(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top], ksize);
- memcpy(ptr, key->mv_data, ksize);
- }
+ ptr = LEAF2KEY(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top], ksize);
+ memcpy(ptr, key->mv_data, ksize);
return MDB_SUCCESS;
}
+more:
leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]);
+ olddata.mv_size = NODEDSZ(leaf);
+ olddata.mv_data = NODEDATA(leaf);
/* DB has dups? */
if (F_ISSET(mc->mc_db->md_flags, MDB_DUPSORT)) {
+ /* Prepare (sub-)page/sub-DB to accept the new item,
+ * if needed. fp: old sub-page or a header faking
+ * it. mp: new (sub-)page. offset: growth in page
+ * size. xdata: node data with new page or DB.
+ */
+ unsigned i, offset = 0;
+ mp = fp = xdata.mv_data = env->me_pbuf;
+ mp->mp_pgno = mc->mc_pg[mc->mc_top]->mp_pgno;
+
/* Was a single item before, must convert now */
-more:
if (!F_ISSET(leaf->mn_flags, F_DUPDATA)) {
/* Just overwrite the current item */
if (flags == MDB_CURRENT)
goto current;
- dkey.mv_size = NODEDSZ(leaf);
- dkey.mv_data = NODEDATA(leaf);
#if UINT_MAX < SIZE_MAX
- if (mc->mc_dbx->md_dcmp == mdb_cmp_int && dkey.mv_size == sizeof(size_t))
-#ifdef MISALIGNED_OK
- mc->mc_dbx->md_dcmp = mdb_cmp_long;
-#else
- mc->mc_dbx->md_dcmp = mdb_cmp_cint;
+ if (mc->mc_dbx->md_dcmp == mdb_cmp_int && olddata.mv_size == sizeof(size_t))
+ mc->mc_dbx->md_dcmp = mdb_cmp_clong;
#endif
-#endif
- /* if data matches, skip it */
- if (!mc->mc_dbx->md_dcmp(data, &dkey)) {
+ /* does data match? */
+ if (!mc->mc_dbx->md_dcmp(data, &olddata)) {
if (flags & MDB_NODUPDATA)
- rc = MDB_KEYEXIST;
- else if (flags & MDB_MULTIPLE)
- goto next_mult;
- else
- rc = MDB_SUCCESS;
- return rc;
+ return MDB_KEYEXIST;
+ /* overwrite it */
+ goto current;
}
- /* create a fake page for the dup items */
- memcpy(dbuf, dkey.mv_data, dkey.mv_size);
- dkey.mv_data = dbuf;
- fp = (MDB_page *)&pbuf;
- fp->mp_pgno = mc->mc_pg[mc->mc_top]->mp_pgno;
+ /* Back up original data item */
+ dkey.mv_size = olddata.mv_size;
+ dkey.mv_data = memcpy(fp+1, olddata.mv_data, olddata.mv_size);
+
+ /* Make sub-page header for the dup items, with dummy body */
fp->mp_flags = P_LEAF|P_DIRTY|P_SUBP;
fp->mp_lower = PAGEHDRSZ;
- fp->mp_upper = PAGEHDRSZ + dkey.mv_size + data->mv_size;
+ xdata.mv_size = PAGEHDRSZ + dkey.mv_size + data->mv_size;
if (mc->mc_db->md_flags & MDB_DUPFIXED) {
fp->mp_flags |= P_LEAF2;
fp->mp_pad = data->mv_size;
- fp->mp_upper += 2 * data->mv_size; /* leave space for 2 more */
+ xdata.mv_size += 2 * data->mv_size; /* leave space for 2 more */
} else {
- fp->mp_upper += 2 * sizeof(indx_t) + 2 * NODESIZE +
+ xdata.mv_size += 2 * (sizeof(indx_t) + NODESIZE) +
(dkey.mv_size & 1) + (data->mv_size & 1);
}
- mdb_node_del(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top], 0);
- do_sub = 1;
- rdata = &xdata;
- xdata.mv_size = fp->mp_upper;
- xdata.mv_data = fp;
- flags |= F_DUPDATA;
- goto new_sub;
- }
- if (!F_ISSET(leaf->mn_flags, F_SUBDATA)) {
- /* See if we need to convert from fake page to subDB */
- MDB_page *mp;
- unsigned int offset;
- unsigned int i;
- uint16_t fp_flags;
-
- fp = NODEDATA(leaf);
- if (flags == MDB_CURRENT) {
-reuse:
+ fp->mp_upper = xdata.mv_size;
+ olddata.mv_size = fp->mp_upper; /* pretend olddata is fp */
+ } else if (leaf->mn_flags & F_SUBDATA) {
+ /* Data is on sub-DB, just store it */
+ flags |= F_DUPDATA|F_SUBDATA;
+ goto put_sub;
+ } else {
+ /* Data is on sub-page */
+ fp = olddata.mv_data;
+ switch (flags) {
+ default:
+ if (!(mc->mc_db->md_flags & MDB_DUPFIXED)) {
+ offset = EVEN(NODESIZE + sizeof(indx_t) +
+ data->mv_size);
+ break;
+ }
+ offset = fp->mp_pad;
+ if (SIZELEFT(fp) < offset) {
+ offset *= 4; /* space for 4 more */
+ break;
+ }
+ /* FALLTHRU: Big enough MDB_DUPFIXED sub-page */
+ case MDB_CURRENT:
fp->mp_flags |= P_DIRTY;
- COPY_PGNO(fp->mp_pgno, mc->mc_pg[mc->mc_top]->mp_pgno);
+ COPY_PGNO(fp->mp_pgno, mp->mp_pgno);
mc->mc_xcursor->mx_cursor.mc_pg[0] = fp;
flags |= F_DUPDATA;
goto put_sub;
}
- if (mc->mc_db->md_flags & MDB_DUPFIXED) {
- offset = fp->mp_pad;
- if (SIZELEFT(fp) >= offset)
- goto reuse;
- offset *= 4; /* space for 4 more */
- } else {
- offset = NODESIZE + sizeof(indx_t) + data->mv_size;
- }
- offset += offset & 1;
- fp_flags = fp->mp_flags;
- if (NODESIZE + sizeof(indx_t) + NODEKSZ(leaf) + NODEDSZ(leaf) +
- offset >= mc->mc_txn->mt_env->me_nodemax) {
- /* yes, convert it */
- dummy.md_flags = 0;
+ xdata.mv_size = olddata.mv_size + offset;
+ }
+
+ fp_flags = fp->mp_flags;
+ if (NODESIZE + NODEKSZ(leaf) + xdata.mv_size > env->me_nodemax) {
+ /* Too big for a sub-page, convert to sub-DB */
+ fp_flags &= ~P_SUBP;
+prep_subDB:
if (mc->mc_db->md_flags & MDB_DUPFIXED) {
+ fp_flags |= P_LEAF2;
dummy.md_pad = fp->mp_pad;
dummy.md_flags = MDB_DUPFIXED;
if (mc->mc_db->md_flags & MDB_INTEGERDUP)
dummy.md_flags |= MDB_INTEGERKEY;
+ } else {
+ dummy.md_pad = 0;
+ dummy.md_flags = 0;
}
dummy.md_depth = 1;
dummy.md_branch_pages = 0;
dummy.md_leaf_pages = 1;
dummy.md_overflow_pages = 0;
dummy.md_entries = NUMKEYS(fp);
- rdata = &xdata;
xdata.mv_size = sizeof(MDB_db);
xdata.mv_data = &dummy;
if ((rc = mdb_page_alloc(mc, 1, &mp)))
return rc;
- offset = mc->mc_txn->mt_env->me_psize - NODEDSZ(leaf);
+ offset = env->me_psize - olddata.mv_size;
flags |= F_DUPDATA|F_SUBDATA;
dummy.md_root = mp->mp_pgno;
- fp_flags &= ~P_SUBP;
- } else {
- /* no, just grow it */
- rdata = &xdata;
- xdata.mv_size = NODEDSZ(leaf) + offset;
- xdata.mv_data = &pbuf;
- mp = (MDB_page *)&pbuf;
- mp->mp_pgno = mc->mc_pg[mc->mc_top]->mp_pgno;
- flags |= F_DUPDATA;
- }
+ }
+ if (mp != fp) {
mp->mp_flags = fp_flags | P_DIRTY;
mp->mp_pad = fp->mp_pad;
mp->mp_lower = fp->mp_lower;
mp->mp_upper = fp->mp_upper + offset;
- if (IS_LEAF2(fp)) {
+ if (fp_flags & P_LEAF2) {
memcpy(METADATA(mp), METADATA(fp), NUMKEYS(fp) * fp->mp_pad);
} else {
- nsize = NODEDSZ(leaf) - fp->mp_upper;
- memcpy((char *)mp + mp->mp_upper, (char *)fp + fp->mp_upper, nsize);
+ memcpy((char *)mp + mp->mp_upper, (char *)fp + fp->mp_upper,
+ olddata.mv_size - fp->mp_upper);
for (i=0; i<NUMKEYS(fp); i++)
mp->mp_ptrs[i] = fp->mp_ptrs[i] + offset;
}
- mdb_node_del(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top], 0);
- do_sub = 1;
- goto new_sub;
}
- /* data is on sub-DB, just store it */
- flags |= F_DUPDATA|F_SUBDATA;
- goto put_sub;
+
+ rdata = &xdata;
+ flags |= F_DUPDATA;
+ do_sub = 1;
+ if (!insert_key)
+ mdb_node_del(mc, 0);
+ goto new_sub;
}
current:
/* overflow page overwrites need special handling */
if (F_ISSET(leaf->mn_flags, F_BIGDATA)) {
MDB_page *omp;
pgno_t pg;
- unsigned psize = mc->mc_txn->mt_env->me_psize;
- int level, ovpages, dpages = OVPAGES(data->mv_size, psize);
+ int level, ovpages, dpages = OVPAGES(data->mv_size, env->me_psize);
- memcpy(&pg, NODEDATA(leaf), sizeof(pg));
+ memcpy(&pg, olddata.mv_data, sizeof(pg));
if ((rc2 = mdb_page_get(mc->mc_txn, pg, &omp, &level)) != 0)
return rc2;
ovpages = omp->mp_pages;
/* Is the ov page large enough? */
if (ovpages >= dpages) {
if (!(omp->mp_flags & P_DIRTY) &&
- (level || (mc->mc_txn->mt_env->me_flags & MDB_WRITEMAP)))
+ (level || (env->me_flags & MDB_WRITEMAP)))
{
rc = mdb_page_unspill(mc->mc_txn, omp, &omp);
if (rc)
*/
if (level > 1) {
/* It is writable only in a parent txn */
- size_t sz = (size_t) psize * ovpages, off;
+ size_t sz = (size_t) env->me_psize * ovpages, off;
MDB_page *np = mdb_page_malloc(mc->mc_txn, ovpages);
MDB_ID2 id2;
if (!np)
return ENOMEM;
id2.mid = pg;
id2.mptr = np;
- mdb_mid2l_insert(mc->mc_txn->mt_u.dirty_list, &id2);
+ rc2 = mdb_mid2l_insert(mc->mc_txn->mt_u.dirty_list, &id2);
+ mdb_cassert(mc, rc2 == 0);
if (!(flags & MDB_RESERVE)) {
/* Copy end of page, adjusting alignment so
* compiler may copy words instead of bytes.
data->mv_data = METADATA(omp);
else
memcpy(METADATA(omp), data->mv_data, data->mv_size);
- goto done;
+ return MDB_SUCCESS;
}
}
if ((rc2 = mdb_ovpage_free(mc, omp)) != MDB_SUCCESS)
return rc2;
- } else if (NODEDSZ(leaf) == data->mv_size) {
+ } else if (data->mv_size == olddata.mv_size) {
/* same size, just replace it. Note that we could
* also reuse this node if the new data is smaller,
* but instead we opt to shrink the node in that case.
*/
if (F_ISSET(flags, MDB_RESERVE))
- data->mv_data = NODEDATA(leaf);
- else if (data->mv_size)
- memcpy(NODEDATA(leaf), data->mv_data, data->mv_size);
+ data->mv_data = olddata.mv_data;
+ else if (!(mc->mc_flags & C_SUB))
+ memcpy(olddata.mv_data, data->mv_data, data->mv_size);
else
memcpy(NODEKEY(leaf), key->mv_data, key->mv_size);
- goto done;
+ return MDB_SUCCESS;
}
- mdb_node_del(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top], 0);
- mc->mc_db->md_entries--;
- } else {
- DPRINTF(("inserting key at index %i", mc->mc_ki[mc->mc_top]));
- insert = 1;
+ mdb_node_del(mc, 0);
}
rdata = data;
new_sub:
nflags = flags & NODE_ADD_FLAGS;
- nsize = IS_LEAF2(mc->mc_pg[mc->mc_top]) ? key->mv_size : mdb_leaf_size(mc->mc_txn->mt_env, key, rdata);
+ nsize = IS_LEAF2(mc->mc_pg[mc->mc_top]) ? key->mv_size : mdb_leaf_size(env, key, rdata);
if (SIZELEFT(mc->mc_pg[mc->mc_top]) < nsize) {
if (( flags & (F_DUPDATA|F_SUBDATA)) == F_DUPDATA )
- nflags &= ~MDB_APPEND;
- if (!insert)
+ nflags &= ~MDB_APPEND; /* sub-page may need room to grow */
+ if (!insert_key)
nflags |= MDB_SPLIT_REPLACE;
rc = mdb_page_split(mc, key, rdata, P_INVALID, nflags);
} else {
/* There is room already in this leaf page. */
rc = mdb_node_add(mc, mc->mc_ki[mc->mc_top], key, rdata, 0, nflags);
- if (rc == 0 && !do_sub && insert) {
+ if (rc == 0 && insert_key) {
/* Adjust other cursors pointing to mp */
MDB_cursor *m2, *m3;
MDB_dbi dbi = mc->mc_dbi;
}
}
- if (rc != MDB_SUCCESS)
- mc->mc_txn->mt_flags |= MDB_TXN_ERROR;
- else {
+ if (rc == MDB_SUCCESS) {
/* Now store the actual data in the child DB. Note that we're
* storing the user data in the keys field, so there are strict
* size limits on dupdata. The actual data fields of the child
*/
if (do_sub) {
int xflags;
+ size_t ecount;
put_sub:
xdata.mv_size = 0;
xdata.mv_data = "";
if (dkey.mv_size) {
rc = mdb_cursor_put(&mc->mc_xcursor->mx_cursor, &dkey, &xdata, xflags);
if (rc)
- return rc;
+ goto bad_sub;
{
/* Adjust other cursors pointing to mp */
MDB_cursor *m2;
/* we've done our job */
dkey.mv_size = 0;
}
+ ecount = mc->mc_xcursor->mx_db.md_entries;
if (flags & MDB_APPENDDUP)
xflags |= MDB_APPEND;
rc = mdb_cursor_put(&mc->mc_xcursor->mx_cursor, data, &xdata, xflags);
void *db = NODEDATA(leaf);
memcpy(db, &mc->mc_xcursor->mx_db, sizeof(MDB_db));
}
+ insert_data = mc->mc_xcursor->mx_db.md_entries - ecount;
}
- /* sub-writes might have failed so check rc again.
- * Don't increment count if we just replaced an existing item.
- */
- if (!rc && !(flags & MDB_CURRENT))
+ /* Increment count unless we just replaced an existing item. */
+ if (insert_data)
mc->mc_db->md_entries++;
+ if (insert_key) {
+ /* Invalidate txn if we created an empty sub-DB */
+ if (rc)
+ goto bad_sub;
+ /* If we succeeded and the key didn't exist before,
+ * make sure the cursor is marked valid.
+ */
+ mc->mc_flags |= C_INITIALIZED;
+ }
if (flags & MDB_MULTIPLE) {
if (!rc) {
-next_mult:
mcount++;
/* let caller know how many succeeded, if any */
data[1].mv_size = mcount;
if (mcount < dcount) {
data[0].mv_data = (char *)data[0].mv_data + data[0].mv_size;
- leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]);
+ insert_key = insert_data = 0;
goto more;
}
}
}
+ return rc;
+bad_sub:
+ if (rc == MDB_KEYEXIST) /* should not happen, we deleted that item */
+ rc = MDB_CORRUPTED;
}
-done:
- /* If we succeeded and the key didn't exist before, make sure
- * the cursor is marked valid.
- */
- if (!rc && insert)
- mc->mc_flags |= C_INITIALIZED;
+ mc->mc_txn->mt_flags |= MDB_TXN_ERROR;
return rc;
}
return rc;
mp = mc->mc_pg[mc->mc_top];
+ if (IS_LEAF2(mp))
+ goto del_key;
leaf = NODEPTR(mp, mc->mc_ki[mc->mc_top]);
- if (!IS_LEAF2(mp) && F_ISSET(leaf->mn_flags, F_DUPDATA)) {
- if (!(flags & MDB_NODUPDATA)) {
+ if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
+ if (flags & MDB_NODUPDATA) {
+ /* mdb_cursor_del0() will subtract the final entry */
+ mc->mc_db->md_entries -= mc->mc_xcursor->mx_db.md_entries - 1;
+ } else {
if (!F_ISSET(leaf->mn_flags, F_SUBDATA)) {
mc->mc_xcursor->mx_cursor.mc_pg[0] = NODEDATA(leaf);
}
rc = mdb_cursor_del(&mc->mc_xcursor->mx_cursor, MDB_NOSPILL);
+ if (rc)
+ return rc;
/* If sub-DB still has entries, we're done */
if (mc->mc_xcursor->mx_db.md_entries) {
if (leaf->mn_flags & F_SUBDATA) {
if (leaf->mn_flags & F_SUBDATA) {
/* add all the child DB's pages to the free list */
rc = mdb_drop0(&mc->mc_xcursor->mx_cursor, 0);
- if (rc == MDB_SUCCESS) {
- mc->mc_db->md_entries -=
- mc->mc_xcursor->mx_db.md_entries;
- }
+ if (rc)
+ goto fail;
}
}
- return mdb_cursor_del0(mc, leaf);
+ /* add overflow pages to free list */
+ if (F_ISSET(leaf->mn_flags, F_BIGDATA)) {
+ MDB_page *omp;
+ pgno_t pg;
+
+ memcpy(&pg, NODEDATA(leaf), sizeof(pg));
+ if ((rc = mdb_page_get(mc->mc_txn, pg, &omp, NULL)) ||
+ (rc = mdb_ovpage_free(mc, omp)))
+ goto fail;
+ }
+
+del_key:
+ return mdb_cursor_del0(mc);
+
+fail:
+ mc->mc_txn->mt_flags |= MDB_TXN_ERROR;
+ return rc;
}
/** Allocate and initialize new pages for a database.
size_t sz;
sz = LEAFSIZE(key, data);
- if (sz >= env->me_nodemax) {
+ if (sz > env->me_nodemax) {
/* put on overflow page */
sz -= data->mv_size - sizeof(pgno_t);
}
- sz += sz & 1;
- return sz + sizeof(indx_t);
+ return EVEN(sz + sizeof(indx_t));
}
/** Calculate the size of a branch node.
size_t sz;
sz = INDXSIZE(key);
- if (sz >= env->me_nodemax) {
+ if (sz > env->me_nodemax) {
/* put on overflow page */
/* not implemented */
/* sz -= key->size - sizeof(pgno_t); */
MDB_page *ofp = NULL; /* overflow page */
DKBUF;
- assert(mp->mp_upper >= mp->mp_lower);
+ mdb_cassert(mc, mp->mp_upper >= mp->mp_lower);
DPRINTF(("add to %s %spage %"Z"u index %i, data size %"Z"u key size %"Z"u [%s]",
IS_LEAF(mp) ? "leaf" : "branch",
IS_SUBP(mp) ? "sub-" : "",
- mp->mp_pgno, indx, data ? data->mv_size : 0,
+ mdb_dbg_pgno(mp), indx, data ? data->mv_size : 0,
key ? key->mv_size : 0, key ? DKEY(key) : "null"));
if (IS_LEAF2(mp)) {
if (key != NULL)
node_size += key->mv_size;
if (IS_LEAF(mp)) {
- assert(data);
+ mdb_cassert(mc, data);
if (F_ISSET(flags, F_BIGDATA)) {
/* Data already on overflow page. */
node_size += sizeof(pgno_t);
- } else if (node_size + data->mv_size >= mc->mc_txn->mt_env->me_nodemax) {
+ } else if (node_size + data->mv_size > mc->mc_txn->mt_env->me_nodemax) {
int ovpages = OVPAGES(data->mv_size, mc->mc_txn->mt_env->me_psize);
int rc;
/* Put data on overflow page. */
DPRINTF(("data size is %"Z"u, node would be %"Z"u, put data on overflow page",
data->mv_size, node_size+data->mv_size));
- node_size += sizeof(pgno_t) + (node_size & 1);
+ node_size = EVEN(node_size + sizeof(pgno_t));
if ((ssize_t)node_size > room)
goto full;
if ((rc = mdb_page_new(mc, P_OVERFLOW, ovpages, &ofp)))
node_size += data->mv_size;
}
}
- node_size += node_size & 1;
+ node_size = EVEN(node_size);
if ((ssize_t)node_size > room)
goto full;
/* Adjust free space offsets. */
ofs = mp->mp_upper - node_size;
- assert(ofs >= mp->mp_lower + sizeof(indx_t));
+ mdb_cassert(mc, ofs >= mp->mp_lower + sizeof(indx_t));
mp->mp_ptrs[indx] = ofs;
mp->mp_upper = ofs;
mp->mp_lower += sizeof(indx_t);
memcpy(NODEKEY(node), key->mv_data, key->mv_size);
if (IS_LEAF(mp)) {
- assert(key);
+ mdb_cassert(mc, key);
if (ofp == NULL) {
if (F_ISSET(flags, F_BIGDATA))
memcpy(node->mn_data + key->mv_size, data->mv_data,
full:
DPRINTF(("not enough room in page %"Z"u, got %u ptrs",
- mp->mp_pgno, NUMKEYS(mp)));
+ mdb_dbg_pgno(mp), NUMKEYS(mp)));
DPRINTF(("upper-lower = %u - %u = %"Z"d", mp->mp_upper,mp->mp_lower,room));
DPRINTF(("node size = %"Z"u", node_size));
+ mc->mc_txn->mt_flags |= MDB_TXN_ERROR;
return MDB_PAGE_FULL;
}
/** Delete the specified node from a page.
- * @param[in] mp The page to operate on.
- * @param[in] indx The index of the node to delete.
+ * @param[in] mc Cursor pointing to the node to delete.
* @param[in] ksize The size of a node. Only used if the page is
* part of a #MDB_DUPFIXED database.
*/
static void
-mdb_node_del(MDB_page *mp, indx_t indx, int ksize)
+mdb_node_del(MDB_cursor *mc, int ksize)
{
+ MDB_page *mp = mc->mc_pg[mc->mc_top];
+ indx_t indx = mc->mc_ki[mc->mc_top];
unsigned int sz;
indx_t i, j, numkeys, ptr;
MDB_node *node;
char *base;
-#if MDB_DEBUG
- {
- pgno_t pgno;
- COPY_PGNO(pgno, mp->mp_pgno);
DPRINTF(("delete node %u on %s page %"Z"u", indx,
- IS_LEAF(mp) ? "leaf" : "branch", pgno));
- }
-#endif
- assert(indx < NUMKEYS(mp));
+ IS_LEAF(mp) ? "leaf" : "branch", mdb_dbg_pgno(mp)));
+ numkeys = NUMKEYS(mp);
+ mdb_cassert(mc, indx < numkeys);
if (IS_LEAF2(mp)) {
- int x = NUMKEYS(mp) - 1 - indx;
+ int x = numkeys - 1 - indx;
base = LEAF2KEY(mp, indx, ksize);
if (x)
memmove(base, base + ksize, x * ksize);
else
sz += NODEDSZ(node);
}
- sz += sz & 1;
+ sz = EVEN(sz);
ptr = mp->mp_ptrs[indx];
- numkeys = NUMKEYS(mp);
for (i = j = 0; i < numkeys; i++) {
if (i != indx) {
mp->mp_ptrs[j] = mp->mp_ptrs[i];
MDB_node *node;
MDB_page *sp, *xp;
char *base;
- int osize, nsize;
- int delta;
+ int nsize, delta;
indx_t i, numkeys, ptr;
node = NODEPTR(mp, indx);
sp = (MDB_page *)NODEDATA(node);
- osize = NODEDSZ(node);
-
- delta = sp->mp_upper - sp->mp_lower;
- SETDSZ(node, osize - delta);
+ delta = SIZELEFT(sp);
xp = (MDB_page *)((char *)sp + delta);
/* shift subpage upward */
if (IS_LEAF2(sp)) {
nsize = NUMKEYS(sp) * sp->mp_pad;
+ if (nsize & 1)
+ return; /* do not make the node uneven-sized */
memmove(METADATA(xp), METADATA(sp), nsize);
} else {
int i;
- nsize = osize - sp->mp_upper;
numkeys = NUMKEYS(sp);
for (i=numkeys-1; i>=0; i--)
xp->mp_ptrs[i] = sp->mp_ptrs[i] - delta;
xp->mp_pad = sp->mp_pad;
COPY_PGNO(xp->mp_pgno, mp->mp_pgno);
+ nsize = NODEDSZ(node) - delta;
+ SETDSZ(node, nsize);
+
/* shift lower nodes upward */
ptr = mp->mp_ptrs[indx];
numkeys = NUMKEYS(mp);
mx->mx_dbflag = DB_VALID|DB_DIRTY; /* DB_DIRTY guides mdb_cursor_touch */
#if UINT_MAX < SIZE_MAX
if (mx->mx_dbx.md_cmp == mdb_cmp_int && mx->mx_db.md_pad == sizeof(size_t))
-#ifdef MISALIGNED_OK
- mx->mx_dbx.md_cmp = mdb_cmp_long;
-#else
- mx->mx_dbx.md_cmp = mdb_cmp_cint;
-#endif
+ mx->mx_dbx.md_cmp = mdb_cmp_clong;
#endif
}
mc->mc_pg[0] = 0;
mc->mc_flags = 0;
if (txn->mt_dbs[dbi].md_flags & MDB_DUPSORT) {
- assert(mx != NULL);
+ mdb_tassert(txn, mx != NULL);
mc->mc_xcursor = mx;
mdb_xcursor_init0(mc);
} else {
MDB_cursor *mc;
size_t size = sizeof(MDB_cursor);
- if (txn == NULL || ret == NULL || dbi >= txn->mt_numdbs || !(txn->mt_dbflags[dbi] & DB_VALID))
+ if (!ret || !TXN_DBI_EXIST(txn, dbi))
return EINVAL;
if (txn->mt_flags & MDB_TXN_ERROR)
int
mdb_cursor_renew(MDB_txn *txn, MDB_cursor *mc)
{
- if (txn == NULL || mc == NULL || mc->mc_dbi >= txn->mt_numdbs)
+ if (!mc || !TXN_DBI_EXIST(txn, mc->mc_dbi))
return EINVAL;
if ((mc->mc_flags & C_UNTRACK) || txn->mt_cursors)
return EINVAL;
+ if (txn->mt_flags & MDB_TXN_ERROR)
+ return MDB_BAD_TXN;
+
mdb_cursor_init(mc, txn, mc->mc_dbi, mc->mc_xcursor);
return MDB_SUCCESS;
}
if (mc->mc_xcursor == NULL)
return MDB_INCOMPATIBLE;
+ if (mc->mc_txn->mt_flags & MDB_TXN_ERROR)
+ return MDB_BAD_TXN;
+
leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]);
if (!F_ISSET(leaf->mn_flags, F_DUPDATA)) {
*countp = 1;
MDB_dbi
mdb_cursor_dbi(MDB_cursor *mc)
{
- assert(mc != NULL);
return mc->mc_dbi;
}
-/** Replace the key for a node with a new key.
+/** Replace the key for a branch node with a new key.
* @param[in] mc Cursor pointing to the node to operate on.
* @param[in] key The new key to use.
* @return 0 on success, non-zero on failure.
MDB_node *node;
char *base;
size_t len;
- int delta, delta0;
+ int delta, ksize, oksize;
indx_t ptr, i, numkeys, indx;
DKBUF;
#if MDB_DEBUG
{
MDB_val k2;
- char kbuf2[(MDB_MAXKEYSIZE*2+1)];
+ char kbuf2[DKBUF_MAXKEYSIZE*2+1];
k2.mv_data = NODEKEY(node);
k2.mv_size = node->mn_ksize;
DPRINTF(("update key %u (ofs %u) [%s] to [%s] on page %"Z"u",
}
#endif
- delta0 = delta = key->mv_size - node->mn_ksize;
+ /* Sizes must be 2-byte aligned. */
+ ksize = EVEN(key->mv_size);
+ oksize = EVEN(node->mn_ksize);
+ delta = ksize - oksize;
- /* Must be 2-byte aligned. If new key is
- * shorter by 1, the shift will be skipped.
- */
- delta += (delta & 1);
+ /* Shift node contents if EVEN(key length) changed. */
if (delta) {
if (delta > 0 && SIZELEFT(mp) < delta) {
pgno_t pgno;
/* not enough space left, do a delete and split */
DPRINTF(("Not enough room, delta = %d, splitting...", delta));
pgno = NODEPGNO(node);
- mdb_node_del(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top], 0);
+ mdb_node_del(mc, 0);
return mdb_page_split(mc, key, NULL, pgno, MDB_SPLIT_REPLACE);
}
}
/* But even if no shift was needed, update ksize */
- if (delta0)
+ if (node->mn_ksize != key->mv_size)
node->mn_ksize = key->mv_size;
if (key->mv_size)
return rc;
if (IS_LEAF2(csrc->mc_pg[csrc->mc_top])) {
- srcnode = NODEPTR(csrc->mc_pg[csrc->mc_top], 0); /* fake */
key.mv_size = csrc->mc_db->md_pad;
key.mv_data = LEAF2KEY(csrc->mc_pg[csrc->mc_top], csrc->mc_ki[csrc->mc_top], key.mv_size);
data.mv_size = 0;
flags = 0;
} else {
srcnode = NODEPTR(csrc->mc_pg[csrc->mc_top], csrc->mc_ki[csrc->mc_top]);
- assert(!((long)srcnode&1));
+ mdb_cassert(csrc, !((size_t)srcnode & 1));
srcpg = NODEPGNO(srcnode);
flags = srcnode->mn_flags;
if (csrc->mc_ki[csrc->mc_top] == 0 && IS_BRANCH(csrc->mc_pg[csrc->mc_top])) {
unsigned int snum = csrc->mc_snum;
MDB_node *s2;
/* must find the lowest key below src */
- mdb_page_search_lowest(csrc);
+ rc = mdb_page_search_lowest(csrc);
+ if (rc)
+ return rc;
if (IS_LEAF2(csrc->mc_pg[csrc->mc_top])) {
key.mv_size = csrc->mc_db->md_pad;
key.mv_data = LEAF2KEY(csrc->mc_pg[csrc->mc_top], 0, key.mv_size);
MDB_node *s2;
MDB_val bkey;
/* must find the lowest key below dst */
- mdb_page_search_lowest(cdst);
- if (IS_LEAF2(cdst->mc_pg[cdst->mc_top])) {
- bkey.mv_size = cdst->mc_db->md_pad;
- bkey.mv_data = LEAF2KEY(cdst->mc_pg[cdst->mc_top], 0, bkey.mv_size);
+ mdb_cursor_copy(cdst, &mn);
+ rc = mdb_page_search_lowest(&mn);
+ if (rc)
+ return rc;
+ if (IS_LEAF2(mn.mc_pg[mn.mc_top])) {
+ bkey.mv_size = mn.mc_db->md_pad;
+ bkey.mv_data = LEAF2KEY(mn.mc_pg[mn.mc_top], 0, bkey.mv_size);
} else {
- s2 = NODEPTR(cdst->mc_pg[cdst->mc_top], 0);
+ s2 = NODEPTR(mn.mc_pg[mn.mc_top], 0);
bkey.mv_size = NODEKSZ(s2);
bkey.mv_data = NODEKEY(s2);
}
- cdst->mc_snum = snum--;
- cdst->mc_top = snum;
- mdb_cursor_copy(cdst, &mn);
+ mn.mc_snum = snum--;
+ mn.mc_top = snum;
mn.mc_ki[snum] = 0;
rc = mdb_update_key(&mn, &bkey);
if (rc)
/* Delete the node from the source page.
*/
- mdb_node_del(csrc->mc_pg[csrc->mc_top], csrc->mc_ki[csrc->mc_top], key.mv_size);
+ mdb_node_del(csrc, key.mv_size);
{
/* Adjust other cursors pointing to mp */
csrc->mc_ki[csrc->mc_top] = 0;
rc = mdb_update_key(csrc, &nullkey);
csrc->mc_ki[csrc->mc_top] = ix;
- assert(rc == MDB_SUCCESS);
+ mdb_cassert(csrc, rc == MDB_SUCCESS);
}
}
cdst->mc_ki[cdst->mc_top] = 0;
rc = mdb_update_key(cdst, &nullkey);
cdst->mc_ki[cdst->mc_top] = ix;
- assert(rc == MDB_SUCCESS);
+ mdb_cassert(csrc, rc == MDB_SUCCESS);
}
}
* the \b csrc page will be freed.
* @param[in] csrc Cursor pointing to the source page.
* @param[in] cdst Cursor pointing to the destination page.
+ * @return 0 on success, non-zero on failure.
*/
static int
mdb_page_merge(MDB_cursor *csrc, MDB_cursor *cdst)
{
- int rc;
- indx_t i, j;
- MDB_node *srcnode;
+ MDB_page *psrc, *pdst;
+ MDB_node *srcnode;
MDB_val key, data;
- unsigned nkeys;
+ unsigned nkeys;
+ int rc;
+ indx_t i, j;
+
+ psrc = csrc->mc_pg[csrc->mc_top];
+ pdst = cdst->mc_pg[cdst->mc_top];
- DPRINTF(("merging page %"Z"u into %"Z"u", csrc->mc_pg[csrc->mc_top]->mp_pgno,
- cdst->mc_pg[cdst->mc_top]->mp_pgno));
+ DPRINTF(("merging page %"Z"u into %"Z"u", psrc->mp_pgno, pdst->mp_pgno));
- assert(csrc->mc_snum > 1); /* can't merge root page */
- assert(cdst->mc_snum > 1);
+ mdb_cassert(csrc, csrc->mc_snum > 1); /* can't merge root page */
+ mdb_cassert(csrc, cdst->mc_snum > 1);
/* Mark dst as dirty. */
if ((rc = mdb_page_touch(cdst)))
/* Move all nodes from src to dst.
*/
- j = nkeys = NUMKEYS(cdst->mc_pg[cdst->mc_top]);
- if (IS_LEAF2(csrc->mc_pg[csrc->mc_top])) {
+ j = nkeys = NUMKEYS(pdst);
+ if (IS_LEAF2(psrc)) {
key.mv_size = csrc->mc_db->md_pad;
- key.mv_data = METADATA(csrc->mc_pg[csrc->mc_top]);
- for (i = 0; i < NUMKEYS(csrc->mc_pg[csrc->mc_top]); i++, j++) {
+ key.mv_data = METADATA(psrc);
+ for (i = 0; i < NUMKEYS(psrc); i++, j++) {
rc = mdb_node_add(cdst, j, &key, NULL, 0, 0);
if (rc != MDB_SUCCESS)
return rc;
key.mv_data = (char *)key.mv_data + key.mv_size;
}
} else {
- for (i = 0; i < NUMKEYS(csrc->mc_pg[csrc->mc_top]); i++, j++) {
- srcnode = NODEPTR(csrc->mc_pg[csrc->mc_top], i);
- if (i == 0 && IS_BRANCH(csrc->mc_pg[csrc->mc_top])) {
- unsigned int snum = csrc->mc_snum;
+ for (i = 0; i < NUMKEYS(psrc); i++, j++) {
+ srcnode = NODEPTR(psrc, i);
+ if (i == 0 && IS_BRANCH(psrc)) {
+ MDB_cursor mn;
MDB_node *s2;
+ mdb_cursor_copy(csrc, &mn);
/* must find the lowest key below src */
- mdb_page_search_lowest(csrc);
- if (IS_LEAF2(csrc->mc_pg[csrc->mc_top])) {
- key.mv_size = csrc->mc_db->md_pad;
- key.mv_data = LEAF2KEY(csrc->mc_pg[csrc->mc_top], 0, key.mv_size);
+ rc = mdb_page_search_lowest(&mn);
+ if (rc)
+ return rc;
+ if (IS_LEAF2(mn.mc_pg[mn.mc_top])) {
+ key.mv_size = mn.mc_db->md_pad;
+ key.mv_data = LEAF2KEY(mn.mc_pg[mn.mc_top], 0, key.mv_size);
} else {
- s2 = NODEPTR(csrc->mc_pg[csrc->mc_top], 0);
+ s2 = NODEPTR(mn.mc_pg[mn.mc_top], 0);
key.mv_size = NODEKSZ(s2);
key.mv_data = NODEKEY(s2);
}
- csrc->mc_snum = snum--;
- csrc->mc_top = snum;
} else {
key.mv_size = srcnode->mn_ksize;
key.mv_data = NODEKEY(srcnode);
}
DPRINTF(("dst page %"Z"u now has %u keys (%.1f%% filled)",
- cdst->mc_pg[cdst->mc_top]->mp_pgno, NUMKEYS(cdst->mc_pg[cdst->mc_top]),
- (float)PAGEFILL(cdst->mc_txn->mt_env, cdst->mc_pg[cdst->mc_top]) / 10));
+ pdst->mp_pgno, NUMKEYS(pdst),
+ (float)PAGEFILL(cdst->mc_txn->mt_env, pdst) / 10));
/* Unlink the src page from parent and add to free list.
*/
- mdb_node_del(csrc->mc_pg[csrc->mc_top-1], csrc->mc_ki[csrc->mc_top-1], 0);
- if (csrc->mc_ki[csrc->mc_top-1] == 0) {
+ csrc->mc_top--;
+ mdb_node_del(csrc, 0);
+ if (csrc->mc_ki[csrc->mc_top] == 0) {
key.mv_size = 0;
- csrc->mc_top--;
rc = mdb_update_key(csrc, &key);
- csrc->mc_top++;
- if (rc)
+ if (rc) {
+ csrc->mc_top++;
return rc;
+ }
}
+ csrc->mc_top++;
- rc = mdb_midl_append(&csrc->mc_txn->mt_free_pgs,
- csrc->mc_pg[csrc->mc_top]->mp_pgno);
- if (rc)
- return rc;
- if (IS_LEAF(csrc->mc_pg[csrc->mc_top]))
+ psrc = csrc->mc_pg[csrc->mc_top];
+ /* If not operating on FreeDB, allow this page to be reused
+ * in this txn.
+ */
+ if ((psrc->mp_flags & P_DIRTY) && csrc->mc_dbi != FREE_DBI) {
+ mdb_page_loose(csrc->mc_txn->mt_env, psrc);
+ } else {
+ rc = mdb_midl_append(&csrc->mc_txn->mt_free_pgs, psrc->mp_pgno);
+ if (rc)
+ return rc;
+ }
+ if (IS_LEAF(psrc))
csrc->mc_db->md_leaf_pages--;
else
csrc->mc_db->md_branch_pages--;
/* Adjust other cursors pointing to mp */
MDB_cursor *m2, *m3;
MDB_dbi dbi = csrc->mc_dbi;
- MDB_page *mp = cdst->mc_pg[cdst->mc_top];
for (m2 = csrc->mc_txn->mt_cursors[dbi]; m2; m2=m2->mc_next) {
if (csrc->mc_flags & C_SUB)
m3 = m2;
if (m3 == csrc) continue;
if (m3->mc_snum < csrc->mc_snum) continue;
- if (m3->mc_pg[csrc->mc_top] == csrc->mc_pg[csrc->mc_top]) {
- m3->mc_pg[csrc->mc_top] = mp;
+ if (m3->mc_pg[csrc->mc_top] == psrc) {
+ m3->mc_pg[csrc->mc_top] = pdst;
m3->mc_ki[csrc->mc_top] += nkeys;
}
}
}
- mdb_cursor_pop(csrc);
-
- return mdb_rebalance(csrc);
+ {
+ unsigned int snum = cdst->mc_snum;
+ uint16_t depth = cdst->mc_db->md_depth;
+ mdb_cursor_pop(cdst);
+ rc = mdb_rebalance(cdst);
+ /* Did the tree shrink? */
+ if (depth > cdst->mc_db->md_depth)
+ snum--;
+ cdst->mc_snum = snum;
+ cdst->mc_top = snum-1;
+ }
+ return rc;
}
/** Copy the contents of a cursor.
int rc;
unsigned int ptop, minkeys;
MDB_cursor mn;
+ indx_t oldki;
minkeys = 1 + (IS_BRANCH(mc->mc_pg[mc->mc_top]));
-#if MDB_DEBUG
- {
- pgno_t pgno;
- COPY_PGNO(pgno, mc->mc_pg[mc->mc_top]->mp_pgno);
DPRINTF(("rebalancing %s page %"Z"u (has %u keys, %.1f%% full)",
IS_LEAF(mc->mc_pg[mc->mc_top]) ? "leaf" : "branch",
- pgno, NUMKEYS(mc->mc_pg[mc->mc_top]),
+ mdb_dbg_pgno(mc->mc_pg[mc->mc_top]), NUMKEYS(mc->mc_pg[mc->mc_top]),
(float)PAGEFILL(mc->mc_txn->mt_env, mc->mc_pg[mc->mc_top]) / 10));
- }
-#endif
if (PAGEFILL(mc->mc_txn->mt_env, mc->mc_pg[mc->mc_top]) >= FILL_THRESHOLD &&
NUMKEYS(mc->mc_pg[mc->mc_top]) >= minkeys) {
-#if MDB_DEBUG
- pgno_t pgno;
- COPY_PGNO(pgno, mc->mc_pg[mc->mc_top]->mp_pgno);
DPRINTF(("no need to rebalance page %"Z"u, above fill threshold",
- pgno));
-#endif
+ mdb_dbg_pgno(mc->mc_pg[mc->mc_top])));
return MDB_SUCCESS;
}
}
}
} else if (IS_BRANCH(mp) && NUMKEYS(mp) == 1) {
+ int i;
DPUTS("collapsing root page!");
rc = mdb_midl_append(&mc->mc_txn->mt_free_pgs, mp->mp_pgno);
if (rc)
mc->mc_db->md_depth--;
mc->mc_db->md_branch_pages--;
mc->mc_ki[0] = mc->mc_ki[1];
+ for (i = 1; i<mc->mc_db->md_depth; i++) {
+ mc->mc_pg[i] = mc->mc_pg[i+1];
+ mc->mc_ki[i] = mc->mc_ki[i+1];
+ }
{
/* Adjust other cursors pointing to mp */
MDB_cursor *m2, *m3;
m3 = m2;
if (m3 == mc || m3->mc_snum < mc->mc_snum) continue;
if (m3->mc_pg[0] == mp) {
- int i;
m3->mc_snum--;
m3->mc_top--;
for (i=0; i<m3->mc_snum; i++) {
* otherwise the tree is invalid.
*/
ptop = mc->mc_top-1;
- assert(NUMKEYS(mc->mc_pg[ptop]) > 1);
+ mdb_cassert(mc, NUMKEYS(mc->mc_pg[ptop]) > 1);
/* Leaf page fill factor is below the threshold.
* Try to move keys from left or right neighbor, or
mdb_cursor_copy(mc, &mn);
mn.mc_xcursor = NULL;
+ oldki = mc->mc_ki[mc->mc_top];
if (mc->mc_ki[ptop] == 0) {
/* We're the leftmost leaf in our parent.
*/
* (A branch page must never have less than 2 keys.)
*/
minkeys = 1 + (IS_BRANCH(mn.mc_pg[mn.mc_top]));
- if (PAGEFILL(mc->mc_txn->mt_env, mn.mc_pg[mn.mc_top]) >= FILL_THRESHOLD && NUMKEYS(mn.mc_pg[mn.mc_top]) > minkeys)
- return mdb_node_move(&mn, mc);
- else {
- if (mc->mc_ki[ptop] == 0)
+ if (PAGEFILL(mc->mc_txn->mt_env, mn.mc_pg[mn.mc_top]) >= FILL_THRESHOLD && NUMKEYS(mn.mc_pg[mn.mc_top]) > minkeys) {
+ rc = mdb_node_move(&mn, mc);
+ if (mc->mc_ki[ptop]) {
+ oldki++;
+ }
+ } else {
+ if (mc->mc_ki[ptop] == 0) {
rc = mdb_page_merge(&mn, mc);
- else {
+ } else {
+ oldki += NUMKEYS(mn.mc_pg[mn.mc_top]);
mn.mc_ki[mn.mc_top] += mc->mc_ki[mn.mc_top] + 1;
rc = mdb_page_merge(mc, &mn);
mdb_cursor_copy(&mn, mc);
}
- mc->mc_flags &= ~(C_INITIALIZED|C_EOF);
+ mc->mc_flags &= ~C_EOF;
}
+ mc->mc_ki[mc->mc_top] = oldki;
return rc;
}
/** Complete a delete operation started by #mdb_cursor_del(). */
static int
-mdb_cursor_del0(MDB_cursor *mc, MDB_node *leaf)
+mdb_cursor_del0(MDB_cursor *mc)
{
int rc;
MDB_page *mp;
indx_t ki;
unsigned int nkeys;
- mp = mc->mc_pg[mc->mc_top];
ki = mc->mc_ki[mc->mc_top];
-
- /* add overflow pages to free list */
- if (!IS_LEAF2(mp) && F_ISSET(leaf->mn_flags, F_BIGDATA)) {
- MDB_page *omp;
- pgno_t pg;
-
- memcpy(&pg, NODEDATA(leaf), sizeof(pg));
- if ((rc = mdb_page_get(mc->mc_txn, pg, &omp, NULL)) ||
- (rc = mdb_ovpage_free(mc, omp)))
- return rc;
- }
- mdb_node_del(mp, ki, mc->mc_db->md_pad);
+ mdb_node_del(mc, mc->mc_db->md_pad);
mc->mc_db->md_entries--;
rc = mdb_rebalance(mc);
- if (rc != MDB_SUCCESS)
- mc->mc_txn->mt_flags |= MDB_TXN_ERROR;
- else {
- MDB_cursor *m2;
+
+ if (rc == MDB_SUCCESS) {
+ MDB_cursor *m2, *m3;
MDB_dbi dbi = mc->mc_dbi;
mp = mc->mc_pg[mc->mc_top];
nkeys = NUMKEYS(mp);
/* if mc points past last node in page, find next sibling */
- if (mc->mc_ki[mc->mc_top] >= nkeys)
- mdb_cursor_sibling(mc, 1);
+ if (mc->mc_ki[mc->mc_top] >= nkeys) {
+ rc = mdb_cursor_sibling(mc, 1);
+ if (rc == MDB_NOTFOUND)
+ rc = MDB_SUCCESS;
+ }
/* Adjust other cursors pointing to mp */
- for (m2 = mc->mc_txn->mt_cursors[dbi]; m2; m2=m2->mc_next) {
- if (m2 == mc)
- continue;
- if (!(m2->mc_flags & C_INITIALIZED))
+ for (m2 = mc->mc_txn->mt_cursors[dbi]; !rc && m2; m2=m2->mc_next) {
+ m3 = (mc->mc_flags & C_SUB) ? &m2->mc_xcursor->mx_cursor : m2;
+ if (! (m2->mc_flags & m3->mc_flags & C_INITIALIZED))
continue;
- if (m2->mc_top < mc->mc_top)
+ if (m3 == mc || m3->mc_snum < mc->mc_snum)
continue;
- if (m2->mc_pg[mc->mc_top] == mp) {
- if (m2->mc_ki[mc->mc_top] >= ki) {
- m2->mc_flags |= C_DEL;
- if (m2->mc_ki[mc->mc_top] > ki)
- m2->mc_ki[mc->mc_top]--;
+ if (m3->mc_pg[mc->mc_top] == mp) {
+ if (m3->mc_ki[mc->mc_top] >= ki) {
+ m3->mc_flags |= C_DEL;
+ if (m3->mc_ki[mc->mc_top] > ki)
+ m3->mc_ki[mc->mc_top]--;
+ }
+ if (m3->mc_ki[mc->mc_top] >= nkeys) {
+ rc = mdb_cursor_sibling(m3, 1);
+ if (rc == MDB_NOTFOUND)
+ rc = MDB_SUCCESS;
}
- if (m2->mc_ki[mc->mc_top] >= nkeys)
- mdb_cursor_sibling(m2, 1);
}
}
mc->mc_flags |= C_DEL;
}
+ if (rc)
+ mc->mc_txn->mt_flags |= MDB_TXN_ERROR;
return rc;
}
int
mdb_del(MDB_txn *txn, MDB_dbi dbi,
MDB_val *key, MDB_val *data)
+{
+ if (!key || dbi == FREE_DBI || !TXN_DBI_EXIST(txn, dbi))
+ return EINVAL;
+
+ if (txn->mt_flags & (MDB_TXN_RDONLY|MDB_TXN_ERROR))
+ return (txn->mt_flags & MDB_TXN_RDONLY) ? EACCES : MDB_BAD_TXN;
+
+ if (!F_ISSET(txn->mt_dbs[dbi].md_flags, MDB_DUPSORT)) {
+ /* must ignore any data */
+ data = NULL;
+ }
+
+ return mdb_del0(txn, dbi, key, data, 0);
+}
+
+static int
+mdb_del0(MDB_txn *txn, MDB_dbi dbi,
+ MDB_val *key, MDB_val *data, unsigned flags)
{
MDB_cursor mc;
MDB_xcursor mx;
MDB_cursor_op op;
MDB_val rdata, *xdata;
- int rc, exact;
+ int rc, exact = 0;
DKBUF;
- assert(key != NULL);
-
DPRINTF(("====> delete db %u key [%s]", dbi, DKEY(key)));
- if (txn == NULL || !dbi || dbi >= txn->mt_numdbs || !(txn->mt_dbflags[dbi] & DB_VALID))
- return EINVAL;
-
- if (txn->mt_flags & (MDB_TXN_RDONLY|MDB_TXN_ERROR))
- return (txn->mt_flags & MDB_TXN_RDONLY) ? EACCES : MDB_BAD_TXN;
-
- if (key->mv_size > MDB_MAXKEYSIZE) {
- return MDB_BAD_VALSIZE;
- }
-
mdb_cursor_init(&mc, txn, dbi, &mx);
- exact = 0;
- if (!F_ISSET(txn->mt_dbs[dbi].md_flags, MDB_DUPSORT)) {
- /* must ignore any data */
- data = NULL;
- }
if (data) {
op = MDB_GET_BOTH;
rdata = *data;
} else {
op = MDB_SET;
xdata = NULL;
+ flags |= MDB_NODUPDATA;
}
rc = mdb_cursor_set(&mc, key, xdata, op, &exact);
if (rc == 0) {
mc.mc_flags |= C_UNTRACK;
mc.mc_next = txn->mt_cursors[dbi];
txn->mt_cursors[dbi] = &mc;
- rc = mdb_cursor_del(&mc, data ? 0 : MDB_NODUPDATA);
+ rc = mdb_cursor_del(&mc, flags);
txn->mt_cursors[dbi] = mc.mc_next;
}
return rc;
if (mc->mc_snum < 2) {
if ((rc = mdb_page_new(mc, P_BRANCH, 1, &pp)))
- return rc;
+ goto done;
/* shift current top to make room for new parent */
mc->mc_pg[1] = mc->mc_pg[0];
mc->mc_ki[1] = mc->mc_ki[0];
mc->mc_ki[0] = mc->mc_ki[1];
mc->mc_db->md_root = mp->mp_pgno;
mc->mc_db->md_depth--;
- return rc;
+ goto done;
}
mc->mc_snum = 2;
mc->mc_top = 1;
int x;
unsigned int lsize, rsize, ksize;
/* Move half of the keys to the right sibling */
- copy = NULL;
x = mc->mc_ki[mc->mc_top] - split_indx;
ksize = mc->mc_db->md_pad;
split = LEAF2KEY(mp, split_indx, ksize);
nsize = mdb_leaf_size(env, newkey, newdata);
else
nsize = mdb_branch_size(env, newkey);
- nsize += nsize & 1;
+ nsize = EVEN(nsize);
/* grab a page to hold a temporary copy */
copy = mdb_page_malloc(mc->mc_txn, 1);
- if (copy == NULL)
- return ENOMEM;
+ if (copy == NULL) {
+ rc = ENOMEM;
+ goto done;
+ }
copy->mp_pgno = mp->mp_pgno;
copy->mp_flags = mp->mp_flags;
copy->mp_lower = PAGEHDRSZ;
else
psize += NODEDSZ(node);
}
- psize += psize & 1;
+ psize = EVEN(psize);
}
if (psize > pmax || i == k-j) {
split_indx = i + (j<0);
mn.mc_top--;
did_split = 1;
rc = mdb_page_split(&mn, &sepkey, NULL, rp->mp_pgno, 0);
+ if (rc)
+ goto done;
/* root split? */
if (mn.mc_snum == mc->mc_snum) {
mc->mc_ki[i] = mn.mc_ki[i];
}
mc->mc_pg[ptop] = mn.mc_pg[ptop];
- mc->mc_ki[ptop] = mn.mc_ki[ptop] - 1;
+ if (mn.mc_ki[ptop]) {
+ mc->mc_ki[ptop] = mn.mc_ki[ptop] - 1;
+ } else {
+ /* find right page's left sibling */
+ mc->mc_ki[ptop] = mn.mc_ki[ptop];
+ mdb_cursor_sibling(mc, 0);
+ }
}
} else {
mn.mc_top--;
}
mc->mc_flags ^= C_SPLITTING;
if (rc != MDB_SUCCESS) {
- return rc;
+ goto done;
}
if (nflags & MDB_APPEND) {
mc->mc_pg[mc->mc_top] = rp;
mc->mc_ki[mc->mc_top] = 0;
rc = mdb_node_add(mc, 0, newkey, newdata, newpgno, nflags);
if (rc)
- return rc;
+ goto done;
for (i=0; i<mc->mc_top; i++)
mc->mc_ki[i] = mn.mc_ki[i];
} else if (!IS_LEAF2(mp)) {
}
rc = mdb_node_add(mc, j, &rkey, rdata, pgno, flags);
- if (rc) {
- /* return tmp page to freelist */
- mdb_page_free(env, copy);
- return rc;
- }
+ if (rc)
+ goto done;
if (i == nkeys) {
i = 0;
j = 0;
*/
if (mn.mc_pg[ptop] != mc->mc_pg[ptop] &&
mc->mc_ki[ptop] >= NUMKEYS(mc->mc_pg[ptop])) {
- for (i=0; i<ptop; i++) {
+ for (i=0; i<=ptop; i++) {
mc->mc_pg[i] = mn.mc_pg[i];
mc->mc_ki[i] = mn.mc_ki[i];
}
- mc->mc_pg[ptop] = mn.mc_pg[ptop];
- mc->mc_ki[ptop] = mn.mc_ki[ptop] - 1;
}
}
- /* return tmp page to freelist */
- mdb_page_free(env, copy);
}
{
}
}
DPRINTF(("mp left: %d, rp left: %d", SIZELEFT(mp), SIZELEFT(rp)));
+
+done:
+ if (copy) /* tmp page */
+ mdb_page_free(env, copy);
+ if (rc)
+ mc->mc_txn->mt_flags |= MDB_TXN_ERROR;
return rc;
}
MDB_cursor mc;
MDB_xcursor mx;
- assert(key != NULL);
- assert(data != NULL);
-
- if (txn == NULL || !dbi || dbi >= txn->mt_numdbs || !(txn->mt_dbflags[dbi] & DB_VALID))
+ if (!key || !data || dbi == FREE_DBI || !TXN_DBI_EXIST(txn, dbi))
return EINVAL;
if ((flags & (MDB_NOOVERWRITE|MDB_NODUPDATA|MDB_RESERVE|MDB_APPEND|MDB_APPENDDUP)) != flags)
return MDB_SUCCESS;
}
+int
+mdb_env_set_userctx(MDB_env *env, void *ctx)
+{
+ if (!env)
+ return EINVAL;
+ env->me_userctx = ctx;
+ return MDB_SUCCESS;
+}
+
+void *
+mdb_env_get_userctx(MDB_env *env)
+{
+ return env ? env->me_userctx : NULL;
+}
+
+int
+mdb_env_set_assert(MDB_env *env, MDB_assert_func *func)
+{
+ if (!env)
+ return EINVAL;
+#ifndef NDEBUG
+ env->me_assert_func = func;
+#endif
+ return MDB_SUCCESS;
+}
+
int
mdb_env_get_path(MDB_env *env, const char **arg)
{
int mdb_stat(MDB_txn *txn, MDB_dbi dbi, MDB_stat *arg)
{
- if (txn == NULL || arg == NULL || dbi >= txn->mt_numdbs)
+ if (!arg || !TXN_DBI_EXIST(txn, dbi))
return EINVAL;
+ if (txn->mt_flags & MDB_TXN_ERROR)
+ return MDB_BAD_TXN;
+
if (txn->mt_dbflags[dbi] & DB_STALE) {
MDB_cursor mc;
MDB_xcursor mx;
int mdb_dbi_flags(MDB_txn *txn, MDB_dbi dbi, unsigned int *flags)
{
/* We could return the flags for the FREE_DBI too but what's the point? */
- if (txn == NULL || dbi < MAIN_DBI || dbi >= txn->mt_numdbs)
+ if (dbi == FREE_DBI || !TXN_DBI_EXIST(txn, dbi))
return EINVAL;
*flags = txn->mt_dbs[dbi].md_flags & PERSISTENT_FLAGS;
return MDB_SUCCESS;
memcpy(&pg, NODEDATA(ni), sizeof(pg));
rc = mdb_page_get(txn, pg, &omp, NULL);
if (rc != 0)
- return rc;
- assert(IS_OVERFLOW(omp));
+ goto done;
+ mdb_cassert(mc, IS_OVERFLOW(omp));
rc = mdb_midl_append_range(&txn->mt_free_pgs,
pg, omp->mp_pages);
if (rc)
- return rc;
+ goto done;
} else if (subs && (ni->mn_flags & F_SUBDATA)) {
mdb_xcursor_init1(mc, ni);
rc = mdb_drop0(&mc->mc_xcursor->mx_cursor, 0);
if (rc)
- return rc;
+ goto done;
}
}
} else {
if ((rc = mdb_midl_need(&txn->mt_free_pgs, n)) != 0)
- return rc;
+ goto done;
for (i=0; i<n; i++) {
pgno_t pg;
ni = NODEPTR(mp, i);
mc->mc_ki[mc->mc_top] = i;
rc = mdb_cursor_sibling(mc, 1);
if (rc) {
+ if (rc != MDB_NOTFOUND)
+ goto done;
/* no more siblings, go back to beginning
* of previous level.
*/
}
/* free it */
rc = mdb_midl_append(&txn->mt_free_pgs, mc->mc_db->md_root);
+done:
+ if (rc)
+ txn->mt_flags |= MDB_TXN_ERROR;
} else if (rc == MDB_NOTFOUND) {
rc = MDB_SUCCESS;
}
MDB_cursor *mc, *m2;
int rc;
- if (!txn || !dbi || dbi >= txn->mt_numdbs || (unsigned)del > 1 || !(txn->mt_dbflags[dbi] & DB_VALID))
+ if ((unsigned)del > 1 || dbi == FREE_DBI || !TXN_DBI_EXIST(txn, dbi))
return EINVAL;
if (F_ISSET(txn->mt_flags, MDB_TXN_RDONLY))
/* Can't delete the main DB */
if (del && dbi > MAIN_DBI) {
- rc = mdb_del(txn, MAIN_DBI, &mc->mc_dbx->md_name, NULL);
+ rc = mdb_del0(txn, MAIN_DBI, &mc->mc_dbx->md_name, NULL, 0);
if (!rc) {
txn->mt_dbflags[dbi] = DB_STALE;
mdb_dbi_close(txn->mt_env, dbi);
+ } else {
+ txn->mt_flags |= MDB_TXN_ERROR;
}
} else {
/* reset the DB record, mark it dirty */
int mdb_set_compare(MDB_txn *txn, MDB_dbi dbi, MDB_cmp_func *cmp)
{
- if (txn == NULL || !dbi || dbi >= txn->mt_numdbs || !(txn->mt_dbflags[dbi] & DB_VALID))
+ if (dbi == FREE_DBI || !TXN_DBI_EXIST(txn, dbi))
return EINVAL;
txn->mt_dbxs[dbi].md_cmp = cmp;
int mdb_set_dupsort(MDB_txn *txn, MDB_dbi dbi, MDB_cmp_func *cmp)
{
- if (txn == NULL || !dbi || dbi >= txn->mt_numdbs || !(txn->mt_dbflags[dbi] & DB_VALID))
+ if (dbi == FREE_DBI || !TXN_DBI_EXIST(txn, dbi))
return EINVAL;
txn->mt_dbxs[dbi].md_dcmp = cmp;
int mdb_set_relfunc(MDB_txn *txn, MDB_dbi dbi, MDB_rel_func *rel)
{
- if (txn == NULL || !dbi || dbi >= txn->mt_numdbs || !(txn->mt_dbflags[dbi] & DB_VALID))
+ if (dbi == FREE_DBI || !TXN_DBI_EXIST(txn, dbi))
return EINVAL;
txn->mt_dbxs[dbi].md_rel = rel;
int mdb_set_relctx(MDB_txn *txn, MDB_dbi dbi, void *ctx)
{
- if (txn == NULL || !dbi || dbi >= txn->mt_numdbs || !(txn->mt_dbflags[dbi] & DB_VALID))
+ if (dbi == FREE_DBI || !TXN_DBI_EXIST(txn, dbi))
return EINVAL;
txn->mt_dbxs[dbi].md_relctx = ctx;
int mdb_env_get_maxkeysize(MDB_env *env)
{
- return MDB_MAXKEYSIZE;
+ return ENV_MAXKEY(env);
}
int mdb_reader_list(MDB_env *env, MDB_msg_func *func, void *ctx)
unsigned int i, rdrs;
MDB_reader *mr;
char buf[64];
- int first = 1;
+ int rc = 0, first = 1;
if (!env || !func)
return -1;
mr = env->me_txns->mti_readers;
for (i=0; i<rdrs; i++) {
if (mr[i].mr_pid) {
- size_t tid;
- int rc;
- tid = mr[i].mr_tid;
- if (mr[i].mr_txnid == (txnid_t)-1) {
- sprintf(buf, "%10d %"Z"x -\n", mr[i].mr_pid, tid);
- } else {
- sprintf(buf, "%10d %"Z"x %"Z"u\n", mr[i].mr_pid, tid, mr[i].mr_txnid);
- }
+ txnid_t txnid = mr[i].mr_txnid;
+ sprintf(buf, txnid == (txnid_t)-1 ?
+ "%10d %"Z"x -\n" : "%10d %"Z"x %"Z"u\n",
+ (int)mr[i].mr_pid, (size_t)mr[i].mr_tid, txnid);
if (first) {
first = 0;
- func(" pid thread txnid\n", ctx);
+ rc = func(" pid thread txnid\n", ctx);
+ if (rc < 0)
+ break;
}
rc = func(buf, ctx);
if (rc < 0)
- return rc;
+ break;
}
}
if (first) {
- func("(no active readers)\n", ctx);
+ rc = func("(no active readers)\n", ctx);
}
- return 0;
+ return rc;
}
/** Insert pid into list if not already present.
* return -1 if already present.
*/
-static int mdb_pid_insert(pid_t *ids, pid_t pid)
+static int mdb_pid_insert(MDB_PID_T *ids, MDB_PID_T pid)
{
/* binary search of pid in list */
unsigned base = 0;
{
unsigned int i, j, rdrs;
MDB_reader *mr;
- pid_t *pids, pid;
+ MDB_PID_T *pids, pid;
int count = 0;
if (!env)
if (!env->me_txns)
return MDB_SUCCESS;
rdrs = env->me_txns->mti_numreaders;
- pids = malloc((rdrs+1) * sizeof(pid_t));
+ pids = malloc((rdrs+1) * sizeof(MDB_PID_T));
if (!pids)
return ENOMEM;
pids[0] = 0;
mr = env->me_txns->mti_readers;
- j = 0;
for (i=0; i<rdrs; i++) {
if (mr[i].mr_pid && mr[i].mr_pid != env->me_pid) {
pid = mr[i].mr_pid;
projects / openldap / blobdiff